geo/server/internal/tenant/app/monitoring_projection.go

package app

import (
	"context"
	"database/sql"
	"time"

	"github.com/jackc/pgx/v5"
)

type monitoringBrandDailyAggregate struct {
	PlannedSampleCount     int64
	ActualSampleCount      int64
	MentionedCount         int64
	Top1MentionedCount     int64
	FirstRecommendedCount  int64
	PositiveMentionedCount int64
	CitedAnswerCount       int64
	SnapshotUpdatedAt      sql.NullTime
	LatestTriggerSource    sql.NullString
}

type monitoringPlatformDailyAggregate struct {
	AIPlatformID       string
	PlannedSampleCount int64
	ActualSampleCount  int64
	MentionedCount     int64
	Top1MentionedCount int64
	LastSampledAt      sql.NullTime
	AccessStatus       string
}

func (s *MonitoringCallbackService) rebuildBrandDailyProjection(ctx context.Context, tx pgx.Tx, tenantID, brandID int64, businessDate time.Time) error {
	platforms, err := loadMonitoringProjectionPlatforms(ctx, tx, tenantID, brandID, businessDate)
	if err != nil {
		return err
	}

	enabledQuestionCount, err := s.loadMonitoringEnabledQuestionCount(ctx, tx, tenantID, brandID, businessDate)
	if err != nil {
		return err
	}

	brandAgg, err := s.loadMonitoringBrandDailyAggregate(ctx, tx, tenantID, brandID, businessDate)
	if err != nil {
		return err
	}

	platformAggs, err := s.loadMonitoringPlatformDailyAggregates(ctx, tx, tenantID, brandID, businessDate, platforms)
	if err != nil {
		return err
	}

	desiredSampleCount := enabledQuestionCount * int64(len(platforms))
	coverageRate := divideAsPointer(brandAgg.ActualSampleCount, brandAgg.PlannedSampleCount)
	mentionRate := divideAsPointer(brandAgg.MentionedCount, brandAgg.ActualSampleCount)
	top1MentionRate := divideAsPointer(brandAgg.Top1MentionedCount, brandAgg.ActualSampleCount)
	firstRecommendRate := divideAsPointer(brandAgg.FirstRecommendedCount, brandAgg.ActualSampleCount)
	positiveMentionRate := divideAsPointer(brandAgg.PositiveMentionedCount, brandAgg.ActualSampleCount)
	citationRate := divideAsPointer(brandAgg.CitedAnswerCount, brandAgg.ActualSampleCount)

	if _, err := tx.Exec(ctx, `
		INSERT INTO monitoring_brand_daily (
			tenant_id, brand_id, collector_type, business_date, sample_status,
			desired_sample_count, planned_sample_count, actual_sample_count,
			coverage_rate, confidence_level, trigger_source, snapshot_updated_at,
			mentioned_count, top1_mentioned_count, first_recommended_count, positive_mentioned_count,
			cited_answer_count, mention_rate, top1_mention_rate, first_recommend_rate,
			positive_mention_rate, citation_rate
		)
		VALUES (
			$1, $2, $3, $4::date, $5,
			$6, $7, $8,
			$9, $10, $11, $12,
			$13, $14, $15, $16,
			$17, $18, $19, $20,
			$21, $22
		)
		ON CONFLICT (tenant_id, brand_id, collector_type, business_date)
		DO UPDATE SET
			sample_status = EXCLUDED.sample_status,
			desired_sample_count = EXCLUDED.desired_sample_count,
			planned_sample_count = EXCLUDED.planned_sample_count,
			actual_sample_count = EXCLUDED.actual_sample_count,
			coverage_rate = EXCLUDED.coverage_rate,
			confidence_level = EXCLUDED.confidence_level,
			trigger_source = EXCLUDED.trigger_source,
			snapshot_updated_at = EXCLUDED.snapshot_updated_at,
			mentioned_count = EXCLUDED.mentioned_count,
			top1_mentioned_count = EXCLUDED.top1_mentioned_count,
			first_recommended_count = EXCLUDED.first_recommended_count,
			positive_mentioned_count = EXCLUDED.positive_mentioned_count,
			cited_answer_count = EXCLUDED.cited_answer_count,
			mention_rate = EXCLUDED.mention_rate,
			top1_mention_rate = EXCLUDED.top1_mention_rate,
			first_recommend_rate = EXCLUDED.first_recommend_rate,
			positive_mention_rate = EXCLUDED.positive_mention_rate,
			citation_rate = EXCLUDED.citation_rate,
			updated_at = NOW()
	`,
		tenantID,
		brandID,
		monitoringCollectorType,
		monitoringBusinessDateText(businessDate),
		deriveMonitoringBrandSampleStatus(brandAgg.ActualSampleCount, brandAgg.PlannedSampleCount, coverageRate),
		desiredSampleCount,
		brandAgg.PlannedSampleCount,
		brandAgg.ActualSampleCount,
		nullableFloat64(coverageRate),
		deriveMonitoringConfidenceLevel(brandAgg.ActualSampleCount, brandAgg.PlannedSampleCount, coverageRate),
		nullableString(stringPointer(brandAgg.LatestTriggerSource)),
		nullableTime(brandAgg.SnapshotUpdatedAt),
		brandAgg.MentionedCount,
		brandAgg.Top1MentionedCount,
		brandAgg.FirstRecommendedCount,
		brandAgg.PositiveMentionedCount,
		brandAgg.CitedAnswerCount,
		nullableFloat64(mentionRate),
		nullableFloat64(top1MentionRate),
		nullableFloat64(firstRecommendRate),
		nullableFloat64(positiveMentionRate),
		nullableFloat64(citationRate),
	); err != nil {
		return monitoringInternalError(50041, "brand_daily_projection_failed", "failed to persist monitoring brand daily projection", err)
	}

	if err := s.replaceMonitoringPlatformDailyProjection(ctx, tx, tenantID, brandID, businessDate, platformAggs); err != nil {
		return err
	}

	return nil
}

