refactor: streamline monitoring daily task logic and remove unused functions

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
+	TenantID        int64
-	WorkspaceID          int64
+	WorkspaceID     int64
-	MaxBrands            int
+	MaxBrands       int
-	EnabledJSON          []byte
+	EnabledJSON     []byte
-	PrimaryClientID      *uuid.UUID
+	PrimaryClientID *uuid.UUID
-	MaxQuestionsPerBrand int
 }
 
 type monitoringDailyPlanClientCandidate struct {
-	TenantID             int64
+	TenantID        int64
-	WorkspaceID          int64
+	WorkspaceID     int64
-	PrimaryClientID      uuid.UUID
+	PrimaryClientID uuid.UUID
-	MaxBrands            int
+	MaxBrands       int
-	MaxQuestionsPerBrand 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,
+			)::int AS max_brands
-			$3::int AS max_questions_per_brand
 		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,
+			TenantID:        candidate.TenantID,
-			WorkspaceID:          candidate.WorkspaceID,
+			WorkspaceID:     candidate.WorkspaceID,
-			MaxBrands:            candidate.MaxBrands,
+			MaxBrands:       candidate.MaxBrands,
-			EnabledJSON:          append([]byte(nil), enabledJSON...),
+			EnabledJSON:     append([]byte(nil), enabledJSON...),
-			PrimaryClientID:      &clientID,
+			PrimaryClientID: &clientID,
-			MaxQuestionsPerBrand: candidate.MaxQuestionsPerBrand,
 		})
 	}
 	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

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) {
+func TestBuildMonitoringDailyTaskCandidatesIncludesQuestionsBeyondLibraryLimit(t *testing.T) {
-	assert.Equal(t, 25, monitoringDailyQuestionLimit(config.BrandLibraryConfig{
+	businessDay := time.Date(2026, 4, 24, 0, 0, 0, 0, monitoringBusinessLocation())
-		MaxKeywords:            5,
+	questions := make([]monitoringConfiguredQuestion, 0, 30)
-		MaxQuestionsPerKeyword: 5,
+	for id := int64(1); id <= 30; id++ {
-	}))
+		questions = append(questions, monitoringConfiguredQuestion{
-	assert.Equal(t, 6, monitoringDailyQuestionLimit(config.BrandLibraryConfig{
+			ID:           id,
-		MaxKeywords:            2,
+			QuestionHash: []byte("q" + int64Text(id)),
-		MaxQuestionsPerKeyword: 3,
+		})
-	}))
-
-	questions := []monitoringConfiguredQuestion{
-		{ID: 1},
-		{ID: 2},
-		{ID: 3},
 	}
-	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,
+			TenantID:        1,
-			WorkspaceID:          10,
+			WorkspaceID:     10,
-			PrimaryClientID:      clientOne,
+			PrimaryClientID: clientOne,
-			MaxBrands:            3,
+			MaxBrands:       3,
-			MaxQuestionsPerBrand: 25,
 		},
 		{
-			TenantID:             1,
+			TenantID:        1,
-			WorkspaceID:          11,
+			WorkspaceID:     11,
-			PrimaryClientID:      clientTwo,
+			PrimaryClientID: clientTwo,
-			MaxBrands:            2,
+			MaxBrands:       2,
-			MaxQuestionsPerBrand: 10,
 		},
 	}
 	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},