From 316aae240eaa85347c23509f0f0c2870361f4955 Mon Sep 17 00:00:00 2001 From: liangxu Date: Sun, 21 Jun 2026 15:34:50 +0800 Subject: [PATCH] refactor: streamline monitoring daily task logic and remove unused functions --- .../app/monitoring_daily_task_worker.go | 188 +++++------------- .../app/monitoring_daily_task_worker_test.go | 86 ++++---- 2 files changed, 101 insertions(+), 173 deletions(-) diff --git a/server/internal/tenant/app/monitoring_daily_task_worker.go b/server/internal/tenant/app/monitoring_daily_task_worker.go index bd81a6b..67ab1df 100644 --- a/server/internal/tenant/app/monitoring_daily_task_worker.go +++ b/server/internal/tenant/app/monitoring_daily_task_worker.go @@ -14,7 +14,6 @@ import ( "github.com/jackc/pgx/v5" "go.uber.org/zap" - "github.com/geo-platform/tenant-api/internal/shared/config" "github.com/geo-platform/tenant-api/internal/shared/response" "github.com/geo-platform/tenant-api/internal/shared/stream" ) @@ -91,20 +90,18 @@ func normalizeMonitoringDailyRunOptions(options ...MonitoringDailyTaskRunOptions } type monitoringDailyPlan struct { - TenantID int64 - WorkspaceID int64 - MaxBrands int - EnabledJSON []byte - PrimaryClientID *uuid.UUID - MaxQuestionsPerBrand int + TenantID int64 + WorkspaceID int64 + MaxBrands int + EnabledJSON []byte + PrimaryClientID *uuid.UUID } type monitoringDailyPlanClientCandidate struct { - TenantID int64 - WorkspaceID int64 - PrimaryClientID uuid.UUID - MaxBrands int - MaxQuestionsPerBrand int + TenantID int64 + WorkspaceID int64 + PrimaryClientID uuid.UUID + MaxBrands int } type monitoringDailyPlanClientKey struct { @@ -327,21 +324,10 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks( s.logDailyMonitoringBrandError("monitoring daily question load failed", plan, brand, businessDate, err) continue } - questions = limitMonitoringDailyQuestions(questions, plan.MaxQuestionsPerBrand) if len(questions) == 0 { continue } dailyHardCap := len(questions) * len(platforms) - materializedCount, err := s.countDailyMonitoringMaterializedTasks(ctx, plan.TenantID, brand.BrandID, businessDay) - if err != nil { - if abortErr := monitoringDailyAbortErr(ctx); abortErr != nil { - return nil, abortErr - } - summary.FailedBrandCount++ - s.logDailyMonitoringBrandError("monitoring daily materialized task count failed", plan, brand, businessDate, err) - continue - } - dailyRemaining := monitoringDailyRemainingCapacity(dailyHardCap, materializedCount) candidates := buildMonitoringDailyTaskCandidates(plan.TenantID, brand.BrandID, businessDay, questions, platforms, dailyHardCap) if len(candidates) == 0 { @@ -349,7 +335,7 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks( } dispatchLimit := minMonitoringDailyInt(options.MaxMaterializePerBrand, planRunRemaining, globalRemaining) - materializeLimit := minMonitoringDailyInt(dispatchLimit, dailyRemaining) + materializeLimit := dispatchLimit createdCount, dueCount, desktopCount, deferredCount, err := s.generateDailyMonitoringTasksForBrand( ctx, projectionService, @@ -372,7 +358,6 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks( continue } - dailyRemaining = decrementMonitoringDailyRemaining(dailyRemaining, createdCount) processedCount := int(maxMonitoringDailyInt(dueCount, desktopCount+deferredCount)) if processedCount > 0 { planRunRemaining -= processedCount @@ -569,10 +554,9 @@ func (s *MonitoringService) loadDailyMonitoringPlans(ctx context.Context) ([]mon _, brandLibraryConfig := s.currentConfig() freeBrandLimit := normalizeMonitoringDailyBrandLimit(brandLibraryConfig.FreeBrandLimit) paidBrandLimit := normalizeMonitoringDailyBrandLimit(brandLibraryConfig.PaidBrandLimit) - questionLimit := monitoringDailyQuestionLimit(brandLibraryConfig) supportedPlatforms := monitoringPlatformIDs(defaultMonitoringPlatforms) - candidates, err := s.loadDailyMonitoringPrimaryClientPlans(ctx, freeBrandLimit, paidBrandLimit, questionLimit, supportedPlatforms) + candidates, err := s.loadDailyMonitoringPrimaryClientPlans(ctx, freeBrandLimit, paidBrandLimit, supportedPlatforms) if err != nil { return nil, err } @@ -591,10 +575,42 @@ func (s *MonitoringService) loadDailyMonitoringPrimaryClientPlans( ctx context.Context, freeBrandLimit int, paidBrandLimit int, - questionLimit int, supportedPlatforms []string, ) ([]monitoringDailyPlanClientCandidate, error) { - rows, err := s.businessPool.Query(ctx, ` + rows, err := s.businessPool.Query(ctx, dailyMonitoringPrimaryClientPlansSQL(), + freeBrandLimit, + paidBrandLimit, + supportedPlatforms, + formatPgInterval(monitoringDailyPlatformActiveWindow), + ) + if err != nil { + appErr := response.ErrInternal(50041, "query_failed", "failed to load daily monitoring plans") + appErr.Cause = err + return nil, appErr + } + defer rows.Close() + + candidates := make([]monitoringDailyPlanClientCandidate, 0) + for rows.Next() { + var item monitoringDailyPlanClientCandidate + if scanErr := rows.Scan( + &item.TenantID, + &item.WorkspaceID, + &item.PrimaryClientID, + &item.MaxBrands, + ); scanErr != nil { + return nil, response.ErrInternal(50041, "scan_failed", "failed to parse daily monitoring plans") + } + candidates = append(candidates, item) + } + if err := rows.Err(); err != nil { + return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate daily monitoring plans") + } + return candidates, nil +} + +func dailyMonitoringPrimaryClientPlansSQL() string { + return ` WITH eligible_clients AS ( SELECT dc.tenant_id, @@ -616,7 +632,7 @@ func (s *MonitoringService) loadDailyMonitoringPrimaryClientPlans( AND p.deleted_at IS NULL WHERE dc.revoked_at IS NULL AND dc.last_seen_at IS NOT NULL - AND dc.last_seen_at >= NOW() - $5::interval + AND dc.last_seen_at >= NOW() - $4::interval AND EXISTS ( SELECT 1 FROM platform_accounts pa @@ -624,7 +640,7 @@ func (s *MonitoringService) loadDailyMonitoringPrimaryClientPlans( AND pa.workspace_id = dc.workspace_id AND pa.client_id = dc.id AND pa.deleted_at IS NULL - AND pa.platform_id = ANY($4::text[]) + AND pa.platform_id = ANY($3::text[]) ) ), selected_clients AS ( @@ -657,36 +673,10 @@ func (s *MonitoringService) loadDailyMonitoringPrimaryClientPlans( CASE WHEN c.plan_code = 'free' THEN $1::int ELSE $2::int END ), 0 - )::int AS max_brands, - $3::int AS max_questions_per_brand + )::int AS max_brands FROM selected_clients c ORDER BY c.tenant_id ASC, c.workspace_id ASC - `, freeBrandLimit, paidBrandLimit, questionLimit, supportedPlatforms, formatPgInterval(monitoringDailyPlatformActiveWindow)) - if err != nil { - appErr := response.ErrInternal(50041, "query_failed", "failed to load daily monitoring plans") - appErr.Cause = err - return nil, appErr - } - defer rows.Close() - - candidates := make([]monitoringDailyPlanClientCandidate, 0) - for rows.Next() { - var item monitoringDailyPlanClientCandidate - if scanErr := rows.Scan( - &item.TenantID, - &item.WorkspaceID, - &item.PrimaryClientID, - &item.MaxBrands, - &item.MaxQuestionsPerBrand, - ); scanErr != nil { - return nil, response.ErrInternal(50041, "scan_failed", "failed to parse daily monitoring plans") - } - candidates = append(candidates, item) - } - if err := rows.Err(); err != nil { - return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate daily monitoring plans") - } - return candidates, nil + ` } func (s *MonitoringService) loadDailyMonitoringPlanPlatformJSON( @@ -796,12 +786,11 @@ func buildDailyMonitoringPlansFromSelectedClients( } clientID := candidate.PrimaryClientID plans = append(plans, monitoringDailyPlan{ - TenantID: candidate.TenantID, - WorkspaceID: candidate.WorkspaceID, - MaxBrands: candidate.MaxBrands, - EnabledJSON: append([]byte(nil), enabledJSON...), - PrimaryClientID: &clientID, - MaxQuestionsPerBrand: candidate.MaxQuestionsPerBrand, + TenantID: candidate.TenantID, + WorkspaceID: candidate.WorkspaceID, + MaxBrands: candidate.MaxBrands, + EnabledJSON: append([]byte(nil), enabledJSON...), + PrimaryClientID: &clientID, }) } return plans @@ -883,27 +872,6 @@ func (s *MonitoringService) loadDailyMonitoringTaskKeys(ctx context.Context, q m return keys, nil } -func (s *MonitoringService) countDailyMonitoringMaterializedTasks(ctx context.Context, tenantID, brandID int64, businessDay time.Time) (int64, error) { - if s == nil || s.monitoringPool == nil { - return 0, nil - } - var count int64 - if err := s.monitoringPool.QueryRow(ctx, ` - SELECT COUNT(*) - FROM monitoring_collect_tasks - WHERE tenant_id = $1 - AND brand_id = $2 - AND collector_type = $3 - AND trigger_source = 'automatic' - AND run_mode = 'desktop_standard' - AND business_date = $4::date - AND COALESCE(execution_owner, 'legacy') = 'desktop_tasks' - `, tenantID, brandID, monitoringCollectorType, monitoringBusinessDateText(businessDay)).Scan(&count); err != nil { - return 0, response.ErrInternal(50041, "query_failed", "failed to count daily monitoring materialized tasks") - } - return count, nil -} - func (s *MonitoringService) loadDailyMonitorDispatchablePlatformIDs( ctx context.Context, tenantID, workspaceID int64, @@ -1344,54 +1312,6 @@ func normalizeMonitoringDailyBrandLimit(value int) int { return value } -func monitoringDailyQuestionLimit(cfg config.BrandLibraryConfig) int { - maxKeywords := cfg.MaxKeywords - if maxKeywords <= 0 { - maxKeywords = 5 - } - maxQuestionsPerKeyword := cfg.MaxQuestionsPerKeyword - if maxQuestionsPerKeyword <= 0 { - maxQuestionsPerKeyword = 5 - } - return maxKeywords * maxQuestionsPerKeyword -} - -func normalizeMonitoringDailyQuestionLimit(value int) int { - if value <= 0 { - return monitoringDailyQuestionLimit(config.BrandLibraryConfig{}) - } - return value -} - -func limitMonitoringDailyQuestions(questions []monitoringConfiguredQuestion, limit int) []monitoringConfiguredQuestion { - limit = normalizeMonitoringDailyQuestionLimit(limit) - if len(questions) <= limit { - return questions - } - return questions[:limit] -} - -func monitoringDailyRemainingCapacity(hardCap int, materializedCount int64) int { - capacity := normalizeMonitoringDailyTaskHardCap(hardCap) - if materializedCount <= 0 { - return capacity - } - if materializedCount >= int64(capacity) { - return 0 - } - return capacity - int(materializedCount) -} - -func decrementMonitoringDailyRemaining(remaining int, createdCount int64) int { - if remaining <= 0 || createdCount <= 0 { - return remaining - } - if createdCount >= int64(remaining) { - return 0 - } - return remaining - int(createdCount) -} - func nullableUUID(value *uuid.UUID) interface{} { if value == nil { return nil 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 00d3dff..f016535 100644 --- a/server/internal/tenant/app/monitoring_daily_task_worker_test.go +++ b/server/internal/tenant/app/monitoring_daily_task_worker_test.go @@ -14,8 +14,6 @@ import ( "github.com/jackc/pgx/v5/pgconn" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - - "github.com/geo-platform/tenant-api/internal/shared/config" ) func TestBuildMonitoringDailyTaskCandidatesMaterializesAfterMidnight(t *testing.T) { @@ -84,11 +82,8 @@ func TestSelectDueMonitoringDailyCandidatesSkipsExistingAndHonorsLimit(t *testin } } -func TestMonitoringDailyRemainingKeepsFullCandidateHorizon(t *testing.T) { +func TestSelectDueMonitoringDailyCandidatesContinuesPastExistingTasks(t *testing.T) { businessDay := time.Date(2026, 4, 24, 0, 0, 0, 0, monitoringBusinessLocation()) - hardCap := 6 - remaining := monitoringDailyRemainingCapacity(hardCap, 4) - assert.Equal(t, 2, remaining) candidates := buildMonitoringDailyTaskCandidates( 7, @@ -103,9 +98,9 @@ func TestMonitoringDailyRemainingKeepsFullCandidateHorizon(t *testing.T) { {ID: "qwen"}, {ID: "doubao"}, }, - hardCap, + 6, ) - require.Len(t, candidates, hardCap) + require.Len(t, candidates, 6) existing := make(map[monitoringDailyTaskKey]struct{}, 4) for _, candidate := range candidates[:4] { @@ -116,9 +111,9 @@ func TestMonitoringDailyRemainingKeepsFullCandidateHorizon(t *testing.T) { } now := monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset + defaultMonitoringDailyMaterializeWindow) - due := selectDueMonitoringDailyCandidates(candidates, existing, now, defaultMonitoringDailyLookahead, remaining) + due := selectDueMonitoringDailyCandidates(candidates, existing, now, defaultMonitoringDailyLookahead, 4) - require.Len(t, due, remaining) + require.Len(t, due, 2) for _, candidate := range due { _, exists := existing[monitoringDailyTaskKey{ QuestionID: candidate.Question.ID, @@ -126,27 +121,35 @@ func TestMonitoringDailyRemainingKeepsFullCandidateHorizon(t *testing.T) { }] assert.False(t, exists) } - assert.Equal(t, 1, decrementMonitoringDailyRemaining(remaining, 1)) - assert.Equal(t, 0, decrementMonitoringDailyRemaining(remaining, 2)) } -func TestMonitoringDailyQuestionLimitUsesBrandLibraryConfig(t *testing.T) { - assert.Equal(t, 25, monitoringDailyQuestionLimit(config.BrandLibraryConfig{ - MaxKeywords: 5, - MaxQuestionsPerKeyword: 5, - })) - assert.Equal(t, 6, monitoringDailyQuestionLimit(config.BrandLibraryConfig{ - MaxKeywords: 2, - MaxQuestionsPerKeyword: 3, - })) - - questions := []monitoringConfiguredQuestion{ - {ID: 1}, - {ID: 2}, - {ID: 3}, +func TestBuildMonitoringDailyTaskCandidatesIncludesQuestionsBeyondLibraryLimit(t *testing.T) { + businessDay := time.Date(2026, 4, 24, 0, 0, 0, 0, monitoringBusinessLocation()) + questions := make([]monitoringConfiguredQuestion, 0, 30) + for id := int64(1); id <= 30; id++ { + questions = append(questions, monitoringConfiguredQuestion{ + ID: id, + QuestionHash: []byte("q" + int64Text(id)), + }) } - assert.Len(t, limitMonitoringDailyQuestions(questions, 2), 2) - assert.Len(t, limitMonitoringDailyQuestions(questions, 25), 3) + + candidates := buildMonitoringDailyTaskCandidates( + 7, + 11, + businessDay, + questions, + []monitoringPlatformMetadata{{ID: "qwen"}}, + len(questions), + ) + + seen := make(map[int64]struct{}, len(candidates)) + for _, candidate := range candidates { + seen[candidate.Question.ID] = struct{}{} + } + + require.Len(t, candidates, len(questions)) + assert.Contains(t, seen, int64(26)) + assert.Contains(t, seen, int64(30)) } func TestDecodeDailyMonitoringEnabledPlatformsReportsMalformedJSON(t *testing.T) { @@ -166,18 +169,16 @@ func TestBuildDailyMonitoringPlansFromSelectedClientsRequiresPlatformSnapshot(t clientTwo := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000002") candidates := []monitoringDailyPlanClientCandidate{ { - TenantID: 1, - WorkspaceID: 10, - PrimaryClientID: clientOne, - MaxBrands: 3, - MaxQuestionsPerBrand: 25, + TenantID: 1, + WorkspaceID: 10, + PrimaryClientID: clientOne, + MaxBrands: 3, }, { - TenantID: 1, - WorkspaceID: 11, - PrimaryClientID: clientTwo, - MaxBrands: 2, - MaxQuestionsPerBrand: 10, + TenantID: 1, + WorkspaceID: 11, + PrimaryClientID: clientTwo, + MaxBrands: 2, }, } enabledByClient := map[monitoringDailyPlanClientKey][]byte{ @@ -190,12 +191,19 @@ func TestBuildDailyMonitoringPlansFromSelectedClientsRequiresPlatformSnapshot(t assert.Equal(t, int64(1), plans[0].TenantID) assert.Equal(t, int64(10), plans[0].WorkspaceID) assert.Equal(t, 3, plans[0].MaxBrands) - assert.Equal(t, 25, plans[0].MaxQuestionsPerBrand) require.NotNil(t, plans[0].PrimaryClientID) assert.Equal(t, clientOne, *plans[0].PrimaryClientID) assert.Equal(t, []byte(`["qwen"]`), plans[0].EnabledJSON) } +func TestDailyMonitoringPrimaryClientPlansSQLUsesContiguousParameters(t *testing.T) { + sql := dailyMonitoringPrimaryClientPlansSQL() + + assert.Contains(t, normalizeSQLWhitespace(sql), "pa.platform_id = ANY($3::text[])") + assert.Contains(t, normalizeSQLWhitespace(sql), "dc.last_seen_at >= NOW() - $4::interval") + assert.NotContains(t, sql, "$5") +} + func TestOrderDailyMonitoringPlansRotatesDeterministicallyByBucket(t *testing.T) { plans := []monitoringDailyPlan{ {TenantID: 1},