func loadMonitoringProjectionPlatforms(_ context.Context, _ pgx.Tx, _ int64, _ int64, _ time.Time) ([]monitoringPlatformMetadata, error) {
	return defaultMonitoringPlatformMetadata(), nil
}

func (s *MonitoringCallbackService) loadMonitoringEnabledQuestionCount(ctx context.Context, tx pgx.Tx, tenantID, brandID int64, businessDate time.Time) (int64, error) {
	dayEnd := monitoringCanonicalBusinessDay(businessDate).AddDate(0, 0, 1)

	var count int64
	if err := tx.QueryRow(ctx, `
		SELECT COUNT(DISTINCT question_id)
		FROM monitoring_question_config_snapshots
		WHERE tenant_id = $1
		  AND brand_id = $2
		  AND monitor_enabled = TRUE
		  AND projected_at < $3
		  AND (superseded_at IS NULL OR superseded_at >= $3)
		  AND (deleted_at IS NULL OR deleted_at >= $3)
	`, tenantID, brandID, dayEnd).Scan(&count); err != nil {
		return 0, monitoringInternalError(50041, "question_snapshot_lookup_failed", "failed to load monitoring question snapshots", err)
	}

	return count, nil
}

func (s *MonitoringCallbackService) loadMonitoringBrandDailyAggregate(ctx context.Context, tx pgx.Tx, tenantID, brandID int64, businessDate time.Time) (monitoringBrandDailyAggregate, error) {
	agg := monitoringBrandDailyAggregate{}
	dateText := monitoringBusinessDateText(businessDate)

	if err := tx.QueryRow(ctx, `
		SELECT COUNT(*)
		FROM monitoring_collect_tasks
		WHERE tenant_id = $1
		  AND brand_id = $2
		  AND collector_type = $3
		  AND business_date = $4::date
		  AND NOT (status = 'skipped' AND skip_reason = ANY($5::text[]))
	`, tenantID, brandID, monitoringCollectorType, dateText, monitoringProjectionExcludedSkipReasons).Scan(&agg.PlannedSampleCount); err != nil {
		return agg, monitoringInternalError(50041, "task_projection_lookup_failed", "failed to load monitoring task counts", err)
	}

	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'
	`, 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)
	}

	if err := tx.QueryRow(ctx, `
		SELECT
			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
		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,
		&agg.FirstRecommendedCount,
		&agg.PositiveMentionedCount,
	); err != nil {
		return agg, monitoringInternalError(50041, "parse_projection_lookup_failed", "failed to load monitoring parse aggregates", err)
	}

	if err := tx.QueryRow(ctx, `
		SELECT COUNT(DISTINCT cf.run_id)
		FROM monitoring_citation_facts cf
		JOIN question_monitor_runs r
			ON r.tenant_id = cf.tenant_id AND r.id = cf.run_id
		WHERE cf.tenant_id = $1
		  AND cf.brand_id = $2
		  AND cf.article_id IS NOT NULL
		  AND r.ai_platform_id <> 'qwen'
		  AND r.collector_type = $3
		  AND r.business_date = $4::date
		  AND r.status = 'succeeded'
	`, tenantID, brandID, monitoringCollectorType, dateText).Scan(&agg.CitedAnswerCount); err != nil {
		return agg, monitoringInternalError(50041, "citation_projection_lookup_failed", "failed to load monitoring citation aggregates", err)
	}

	if err := tx.QueryRow(ctx, `
		SELECT trigger_source
		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 completed_at DESC NULLS LAST, id DESC
		LIMIT 1
	`, tenantID, brandID, monitoringCollectorType, dateText).Scan(&agg.LatestTriggerSource); err != nil && err != pgx.ErrNoRows {
		return agg, monitoringInternalError(50041, "trigger_source_lookup_failed", "failed to load monitoring trigger source", err)
	}

	return agg, nil
}

