From 4040d22605d6578a87892b42a0abf3c7fc3ca693 Mon Sep 17 00:00:00 2001 From: liangxu Date: Mon, 29 Jun 2026 17:21:38 +0800 Subject: [PATCH] fix: enhance monitoring metrics with latest sample aggregation and derived metrics calculations --- .../app/monitoring_callback_service_test.go | 39 +++ .../app/monitoring_daily_task_worker_test.go | 27 +- .../tenant/app/monitoring_projection.go | 68 +++-- .../internal/tenant/app/monitoring_service.go | 240 +++++++++++++----- .../app/monitoring_service_date_test.go | 71 ++++++ 5 files changed, 356 insertions(+), 89 deletions(-) diff --git a/server/internal/tenant/app/monitoring_callback_service_test.go b/server/internal/tenant/app/monitoring_callback_service_test.go index b436c2b..c3d5665 100644 --- a/server/internal/tenant/app/monitoring_callback_service_test.go +++ b/server/internal/tenant/app/monitoring_callback_service_test.go @@ -1,6 +1,7 @@ package app import ( + "context" "database/sql" "encoding/json" "errors" @@ -9,6 +10,8 @@ import ( "time" "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestDecideHeartbeatPrimaryLease(t *testing.T) { @@ -111,6 +114,42 @@ func TestMonitoringHeartbeatPrimaryLeaseMetrics(t *testing.T) { } } +func TestLoadMonitoringBrandDailyAggregateUsesLatestQuestionPlatformSamples(t *testing.T) { + tx := &captureDailyMonitorTaskTx{} + + _, err := (&MonitoringCallbackService{}).loadMonitoringBrandDailyAggregate( + context.Background(), + tx, + 3, + 3, + time.Date(2026, 6, 29, 0, 0, 0, 0, monitoringBusinessLocation()), + ) + + require.NoError(t, err) + require.Len(t, tx.queryRowSQLs, 5) + combined := normalizeSQLWhitespace(strings.Join(tx.queryRowSQLs, " ")) + assert.Contains(t, combined, "SELECT DISTINCT ON (business_date, question_id, ai_platform_id)") + assert.Contains(t, combined, "p.brand_mentioned IS TRUE AND p.sentiment_label = 'positive'") +} + +func TestLoadMonitoringPlatformDailyAggregatesUsesLatestQuestionPlatformSamples(t *testing.T) { + tx := &captureDailyMonitorTaskTx{} + + _, err := (&MonitoringCallbackService{}).loadMonitoringPlatformDailyAggregates( + context.Background(), + tx, + 3, + 3, + time.Date(2026, 6, 29, 0, 0, 0, 0, monitoringBusinessLocation()), + []monitoringPlatformMetadata{{ID: "doubao", Name: "豆包"}}, + ) + + require.NoError(t, err) + require.GreaterOrEqual(t, len(tx.querySQLs), 2) + combined := normalizeSQLWhitespace(strings.Join(tx.querySQLs, " ")) + assert.Contains(t, combined, "SELECT DISTINCT ON (business_date, question_id, ai_platform_id)") +} + func TestDecideMonitoringHeartbeatSnapshot(t *testing.T) { now := time.Date(2026, 4, 24, 3, 0, 0, 0, time.UTC) base := monitoringPlatformAccessSnapshotState{ diff --git a/server/internal/tenant/app/monitoring_daily_task_worker_test.go b/server/internal/tenant/app/monitoring_daily_task_worker_test.go index 0d9af31..4eadf10 100644 --- a/server/internal/tenant/app/monitoring_daily_task_worker_test.go +++ b/server/internal/tenant/app/monitoring_daily_task_worker_test.go @@ -2,6 +2,7 @@ package app import ( "context" + "database/sql" "net/http" "net/http/httptest" "runtime" @@ -745,7 +746,12 @@ type captureDailyMonitorTaskTx struct { queryCalled bool querySQL string queryArgs []any - commandTag pgconn.CommandTag + querySQLs []string + queryArgses []any + + queryRowSQLs []string + queryRowArgses [][]any + commandTag pgconn.CommandTag } func (tx *captureDailyMonitorTaskTx) Begin(context.Context) (pgx.Tx, error) { return tx, nil } @@ -773,9 +779,13 @@ func (tx *captureDailyMonitorTaskTx) Query(_ context.Context, sql string, argume tx.queryCalled = true tx.querySQL = sql tx.queryArgs = arguments + tx.querySQLs = append(tx.querySQLs, sql) + tx.queryArgses = append(tx.queryArgses, arguments...) return emptyDailyMonitorRows{}, nil } -func (tx *captureDailyMonitorTaskTx) QueryRow(context.Context, string, ...any) pgx.Row { +func (tx *captureDailyMonitorTaskTx) QueryRow(_ context.Context, sql string, arguments ...any) pgx.Row { + tx.queryRowSQLs = append(tx.queryRowSQLs, sql) + tx.queryRowArgses = append(tx.queryRowArgses, arguments) return boolDailyMonitorRow(true) } func (tx *captureDailyMonitorTaskTx) Conn() *pgx.Conn { return nil } @@ -785,9 +795,18 @@ var _ pgx.Tx = (*captureDailyMonitorTaskTx)(nil) type boolDailyMonitorRow bool func (r boolDailyMonitorRow) Scan(dest ...any) error { - if len(dest) == 1 { - if value, ok := dest[0].(*bool); ok { + for _, target := range dest { + switch value := target.(type) { + case *bool: *value = bool(r) + case *int64: + *value = 0 + case *string: + *value = "" + case *sql.NullTime: + *value = sql.NullTime{} + case *sql.NullString: + *value = sql.NullString{} } } return nil diff --git a/server/internal/tenant/app/monitoring_projection.go b/server/internal/tenant/app/monitoring_projection.go index 3801517..8a15708 100644 --- a/server/internal/tenant/app/monitoring_projection.go +++ b/server/internal/tenant/app/monitoring_projection.go @@ -173,12 +173,20 @@ func (s *MonitoringCallbackService) loadMonitoringBrandDailyAggregate(ctx contex if err := tx.QueryRow(ctx, ` SELECT COUNT(*), MAX(completed_at) - FROM question_monitor_runs - WHERE tenant_id = $1 - AND brand_id = $2 - AND collector_type = $3 - AND business_date = $4::date - AND status = 'succeeded' + FROM ( + SELECT DISTINCT ON (business_date, question_id, ai_platform_id) + completed_at, + updated_at, + created_at, + id + FROM question_monitor_runs + WHERE tenant_id = $1 + AND brand_id = $2 + AND collector_type = $3 + AND business_date = $4::date + AND status = 'succeeded' + ORDER BY business_date, question_id, ai_platform_id, COALESCE(completed_at, updated_at, created_at) DESC NULLS LAST, id DESC + ) latest_runs `, tenantID, brandID, monitoringCollectorType, dateText).Scan(&agg.ActualSampleCount, &agg.SnapshotUpdatedAt); err != nil { return agg, monitoringInternalError(50041, "run_projection_lookup_failed", "failed to load monitoring run counts", err) } @@ -188,15 +196,27 @@ func (s *MonitoringCallbackService) loadMonitoringBrandDailyAggregate(ctx contex COUNT(*) FILTER (WHERE p.brand_mentioned IS TRUE)::bigint, COUNT(*) FILTER (WHERE p.brand_mention_position = 'top1')::bigint, COUNT(*) FILTER (WHERE p.first_recommended IS TRUE)::bigint, - COUNT(*) FILTER (WHERE p.sentiment_label = 'positive')::bigint - FROM question_monitor_runs r + COUNT(*) FILTER (WHERE p.brand_mentioned IS TRUE AND p.sentiment_label = 'positive')::bigint + FROM ( + SELECT DISTINCT ON (business_date, question_id, ai_platform_id) + id, + tenant_id, + business_date, + question_id, + ai_platform_id, + completed_at, + updated_at, + created_at + FROM question_monitor_runs + WHERE tenant_id = $1 + AND brand_id = $2 + AND collector_type = $3 + AND business_date = $4::date + AND status = 'succeeded' + ORDER BY business_date, question_id, ai_platform_id, COALESCE(completed_at, updated_at, created_at) DESC NULLS LAST, id DESC + ) r LEFT JOIN question_monitor_parse_results p ON p.tenant_id = r.tenant_id AND p.run_id = r.id - WHERE r.tenant_id = $1 - AND r.brand_id = $2 - AND r.collector_type = $3 - AND r.business_date = $4::date - AND r.status = 'succeeded' `, tenantID, brandID, monitoringCollectorType, dateText).Scan( &agg.MentionedCount, &agg.Top1MentionedCount, @@ -280,14 +300,24 @@ func (s *MonitoringCallbackService) loadMonitoringPlatformDailyAggregates(ctx co COUNT(*) FILTER (WHERE p.brand_mentioned IS TRUE)::bigint AS mentioned_count, COUNT(*) FILTER (WHERE p.brand_mention_position = 'top1')::bigint AS top1_mentioned_count, MAX(r.completed_at) AS last_sampled_at - FROM question_monitor_runs r + FROM ( + SELECT DISTINCT ON (business_date, question_id, ai_platform_id) + id, + tenant_id, + ai_platform_id, + completed_at, + updated_at, + created_at + FROM question_monitor_runs + WHERE tenant_id = $1 + AND brand_id = $2 + AND collector_type = $3 + AND business_date = $4::date + AND status = 'succeeded' + ORDER BY business_date, question_id, ai_platform_id, COALESCE(completed_at, updated_at, created_at) DESC NULLS LAST, id DESC + ) r LEFT JOIN question_monitor_parse_results p ON p.tenant_id = r.tenant_id AND p.run_id = r.id - WHERE r.tenant_id = $1 - AND r.brand_id = $2 - AND r.collector_type = $3 - AND r.business_date = $4::date - AND r.status = 'succeeded' GROUP BY r.ai_platform_id `, tenantID, brandID, monitoringCollectorType, dateText) if err != nil { diff --git a/server/internal/tenant/app/monitoring_service.go b/server/internal/tenant/app/monitoring_service.go index 0abd64d..cc28133 100644 --- a/server/internal/tenant/app/monitoring_service.go +++ b/server/internal/tenant/app/monitoring_service.go @@ -349,6 +349,21 @@ type monitoringDerivedMetrics struct { ByQuestion map[int64]monitoringDerivedRateStats } +type monitoringDerivedSampleKey struct { + Date string + PlatformID string + QuestionID int64 +} + +type monitoringDerivedParseSample struct { + BusinessDate time.Time + PlatformID string + QuestionID int64 + EventAt sql.NullTime + RunID int64 + Summary monitoringAnswerParseSummary +} + type monitoringPlatformMetadata struct { ID string Name string @@ -2239,7 +2254,7 @@ func (stats *monitoringDerivedRateStats) add(summary monitoringAnswerParseSummar if summary.FirstRecommended { stats.FirstRecommend++ } - if summary.SentimentLabel == "positive" { + if summary.BrandMentioned && summary.SentimentLabel == "positive" { stats.PositiveMentions++ } } @@ -2264,6 +2279,71 @@ func (stats monitoringDerivedRateStats) positiveMentionRate() *float64 { return divideAsPointer(stats.PositiveMentions, stats.Total) } +func addMonitoringDerivedParseSample(metrics *monitoringDerivedMetrics, sample monitoringDerivedParseSample) { + if metrics == nil { + return + } + + dateKey := sample.BusinessDate.Format("2006-01-02") + dateStats := metrics.ByDate[dateKey] + dateStats.add(sample.Summary) + metrics.ByDate[dateKey] = dateStats + + platformStatsByDate, ok := metrics.ByDatePlatform[dateKey] + if !ok { + platformStatsByDate = make(map[string]monitoringDerivedRateStats) + metrics.ByDatePlatform[dateKey] = platformStatsByDate + } + platformStats := platformStatsByDate[sample.PlatformID] + platformStats.add(sample.Summary) + platformStatsByDate[sample.PlatformID] = platformStats + + questionStatsByDate, ok := metrics.ByDateQuestion[dateKey] + if !ok { + questionStatsByDate = make(map[int64]monitoringDerivedRateStats) + metrics.ByDateQuestion[dateKey] = questionStatsByDate + } + questionStatsForDate := questionStatsByDate[sample.QuestionID] + questionStatsForDate.add(sample.Summary) + questionStatsByDate[sample.QuestionID] = questionStatsForDate + + questionStats := metrics.ByQuestion[sample.QuestionID] + questionStats.add(sample.Summary) + metrics.ByQuestion[sample.QuestionID] = questionStats +} + +func monitoringDerivedParseSampleIsNewer(candidate, existing monitoringDerivedParseSample) bool { + if candidate.EventAt.Valid && existing.EventAt.Valid && !candidate.EventAt.Time.Equal(existing.EventAt.Time) { + return candidate.EventAt.Time.After(existing.EventAt.Time) + } + if candidate.EventAt.Valid != existing.EventAt.Valid { + return candidate.EventAt.Valid + } + return candidate.RunID > existing.RunID +} + +func upsertLatestMonitoringDerivedParseSample(samples map[monitoringDerivedSampleKey]monitoringDerivedParseSample, sample monitoringDerivedParseSample) { + if samples == nil { + return + } + key := monitoringDerivedSampleKey{ + Date: sample.BusinessDate.Format("2006-01-02"), + PlatformID: sample.PlatformID, + QuestionID: sample.QuestionID, + } + if existing, ok := samples[key]; !ok || monitoringDerivedParseSampleIsNewer(sample, existing) { + samples[key] = sample + } +} + +func buildMonitoringDerivedMetricsFromLatestSamples(samples map[monitoringDerivedSampleKey]monitoringDerivedParseSample) monitoringDerivedMetrics { + metrics := newMonitoringDerivedMetrics() + for _, sample := range samples { + addMonitoringDerivedParseSample(&metrics, sample) + } + return metrics +} + func (s *MonitoringService) loadDerivedParseMetrics( ctx context.Context, tenantID, brandID int64, @@ -2280,35 +2360,51 @@ func (s *MonitoringService) loadDerivedParseMetrics( rows, err := s.monitoringPool.Query(ctx, ` SELECT + r.id, r.business_date, r.ai_platform_id, r.question_id, + r.event_at, r.raw_answer_text, p.brand_mentioned, p.brand_mention_position, p.first_recommended, p.sentiment_label, p.matched_brand_terms - FROM question_monitor_runs r + FROM ( + SELECT DISTINCT ON (business_date, question_id, ai_platform_id) + id, + tenant_id, + business_date, + ai_platform_id, + question_id, + raw_answer_text, + COALESCE(completed_at, updated_at, created_at) AS event_at + FROM question_monitor_runs + WHERE tenant_id = $1 + AND brand_id = $2 + AND collector_type = $3 + AND status = 'succeeded' + AND business_date BETWEEN $4::date AND $5::date + AND ($6::bigint[] IS NULL OR question_id = ANY($6)) + AND ($7::text[] IS NULL OR ai_platform_id = ANY($7)) + ORDER BY business_date, question_id, ai_platform_id, COALESCE(completed_at, updated_at, created_at) DESC NULLS LAST, id DESC + ) r LEFT JOIN question_monitor_parse_results p ON p.tenant_id = r.tenant_id AND p.run_id = r.id - WHERE r.tenant_id = $1 - AND r.brand_id = $2 - AND r.collector_type = $3 - AND r.status = 'succeeded' - AND r.business_date BETWEEN $4::date AND $5::date - AND ($6::bigint[] IS NULL OR r.question_id = ANY($6)) - AND ($7::text[] IS NULL OR r.ai_platform_id = ANY($7)) `, tenantID, brandID, monitoringCollectorType, startDate.Format("2006-01-02"), endDate.Format("2006-01-02"), nullableInt64Array(questionIDs), nullableStringArray(platformQueryIDs)) if err != nil { return metrics, response.ErrInternal(50041, "query_failed", "failed to load monitoring derived parse metrics") } defer rows.Close() + samples := make(map[monitoringDerivedSampleKey]monitoringDerivedParseSample) for rows.Next() { + var runID int64 var businessDate time.Time var platformID string var questionID int64 + var eventAt sql.NullTime var answerText sql.NullString var brandMentioned sql.NullBool var brandMentionPosition sql.NullString @@ -2316,9 +2412,11 @@ func (s *MonitoringService) loadDerivedParseMetrics( var sentimentLabel sql.NullString var matchedBrandTermsJSON []byte if scanErr := rows.Scan( + &runID, &businessDate, &platformID, &questionID, + &eventAt, &answerText, &brandMentioned, &brandMentionPosition, @@ -2345,39 +2443,21 @@ func (s *MonitoringService) loadDerivedParseMetrics( }, ) - dateKey := businessDate.Format("2006-01-02") - dateStats := metrics.ByDate[dateKey] - dateStats.add(summary) - metrics.ByDate[dateKey] = dateStats - - platformStatsByDate, ok := metrics.ByDatePlatform[dateKey] - if !ok { - platformStatsByDate = make(map[string]monitoringDerivedRateStats) - metrics.ByDatePlatform[dateKey] = platformStatsByDate - } - platformStats := platformStatsByDate[platformID] - platformStats.add(summary) - platformStatsByDate[platformID] = platformStats - - questionStatsByDate, ok := metrics.ByDateQuestion[dateKey] - if !ok { - questionStatsByDate = make(map[int64]monitoringDerivedRateStats) - metrics.ByDateQuestion[dateKey] = questionStatsByDate - } - questionStatsForDate := questionStatsByDate[questionID] - questionStatsForDate.add(summary) - questionStatsByDate[questionID] = questionStatsForDate - - questionStats := metrics.ByQuestion[questionID] - questionStats.add(summary) - metrics.ByQuestion[questionID] = questionStats + upsertLatestMonitoringDerivedParseSample(samples, monitoringDerivedParseSample{ + BusinessDate: businessDate, + PlatformID: platformID, + QuestionID: questionID, + EventAt: eventAt, + RunID: runID, + Summary: summary, + }) } if err := rows.Err(); err != nil { return metrics, response.ErrInternal(50041, "scan_failed", "failed to iterate monitoring derived metrics") } - return metrics, nil + return buildMonitoringDerivedMetricsFromLatestSamples(samples), nil } func decodeMonitoringMatchedBrandTerms(raw []byte) []string { @@ -2608,18 +2688,30 @@ func (s *MonitoringService) loadOverviewRunAggregate( COUNT(*) FILTER (WHERE p.brand_mentioned IS TRUE)::bigint AS mentioned_count, COUNT(*) FILTER (WHERE p.brand_mention_position = 'top1')::bigint AS top1_mentioned_count, COUNT(*) FILTER (WHERE p.first_recommended IS TRUE)::bigint AS first_recommended_count, - COUNT(*) FILTER (WHERE p.sentiment_label = 'positive')::bigint AS positive_mentioned_count, + COUNT(*) FILTER (WHERE p.brand_mentioned IS TRUE AND p.sentiment_label = 'positive')::bigint AS positive_mentioned_count, MAX(r.completed_at) AS last_sampled_at - FROM question_monitor_runs r + FROM ( + SELECT DISTINCT ON (business_date, question_id, ai_platform_id) + id, + tenant_id, + business_date, + question_id, + ai_platform_id, + completed_at, + updated_at, + created_at + FROM question_monitor_runs + WHERE tenant_id = $1 + AND brand_id = $2 + AND collector_type = $3 + AND status = 'succeeded' + AND business_date BETWEEN $4::date AND $5::date + AND question_id = ANY($6) + AND ($7::text[] IS NULL OR ai_platform_id = ANY($7)) + ORDER BY business_date, question_id, ai_platform_id, COALESCE(completed_at, updated_at, created_at) DESC NULLS LAST, id DESC + ) r LEFT JOIN question_monitor_parse_results p ON p.tenant_id = r.tenant_id AND p.run_id = r.id - WHERE r.tenant_id = $1 - AND r.brand_id = $2 - AND r.collector_type = $3 - AND r.status = 'succeeded' - AND r.business_date BETWEEN $4::date AND $5::date - AND r.question_id = ANY($6) - AND ($7::text[] IS NULL OR r.ai_platform_id = ANY($7)) GROUP BY r.business_date ORDER BY r.business_date DESC `, tenantID, brandID, monitoringCollectorType, startDate.Format("2006-01-02"), endDate.Format("2006-01-02"), questionIDs, nullableStringArray(platformIDs)) @@ -3136,15 +3228,25 @@ func (s *MonitoringService) loadPlatformBreakdownFromRuns(ctx context.Context, t COUNT(*) FILTER (WHERE p.brand_mentioned IS TRUE)::bigint AS mentioned_count, COUNT(*) FILTER (WHERE p.brand_mention_position = 'top1')::bigint AS top1_mentioned_count, MAX(r.completed_at) AS last_sampled_at - FROM question_monitor_runs r + FROM ( + SELECT DISTINCT ON (business_date, question_id, ai_platform_id) + id, + tenant_id, + ai_platform_id, + completed_at, + updated_at, + created_at + FROM question_monitor_runs + WHERE tenant_id = $1 + AND brand_id = $2 + AND collector_type = $3 + AND business_date = $4::date + AND status = 'succeeded' + AND question_id = ANY($5) + ORDER BY business_date, question_id, ai_platform_id, COALESCE(completed_at, updated_at, created_at) DESC NULLS LAST, id DESC + ) r LEFT JOIN question_monitor_parse_results p ON p.tenant_id = r.tenant_id AND p.run_id = r.id - WHERE r.tenant_id = $1 - AND r.brand_id = $2 - AND r.collector_type = $3 - AND r.business_date = $4::date - AND r.status = 'succeeded' - AND r.question_id = ANY($5) GROUP BY r.ai_platform_id `, tenantID, brandID, monitoringCollectorType, dateKey, questionIDs) if err != nil { @@ -3192,6 +3294,7 @@ func (s *MonitoringService) loadPlatformBreakdownFromRuns(ctx context.Context, t if platformStats, ok := derived.ByDatePlatform[dateKey][platform.ID]; ok && platformStats.hasSamples() { item.MentionRate = pointerFloat64ToNull(platformStats.mentionRate()) item.Top1MentionRate = pointerFloat64ToNull(platformStats.top1MentionRate()) + item.ActualSampleCount = platformStats.Total } breakdown = append(breakdown, MonitoringPlatformDaily{ AIPlatformID: platform.ID, @@ -3236,11 +3339,15 @@ func (s *MonitoringService) loadHotQuestions( platformQueryIDs := monitoringPlatformQueryIDs(aiPlatformID) rows, err := s.monitoringPool.Query(ctx, ` - WITH stats AS ( - SELECT + WITH latest_runs AS ( + SELECT DISTINCT ON (r.question_id, r.ai_platform_id) r.question_id, - AVG(CASE WHEN p.brand_mentioned IS TRUE THEN 1 ELSE 0 END)::double precision AS mention_rate, - MAX(r.completed_at) AS last_sampled_at + r.ai_platform_id, + r.question_hash, + r.completed_at, + r.updated_at, + r.created_at, + p.brand_mentioned FROM question_monitor_runs r LEFT JOIN question_monitor_parse_results p ON p.tenant_id = r.tenant_id AND p.run_id = r.id @@ -3251,21 +3358,22 @@ func (s *MonitoringService) loadHotQuestions( AND r.business_date = $4::date AND r.question_id = ANY($5) AND ($6::text[] IS NULL OR r.ai_platform_id = ANY($6)) - GROUP BY r.question_id + ORDER BY r.question_id, r.ai_platform_id, COALESCE(r.completed_at, r.updated_at, r.created_at) DESC NULLS LAST, r.id DESC + ), + stats AS ( + SELECT + question_id, + AVG(CASE WHEN brand_mentioned IS TRUE THEN 1 ELSE 0 END)::double precision AS mention_rate, + MAX(completed_at) AS last_sampled_at + FROM latest_runs + GROUP BY question_id ), latest_hash AS ( SELECT DISTINCT ON (r.question_id) r.question_id, r.question_hash - FROM question_monitor_runs r - WHERE r.tenant_id = $1 - AND r.brand_id = $2 - AND r.collector_type = $3 - AND r.status = 'succeeded' - AND r.business_date = $4::date - AND r.question_id = ANY($5) - AND ($6::text[] IS NULL OR r.ai_platform_id = ANY($6)) - ORDER BY r.question_id, r.business_date DESC, COALESCE(r.completed_at, r.updated_at, r.created_at) DESC NULLS LAST, r.id DESC + FROM latest_runs r + ORDER BY r.question_id, COALESCE(r.completed_at, r.updated_at, r.created_at) DESC NULLS LAST ) SELECT s.question_id, diff --git a/server/internal/tenant/app/monitoring_service_date_test.go b/server/internal/tenant/app/monitoring_service_date_test.go index 0c08465..eb361bd 100644 --- a/server/internal/tenant/app/monitoring_service_date_test.go +++ b/server/internal/tenant/app/monitoring_service_date_test.go @@ -1,6 +1,7 @@ package app import ( + "database/sql" "testing" "time" @@ -234,3 +235,73 @@ func TestApplyDerivedRatesToOverviewSnapshotUsesAllSamplesForMentionRate(t *test assert.InDelta(t, 0.25, snapshot.FirstRecommendRate.Float64, 0.0001) assert.InDelta(t, 0.25, snapshot.PositiveMentionRate.Float64, 0.0001) } + +func TestBuildMonitoringDerivedMetricsUsesLatestSamplePerQuestionPlatform(t *testing.T) { + day := time.Date(2026, 6, 29, 0, 0, 0, 0, monitoringBusinessLocation()) + questionID := int64(177) + samples := map[monitoringDerivedSampleKey]monitoringDerivedParseSample{} + + upsertLatestMonitoringDerivedParseSample(samples, monitoringDerivedParseSample{ + BusinessDate: day, + PlatformID: "doubao", + QuestionID: questionID, + EventAt: sql.NullTime{Time: day.Add(10 * time.Hour), Valid: true}, + RunID: 2507, + Summary: monitoringAnswerParseSummary{ + BrandMentioned: true, + BrandMentionPosition: "mentioned", + }, + }) + upsertLatestMonitoringDerivedParseSample(samples, monitoringDerivedParseSample{ + BusinessDate: day, + PlatformID: "doubao", + QuestionID: questionID, + EventAt: sql.NullTime{Time: day.Add(11 * time.Hour), Valid: true}, + RunID: 2628, + Summary: monitoringAnswerParseSummary{ + BrandMentioned: false, + BrandMentionPosition: "not_mentioned", + }, + }) + upsertLatestMonitoringDerivedParseSample(samples, monitoringDerivedParseSample{ + BusinessDate: day, + PlatformID: "qwen", + QuestionID: questionID, + EventAt: sql.NullTime{Time: day.Add(10*time.Hour + 30*time.Minute), Valid: true}, + RunID: 2510, + Summary: monitoringAnswerParseSummary{ + BrandMentioned: true, + BrandMentionPosition: "mentioned", + }, + }) + + metrics := buildMonitoringDerivedMetricsFromLatestSamples(samples) + stats := metrics.ByDateQuestion["2026-06-29"][questionID] + + require.True(t, stats.hasSamples()) + assert.Equal(t, int64(2), stats.Total) + assert.Equal(t, int64(1), stats.MentionedCount) + require.NotNil(t, stats.mentionRate()) + assert.InDelta(t, 0.5, *stats.mentionRate(), 0.0001) +} + +func TestMonitoringDerivedPositiveMentionRequiresBrandMention(t *testing.T) { + var stats monitoringDerivedRateStats + + stats.add(monitoringAnswerParseSummary{ + BrandMentioned: false, + BrandMentionPosition: "not_mentioned", + SentimentLabel: "positive", + }) + stats.add(monitoringAnswerParseSummary{ + BrandMentioned: true, + BrandMentionPosition: "mentioned", + SentimentLabel: "positive", + }) + + assert.Equal(t, int64(2), stats.Total) + assert.Equal(t, int64(1), stats.MentionedCount) + assert.Equal(t, int64(1), stats.PositiveMentions) + require.NotNil(t, stats.positiveMentionRate()) + assert.InDelta(t, 0.5, *stats.positiveMentionRate(), 0.0001) +}