func (s *MonitoringCallbackService) loadMonitoringPlatformDailyAggregates(ctx context.Context, tx pgx.Tx, tenantID, brandID int64, businessDate time.Time, platforms []monitoringPlatformMetadata) ([]monitoringPlatformDailyAggregate, error) {
	dateText := monitoringBusinessDateText(businessDate)

	taskCounts := make(map[string]int64, len(platforms))
	taskRows, err := tx.Query(ctx, `
		SELECT ai_platform_id, COUNT(*)
		FROM monitoring_collect_tasks
		WHERE tenant_id = $1
		  AND brand_id = $2
		  AND collector_type = $3
		  AND business_date = $4::date
		  AND NOT (status = 'skipped' AND skip_reason = ANY($5::text[]))
		GROUP BY ai_platform_id
	`, tenantID, brandID, monitoringCollectorType, dateText, monitoringProjectionExcludedSkipReasons)
	if err != nil {
		return nil, monitoringInternalError(50041, "platform_task_projection_lookup_failed", "failed to load monitoring platform task counts", err)
	}
	for taskRows.Next() {
		var platformID string
		var plannedSampleCount int64
		if scanErr := taskRows.Scan(&platformID, &plannedSampleCount); scanErr != nil {
			taskRows.Close()
			return nil, monitoringInternalError(50041, "platform_task_projection_scan_failed", "failed to parse monitoring platform task counts", scanErr)
		}
		taskCounts[platformID] = plannedSampleCount
	}
	if err := taskRows.Err(); err != nil {
		taskRows.Close()
		return nil, monitoringInternalError(50041, "platform_task_projection_scan_failed", "failed to iterate monitoring platform task counts", err)
	}
	taskRows.Close()

	runAggs := make(map[string]monitoringPlatformDailyAggregate, len(platforms))
	runRows, err := tx.Query(ctx, `
		SELECT
			r.ai_platform_id,
			COUNT(*)::bigint AS actual_sample_count,
			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
		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 {
		return nil, monitoringInternalError(50041, "platform_run_projection_lookup_failed", "failed to load monitoring platform run aggregates", err)
	}
	for runRows.Next() {
		var item monitoringPlatformDailyAggregate
		if scanErr := runRows.Scan(
			&item.AIPlatformID,
			&item.ActualSampleCount,
			&item.MentionedCount,
			&item.Top1MentionedCount,
			&item.LastSampledAt,
		); scanErr != nil {
			runRows.Close()
			return nil, monitoringInternalError(50041, "platform_run_projection_scan_failed", "failed to parse monitoring platform run aggregates", scanErr)
		}
		runAggs[item.AIPlatformID] = item
	}
	if err := runRows.Err(); err != nil {
		runRows.Close()
		return nil, monitoringInternalError(50041, "platform_run_projection_scan_failed", "failed to iterate monitoring platform run aggregates", err)
	}
	runRows.Close()

	accessStates := make(map[string]string, len(platforms))
	accessRows, err := tx.Query(ctx, `
		SELECT DISTINCT ON (ai_platform_id) ai_platform_id, access_status
		FROM monitoring_platform_access_snapshots
		WHERE tenant_id = $1
		  AND business_date = $2::date
		ORDER BY ai_platform_id ASC, detected_at DESC, id DESC
	`, tenantID, dateText)
	if err != nil {
		return nil, monitoringInternalError(50041, "platform_access_projection_lookup_failed", "failed to load monitoring platform access states", err)
	}
	for accessRows.Next() {
		var platformID string
		var accessStatus string
		if scanErr := accessRows.Scan(&platformID, &accessStatus); scanErr != nil {
			accessRows.Close()
			return nil, monitoringInternalError(50041, "platform_access_projection_scan_failed", "failed to parse monitoring platform access states", scanErr)
		}
		accessStates[platformID] = accessStatus
	}
	if err := accessRows.Err(); err != nil {
		accessRows.Close()
		return nil, monitoringInternalError(50041, "platform_access_projection_scan_failed", "failed to iterate monitoring platform access states", err)
	}
	accessRows.Close()

	items := make([]monitoringPlatformDailyAggregate, 0, len(platforms))
	for _, platform := range platforms {
		item := runAggs[platform.ID]
		item.AIPlatformID = platform.ID
		item.PlannedSampleCount = taskCounts[platform.ID]
		item.AccessStatus = accessStates[platform.ID]
		items = append(items, item)
	}

	return items, nil
}

func (s *MonitoringCallbackService) replaceMonitoringPlatformDailyProjection(ctx context.Context, tx pgx.Tx, tenantID, brandID int64, businessDate time.Time, items []monitoringPlatformDailyAggregate) error {
	dateText := monitoringBusinessDateText(businessDate)

	if _, err := tx.Exec(ctx, `
		DELETE FROM monitoring_brand_platform_daily
		WHERE tenant_id = $1
		  AND brand_id = $2
		  AND collector_type = $3
		  AND business_date = $4::date
	`, tenantID, brandID, monitoringCollectorType, dateText); err != nil {
		return monitoringInternalError(50041, "platform_daily_projection_cleanup_failed", "failed to reset monitoring platform daily projection", err)
	}

	for _, item := range items {
		mentionRate := divideAsPointer(item.MentionedCount, item.ActualSampleCount)
		top1MentionRate := divideAsPointer(item.Top1MentionedCount, item.ActualSampleCount)

		if _, err := tx.Exec(ctx, `
			INSERT INTO monitoring_brand_platform_daily (
				tenant_id, brand_id, ai_platform_id, collector_type, business_date,
				platform_sample_status, planned_sample_count, actual_sample_count,
				mention_rate, top1_mention_rate, last_sampled_at
			)
			VALUES (
				$1, $2, $3, $4, $5::date,
				$6, $7, $8,
				$9, $10, $11
			)
		`,
			tenantID,
			brandID,
			item.AIPlatformID,
			monitoringCollectorType,
			dateText,
			deriveMonitoringPlatformSampleStatus(item.ActualSampleCount, item.PlannedSampleCount, item.AccessStatus),
			item.PlannedSampleCount,
			item.ActualSampleCount,
			nullableFloat64(mentionRate),
			nullableFloat64(top1MentionRate),
			nullableTime(item.LastSampledAt),
		); err != nil {
			return monitoringInternalError(50041, "platform_daily_projection_failed", "failed to persist monitoring platform daily projection", err)
		}
	}

	return nil
}

func (s *MonitoringCallbackService) loadMonitoringAffectedBrandIDs(ctx context.Context, tx pgx.Tx, tenantID int64, businessDate time.Time, aiPlatformID string) ([]int64, error) {
	rows, err := tx.Query(ctx, `
		SELECT DISTINCT brand_id
		FROM monitoring_collect_tasks
		WHERE tenant_id = $1
		  AND ai_platform_id = $2
		  AND collector_type = $3
		  AND business_date = $4::date
		ORDER BY brand_id ASC
	`, tenantID, aiPlatformID, monitoringCollectorType, monitoringBusinessDateText(businessDate))
	if err != nil {
		return nil, monitoringInternalError(50041, "affected_brand_lookup_failed", "failed to load monitoring brands for projection", err)
	}
	defer rows.Close()

	brandIDs := make([]int64, 0)
	for rows.Next() {
		var brandID int64
		if scanErr := rows.Scan(&brandID); scanErr != nil {
			return nil, monitoringInternalError(50041, "affected_brand_scan_failed", "failed to parse monitoring brands for projection", scanErr)
		}
		brandIDs = append(brandIDs, brandID)
	}
	if err := rows.Err(); err != nil {
		return nil, monitoringInternalError(50041, "affected_brand_scan_failed", "failed to iterate monitoring brands for projection", err)
	}

	return brandIDs, nil
}

func deriveMonitoringBrandSampleStatus(actualSampleCount, plannedSampleCount int64, coverageRate *float64) string {
	if actualSampleCount <= 0 {
		return "unsampled"
	}
	if plannedSampleCount <= 0 {
		return "sampled"
	}
	if coverageRate != nil && *coverageRate >= 0.7 {
		return "sampled"
	}
	return "partial"
}

func deriveMonitoringConfidenceLevel(actualSampleCount, plannedSampleCount int64, coverageRate *float64) string {
	if actualSampleCount <= 0 {
		return "low"
	}
	if plannedSampleCount <= 0 {
		return "high"
	}
	if coverageRate == nil {
		return "low"
	}
	switch {
	case *coverageRate >= 0.7:
		return "high"
	case *coverageRate >= 0.4:
		return "medium"
	default:
		return "low"
	}
}

func deriveMonitoringPlatformSampleStatus(actualSampleCount, plannedSampleCount int64, accessStatus string) string {
	if actualSampleCount > 0 {
		return "sampled"
	}
	switch accessStatus {
	case "not_logged_in":
		return "not_logged_in"
	case "unavailable":
		return "unavailable"
	}
	if plannedSampleCount > 0 {
		return "pending"
	}
	if accessStatus == "accessible" {
		return "pending"
	}
	return "pending"
}

func nullableFloat64(value *float64) interface{} {
	if value == nil {
		return nil
	}
	return *value
}

func nullableTime(value sql.NullTime) interface{} {
	if !value.Valid {
		return nil
	}
	return value.Time
}