package app import ( "context" "crypto/md5" crand "crypto/rand" "database/sql" "encoding/hex" "encoding/json" "fmt" "math/big" "sort" "strconv" "strings" "sync" "time" "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" goredis "github.com/redis/go-redis/v9" "go.uber.org/zap" "github.com/geo-platform/tenant-api/internal/shared/auth" "github.com/geo-platform/tenant-api/internal/shared/config" "github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq" "github.com/geo-platform/tenant-api/internal/shared/response" "github.com/geo-platform/tenant-api/internal/shared/stream" "github.com/geo-platform/tenant-api/internal/tenant/repository" ) const ( defaultTrackingDays = 7 maxTrackingDays = 30 defaultMonitoringListPageSize = 10 maxMonitoringListPageSize = 100 monitoringCollectorType = "desktop" monitoringOnlineDuration = 20 * time.Minute monitoringCollectNowQuestionLimit = 5 monitoringPlatformAuthorizationAuthorized = "authorized" monitoringPlatformAuthorizationNoDesktopClient = "no_desktop_client" monitoringPlatformAuthorizationNoAuthorizedPlatforms = "no_authorized_platforms" ) type MonitoringService struct { businessPool *pgxpool.Pool monitoringPool *pgxpool.Pool rabbitMQ *rabbitmq.Client redis *goredis.Client logger *zap.Logger configMu sync.RWMutex desktopTasksRolloutPercentage int brandLibraryConfig config.BrandLibraryConfig } func NewMonitoringService( businessPool, monitoringPool *pgxpool.Pool, rabbitMQClient *rabbitmq.Client, dispatchCfg config.MonitoringDispatchConfig, brandLibraryCfg config.BrandLibraryConfig, logger *zap.Logger, ) *MonitoringService { return &MonitoringService{ businessPool: businessPool, monitoringPool: monitoringPool, rabbitMQ: rabbitMQClient, logger: logger, desktopTasksRolloutPercentage: dispatchCfg.DesktopTasksRolloutPercentage, brandLibraryConfig: brandLibraryCfg, } } func (s *MonitoringService) WithRedis(redis *goredis.Client) *MonitoringService { if s != nil { s.redis = redis } return s } func (s *MonitoringService) UpdateConfig(dispatchCfg config.MonitoringDispatchConfig, brandLibraryCfg config.BrandLibraryConfig) { if s == nil { return } s.configMu.Lock() s.desktopTasksRolloutPercentage = dispatchCfg.DesktopTasksRolloutPercentage s.brandLibraryConfig = brandLibraryCfg s.configMu.Unlock() } func (s *MonitoringService) currentConfig() (int, config.BrandLibraryConfig) { if s == nil { return 0, config.BrandLibraryConfig{} } s.configMu.RLock() defer s.configMu.RUnlock() return s.desktopTasksRolloutPercentage, s.brandLibraryConfig } type MonitoringDashboardCompositeResponse struct { Overview MonitoringOverview `json:"overview"` Runtime MonitoringDashboardRuntime `json:"runtime"` CitationWindow MonitoringDashboardWindow `json:"citation_window"` PlatformBreakdown []MonitoringPlatformDaily `json:"platform_breakdown"` HotQuestions []MonitoringHotQuestion `json:"hot_questions"` HotQuestionsPage MonitoringPage `json:"hot_questions_page"` CitationRanking []MonitoringCitationRanking `json:"citation_ranking"` CitedArticles []MonitoringCitedArticle `json:"cited_articles"` CitedArticlesPage MonitoringPage `json:"cited_articles_page"` } type MonitoringCitationSummaryResponse struct { CitationWindow MonitoringDashboardWindow `json:"citation_window"` CitationRanking []MonitoringCitationRanking `json:"citation_ranking"` CitedArticles []MonitoringCitedArticle `json:"cited_articles"` CitedArticlesPage MonitoringPage `json:"cited_articles_page"` } type MonitoringPage struct { Page int `json:"page"` PageSize int `json:"page_size"` Total int64 `json:"total"` } type MonitoringPageRequest struct { Page int PageSize int Query string } type MonitoringDashboardWindow struct { Days int `json:"days"` From string `json:"from"` To string `json:"to"` } type MonitoringDashboardRuntime struct { CurrentUserClientOnline bool `json:"current_user_client_online"` PlatformAuthorizationStatus string `json:"platform_authorization_status"` AuthorizedMonitoringPlatformIDs []string `json:"authorized_monitoring_platform_ids"` AuthorizedMonitoringPlatformCount int `json:"authorized_monitoring_platform_count"` OnlineAuthorizedMonitoringPlatformIDs []string `json:"online_authorized_monitoring_platform_ids"` } type MonitoringOverview struct { CollectionMode string `json:"collection_mode"` MentionRate *float64 `json:"mention_rate"` Top1MentionRate *float64 `json:"top1_mention_rate"` FirstRecommendRate *float64 `json:"first_recommend_rate"` PositiveMentionRate *float64 `json:"positive_mention_rate"` CitationRate *float64 `json:"citation_rate"` DesiredSampleCount int64 `json:"desired_sample_count"` PlannedSampleCount int64 `json:"planned_sample_count"` ActualSampleCount int64 `json:"actual_sample_count"` CoverageRate *float64 `json:"coverage_rate"` ConfidenceLevel string `json:"confidence_level"` LastSampledAt *string `json:"last_sampled_at"` LastTriggerSource *string `json:"last_trigger_source"` PrevSnapshotMentionRate *float64 `json:"prev_snapshot_mention_rate"` PrevSnapshotTop1MentionRate *float64 `json:"prev_snapshot_top1_mention_rate"` PrevSnapshotFirstRecommendRate *float64 `json:"prev_snapshot_first_recommend_rate"` PrevSnapshotPositiveMentionRate *float64 `json:"prev_snapshot_positive_mention_rate"` } type MonitoringPlatformDaily struct { AIPlatformID string `json:"ai_platform_id"` PlatformName string `json:"platform_name"` MentionRate *float64 `json:"mention_rate"` Top1MentionRate *float64 `json:"top1_mention_rate"` PlatformSampleStatus string `json:"platform_sample_status"` ActualSampleCount int64 `json:"actual_sample_count"` LastSampledAt *string `json:"last_sampled_at"` } type MonitoringHotQuestion struct { QuestionID int64 `json:"question_id"` QuestionHash string `json:"question_hash"` QuestionText string `json:"question_text"` MentionRate *float64 `json:"mention_rate"` LastSampledAt *string `json:"last_sampled_at"` } type MonitoringCitationRanking struct { AIPlatformID string `json:"ai_platform_id"` PlatformName string `json:"platform_name"` SampleStatus string `json:"sample_status"` CitedAnswerCount int64 `json:"cited_answer_count"` CitedArticleCount int64 `json:"cited_article_count"` CitationSourceCount int64 `json:"citation_source_count"` SaaSSourceCount int64 `json:"saas_source_count"` SaaSSourceRate *float64 `json:"saas_source_rate"` CitationRate *float64 `json:"citation_rate"` } type MonitoringCitedArticle struct { ArticleID *int64 `json:"article_id"` ArticleTitle string `json:"article_title"` PublishPlatform string `json:"publish_platform"` SourceType string `json:"source_type"` SourceID int64 `json:"source_id"` SourceURL string `json:"source_url"` CitationCount int64 `json:"citation_count"` CitationRate *float64 `json:"citation_rate"` SourceShare *float64 `json:"source_share"` } type MonitoringQuestionDetailResponse struct { QuestionID int64 `json:"question_id"` QuestionHash string `json:"question_hash"` QuestionText string `json:"question_text"` PlatformAuthorizationStatus string `json:"platform_authorization_status"` TimeWindow MonitoringQuestionDetailTimeWindow `json:"time_window"` Platforms []MonitoringQuestionDetailPlatform `json:"platforms"` CitationAnalysis []MonitoringQuestionCitationStats `json:"citation_analysis"` ContentCitations []MonitoringQuestionContentCitation `json:"content_citations"` } type MonitoringQuestionDetailTimeWindow struct { DateFrom string `json:"date_from"` DateTo string `json:"date_to"` } type MonitoringQuestionDetailPlatform struct { AIPlatformID string `json:"ai_platform_id"` PlatformName string `json:"platform_name"` SampleStatus string `json:"sample_status"` RunID *int64 `json:"run_id"` SampledAt *string `json:"sampled_at"` ProviderModel *string `json:"provider_model"` AnswerText *string `json:"answer_text"` BrandMentioned *bool `json:"brand_mentioned"` BrandMentionPosition *string `json:"brand_mention_position"` Citations []MonitoringQuestionDetailCitation `json:"citations"` InlineCitations []MonitoringSourceItem `json:"inline_citations,omitempty"` } type MonitoringQuestionDetailCitation struct { CitedURL string `json:"cited_url"` CitedTitle *string `json:"cited_title"` SiteName string `json:"site_name"` SiteKey string `json:"site_key"` FaviconURL string `json:"favicon_url"` ArticleID *int64 `json:"article_id"` SourceType *string `json:"source_type,omitempty"` SourceID *int64 `json:"source_id,omitempty"` SourceURL *string `json:"source_url,omitempty"` ArticleTitle *string `json:"article_title,omitempty"` ResolutionStatus string `json:"resolution_status"` ResolutionConfidence string `json:"resolution_confidence"` } type MonitoringQuestionCitationStats struct { AIPlatformID string `json:"ai_platform_id"` PlatformName string `json:"platform_name"` SiteName string `json:"site_name"` SiteKey string `json:"site_key"` SiteDomain string `json:"site_domain"` CitationCount int64 `json:"citation_count"` CitationRate *float64 `json:"citation_rate"` ContentCitationCount int64 `json:"content_citation_count"` } type MonitoringQuestionContentCitation struct { ArticleID *int64 `json:"article_id"` ArticleTitle string `json:"article_title"` PublishPlatform string `json:"publish_platform"` SourceType string `json:"source_type"` SourceID int64 `json:"source_id"` SourceURL string `json:"source_url"` AIPlatformID string `json:"ai_platform_id"` PlatformName string `json:"platform_name"` CitationCount int64 `json:"citation_count"` CitationRate *float64 `json:"citation_rate"` } type MonitoringCollectNowResponse struct { CollectionMode string `json:"collection_mode"` RefreshedTaskCount int64 `json:"refreshed_task_count"` CreatedTaskCount int64 `json:"created_task_count"` LeasedTaskCount int64 `json:"leased_task_count"` CompletedTaskCount int64 `json:"completed_task_count"` HasEffectiveSnapshot bool `json:"has_effective_snapshot"` RequestID string `json:"request_id,omitempty"` TargetClientID string `json:"target_client_id,omitempty"` AffectedTaskCount int64 `json:"affected_task_count,omitempty"` PromotedTaskCount int64 `json:"promoted_task_count,omitempty"` AbortedQueuedCount int64 `json:"aborted_queued_count,omitempty"` InterruptRequestedCount int64 `json:"interrupt_requested_count,omitempty"` InterruptGeneration int `json:"interrupt_generation,omitempty"` TTLExpiresAt string `json:"ttl_expires_at,omitempty"` Message string `json:"message"` } type MonitoringCollectNowOptions struct { PlatformIDs []string Preempt bool WaitForFirstDispatch bool TargetClientID *uuid.UUID } type monitoringQuotaConfig struct { CollectionMode string PrimaryClientID *uuid.UUID EnabledPlatforms []string } func monitoringPlatformAuthorizationStatus(quota monitoringQuotaConfig) string { if quota.PrimaryClientID == nil { return monitoringPlatformAuthorizationNoDesktopClient } if len(quota.EnabledPlatforms) == 0 { return monitoringPlatformAuthorizationNoAuthorizedPlatforms } return monitoringPlatformAuthorizationAuthorized } type monitoringBrandIdentity struct { ID int64 Name string } type monitoringAccessState struct { AccessStatus string } type monitoringLatestPlatformRun struct { PlatformID string RunID int64 CompletedAt sql.NullTime ProviderModel sql.NullString AnswerText sql.NullString BrandMentioned sql.NullBool BrandMentionPosition sql.NullString InlineCitations []MonitoringSourceItem } type monitoringDerivedRateStats struct { Total int64 MentionedCount int64 Top1Count int64 FirstRecommend int64 PositiveMentions int64 } type monitoringDerivedMetrics struct { ByDate map[string]monitoringDerivedRateStats ByDatePlatform map[string]map[string]monitoringDerivedRateStats ByDateQuestion map[string]map[int64]monitoringDerivedRateStats 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 } type monitoringConfiguredQuestion struct { ID int64 KeywordID int64 QuestionText string QuestionHash []byte } func monitoringQuestionTextSnapshot(question monitoringConfiguredQuestion) (string, error) { questionText := strings.TrimSpace(question.QuestionText) if questionText == "" { return "", response.ErrInternal(50041, "missing_question_text_snapshot", "configured monitoring question text is empty") } return questionText, nil } var defaultMonitoringPlatforms = []monitoringPlatformMetadata{ {ID: "yuanbao", Name: "混元 / 元宝"}, {ID: "kimi", Name: "Kimi"}, {ID: "wenxin", Name: "文心一言"}, {ID: "deepseek", Name: "DeepSeek"}, {ID: "doubao", Name: "豆包"}, {ID: "qwen", Name: "通义千问"}, } var monitoringPlatformNames = map[string]string{ "deepseek": "DeepSeek", "doubao": "豆包", "kimi": "Kimi", "qwen": "通义千问", "wenxin": "文心一言", "yuanbao": "混元 / 元宝", } func (s *MonitoringService) DashboardComposite( ctx context.Context, brandID int64, keywordID *int64, questionID *int64, days int, businessDate string, aiPlatformID *string, hotQuestionPage MonitoringPageRequest, ) (*MonitoringDashboardCompositeResponse, error) { actor := auth.MustActor(ctx) workspaceID := auth.CurrentWorkspaceID(ctx) brand, err := s.resolveBrand(ctx, actor.TenantID, brandID) if err != nil { return nil, err } if keywordID != nil { if err := s.ensureKeywordBelongsToBrand(ctx, actor.TenantID, brand.ID, *keywordID); err != nil { return nil, err } } quota, err := s.loadQuota(ctx, actor, workspaceID) if err != nil { return nil, err } runtime, err := s.loadDashboardRuntime(ctx, actor, workspaceID, quota) if err != nil { return nil, err } configuredQuestions, err := s.loadConfiguredQuestions(ctx, actor.TenantID, brand.ID, keywordID) if err != nil { return nil, err } configuredQuestions, err = filterConfiguredQuestionsByQuestionID(configuredQuestions, questionID) if err != nil { return nil, err } platforms := defaultMonitoringPlatformMetadata() selectedPlatformID := normalizedOptionalMonitoringPlatformPointer(aiPlatformID) filteredPlatforms := filterMonitoringPlatforms(platforms, selectedPlatformID) startDate, endDate, err := resolveDashboardDateWindow(days, businessDate) if err != nil { return nil, err } questionIDs := configuredQuestionIDs(configuredQuestions) derivedMetrics, err := s.loadDerivedParseMetrics(ctx, actor.TenantID, brand.ID, questionIDs, brand.Name, startDate, endDate, selectedPlatformID) if err != nil { return nil, err } overview, _, err := s.loadOverview(ctx, actor.TenantID, brand.ID, questionIDs, startDate, endDate, quota.CollectionMode, selectedPlatformID, derivedMetrics) if err != nil { return nil, err } accessStates, err := s.loadAccessStates(ctx, actor.TenantID, workspaceID, quota.PrimaryClientID, endDate) if err != nil { return nil, err } platformBreakdown, err := s.loadPlatformBreakdown(ctx, actor.TenantID, brand.ID, questionIDs, endDate, quota.CollectionMode, filteredPlatforms, accessStates, derivedMetrics) if err != nil { return nil, err } hotQuestions, hotQuestionsPage, err := s.loadHotQuestionsPage( ctx, actor.TenantID, brand.ID, keywordID, questionID, endDate, selectedPlatformID, derivedMetrics, hotQuestionPage, ) if err != nil { return nil, err } return &MonitoringDashboardCompositeResponse{ Overview: overview, Runtime: runtime, CitationWindow: MonitoringDashboardWindow{ Days: defaultTrackingDays, From: endDate.AddDate(0, 0, -(defaultTrackingDays - 1)).Format("2006-01-02"), To: endDate.Format("2006-01-02"), }, PlatformBreakdown: platformBreakdown, HotQuestions: hotQuestions, HotQuestionsPage: hotQuestionsPage, CitationRanking: []MonitoringCitationRanking{}, CitedArticles: []MonitoringCitedArticle{}, CitedArticlesPage: emptyMonitoringPage(normalizeMonitoringPageRequest(MonitoringPageRequest{}, defaultMonitoringListPageSize)), }, nil } func (s *MonitoringService) CitationSummary( ctx context.Context, days int, brandID int64, keywordID *int64, questionID *int64, businessDate string, aiPlatformID *string, citedArticlePage MonitoringPageRequest, ) (*MonitoringCitationSummaryResponse, error) { actor := auth.MustActor(ctx) workspaceID := auth.CurrentWorkspaceID(ctx) var questionIDs []int64 if brandID != 0 { brand, err := s.resolveBrand(ctx, actor.TenantID, brandID) if err != nil { return nil, err } if keywordID != nil { if err := s.ensureKeywordBelongsToBrand(ctx, actor.TenantID, brand.ID, *keywordID); err != nil { return nil, err } } configuredQuestions, err := s.loadConfiguredQuestions(ctx, actor.TenantID, brand.ID, keywordID) if err != nil { return nil, err } configuredQuestions, err = filterConfiguredQuestionsByQuestionID(configuredQuestions, questionID) if err != nil { return nil, err } questionIDs = configuredQuestionIDs(configuredQuestions) } quota, err := s.loadQuota(ctx, actor, workspaceID) if err != nil { return nil, err } citationWindowDays := normalizeCitationWindowDays(days) startDate, endDate, err := resolveDashboardDateWindow(citationWindowDays, businessDate) if err != nil { return nil, err } accessStates, err := s.loadAccessStates(ctx, actor.TenantID, workspaceID, quota.PrimaryClientID, endDate) if err != nil { return nil, err } selectedPlatformID := normalizedOptionalMonitoringPlatformPointer(aiPlatformID) citationRanking, err := s.loadCitationRanking(ctx, actor.TenantID, brandID, questionIDs, startDate, endDate, accessStates, selectedPlatformID) if err != nil { return nil, err } citedArticles, citedArticlesPage, err := s.loadCitedArticlesPage(ctx, actor.TenantID, brandID, questionIDs, startDate, endDate, selectedPlatformID, citedArticlePage) if err != nil { return nil, err } return &MonitoringCitationSummaryResponse{ CitationWindow: MonitoringDashboardWindow{ Days: citationWindowDays, From: startDate.Format("2006-01-02"), To: endDate.Format("2006-01-02"), }, CitationRanking: citationRanking, CitedArticles: citedArticles, CitedArticlesPage: citedArticlesPage, }, nil } func (s *MonitoringService) loadDashboardRuntime(ctx context.Context, actor auth.Actor, workspaceID int64, quota monitoringQuotaConfig) (MonitoringDashboardRuntime, error) { online, err := s.hasOnlineClientForUser(ctx, actor.TenantID, workspaceID, actor.UserID) if err != nil { return MonitoringDashboardRuntime{}, err } onlinePlatformIDs, err := s.loadOnlineMonitoringPlatformIDsForUser(ctx, actor.TenantID, workspaceID, actor.UserID, quota.EnabledPlatforms) if err != nil { return MonitoringDashboardRuntime{}, err } return MonitoringDashboardRuntime{ CurrentUserClientOnline: online, PlatformAuthorizationStatus: monitoringPlatformAuthorizationStatus(quota), AuthorizedMonitoringPlatformIDs: append([]string(nil), quota.EnabledPlatforms...), AuthorizedMonitoringPlatformCount: len(quota.EnabledPlatforms), OnlineAuthorizedMonitoringPlatformIDs: onlinePlatformIDs, }, nil } func (s *MonitoringService) QuestionDetail(ctx context.Context, brandID, questionID int64, dateFrom, dateTo string, questionHash string, aiPlatformID *string) (*MonitoringQuestionDetailResponse, error) { actor := auth.MustActor(ctx) workspaceID := auth.CurrentWorkspaceID(ctx) brand, err := s.resolveBrand(ctx, actor.TenantID, brandID) if err != nil { return nil, err } fromDate, toDate, err := parseDetailDateRange(dateFrom, dateTo) if err != nil { return nil, err } quota, err := s.loadQuota(ctx, actor, workspaceID) if err != nil { return nil, err } selectedPlatformID := normalizedOptionalMonitoringPlatformPointer(aiPlatformID) platforms := filterMonitoringPlatforms(defaultMonitoringPlatformMetadata(), selectedPlatformID) questionText, err := s.loadQuestionText(ctx, actor.TenantID, brand.ID, questionID) if err != nil { return nil, err } hashBytes := []byte(nil) if strings.TrimSpace(questionHash) != "" { hashBytes, err = decodeQuestionHash(questionHash) if err != nil { return nil, response.ErrBadRequest(40031, "invalid_question_hash", "question_hash must be a v1 hex digest") } } else { hashBytes, err = s.loadLatestQuestionHash(ctx, actor.TenantID, brand.ID, questionID, fromDate, toDate) if err != nil { return nil, err } } accessStates, err := s.loadAccessStates(ctx, actor.TenantID, workspaceID, quota.PrimaryClientID, toDate) if err != nil { return nil, err } latestRuns, err := s.loadLatestPlatformRuns(ctx, actor.TenantID, brand.ID, brand.Name, questionID, hashBytes, fromDate, toDate, selectedPlatformID) if err != nil { return nil, err } runIDs := make([]int64, 0, len(latestRuns)) for _, item := range latestRuns { runIDs = append(runIDs, item.RunID) } citationMap, err := s.loadCitationsForRuns(ctx, actor.TenantID, runIDs) if err != nil { return nil, err } platformPayload := make([]MonitoringQuestionDetailPlatform, 0, len(platforms)) for _, platform := range platforms { if selectedPlatformID != nil && platform.ID != *selectedPlatformID { continue } if run, ok := latestRuns[platform.ID]; ok { runID := run.RunID platformPayload = append(platformPayload, MonitoringQuestionDetailPlatform{ AIPlatformID: platform.ID, PlatformName: platform.Name, SampleStatus: "sampled", RunID: &runID, SampledAt: formatNullTime(run.CompletedAt), ProviderModel: stringPointer(run.ProviderModel), AnswerText: stringPointer(run.AnswerText), BrandMentioned: boolPointer(run.BrandMentioned), BrandMentionPosition: stringPointer(run.BrandMentionPosition), Citations: citationMap[run.RunID], InlineCitations: append([]MonitoringSourceItem(nil), run.InlineCitations...), }) continue } platformPayload = append(platformPayload, MonitoringQuestionDetailPlatform{ AIPlatformID: platform.ID, PlatformName: platform.Name, SampleStatus: derivePlatformSampleStatus("", accessStates[platform.ID]), Citations: []MonitoringQuestionDetailCitation{}, }) } citationAnalysis, err := s.loadQuestionCitationAnalysis(ctx, actor.TenantID, runIDs) if err != nil { return nil, err } contentCitations, err := s.loadQuestionContentCitations(ctx, actor.TenantID, runIDs) if err != nil { return nil, err } return &MonitoringQuestionDetailResponse{ QuestionID: questionID, QuestionHash: encodeQuestionHash(hashBytes), QuestionText: questionText, PlatformAuthorizationStatus: monitoringPlatformAuthorizationStatus(quota), TimeWindow: MonitoringQuestionDetailTimeWindow{ DateFrom: fromDate.Format("2006-01-02"), DateTo: toDate.Format("2006-01-02"), }, Platforms: platformPayload, CitationAnalysis: citationAnalysis, ContentCitations: contentCitations, }, nil } func (s *MonitoringService) CollectNow( ctx context.Context, brandID int64, keywordID *int64, questionID *int64, options MonitoringCollectNowOptions, ) (*MonitoringCollectNowResponse, error) { actor := auth.MustActor(ctx) workspaceID := auth.CurrentWorkspaceID(ctx) brand, err := s.resolveBrand(ctx, actor.TenantID, brandID) if err != nil { return nil, err } if keywordID != nil { if err := s.ensureKeywordBelongsToBrand(ctx, actor.TenantID, brand.ID, *keywordID); err != nil { return nil, err } } quota, err := s.loadQuota(ctx, actor, workspaceID) if err != nil { return nil, err } configuredQuestions, err := s.loadConfiguredQuestions(ctx, actor.TenantID, brand.ID, keywordID) if err != nil { return nil, err } configuredQuestions, err = filterConfiguredQuestionsByQuestionID(configuredQuestions, questionID) if err != nil { return nil, err } limitApplied := false if questionID == nil && len(configuredQuestions) > monitoringCollectNowQuestionLimit { configuredQuestions = configuredQuestions[:monitoringCollectNowQuestionLimit] limitApplied = true } if len(configuredQuestions) == 0 { message := "当前品牌下暂无关键词库（搜索词），未创建采集任务" if keywordID != nil { message = "当前关键词库（搜索词）下暂无搜索词，未创建采集任务" } if questionID != nil { message = "当前搜索词不可用，未创建采集任务" } return &MonitoringCollectNowResponse{ CollectionMode: quota.CollectionMode, RefreshedTaskCount: 0, CreatedTaskCount: 0, LeasedTaskCount: 0, CompletedTaskCount: 0, HasEffectiveSnapshot: true, Message: message, }, nil } platforms := resolveMonitoringPlatforms(quota.EnabledPlatforms) platforms, err = intersectMonitoringPlatforms(platforms, options.PlatformIDs) if err != nil { return nil, err } if len(platforms) == 0 { return &MonitoringCollectNowResponse{ CollectionMode: quota.CollectionMode, RefreshedTaskCount: 0, CreatedTaskCount: 0, LeasedTaskCount: 0, CompletedTaskCount: 0, HasEffectiveSnapshot: true, Message: "当前桌面端暂无已授权的 AI 平台，未创建采集任务", }, nil } questionIDs := configuredQuestionIDs(configuredQuestions) platformIDs := monitoringPlatformIDs(platforms) targetClientIDsByPlatform, err := s.loadRandomOnlineMonitoringClientIDsByPlatformForUser(ctx, actor.TenantID, workspaceID, actor.UserID, platformIDs) if err != nil { return nil, err } platforms = filterMonitoringPlatformsByIDSet(platforms, platformIDsFromClientTargetMap(targetClientIDsByPlatform)) if len(platforms) == 0 { return nil, response.ErrConflict(40941, "desktop_client_offline", "current user's monitoring desktop client is offline for selected platforms") } platformIDs = monitoringPlatformIDs(platforms) dispatchTargetClientIDs := collectNowDispatchTargetClientIDs(targetClientIDsByPlatform, nil) executionOwner := "legacy" if s.shouldUseDesktopTasksExecution(actor.TenantID) { executionOwner = "desktop_tasks" } clientID, err := s.resolveCollectNowClientID( ctx, actor.TenantID, workspaceID, actor.UserID, quota.PrimaryClientID, options.TargetClientID, platformIDs, ) if err != nil { return nil, err } if clientID == nil { return nil, response.ErrConflict(40941, "desktop_client_offline", "current user's monitoring desktop client is offline") } dispatchTargetClientIDs = collectNowDispatchTargetClientIDs(targetClientIDsByPlatform, clientID) now := time.Now().UTC() todayTime, _ := monitoringBusinessDayAndDateAt(now) requestID := uuid.New() ttlExpiresAt := now.Add(monitoringCollectNowTTL) scopeHash := hashMonitoringCollectNowScope(actor.UserID, brand.ID, keywordID, questionIDs, monitoringPlatformIDs(platforms)) tx, err := s.monitoringPool.Begin(ctx) if err != nil { return nil, response.ErrInternal(50041, "begin_failed", "failed to start monitoring collect-now transaction") } defer tx.Rollback(ctx) if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, collectNowScopeLockKey(actor.UserID, brand.ID, scopeHash)); err != nil { return nil, response.ErrInternal(50041, "lock_failed", "failed to acquire collect-now scope lock") } existingRequest, err := s.findActiveCollectNowRequest( ctx, tx, actor.TenantID, workspaceID, brand.ID, actor.UserID, scopeHash, now, ) if err != nil { return nil, err } if existingRequest != nil { base := buildCollectNowResponseFromState(quota.CollectionMode, existingRequest) executing, execErr := s.hasExecutingCollectNowTask(ctx, existingRequest.RequestID) if execErr != nil && s.logger != nil { s.logger.Warn("collect-now executing-state inspection failed", zap.String("request_id", existingRequest.RequestID.String()), zap.Error(execErr), ) } if executing { if base == nil { base = &MonitoringCollectNowResponse{CollectionMode: quota.CollectionMode} } base.Message = "任务正在执行中，请等待" _ = tx.Rollback(ctx) return base, nil } if _, err := tx.Exec(ctx, ` UPDATE monitoring_collect_requests SET superseded_by_request_id = $2, status = CASE WHEN status IN ('accepted', 'dispatching') THEN 'superseded' ELSE status END, updated_at = NOW() WHERE request_id = $1 AND superseded_by_request_id IS NULL `, existingRequest.RequestID, requestID); err != nil { return nil, response.ErrInternal(50041, "request_update_failed", "failed to supersede active collect-now request after client drift") } } if _, err := expireMonitoringLeasedTasks(ctx, tx, &actor.TenantID, now); err != nil { return nil, err } if err := s.syncMonitoringQuestionSnapshots(ctx, tx, actor.TenantID, brand.ID, configuredQuestions, keywordID == nil && questionID == nil); err != nil { return nil, err } interruptGeneration, err := s.nextCollectNowInterruptGeneration(ctx, tx, dispatchTargetClientIDs, executionOwner) if err != nil { return nil, err } refreshedCount, createdCount, leasedCount, completedCount, err := s.ensureCollectNowTasks( ctx, tx, actor.TenantID, brand.ID, configuredQuestions, platforms, todayTime, targetClientIDsByPlatform, *clientID, interruptGeneration, executionOwner, ) if err != nil { return nil, err } phase2TaskSpecs := make([]monitorDesktopTaskSpec, 0) if executionOwner == "desktop_tasks" { phase2TaskSpecs, err = s.loadMonitorDesktopTaskSpecs( ctx, tx, actor.TenantID, workspaceID, todayTime, questionIDs, platformIDs, *clientID, actor.UserID, ) if err != nil { return nil, err } } abortedQueuedCount := int64(0) if options.Preempt { for _, targetClientID := range dispatchTargetClientIDs { deferredCount, deferErr := s.deferQueuedNormalTasks( ctx, tx, actor.TenantID, todayTime, brand.ID, configuredQuestionIDs(configuredQuestions), monitoringPlatformIDs(platforms), targetClientID, requestID, ttlExpiresAt, ) if deferErr != nil { return nil, deferErr } abortedQueuedCount += deferredCount } } requestScopeJSON, err := json.Marshal(map[string]any{ "brand_id": brand.ID, "keyword_id": keywordID, "question_ids": questionIDs, "platform_ids": platformIDs, "target_client_ids": uuidStrings(dispatchTargetClientIDs), "preempt": options.Preempt, }) if err != nil { return nil, response.ErrInternal(50041, "payload_encode_failed", "failed to encode collect-now request scope") } affectedTaskCount := refreshedCount + createdCount + leasedCount message := "已提升为高优先级并通知桌面端优先执行" if limitApplied && affectedTaskCount > 0 { message = fmt.Sprintf("%s（本次最多执行 %d 个问题）", message, monitoringCollectNowQuestionLimit) } if _, err := tx.Exec(ctx, ` INSERT INTO monitoring_collect_requests ( request_id, tenant_id, workspace_id, brand_id, keyword_id, requested_by_user_id, target_client_id, scope_hash, request_scope, status, requested_task_count, dispatched_task_count, promoted_task_count, aborted_queued_count, interrupt_requested_count, interrupt_generation, message, ttl_expires_at ) VALUES ( $1, $2, $3, $4, $5, $6, $7, $8, $9, 'accepted', $10, 0, $11, $12, $13, $14, $15, $16 ) `, requestID, actor.TenantID, workspaceID, brand.ID, nullableInt64(keywordID), actor.UserID, *clientID, scopeHash, requestScopeJSON, affectedTaskCount, affectedTaskCount, abortedQueuedCount, int64(0), interruptGeneration, message, ttlExpiresAt); err != nil { return nil, response.ErrInternal(50041, "request_create_failed", "failed to persist monitoring collect-now request") } if err := s.enqueueCollectNowOutbox( ctx, tx, requestID, workspaceID, dispatchTargetClientIDs, affectedTaskCount, interruptGeneration, ); err != nil { return nil, err } if err := tx.Commit(ctx); err != nil { return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring collect-now") } phase2PublishedTasks := make([]*repository.DesktopTask, 0) phase2DispatchReady := executionOwner == "desktop_tasks" if len(phase2TaskSpecs) > 0 { phase2TaskCount := len(phase2TaskSpecs) publishedTasks, deferredTaskIDs, dispatchErr := s.dispatchMonitorDesktopTasks(ctx, phase2TaskSpecs) if dispatchErr != nil { phase2DispatchReady = false phase2TaskSpecs = nil if s.logger != nil { s.logger.Warn("phase2 monitor desktop dispatch failed; pending tasks will retry when a desktop client is online", zap.Error(dispatchErr), zap.Int("phase2_task_count", phase2TaskCount), ) } } else { if len(deferredTaskIDs) > 0 && s.logger != nil { s.logger.Info("phase2 monitor desktop tasks deferred until a desktop client is online", zap.Int("deferred_task_count", len(deferredTaskIDs)), ) } phase2PublishedTasks = publishedTasks } } phase2DeferredTasks := make([]*repository.DesktopTask, 0) if options.Preempt && phase2DispatchReady { excludedMonitorTaskIDs := make([]int64, 0, len(phase2TaskSpecs)) for _, spec := range phase2TaskSpecs { excludedMonitorTaskIDs = append(excludedMonitorTaskIDs, spec.MonitorTaskID) } phase2TargetClientIDs := desktopTaskClientIDs(phase2PublishedTasks) if len(phase2TargetClientIDs) == 0 { phase2TargetClientIDs = []uuid.UUID{*clientID} } for _, phase2TargetClientID := range phase2TargetClientIDs { deferredTasks, deferErr := s.deferQueuedPhase2MonitorTasks(ctx, phase2TargetClientID, excludedMonitorTaskIDs) if deferErr != nil { return nil, deferErr } phase2DeferredTasks = append(phase2DeferredTasks, deferredTasks...) } if len(phase2DeferredTasks) > 0 { abortedQueuedCount += int64(len(phase2DeferredTasks)) if _, updateErr := s.monitoringPool.Exec(ctx, ` UPDATE monitoring_collect_requests SET aborted_queued_count = aborted_queued_count + $2, updated_at = NOW() WHERE request_id = $1 `, requestID, len(phase2DeferredTasks)); updateErr != nil { return nil, response.ErrInternal(50041, "request_update_failed", "failed to persist phase2 deferred queue count") } } } for _, task := range phase2PublishedTasks { if publishErr := publishMonitorDesktopTaskAvailable( context.Background(), task.TargetClientID.String(), task, func(publishCtx context.Context, event DesktopTaskEvent) error { return publishDesktopTaskEvent(publishCtx, s.rabbitMQ, event) }, func(publishCtx context.Context, event stream.DesktopDispatchEvent) error { return publishDesktopDispatchEvent(publishCtx, s.rabbitMQ, s.logger, event) }, ); publishErr != nil && s.logger != nil { s.logger.Warn("phase2 monitor desktop task available publish failed") } } for _, task := range phase2DeferredTasks { if publishErr := publishMonitorDesktopTaskAvailable( context.Background(), task.TargetClientID.String(), task, func(publishCtx context.Context, event DesktopTaskEvent) error { return publishDesktopTaskEvent(publishCtx, s.rabbitMQ, event) }, func(publishCtx context.Context, event stream.DesktopDispatchEvent) error { return publishDesktopDispatchEvent(publishCtx, s.rabbitMQ, s.logger, event) }, ); publishErr != nil && s.logger != nil { s.logger.Warn("phase2 deferred monitor desktop task publish failed", zap.String("task_id", task.DesktopID.String()), zap.Error(publishErr), ) } } hasEffectiveSnapshot := leasedCount > 0 || completedCount > 0 responseData := &MonitoringCollectNowResponse{ CollectionMode: quota.CollectionMode, RefreshedTaskCount: refreshedCount, CreatedTaskCount: createdCount, LeasedTaskCount: leasedCount, CompletedTaskCount: completedCount, HasEffectiveSnapshot: hasEffectiveSnapshot, RequestID: requestID.String(), TargetClientID: clientID.String(), AffectedTaskCount: affectedTaskCount, PromotedTaskCount: affectedTaskCount, AbortedQueuedCount: abortedQueuedCount, InterruptRequestedCount: 0, InterruptGeneration: interruptGeneration, TTLExpiresAt: ttlExpiresAt.Format(time.RFC3339), Message: message, } return s.finalizeCollectNowResponse(ctx, requestID, quota.CollectionMode, options.WaitForFirstDispatch, responseData), nil } func (s *MonitoringService) resolveCollectNowClientID( ctx context.Context, tenantID, workspaceID, userID int64, defaultClientID *uuid.UUID, preferredClientID *uuid.UUID, platformIDs []string, ) (*uuid.UUID, error) { if preferredClientID != nil { if err := s.ensureClientBelongsToUser(ctx, tenantID, workspaceID, userID, *preferredClientID, platformIDs); err != nil { return nil, err } return preferredClientID, nil } clientID, err := s.findRandomOnlineClientForUserWithMonitoringPlatforms(ctx, tenantID, workspaceID, userID, platformIDs) if err != nil { return nil, err } if clientID != nil { return clientID, nil } if defaultClientID == nil { return nil, nil } if err := s.ensureClientBelongsToUser(ctx, tenantID, workspaceID, userID, *defaultClientID, platformIDs); err != nil { return nil, err } return defaultClientID, nil } func (s *MonitoringService) hasOnlineClientForUser(ctx context.Context, tenantID, workspaceID, userID int64) (bool, error) { clientID, err := s.findLatestOnlineClientForUser(ctx, tenantID, workspaceID, userID) if err != nil { return false, err } return clientID != nil, nil } func (s *MonitoringService) findRandomOnlineClientForUser( ctx context.Context, tenantID, workspaceID, userID int64, ) (*uuid.UUID, error) { candidates, err := s.loadOnlineClientIDsForUser(ctx, tenantID, workspaceID, userID) if err != nil { return nil, err } if len(candidates) == 0 { return nil, nil } return randomUUIDFromSlice(candidates), nil } func (s *MonitoringService) findRandomOnlineClientForUserWithMonitoringPlatforms( ctx context.Context, tenantID, workspaceID, userID int64, platformIDs []string, ) (*uuid.UUID, error) { targets, err := s.loadRandomOnlineMonitoringClientIDsByPlatformForUser(ctx, tenantID, workspaceID, userID, platformIDs) if err != nil { return nil, err } return randomUUIDFromSlice(collectNowDispatchTargetClientIDs(targets, nil)), nil } func (s *MonitoringService) loadOnlineMonitoringPlatformIDsForUser( ctx context.Context, tenantID, workspaceID, userID int64, platformIDs []string, ) ([]string, error) { onlineClientIDs, err := s.loadOnlineClientIDsForUser(ctx, tenantID, workspaceID, userID) if err != nil { return nil, err } if len(onlineClientIDs) == 0 { return nil, nil } return s.loadMonitoringPlatformIDsForClients(ctx, tenantID, workspaceID, userID, onlineClientIDs, platformIDs) } func (s *MonitoringService) loadRandomOnlineMonitoringClientIDsByPlatformForUser( ctx context.Context, tenantID, workspaceID, userID int64, platformIDs []string, ) (map[string]uuid.UUID, error) { platformIDs = reconcileEnabledMonitoringPlatforms(platformIDs) if len(platformIDs) == 0 { platformIDs = monitoringPlatformIDs(defaultMonitoringPlatforms) } onlineClientIDs, err := s.loadOnlineClientIDsForUser(ctx, tenantID, workspaceID, userID) if err != nil { return nil, err } if len(onlineClientIDs) == 0 { return nil, nil } candidatesByPlatform, err := s.loadOnlineMonitoringClientIDCandidatesByPlatformForUser(ctx, tenantID, workspaceID, userID, onlineClientIDs, platformIDs) if err != nil { return nil, err } result := make(map[string]uuid.UUID, len(candidatesByPlatform)) for platformID, candidates := range candidatesByPlatform { if selected := randomUUIDFromSlice(candidates); selected != nil { result[platformID] = *selected } } return result, nil } func (s *MonitoringService) ensureClientBelongsToUser( ctx context.Context, tenantID, workspaceID, userID int64, clientID uuid.UUID, platformIDs []string, ) error { var exists bool if err := s.businessPool.QueryRow(ctx, ` SELECT EXISTS ( SELECT 1 FROM desktop_clients dc WHERE dc.id = $1 AND dc.tenant_id = $2 AND dc.workspace_id = $3 AND dc.user_id = $4 AND dc.revoked_at IS NULL AND dc.last_seen_at IS NOT NULL AND dc.last_seen_at >= NOW() - $5::interval AND EXISTS ( SELECT 1 FROM platform_accounts pa WHERE pa.tenant_id = dc.tenant_id AND pa.workspace_id = dc.workspace_id AND pa.user_id = dc.user_id AND pa.client_id = dc.id AND pa.deleted_at IS NULL AND pa.platform_id = ANY($6::text[]) ) ) `, clientID, tenantID, workspaceID, userID, formatPgInterval(desktopClientPresenceTTL), reconcileEnabledMonitoringPlatforms(platformIDs)).Scan(&exists); err != nil { return response.ErrInternal(50041, "query_failed", "failed to validate target monitoring desktop client") } if !exists { return response.ErrConflict(40941, "desktop_client_offline", "current user's monitoring desktop client is offline") } return nil } func (s *MonitoringService) nextCollectNowInterruptGeneration( ctx context.Context, tx pgx.Tx, clientIDs []uuid.UUID, executionOwner string, ) (int, error) { var next int clientIDs = uniqueUUIDs(clientIDs) if len(clientIDs) == 0 { return 1, nil } if strings.TrimSpace(executionOwner) == "desktop_tasks" && s.businessPool != nil { if err := s.businessPool.QueryRow(ctx, ` SELECT COALESCE(MAX(interrupt_generation), 0) + 1 FROM desktop_tasks WHERE kind = 'monitor' AND target_client_id = ANY($1::uuid[]) `, clientIDs).Scan(&next); err != nil { return 0, response.ErrInternal(50041, "query_failed", "failed to allocate collect-now interrupt generation") } } else { if err := tx.QueryRow(ctx, ` SELECT COALESCE(MAX(interrupt_generation), 0) + 1 FROM monitoring_collect_tasks WHERE target_client_id = ANY($1::uuid[]) `, clientIDs).Scan(&next); err != nil { return 0, response.ErrInternal(50041, "query_failed", "failed to allocate collect-now interrupt generation") } } if next <= 0 { next = 1 } return next, nil } func monitoringPlatformIDs(items []monitoringPlatformMetadata) []string { if len(items) == 0 { return nil } result := make([]string, 0, len(items)) for _, item := range items { if strings.TrimSpace(item.ID) == "" { continue } result = append(result, item.ID) } return result } func filterMonitoringPlatformsByIDSet(items []monitoringPlatformMetadata, platformIDs []string) []monitoringPlatformMetadata { if len(items) == 0 || len(platformIDs) == 0 { return nil } enabled := make(map[string]struct{}, len(platformIDs)) for _, platformID := range platformIDs { platformID = normalizeMonitoringPlatformID(platformID) if platformID == "" { continue } enabled[platformID] = struct{}{} } if len(enabled) == 0 { return nil } result := make([]monitoringPlatformMetadata, 0, len(items)) for _, item := range items { platformID := normalizeMonitoringPlatformID(item.ID) if _, ok := enabled[platformID]; !ok { continue } item.ID = platformID result = append(result, item) } return result } func defaultMonitoringPlatformMetadata() []monitoringPlatformMetadata { return append([]monitoringPlatformMetadata(nil), defaultMonitoringPlatforms...) } func normalizeMonitoringPlatformID(platformID string) string { return strings.ToLower(strings.TrimSpace(platformID)) } func normalizedOptionalMonitoringPlatformID(value *string) string { if value == nil { return "" } return normalizeMonitoringPlatformID(*value) } func normalizedOptionalMonitoringPlatformPointer(value *string) *string { normalized := normalizedOptionalMonitoringPlatformID(value) if normalized == "" { return nil } return &normalized } func monitoringPlatformQueryIDs(value *string) []string { platformID := normalizedOptionalMonitoringPlatformID(value) if platformID == "" { return nil } return []string{platformID} } func reconcileEnabledMonitoringPlatforms(enabled []string) []string { if len(enabled) == 0 { return nil } requested := make(map[string]struct{}, len(enabled)) for _, item := range enabled { id := normalizeMonitoringPlatformID(item) if id == "" { continue } if _, supported := monitoringPlatformNames[id]; !supported { continue } requested[id] = struct{}{} } if len(requested) == 0 { return nil } result := make([]string, 0, len(requested)) for _, platform := range defaultMonitoringPlatforms { if _, ok := requested[platform.ID]; ok { result = append(result, platform.ID) } } return result } func monitoringPlatformSortIndex(platformID string) int { normalized := normalizeMonitoringPlatformID(platformID) for index, item := range defaultMonitoringPlatforms { if item.ID == normalized { return index } } return len(defaultMonitoringPlatforms) } func hashMonitoringCollectNowScope( userID int64, brandID int64, keywordID *int64, questionIDs []int64, platformIDs []string, ) string { payload := map[string]any{ "user_id": userID, "brand_id": brandID, "keyword_id": keywordID, "question_ids": questionIDs, "platform_ids": platformIDs, } body, _ := json.Marshal(payload) sum := md5.Sum(body) return hex.EncodeToString(sum[:]) } func (s *MonitoringService) findLatestOnlineClientForUser(ctx context.Context, tenantID, workspaceID, userID int64) (*uuid.UUID, error) { var clientID uuid.UUID err := s.businessPool.QueryRow(ctx, ` SELECT id FROM desktop_clients WHERE tenant_id = $1 AND workspace_id = $2 AND user_id = $3 AND revoked_at IS NULL AND last_seen_at IS NOT NULL AND last_seen_at >= NOW() - $4::interval ORDER BY last_seen_at DESC, created_at DESC, id DESC LIMIT 1 `, tenantID, workspaceID, userID, formatPgInterval(desktopClientPresenceTTL)).Scan(&clientID) if err != nil { if err == pgx.ErrNoRows { return nil, nil } return nil, response.ErrInternal(50041, "query_failed", "failed to inspect online monitoring desktop client") } return &clientID, nil } func formatPgInterval(duration time.Duration) string { return fmt.Sprintf("%d seconds", int64(duration/time.Second)) } func (s *MonitoringService) resolveBrand(ctx context.Context, tenantID, brandID int64) (*monitoringBrandIdentity, error) { targetBrandID := brandID if targetBrandID == 0 { if err := s.businessPool.QueryRow(ctx, ` SELECT id FROM brands WHERE tenant_id = $1 AND deleted_at IS NULL ORDER BY created_at ASC LIMIT 1 `, tenantID).Scan(&targetBrandID); err != nil { if err == pgx.ErrNoRows { return nil, response.ErrNotFound(40441, "brand_not_found", "no brand found for tenant") } return nil, response.ErrInternal(50041, "query_failed", "failed to resolve brand") } } brand := &monitoringBrandIdentity{} if err := s.businessPool.QueryRow(ctx, ` SELECT id, name FROM brands WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL `, targetBrandID, tenantID).Scan(&brand.ID, &brand.Name); err != nil { if err == pgx.ErrNoRows { return nil, response.ErrNotFound(40441, "brand_not_found", "brand not found") } return nil, response.ErrInternal(50041, "query_failed", "failed to resolve brand") } return brand, nil } func (s *MonitoringService) ensureKeywordBelongsToBrand(ctx context.Context, tenantID, brandID, keywordID int64) error { var exists bool if err := s.businessPool.QueryRow(ctx, ` SELECT EXISTS ( SELECT 1 FROM brand_keywords WHERE id = $1 AND tenant_id = $2 AND brand_id = $3 AND deleted_at IS NULL ) `, keywordID, tenantID, brandID).Scan(&exists); err != nil { return response.ErrInternal(50041, "query_failed", "failed to validate keyword") } if !exists { return response.ErrBadRequest(40041, "invalid_keyword", "keyword does not belong to the selected brand") } return nil } func (s *MonitoringService) loadConfiguredQuestions(ctx context.Context, tenantID, brandID int64, keywordID *int64) ([]monitoringConfiguredQuestion, error) { rows, err := s.businessPool.Query(ctx, ` SELECT q.id, q.keyword_id, q.question_text FROM brand_questions q JOIN brand_keywords k ON k.id = q.keyword_id AND k.tenant_id = q.tenant_id AND k.brand_id = q.brand_id AND k.deleted_at IS NULL AND k.status = 'active' WHERE q.tenant_id = $1 AND q.brand_id = $2 AND q.deleted_at IS NULL AND q.status = 'active' AND ($3::bigint IS NULL OR q.keyword_id = $3) ORDER BY q.keyword_id ASC, q.id ASC `, tenantID, brandID, nullableInt64(keywordID)) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load configured monitoring questions") } defer rows.Close() items := make([]monitoringConfiguredQuestion, 0) for rows.Next() { var item monitoringConfiguredQuestion if scanErr := rows.Scan(&item.ID, &item.KeywordID, &item.QuestionText); scanErr != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to parse configured monitoring questions") } item.QuestionText = strings.TrimSpace(item.QuestionText) if item.QuestionText == "" { return nil, response.ErrInternal(50041, "missing_question_text_snapshot", "configured monitoring question text is empty") } item.QuestionHash = seededQuestionHash(item.QuestionText) items = append(items, item) } if err := rows.Err(); err != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate configured monitoring questions") } return items, nil } func configuredQuestionIDs(questions []monitoringConfiguredQuestion) []int64 { ids := make([]int64, 0, len(questions)) for _, item := range questions { ids = append(ids, item.ID) } return ids } func filterConfiguredQuestionsByQuestionID(questions []monitoringConfiguredQuestion, questionID *int64) ([]monitoringConfiguredQuestion, error) { if questionID == nil { return questions, nil } if *questionID <= 0 { return nil, response.ErrBadRequest(40031, "invalid_question_id", "question_id must be a positive number") } for _, question := range questions { if question.ID == *questionID { return []monitoringConfiguredQuestion{question}, nil } } return nil, response.ErrBadRequest(40041, "invalid_question", "question does not belong to the selected brand or question set") } func normalizeMonitoringPageRequest(req MonitoringPageRequest, defaultPageSize int) MonitoringPageRequest { if req.Page <= 0 { req.Page = 1 } if req.PageSize <= 0 { req.PageSize = defaultPageSize } if req.PageSize > maxMonitoringListPageSize { req.PageSize = maxMonitoringListPageSize } req.Query = strings.TrimSpace(req.Query) return req } func emptyMonitoringPage(req MonitoringPageRequest) MonitoringPage { req = normalizeMonitoringPageRequest(req, defaultMonitoringListPageSize) return MonitoringPage{ Page: req.Page, PageSize: req.PageSize, Total: 0, } } func monitoringPageFromRequest(req MonitoringPageRequest, total int64) MonitoringPage { req = normalizeMonitoringPageRequest(req, defaultMonitoringListPageSize) return MonitoringPage{ Page: req.Page, PageSize: req.PageSize, Total: total, } } func monitoringPageOffset(req MonitoringPageRequest) int { req = normalizeMonitoringPageRequest(req, defaultMonitoringListPageSize) return (req.Page - 1) * req.PageSize } func (s *MonitoringService) syncMonitoringQuestionSnapshots(ctx context.Context, tx pgx.Tx, tenantID, brandID int64, questions []monitoringConfiguredQuestion, pruneMissing bool) error { if pruneMissing { if len(questions) == 0 { if _, err := tx.Exec(ctx, ` UPDATE monitoring_question_config_snapshots SET deleted_at = COALESCE(deleted_at, NOW()), updated_at = NOW() WHERE tenant_id = $1 AND brand_id = $2 AND deleted_at IS NULL `, tenantID, brandID); err != nil { return response.ErrInternal(50041, "snapshot_prune_failed", "failed to prune monitoring question snapshots") } } else if _, err := tx.Exec(ctx, ` UPDATE monitoring_question_config_snapshots SET deleted_at = COALESCE(deleted_at, NOW()), updated_at = NOW() WHERE tenant_id = $1 AND brand_id = $2 AND deleted_at IS NULL AND NOT (question_id = ANY($3)) `, tenantID, brandID, configuredQuestionIDs(questions)); err != nil { return response.ErrInternal(50041, "snapshot_prune_failed", "failed to prune monitoring question snapshots") } } for _, question := range questions { questionText, textErr := monitoringQuestionTextSnapshot(question) if textErr != nil { return textErr } if _, err := tx.Exec(ctx, ` UPDATE monitoring_question_config_snapshots SET superseded_at = COALESCE(superseded_at, NOW()), deleted_at = NULL, updated_at = NOW() WHERE tenant_id = $1 AND brand_id = $2 AND question_id = $3 AND question_hash <> $4 AND superseded_at IS NULL `, tenantID, brandID, question.ID, question.QuestionHash); err != nil { return response.ErrInternal(50041, "snapshot_update_failed", "failed to supersede monitoring question snapshot") } if _, err := tx.Exec(ctx, ` INSERT INTO monitoring_question_config_snapshots ( tenant_id, brand_id, question_id, keyword_id, question_hash, question_text_snapshot, monitor_enabled, projected_at ) VALUES ($1, $2, $3, $4, $5, $6, TRUE, NOW()) ON CONFLICT (tenant_id, question_id, question_hash) DO UPDATE SET brand_id = EXCLUDED.brand_id, keyword_id = EXCLUDED.keyword_id, question_text_snapshot = EXCLUDED.question_text_snapshot, monitor_enabled = TRUE, superseded_at = NULL, deleted_at = NULL, projected_at = NOW(), updated_at = NOW() `, tenantID, brandID, question.ID, question.KeywordID, question.QuestionHash, questionText); err != nil { return response.ErrInternal(50041, "snapshot_upsert_failed", "failed to persist monitoring question snapshot") } } return nil } func (s *MonitoringService) ensureCollectNowTasks( ctx context.Context, tx pgx.Tx, tenantID, brandID int64, questions []monitoringConfiguredQuestion, platforms []monitoringPlatformMetadata, businessDate time.Time, targetClientIDsByPlatform map[string]uuid.UUID, targetClientID uuid.UUID, interruptGeneration int, executionOwner string, ) (int64, int64, int64, int64, error) { const runMode = "plugin_standard" if strings.TrimSpace(executionOwner) == "" { executionOwner = "legacy" } dateText := businessDate.Format("2006-01-02") var refreshedCount int64 var createdCount int64 for _, question := range questions { questionText, textErr := monitoringQuestionTextSnapshot(question) if textErr != nil { return 0, 0, 0, 0, textErr } for _, platform := range platforms { platformID := normalizeMonitoringPlatformID(platform.ID) if platformID == "" { continue } platformTargetClientID, ok := targetClientIDsByPlatform[platformID] if !ok || platformTargetClientID == uuid.Nil { platformTargetClientID = targetClientID } tag, err := tx.Exec(ctx, ` UPDATE monitoring_collect_tasks SET question_hash = $7, question_text_snapshot = $12, status = 'pending', planned_at = NOW(), trigger_source = 'manual', dispatch_priority = 5000, dispatch_lane = 'high', target_client_id = $9, dispatch_after = NOW(), interrupt_generation = $10, superseded_by_request_id = NULL, last_dispatched_at = NULL, execution_owner = $11, ingest_shard_key = COALESCE( NULLIF(ingest_shard_key, ''), tenant_id::text || ':' || brand_id::text || ':' || business_date::text ), lease_token_hash = NULL, leased_to_executor = NULL, leased_at = NULL, lease_expires_at = NULL, callback_received_at = NULL, completed_at = NULL, skip_reason = NULL, request_payload_json = NULL, error_message = NULL, updated_at = NOW() WHERE tenant_id = $1 AND brand_id = $2 AND question_id = $3 AND ai_platform_id = $4 AND collector_type = $5 AND run_mode = $6 AND business_date = $8::date AND status IN ('pending', 'expired', 'failed', 'completed', 'skipped') `, tenantID, brandID, question.ID, platformID, monitoringCollectorType, runMode, question.QuestionHash, dateText, platformTargetClientID, interruptGeneration, executionOwner, questionText) if err != nil { return 0, 0, 0, 0, response.ErrInternal(50041, "update_failed", "failed to refresh monitoring tasks") } refreshedCount += tag.RowsAffected() tag, err = tx.Exec(ctx, ` UPDATE monitoring_collect_tasks SET question_hash = $7, question_text_snapshot = $11, trigger_source = 'manual', dispatch_priority = 5000, dispatch_lane = 'high', target_client_id = $9, dispatch_after = NOW(), interrupt_generation = $10, superseded_by_request_id = NULL, last_dispatched_at = NULL, execution_owner = COALESCE(execution_owner, 'legacy'), ingest_shard_key = COALESCE( NULLIF(ingest_shard_key, ''), tenant_id::text || ':' || brand_id::text || ':' || business_date::text ), updated_at = NOW() WHERE tenant_id = $1 AND brand_id = $2 AND question_id = $3 AND ai_platform_id = $4 AND collector_type = $5 AND run_mode = $6 AND business_date = $8::date AND status = 'leased' AND COALESCE(execution_owner, 'legacy') = 'legacy' `, tenantID, brandID, question.ID, platformID, monitoringCollectorType, runMode, question.QuestionHash, dateText, platformTargetClientID, interruptGeneration, questionText) if err != nil { return 0, 0, 0, 0, response.ErrInternal(50041, "update_failed", "failed to promote leased monitoring tasks") } tag, err = tx.Exec(ctx, ` INSERT INTO monitoring_collect_tasks ( tenant_id, brand_id, question_id, question_hash, question_text_snapshot, ai_platform_id, collector_type, trigger_source, run_mode, business_date, planned_at, status, dispatch_priority, dispatch_lane, target_client_id, dispatch_after, interrupt_generation, ingest_shard_key, execution_owner ) VALUES ( $1, $2, $3, $4, $13, $5, $6, 'manual', $7, $8::date, NOW(), 'pending', 5000, 'high', $9, NOW(), $10, $11, $12 ) ON CONFLICT ( tenant_id, brand_id, question_id, ai_platform_id, collector_type, run_mode, business_date ) DO NOTHING `, tenantID, brandID, question.ID, question.QuestionHash, platformID, monitoringCollectorType, runMode, dateText, platformTargetClientID, interruptGeneration, fmt.Sprintf("%d:%d:%s", tenantID, brandID, dateText), executionOwner, questionText) if err != nil { return 0, 0, 0, 0, response.ErrInternal(50041, "insert_failed", "failed to create monitoring tasks") } createdCount += tag.RowsAffected() } } var leasedCount int64 var completedCount int64 if err := tx.QueryRow(ctx, ` SELECT COUNT(*) FILTER (WHERE status = 'leased') AS leased_count, COUNT(*) FILTER (WHERE status = 'completed') AS completed_count FROM monitoring_collect_tasks WHERE tenant_id = $1 AND brand_id = $2 AND collector_type = $3 AND business_date = $4::date AND question_id = ANY($5) `, tenantID, brandID, monitoringCollectorType, dateText, configuredQuestionIDs(questions)).Scan(&leasedCount, &completedCount); err != nil { return 0, 0, 0, 0, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring tasks") } return refreshedCount, createdCount, leasedCount, completedCount, nil } func (s *MonitoringService) deferQueuedNormalTasks( ctx context.Context, tx pgx.Tx, tenantID int64, businessDate time.Time, brandID int64, questionIDs []int64, platformIDs []string, targetClientID uuid.UUID, requestID uuid.UUID, ttlExpiresAt time.Time, ) (int64, error) { if tx == nil || len(questionIDs) == 0 || len(platformIDs) == 0 { return 0, nil } tag, err := tx.Exec(ctx, ` UPDATE monitoring_collect_tasks SET dispatch_after = GREATEST(COALESCE(dispatch_after, NOW()), $6), superseded_by_request_id = $5, updated_at = NOW() WHERE tenant_id = $1 AND collector_type = $2 AND business_date = $3::date AND COALESCE(execution_owner, 'legacy') = 'legacy' AND status = 'pending' AND target_client_id = $4 AND dispatch_lane IN ('normal', 'normal_boosted', 'retry') AND NOT ( brand_id = $7 AND question_id = ANY($8::bigint[]) AND ai_platform_id = ANY($9::text[]) ) `, tenantID, monitoringCollectorType, businessDate.Format("2006-01-02"), targetClientID, requestID, ttlExpiresAt, brandID, questionIDs, platformIDs) if err != nil { return 0, response.ErrInternal(50041, "update_failed", "failed to defer queued normal monitoring tasks") } return tag.RowsAffected(), nil } func (s *MonitoringService) loadQuota(ctx context.Context, actor auth.Actor, workspaceID int64) (monitoringQuotaConfig, error) { cfg := monitoringQuotaConfig{ CollectionMode: monitoringCollectorType, } clientID, err := s.findLatestOnlineClientWithMonitoringAccountForUser(ctx, actor.TenantID, workspaceID, actor.UserID) if err != nil { return cfg, err } if clientID == nil { clientID, err = s.findLatestRegisteredClientWithMonitoringAccountForUser(ctx, actor.TenantID, workspaceID, actor.UserID) } if err != nil { return cfg, err } if clientID == nil { clientID, err = s.findLatestRegisteredClientForUser(ctx, actor.TenantID, workspaceID, actor.UserID) if err != nil { return cfg, err } } cfg.PrimaryClientID = clientID if clientID == nil { return cfg, nil } platforms, err := s.loadMonitoringPlatformIDsForUser(ctx, actor.TenantID, workspaceID, actor.UserID) if err != nil { return cfg, err } cfg.EnabledPlatforms = platforms return cfg, nil } // findLatestOnlineClientWithMonitoringAccountForUser returns an online client for // dashboard runtime defaults. Collect-now may still randomize among all online // clients that can execute the requested platforms. func (s *MonitoringService) findLatestOnlineClientWithMonitoringAccountForUser(ctx context.Context, tenantID, workspaceID, userID int64) (*uuid.UUID, error) { clientIDs, err := s.loadOnlineClientIDsForUser(ctx, tenantID, workspaceID, userID) if err != nil { return nil, err } if len(clientIDs) == 0 { return nil, nil } return &clientIDs[0], nil } // findLatestRegisteredClientWithMonitoringAccountForUser returns the newest // non-revoked desktop client that has at least one bound monitoring platform // account. Platform account health is not authorization. func (s *MonitoringService) findLatestRegisteredClientWithMonitoringAccountForUser(ctx context.Context, tenantID, workspaceID, userID int64) (*uuid.UUID, error) { var clientID uuid.UUID err := s.businessPool.QueryRow(ctx, ` SELECT dc.id FROM desktop_clients dc WHERE dc.tenant_id = $1 AND dc.workspace_id = $2 AND dc.user_id = $3 AND dc.revoked_at IS NULL AND EXISTS ( SELECT 1 FROM platform_accounts pa WHERE pa.tenant_id = dc.tenant_id 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[]) ) ORDER BY dc.created_at DESC, dc.id DESC LIMIT 1 `, tenantID, workspaceID, userID, monitoringPlatformIDs(defaultMonitoringPlatforms)).Scan(&clientID) if err != nil { if err == pgx.ErrNoRows { return nil, nil } return nil, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring desktop client authorization") } return &clientID, nil } // findLatestRegisteredClientForUser returns the newest non-revoked desktop client // registered by creation order. It intentionally ignores last_seen_at and health; // use findLatestOnlineClientForUser when online presence is required. func (s *MonitoringService) findLatestRegisteredClientForUser(ctx context.Context, tenantID, workspaceID, userID int64) (*uuid.UUID, error) { var clientID uuid.UUID err := s.businessPool.QueryRow(ctx, ` SELECT id FROM desktop_clients WHERE tenant_id = $1 AND workspace_id = $2 AND user_id = $3 AND revoked_at IS NULL ORDER BY created_at DESC, id DESC LIMIT 1 `, tenantID, workspaceID, userID).Scan(&clientID) if err != nil { if err == pgx.ErrNoRows { return nil, nil } return nil, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring desktop client") } return &clientID, nil } func (s *MonitoringService) loadMonitoringPlatformIDsForUser(ctx context.Context, tenantID, workspaceID, userID int64) ([]string, error) { rows, err := s.businessPool.Query(ctx, ` SELECT DISTINCT pa.platform_id FROM platform_accounts pa JOIN desktop_clients dc ON dc.id = pa.client_id AND dc.tenant_id = pa.tenant_id AND dc.workspace_id = pa.workspace_id WHERE pa.tenant_id = $1 AND pa.workspace_id = $2 AND pa.user_id = $3 AND pa.deleted_at IS NULL AND pa.platform_id = ANY($4::text[]) AND dc.user_id = $3 AND dc.revoked_at IS NULL `, tenantID, workspaceID, userID, monitoringPlatformIDs(defaultMonitoringPlatforms)) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load monitoring desktop platform accounts") } defer rows.Close() platformIDs := make([]string, 0) for rows.Next() { var platformID string if scanErr := rows.Scan(&platformID); scanErr != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to parse monitoring desktop platform accounts") } platformIDs = append(platformIDs, platformID) } if err := rows.Err(); err != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate monitoring desktop platform accounts") } return reconcileEnabledMonitoringPlatforms(platformIDs), nil } func (s *MonitoringService) loadMonitoringClientPlatformIDs(ctx context.Context, tenantID, workspaceID int64, clientID uuid.UUID) ([]string, error) { rows, err := s.businessPool.Query(ctx, ` SELECT DISTINCT platform_id FROM platform_accounts WHERE tenant_id = $1 AND workspace_id = $2 AND client_id = $3 AND deleted_at IS NULL AND platform_id = ANY($4::text[]) `, tenantID, workspaceID, clientID, monitoringPlatformIDs(defaultMonitoringPlatforms)) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load monitoring desktop platform accounts") } defer rows.Close() platformIDs := make([]string, 0) for rows.Next() { var platformID string if scanErr := rows.Scan(&platformID); scanErr != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to parse monitoring desktop platform accounts") } platformIDs = append(platformIDs, platformID) } if err := rows.Err(); err != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate monitoring desktop platform accounts") } return reconcileEnabledMonitoringPlatforms(platformIDs), nil } func (s *MonitoringService) loadMonitoringPlatformIDsForClients( ctx context.Context, tenantID, workspaceID, userID int64, clientIDs []uuid.UUID, platformIDs []string, ) ([]string, error) { clientIDs = uniqueUUIDs(clientIDs) if len(clientIDs) == 0 { return nil, nil } platformIDs = reconcileEnabledMonitoringPlatforms(platformIDs) if len(platformIDs) == 0 { platformIDs = monitoringPlatformIDs(defaultMonitoringPlatforms) } rows, err := s.businessPool.Query(ctx, ` SELECT DISTINCT platform_id FROM platform_accounts WHERE tenant_id = $1 AND workspace_id = $2 AND user_id = $3 AND client_id = ANY($4::uuid[]) AND deleted_at IS NULL AND platform_id = ANY($5::text[]) `, tenantID, workspaceID, userID, clientIDs, platformIDs) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load online monitoring desktop platform accounts") } defer rows.Close() result := make([]string, 0) for rows.Next() { var platformID string if scanErr := rows.Scan(&platformID); scanErr != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to parse online monitoring desktop platform accounts") } result = append(result, platformID) } if err := rows.Err(); err != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate online monitoring desktop platform accounts") } return reconcileEnabledMonitoringPlatforms(result), nil } func (s *MonitoringService) loadOnlineMonitoringClientIDCandidatesByPlatformForUser( ctx context.Context, tenantID, workspaceID, userID int64, clientIDs []uuid.UUID, platformIDs []string, ) (map[string][]uuid.UUID, error) { clientIDs = uniqueUUIDs(clientIDs) if len(clientIDs) == 0 { return nil, nil } platformIDs = reconcileEnabledMonitoringPlatforms(platformIDs) if len(platformIDs) == 0 { platformIDs = monitoringPlatformIDs(defaultMonitoringPlatforms) } rows, err := s.businessPool.Query(ctx, ` SELECT DISTINCT platform_id, client_id FROM platform_accounts WHERE tenant_id = $1 AND workspace_id = $2 AND user_id = $3 AND client_id = ANY($4::uuid[]) AND deleted_at IS NULL AND platform_id = ANY($5::text[]) `, tenantID, workspaceID, userID, clientIDs, platformIDs) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load online monitoring desktop platform targets") } defer rows.Close() result := make(map[string][]uuid.UUID) for rows.Next() { var ( platformID string clientID uuid.UUID ) if scanErr := rows.Scan(&platformID, &clientID); scanErr != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to parse online monitoring desktop platform targets") } platformID = normalizeMonitoringPlatformID(platformID) if platformID == "" || clientID == uuid.Nil { continue } result[platformID] = append(result[platformID], clientID) } if err := rows.Err(); err != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate online monitoring desktop platform targets") } for platformID, candidates := range result { result[platformID] = uniqueUUIDs(candidates) } return result, nil } func (s *MonitoringService) loadOnlineClientIDsForUser( ctx context.Context, tenantID, workspaceID, userID int64, ) ([]uuid.UUID, error) { rows, err := s.businessPool.Query(ctx, ` SELECT dc.id FROM desktop_clients dc WHERE dc.tenant_id = $1 AND dc.workspace_id = $2 AND dc.user_id = $3 AND dc.revoked_at IS NULL AND dc.last_seen_at IS NOT NULL AND dc.last_seen_at >= NOW() - $4::interval ORDER BY dc.last_seen_at DESC, dc.created_at DESC, dc.id DESC `, tenantID, workspaceID, userID, formatPgInterval(desktopClientPresenceTTL)) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to inspect online monitoring desktop clients") } defer rows.Close() clientIDs := make([]uuid.UUID, 0) for rows.Next() { var clientID uuid.UUID if scanErr := rows.Scan(&clientID); scanErr != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to parse online monitoring desktop clients") } clientIDs = append(clientIDs, clientID) } if err := rows.Err(); err != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate online monitoring desktop clients") } return clientIDs, nil } func randomUUIDFromSlice(values []uuid.UUID) *uuid.UUID { if len(values) == 0 { return nil } selected := values[randomIndex(len(values))] return &selected } func randomIndex(length int) int { if length <= 1 { return 0 } if value, err := crand.Int(crand.Reader, big.NewInt(int64(length))); err == nil { return int(value.Int64()) } return int(time.Now().UnixNano() % int64(length)) } func platformIDsFromClientTargetMap(targets map[string]uuid.UUID) []string { if len(targets) == 0 { return nil } result := make([]string, 0, len(targets)) for _, platform := range defaultMonitoringPlatforms { if clientID, ok := targets[platform.ID]; ok && clientID != uuid.Nil { result = append(result, platform.ID) } } return result } func collectNowDispatchTargetClientIDs(targets map[string]uuid.UUID, fallback *uuid.UUID) []uuid.UUID { result := make([]uuid.UUID, 0, len(targets)+1) for _, platform := range defaultMonitoringPlatforms { if clientID, ok := targets[platform.ID]; ok && clientID != uuid.Nil { result = append(result, clientID) } } if fallback != nil && *fallback != uuid.Nil { result = append(result, *fallback) } return uniqueUUIDs(result) } func uuidStrings(values []uuid.UUID) []string { values = uniqueUUIDs(values) if len(values) == 0 { return nil } result := make([]string, 0, len(values)) for _, value := range values { if value == uuid.Nil { continue } result = append(result, value.String()) } return result } func desktopTaskClientIDs(tasks []*repository.DesktopTask) []uuid.UUID { if len(tasks) == 0 { return nil } result := make([]uuid.UUID, 0, len(tasks)) for _, task := range tasks { if task == nil || task.TargetClientID == uuid.Nil { continue } result = append(result, task.TargetClientID) } return uniqueUUIDs(result) } func newMonitoringDerivedMetrics() monitoringDerivedMetrics { return monitoringDerivedMetrics{ ByDate: make(map[string]monitoringDerivedRateStats), ByDatePlatform: make(map[string]map[string]monitoringDerivedRateStats), ByDateQuestion: make(map[string]map[int64]monitoringDerivedRateStats), ByQuestion: make(map[int64]monitoringDerivedRateStats), } } func (stats *monitoringDerivedRateStats) add(summary monitoringAnswerParseSummary) { stats.Total++ if summary.BrandMentioned { stats.MentionedCount++ } if summary.BrandMentionPosition == "top1" { stats.Top1Count++ } if summary.FirstRecommended { stats.FirstRecommend++ } if summary.BrandMentioned && summary.SentimentLabel == "positive" { stats.PositiveMentions++ } } func (stats monitoringDerivedRateStats) hasSamples() bool { return stats.Total > 0 } func (stats monitoringDerivedRateStats) mentionRate() *float64 { return divideAsPointer(stats.MentionedCount, stats.Total) } func (stats monitoringDerivedRateStats) top1MentionRate() *float64 { return divideAsPointer(stats.Top1Count, stats.Total) } func (stats monitoringDerivedRateStats) firstRecommendRate() *float64 { return divideAsPointer(stats.FirstRecommend, stats.Total) } 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, questionIDs []int64, brandName string, startDate, endDate time.Time, aiPlatformID *string, ) (monitoringDerivedMetrics, error) { metrics := newMonitoringDerivedMetrics() if questionIDs != nil && len(questionIDs) == 0 { return metrics, nil } platformQueryIDs := monitoringPlatformQueryIDs(aiPlatformID) 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 ( 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 `, 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 var firstRecommended sql.NullBool var sentimentLabel sql.NullString var matchedBrandTermsJSON []byte if scanErr := rows.Scan( &runID, &businessDate, &platformID, &questionID, &eventAt, &answerText, &brandMentioned, &brandMentionPosition, &firstRecommended, &sentimentLabel, &matchedBrandTermsJSON, ); scanErr != nil { return metrics, response.ErrInternal(50041, "scan_failed", "failed to parse monitoring derived metrics") } platformID = normalizeMonitoringPlatformID(platformID) if platformID == "" { continue } summary := resolveMonitoringAnswerParseSummary( answerText.String, brandName, monitoringStoredParseFields{ BrandMentioned: boolPointer(brandMentioned), BrandMentionPosition: stringPointer(brandMentionPosition), FirstRecommended: boolPointer(firstRecommended), SentimentLabel: stringPointer(sentimentLabel), MatchedBrandTerms: decodeMonitoringMatchedBrandTerms(matchedBrandTermsJSON), }, ) 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 buildMonitoringDerivedMetricsFromLatestSamples(samples), nil } func decodeMonitoringMatchedBrandTerms(raw []byte) []string { if len(raw) == 0 { return nil } var items []string if err := json.Unmarshal(raw, &items); err != nil { return nil } return normalizeMonitoringBrandTermList(items) } func (s *MonitoringService) loadOverview( ctx context.Context, tenantID, brandID int64, questionIDs []int64, startDate, endDate time.Time, collectionMode string, aiPlatformID *string, derived monitoringDerivedMetrics, ) (MonitoringOverview, time.Time, error) { if questionIDs != nil { return s.loadOverviewFromRuns(ctx, tenantID, brandID, questionIDs, startDate, endDate, collectionMode, aiPlatformID, derived) } if platformID := normalizedOptionalMonitoringPlatformID(aiPlatformID); platformID != "" { return s.loadOverviewForPlatform(ctx, tenantID, brandID, startDate, endDate, collectionMode, platformID, derived) } overview := MonitoringOverview{ CollectionMode: collectionMode, ConfidenceLevel: "low", } latestDate := endDate current, found, err := s.loadOverviewSnapshotForDate(ctx, tenantID, brandID, collectionMode, endDate) if err != nil { return overview, latestDate, err } if !found { return overview, latestDate, nil } applyOverviewSnapshot(&overview, current, derived) if current.ActualSampleCount <= 0 { return overview, latestDate, nil } latestDate = current.Date previous, foundPrevious, err := s.loadLatestSampledOverviewSnapshot(ctx, tenantID, brandID, collectionMode, startDate, current.Date.AddDate(0, 0, -1)) if err != nil { return overview, latestDate, err } if !foundPrevious { return overview, latestDate, nil } applyDerivedRatesToOverviewSnapshot(&previous, derived) overview.PrevSnapshotMentionRate = floatPointer(previous.MentionRate) overview.PrevSnapshotTop1MentionRate = floatPointer(previous.Top1MentionRate) overview.PrevSnapshotFirstRecommendRate = floatPointer(previous.FirstRecommendRate) overview.PrevSnapshotPositiveMentionRate = floatPointer(previous.PositiveMentionRate) return overview, latestDate, nil } func (s *MonitoringService) loadOverviewFromRuns( ctx context.Context, tenantID, brandID int64, questionIDs []int64, startDate, endDate time.Time, collectionMode string, aiPlatformID *string, derived monitoringDerivedMetrics, ) (MonitoringOverview, time.Time, error) { overview := MonitoringOverview{ CollectionMode: collectionMode, ConfidenceLevel: "low", } latestDate := endDate if len(questionIDs) == 0 { return overview, latestDate, nil } platformQueryIDs := monitoringPlatformQueryIDs(aiPlatformID) current, found, err := s.loadOverviewRunAggregateForDate(ctx, tenantID, brandID, questionIDs, endDate, platformQueryIDs) if err != nil { return overview, latestDate, err } if !found { return overview, latestDate, nil } applyOverviewSnapshot(&overview, current, derived) if current.ActualSampleCount <= 0 { return overview, latestDate, nil } latestDate = current.Date previous, foundPrevious, err := s.loadLatestOverviewRunAggregate(ctx, tenantID, brandID, questionIDs, startDate, current.Date.AddDate(0, 0, -1), platformQueryIDs) if err != nil { return overview, latestDate, err } if !foundPrevious { return overview, latestDate, nil } applyDerivedRatesToOverviewSnapshot(&previous, derived) overview.PrevSnapshotMentionRate = floatPointer(previous.MentionRate) overview.PrevSnapshotTop1MentionRate = floatPointer(previous.Top1MentionRate) overview.PrevSnapshotFirstRecommendRate = floatPointer(previous.FirstRecommendRate) overview.PrevSnapshotPositiveMentionRate = floatPointer(previous.PositiveMentionRate) return overview, latestDate, nil } func (s *MonitoringService) loadOverviewForPlatform( ctx context.Context, tenantID, brandID int64, startDate, endDate time.Time, collectionMode string, aiPlatformID string, derived monitoringDerivedMetrics, ) (MonitoringOverview, time.Time, error) { overview := MonitoringOverview{ CollectionMode: collectionMode, ConfidenceLevel: "low", } latestDate := endDate platformQueryIDs := monitoringPlatformQueryIDs(&aiPlatformID) current, found, err := s.loadPlatformOverviewSnapshotForDate(ctx, tenantID, brandID, platformQueryIDs, collectionMode, endDate) if err != nil { return overview, latestDate, err } if !found { return overview, latestDate, nil } applyOverviewSnapshot(&overview, current, derived) if current.ActualSampleCount <= 0 { return overview, latestDate, nil } latestDate = current.Date previous, foundPrevious, err := s.loadLatestSampledPlatformOverviewSnapshot( ctx, tenantID, brandID, platformQueryIDs, collectionMode, startDate, current.Date.AddDate(0, 0, -1), ) if err != nil { return overview, latestDate, err } if !foundPrevious { return overview, latestDate, nil } applyDerivedRatesToOverviewSnapshot(&previous, derived) overview.PrevSnapshotMentionRate = floatPointer(previous.MentionRate) overview.PrevSnapshotTop1MentionRate = floatPointer(previous.Top1MentionRate) overview.PrevSnapshotFirstRecommendRate = floatPointer(previous.FirstRecommendRate) overview.PrevSnapshotPositiveMentionRate = floatPointer(previous.PositiveMentionRate) return overview, latestDate, nil } type monitoringOverviewSnapshot struct { Date time.Time DesiredSampleCount int64 PlannedSampleCount int64 ActualSampleCount int64 CoverageRate sql.NullFloat64 ConfidenceLevel string TriggerSource sql.NullString SnapshotUpdatedAt sql.NullTime MentionRate sql.NullFloat64 Top1MentionRate sql.NullFloat64 FirstRecommendRate sql.NullFloat64 PositiveMentionRate sql.NullFloat64 CitationRate sql.NullFloat64 } func (s *MonitoringService) loadOverviewRunAggregateForDate( ctx context.Context, tenantID, brandID int64, questionIDs []int64, businessDate time.Time, platformIDs []string, ) (monitoringOverviewSnapshot, bool, error) { return s.loadOverviewRunAggregate(ctx, tenantID, brandID, questionIDs, businessDate, businessDate, platformIDs, false) } func (s *MonitoringService) loadLatestOverviewRunAggregate( ctx context.Context, tenantID, brandID int64, questionIDs []int64, startDate, endDate time.Time, platformIDs []string, ) (monitoringOverviewSnapshot, bool, error) { if endDate.Before(startDate) { return monitoringOverviewSnapshot{}, false, nil } return s.loadOverviewRunAggregate(ctx, tenantID, brandID, questionIDs, startDate, endDate, platformIDs, true) } func (s *MonitoringService) loadOverviewRunAggregate( ctx context.Context, tenantID, brandID int64, questionIDs []int64, startDate, endDate time.Time, platformIDs []string, requireSamples bool, ) (monitoringOverviewSnapshot, bool, error) { var item monitoringOverviewSnapshot if len(questionIDs) == 0 { return item, false, nil } rows, err := s.monitoringPool.Query(ctx, ` SELECT r.business_date, 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, COUNT(*) FILTER (WHERE p.first_recommended IS TRUE)::bigint AS first_recommended_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 ( 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 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)) if err != nil { return item, false, response.ErrInternal(50041, "query_failed", "failed to load monitoring overview") } defer rows.Close() for rows.Next() { var mentionedCount int64 var top1MentionedCount int64 var firstRecommendedCount int64 var positiveMentionedCount int64 if scanErr := rows.Scan( &item.Date, &item.ActualSampleCount, &mentionedCount, &top1MentionedCount, &firstRecommendedCount, &positiveMentionedCount, &item.SnapshotUpdatedAt, ); scanErr != nil { return monitoringOverviewSnapshot{}, false, response.ErrInternal(50041, "scan_failed", "failed to parse monitoring overview") } if requireSamples && item.ActualSampleCount <= 0 { continue } plannedSampleCount, planErr := s.loadOverviewRunPlannedCount(ctx, tenantID, brandID, questionIDs, item.Date, platformIDs) if planErr != nil { return monitoringOverviewSnapshot{}, false, planErr } item.DesiredSampleCount = plannedSampleCount item.PlannedSampleCount = plannedSampleCount item.CoverageRate = pointerFloat64ToNull(divideAsPointer(item.ActualSampleCount, plannedSampleCount)) item.ConfidenceLevel = deriveMonitoringConfidenceLevel(item.ActualSampleCount, plannedSampleCount, floatPointer(item.CoverageRate)) item.MentionRate = pointerFloat64ToNull(divideAsPointer(mentionedCount, item.ActualSampleCount)) item.Top1MentionRate = pointerFloat64ToNull(divideAsPointer(top1MentionedCount, item.ActualSampleCount)) item.FirstRecommendRate = pointerFloat64ToNull(divideAsPointer(firstRecommendedCount, item.ActualSampleCount)) item.PositiveMentionRate = pointerFloat64ToNull(divideAsPointer(positiveMentionedCount, item.ActualSampleCount)) return item, true, nil } if err := rows.Err(); err != nil { return monitoringOverviewSnapshot{}, false, response.ErrInternal(50041, "scan_failed", "failed to iterate monitoring overview") } return monitoringOverviewSnapshot{}, false, nil } func (s *MonitoringService) loadOverviewRunPlannedCount( ctx context.Context, tenantID, brandID int64, questionIDs []int64, businessDate time.Time, platformIDs []string, ) (int64, error) { var plannedSampleCount int64 if err := s.monitoringPool.QueryRow(ctx, ` SELECT COUNT(*)::bigint FROM monitoring_collect_tasks WHERE tenant_id = $1 AND brand_id = $2 AND collector_type = $3 AND business_date = $4::date AND question_id = ANY($5) AND ($6::text[] IS NULL OR ai_platform_id = ANY($6)) AND NOT (status = 'skipped' AND skip_reason = ANY($7::text[])) `, tenantID, brandID, monitoringCollectorType, businessDate.Format("2006-01-02"), questionIDs, nullableStringArray(platformIDs), monitoringProjectionExcludedSkipReasons).Scan(&plannedSampleCount); err != nil { return 0, response.ErrInternal(50041, "query_failed", "failed to load monitoring overview") } if plannedSampleCount > 0 { return plannedSampleCount, nil } if platformIDs != nil { return int64(len(questionIDs) * len(platformIDs)), nil } return int64(len(questionIDs) * len(defaultMonitoringPlatforms)), nil } func (s *MonitoringService) loadOverviewSnapshotForDate(ctx context.Context, tenantID, brandID int64, collectionMode string, businessDate time.Time) (monitoringOverviewSnapshot, bool, error) { var item monitoringOverviewSnapshot err := s.monitoringPool.QueryRow(ctx, ` SELECT business_date, desired_sample_count, planned_sample_count, actual_sample_count, coverage_rate::double precision, confidence_level, trigger_source, snapshot_updated_at, mention_rate::double precision, top1_mention_rate::double precision, first_recommend_rate::double precision, positive_mention_rate::double precision, citation_rate::double precision FROM monitoring_brand_daily WHERE tenant_id = $1 AND brand_id = $2 AND collector_type = $3 AND business_date = $4::date `, tenantID, brandID, collectionMode, businessDate.Format("2006-01-02")).Scan( &item.Date, &item.DesiredSampleCount, &item.PlannedSampleCount, &item.ActualSampleCount, &item.CoverageRate, &item.ConfidenceLevel, &item.TriggerSource, &item.SnapshotUpdatedAt, &item.MentionRate, &item.Top1MentionRate, &item.FirstRecommendRate, &item.PositiveMentionRate, &item.CitationRate, ) if err != nil { if err == pgx.ErrNoRows { return monitoringOverviewSnapshot{}, false, nil } return monitoringOverviewSnapshot{}, false, response.ErrInternal(50041, "query_failed", "failed to load monitoring overview") } return item, true, nil } func (s *MonitoringService) loadLatestSampledOverviewSnapshot(ctx context.Context, tenantID, brandID int64, collectionMode string, startDate, endDate time.Time) (monitoringOverviewSnapshot, bool, error) { if endDate.Before(startDate) { return monitoringOverviewSnapshot{}, false, nil } var item monitoringOverviewSnapshot err := s.monitoringPool.QueryRow(ctx, ` SELECT business_date, desired_sample_count, planned_sample_count, actual_sample_count, coverage_rate::double precision, confidence_level, trigger_source, snapshot_updated_at, mention_rate::double precision, top1_mention_rate::double precision, first_recommend_rate::double precision, positive_mention_rate::double precision, citation_rate::double precision FROM monitoring_brand_daily WHERE tenant_id = $1 AND brand_id = $2 AND collector_type = $3 AND business_date BETWEEN $4::date AND $5::date AND actual_sample_count > 0 ORDER BY business_date DESC LIMIT 1 `, tenantID, brandID, collectionMode, startDate.Format("2006-01-02"), endDate.Format("2006-01-02")).Scan( &item.Date, &item.DesiredSampleCount, &item.PlannedSampleCount, &item.ActualSampleCount, &item.CoverageRate, &item.ConfidenceLevel, &item.TriggerSource, &item.SnapshotUpdatedAt, &item.MentionRate, &item.Top1MentionRate, &item.FirstRecommendRate, &item.PositiveMentionRate, &item.CitationRate, ) if err != nil { if err == pgx.ErrNoRows { return monitoringOverviewSnapshot{}, false, nil } return monitoringOverviewSnapshot{}, false, response.ErrInternal(50041, "query_failed", "failed to load monitoring overview") } return item, true, nil } func (s *MonitoringService) loadPlatformOverviewSnapshotForDate( ctx context.Context, tenantID, brandID int64, aiPlatformIDs []string, collectionMode string, businessDate time.Time, ) (monitoringOverviewSnapshot, bool, error) { var item monitoringOverviewSnapshot err := s.monitoringPool.QueryRow(ctx, ` SELECT business_date, SUM(planned_sample_count) AS planned_sample_count, SUM(actual_sample_count) AS actual_sample_count, CASE WHEN SUM(actual_sample_count) > 0 THEN SUM(COALESCE(mention_rate::double precision, 0) * actual_sample_count)::double precision / SUM(actual_sample_count)::double precision ELSE NULL END AS mention_rate, CASE WHEN SUM(actual_sample_count) > 0 THEN SUM(COALESCE(top1_mention_rate::double precision, 0) * actual_sample_count)::double precision / SUM(actual_sample_count)::double precision ELSE NULL END AS top1_mention_rate, MAX(last_sampled_at) AS last_sampled_at FROM monitoring_brand_platform_daily WHERE tenant_id = $1 AND brand_id = $2 AND ai_platform_id = ANY($3) AND collector_type = $4 AND business_date = $5::date GROUP BY business_date `, tenantID, brandID, aiPlatformIDs, collectionMode, businessDate.Format("2006-01-02")).Scan( &item.Date, &item.PlannedSampleCount, &item.ActualSampleCount, &item.MentionRate, &item.Top1MentionRate, &item.SnapshotUpdatedAt, ) if err != nil { if err == pgx.ErrNoRows { return monitoringOverviewSnapshot{}, false, nil } return monitoringOverviewSnapshot{}, false, response.ErrInternal(50041, "query_failed", "failed to load monitoring platform overview") } coverageRate := divideAsPointer(item.ActualSampleCount, item.PlannedSampleCount) item.DesiredSampleCount = item.PlannedSampleCount item.CoverageRate = pointerFloat64ToNull(coverageRate) item.ConfidenceLevel = deriveMonitoringConfidenceLevel(item.ActualSampleCount, item.PlannedSampleCount, coverageRate) return item, true, nil } func (s *MonitoringService) loadLatestSampledPlatformOverviewSnapshot( ctx context.Context, tenantID, brandID int64, aiPlatformIDs []string, collectionMode string, startDate, endDate time.Time, ) (monitoringOverviewSnapshot, bool, error) { if endDate.Before(startDate) { return monitoringOverviewSnapshot{}, false, nil } var item monitoringOverviewSnapshot err := s.monitoringPool.QueryRow(ctx, ` SELECT business_date, SUM(planned_sample_count) AS planned_sample_count, SUM(actual_sample_count) AS actual_sample_count, CASE WHEN SUM(actual_sample_count) > 0 THEN SUM(COALESCE(mention_rate::double precision, 0) * actual_sample_count)::double precision / SUM(actual_sample_count)::double precision ELSE NULL END AS mention_rate, CASE WHEN SUM(actual_sample_count) > 0 THEN SUM(COALESCE(top1_mention_rate::double precision, 0) * actual_sample_count)::double precision / SUM(actual_sample_count)::double precision ELSE NULL END AS top1_mention_rate, MAX(last_sampled_at) AS last_sampled_at FROM monitoring_brand_platform_daily WHERE tenant_id = $1 AND brand_id = $2 AND ai_platform_id = ANY($3) AND collector_type = $4 AND business_date BETWEEN $5::date AND $6::date AND actual_sample_count > 0 GROUP BY business_date ORDER BY business_date DESC LIMIT 1 `, tenantID, brandID, aiPlatformIDs, collectionMode, startDate.Format("2006-01-02"), endDate.Format("2006-01-02")).Scan( &item.Date, &item.PlannedSampleCount, &item.ActualSampleCount, &item.MentionRate, &item.Top1MentionRate, &item.SnapshotUpdatedAt, ) if err != nil { if err == pgx.ErrNoRows { return monitoringOverviewSnapshot{}, false, nil } return monitoringOverviewSnapshot{}, false, response.ErrInternal(50041, "query_failed", "failed to load monitoring platform overview") } coverageRate := divideAsPointer(item.ActualSampleCount, item.PlannedSampleCount) item.DesiredSampleCount = item.PlannedSampleCount item.CoverageRate = pointerFloat64ToNull(coverageRate) item.ConfidenceLevel = deriveMonitoringConfidenceLevel(item.ActualSampleCount, item.PlannedSampleCount, coverageRate) return item, true, nil } func applyOverviewSnapshot(overview *MonitoringOverview, snapshot monitoringOverviewSnapshot, derived monitoringDerivedMetrics) { if overview == nil { return } overview.DesiredSampleCount = snapshot.DesiredSampleCount overview.PlannedSampleCount = snapshot.PlannedSampleCount overview.ActualSampleCount = snapshot.ActualSampleCount overview.CoverageRate = floatPointer(snapshot.CoverageRate) overview.LastTriggerSource = stringPointer(snapshot.TriggerSource) if strings.TrimSpace(snapshot.ConfidenceLevel) != "" { overview.ConfidenceLevel = snapshot.ConfidenceLevel } if snapshot.ActualSampleCount <= 0 { overview.LastSampledAt = nil overview.MentionRate = nil overview.Top1MentionRate = nil overview.FirstRecommendRate = nil overview.PositiveMentionRate = nil overview.CitationRate = nil return } applyDerivedRatesToOverviewSnapshot(&snapshot, derived) overview.LastSampledAt = formatNullTime(snapshot.SnapshotUpdatedAt) overview.MentionRate = floatPointer(snapshot.MentionRate) overview.Top1MentionRate = floatPointer(snapshot.Top1MentionRate) overview.FirstRecommendRate = floatPointer(snapshot.FirstRecommendRate) overview.PositiveMentionRate = floatPointer(snapshot.PositiveMentionRate) overview.CitationRate = floatPointer(snapshot.CitationRate) } func applyDerivedRatesToOverviewSnapshot(snapshot *monitoringOverviewSnapshot, derived monitoringDerivedMetrics) { if snapshot == nil { return } dateKey := snapshot.Date.Format("2006-01-02") if stats, ok := derived.ByDate[dateKey]; ok && stats.hasSamples() { snapshot.MentionRate = pointerFloat64ToNull(stats.mentionRate()) snapshot.Top1MentionRate = pointerFloat64ToNull(stats.top1MentionRate()) snapshot.FirstRecommendRate = pointerFloat64ToNull(stats.firstRecommendRate()) snapshot.PositiveMentionRate = pointerFloat64ToNull(stats.positiveMentionRate()) } } func (s *MonitoringService) loadAccessStates(ctx context.Context, tenantID, workspaceID int64, clientID *uuid.UUID, businessDate time.Time) (map[string]monitoringAccessState, error) { states := map[string]monitoringAccessState{} if clientID == nil { return states, nil } belongs, err := s.monitoringClientBelongsToWorkspace(ctx, tenantID, workspaceID, *clientID) if err != nil { return nil, err } if !belongs { return states, nil } rows, err := s.monitoringPool.Query(ctx, ` SELECT ai_platform_id, access_status FROM monitoring_platform_access_snapshots WHERE tenant_id = $1 AND client_id = $2 AND business_date = $3::date `, tenantID, *clientID, businessDate.Format("2006-01-02")) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load platform access states") } defer rows.Close() for rows.Next() { var platformID string var accessStatus string if scanErr := rows.Scan(&platformID, &accessStatus); scanErr != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to parse platform access states") } platformID = normalizeMonitoringPlatformID(platformID) if platformID == "" { continue } if existing, ok := states[platformID]; ok && strings.TrimSpace(existing.AccessStatus) != "" { continue } states[platformID] = monitoringAccessState{AccessStatus: accessStatus} } if err := rows.Err(); err != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate platform access states") } return states, nil } func (s *MonitoringService) monitoringClientBelongsToWorkspace(ctx context.Context, tenantID, workspaceID int64, clientID uuid.UUID) (bool, error) { var exists bool if err := s.businessPool.QueryRow(ctx, ` SELECT EXISTS ( SELECT 1 FROM desktop_clients WHERE id = $1 AND tenant_id = $2 AND workspace_id = $3 AND revoked_at IS NULL ) `, clientID, tenantID, workspaceID).Scan(&exists); err != nil { return false, response.ErrInternal(50041, "query_failed", "failed to validate monitoring desktop client workspace") } return exists, nil } func (s *MonitoringService) loadPlatformBreakdown(ctx context.Context, tenantID, brandID int64, questionIDs []int64, businessDate time.Time, collectionMode string, platforms []monitoringPlatformMetadata, accessStates map[string]monitoringAccessState, derived monitoringDerivedMetrics) ([]MonitoringPlatformDaily, error) { if questionIDs != nil { return s.loadPlatformBreakdownFromRuns(ctx, tenantID, brandID, questionIDs, businessDate, platforms, accessStates, derived) } rows, err := s.monitoringPool.Query(ctx, ` SELECT ai_platform_id, mention_rate::double precision, top1_mention_rate::double precision, platform_sample_status, actual_sample_count, last_sampled_at FROM monitoring_brand_platform_daily WHERE tenant_id = $1 AND brand_id = $2 AND collector_type = $3 AND business_date = $4::date `, tenantID, brandID, collectionMode, businessDate.Format("2006-01-02")) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load platform breakdown") } defer rows.Close() type row struct { MentionRate sql.NullFloat64 Top1MentionRate sql.NullFloat64 PlatformSampleStatus string ActualSampleCount int64 LastSampledAt sql.NullTime } items := make(map[string]row) for rows.Next() { var platformID string var item row if scanErr := rows.Scan( &platformID, &item.MentionRate, &item.Top1MentionRate, &item.PlatformSampleStatus, &item.ActualSampleCount, &item.LastSampledAt, ); scanErr != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to parse platform breakdown") } platformID = normalizeMonitoringPlatformID(platformID) if platformID == "" { continue } if existing, ok := items[platformID]; ok { totalSamples := existing.ActualSampleCount + item.ActualSampleCount existing.MentionRate = mergeWeightedNullFloat(existing.MentionRate, existing.ActualSampleCount, item.MentionRate, item.ActualSampleCount) existing.Top1MentionRate = mergeWeightedNullFloat(existing.Top1MentionRate, existing.ActualSampleCount, item.Top1MentionRate, item.ActualSampleCount) existing.PlatformSampleStatus = mergeMonitoringSampleStatus(existing.PlatformSampleStatus, item.PlatformSampleStatus) existing.ActualSampleCount = totalSamples existing.LastSampledAt = maxNullTime(existing.LastSampledAt, item.LastSampledAt) items[platformID] = existing continue } items[platformID] = item } dateKey := businessDate.Format("2006-01-02") breakdown := make([]MonitoringPlatformDaily, 0, len(platforms)) for _, platform := range platforms { if item, ok := items[platform.ID]; ok { if platformStats, ok := derived.ByDatePlatform[dateKey][platform.ID]; ok && platformStats.hasSamples() { item.MentionRate = pointerFloat64ToNull(platformStats.mentionRate()) item.Top1MentionRate = pointerFloat64ToNull(platformStats.top1MentionRate()) } breakdown = append(breakdown, MonitoringPlatformDaily{ AIPlatformID: platform.ID, PlatformName: platform.Name, MentionRate: floatPointer(item.MentionRate), Top1MentionRate: floatPointer(item.Top1MentionRate), PlatformSampleStatus: derivePlatformSampleStatus(item.PlatformSampleStatus, accessStates[platform.ID]), ActualSampleCount: item.ActualSampleCount, LastSampledAt: formatNullTime(item.LastSampledAt), }) continue } breakdown = append(breakdown, MonitoringPlatformDaily{ AIPlatformID: platform.ID, PlatformName: platform.Name, PlatformSampleStatus: derivePlatformSampleStatus("", accessStates[platform.ID]), }) } return breakdown, nil } func (s *MonitoringService) loadPlatformBreakdownFromRuns(ctx context.Context, tenantID, brandID int64, questionIDs []int64, businessDate time.Time, platforms []monitoringPlatformMetadata, accessStates map[string]monitoringAccessState, derived monitoringDerivedMetrics) ([]MonitoringPlatformDaily, error) { if len(questionIDs) == 0 { return emptyPlatformBreakdown(platforms, accessStates), nil } dateKey := businessDate.Format("2006-01-02") rows, err := s.monitoringPool.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 ( 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 GROUP BY r.ai_platform_id `, tenantID, brandID, monitoringCollectorType, dateKey, questionIDs) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load platform breakdown") } defer rows.Close() type row struct { MentionRate sql.NullFloat64 Top1MentionRate sql.NullFloat64 ActualSampleCount int64 LastSampledAt sql.NullTime } items := make(map[string]row) for rows.Next() { var platformID string var mentionedCount int64 var top1MentionedCount int64 var item row if scanErr := rows.Scan( &platformID, &item.ActualSampleCount, &mentionedCount, &top1MentionedCount, &item.LastSampledAt, ); scanErr != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to parse platform breakdown") } platformID = normalizeMonitoringPlatformID(platformID) if platformID == "" { continue } item.MentionRate = pointerFloat64ToNull(divideAsPointer(mentionedCount, item.ActualSampleCount)) item.Top1MentionRate = pointerFloat64ToNull(divideAsPointer(top1MentionedCount, item.ActualSampleCount)) items[platformID] = item } if err := rows.Err(); err != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate platform breakdown") } breakdown := make([]MonitoringPlatformDaily, 0, len(platforms)) for _, platform := range platforms { item := items[platform.ID] 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, PlatformName: platform.Name, MentionRate: floatPointer(item.MentionRate), Top1MentionRate: floatPointer(item.Top1MentionRate), PlatformSampleStatus: deriveMonitoringPlatformSampleStatus(item.ActualSampleCount, item.ActualSampleCount, accessStates[platform.ID].AccessStatus), ActualSampleCount: item.ActualSampleCount, LastSampledAt: formatNullTime(item.LastSampledAt), }) } return breakdown, nil } func emptyPlatformBreakdown(platforms []monitoringPlatformMetadata, accessStates map[string]monitoringAccessState) []MonitoringPlatformDaily { breakdown := make([]MonitoringPlatformDaily, 0, len(platforms)) for _, platform := range platforms { breakdown = append(breakdown, MonitoringPlatformDaily{ AIPlatformID: platform.ID, PlatformName: platform.Name, PlatformSampleStatus: derivePlatformSampleStatus("", accessStates[platform.ID]), }) } return breakdown } func (s *MonitoringService) loadHotQuestions( ctx context.Context, tenantID, brandID int64, questions []monitoringConfiguredQuestion, businessDate time.Time, aiPlatformID *string, derived monitoringDerivedMetrics, ) ([]MonitoringHotQuestion, error) { if len(questions) == 0 { return []MonitoringHotQuestion{}, nil } questionIDs := configuredQuestionIDs(questions) selectedDate := businessDate.Format("2006-01-02") platformQueryIDs := monitoringPlatformQueryIDs(aiPlatformID) rows, err := s.monitoringPool.Query(ctx, ` WITH latest_runs AS ( SELECT DISTINCT ON (r.question_id, r.ai_platform_id) r.question_id, 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 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.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 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, h.question_hash, s.mention_rate, s.last_sampled_at FROM stats s LEFT JOIN latest_hash h ON h.question_id = s.question_id `, tenantID, brandID, monitoringCollectorType, selectedDate, questionIDs, nullableStringArray(platformQueryIDs)) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load hot questions") } defer rows.Close() type questionStat struct { QuestionHash []byte MentionRate sql.NullFloat64 LastSampledAt sql.NullTime } stats := make(map[int64]questionStat, len(questions)) for rows.Next() { var questionID int64 var item questionStat if scanErr := rows.Scan(&questionID, &item.QuestionHash, &item.MentionRate, &item.LastSampledAt); scanErr != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to parse hot questions") } stats[questionID] = item } if err := rows.Err(); err != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate hot questions") } items := make([]MonitoringHotQuestion, 0, len(questions)) for _, question := range questions { item := MonitoringHotQuestion{ QuestionID: question.ID, QuestionHash: encodeQuestionHash(question.QuestionHash), QuestionText: question.QuestionText, } if stat, ok := stats[question.ID]; ok { if len(stat.QuestionHash) > 0 { item.QuestionHash = encodeQuestionHash(stat.QuestionHash) } item.MentionRate = floatPointer(stat.MentionRate) item.LastSampledAt = formatNullTime(stat.LastSampledAt) } if questionStatsByDate, ok := derived.ByDateQuestion[selectedDate]; ok { if derivedStats, ok := questionStatsByDate[question.ID]; ok && derivedStats.hasSamples() { item.MentionRate = derivedStats.mentionRate() } } items = append(items, item) } return items, nil } func (s *MonitoringService) loadHotQuestionsPage( ctx context.Context, tenantID, brandID int64, keywordID *int64, questionID *int64, businessDate time.Time, aiPlatformID *string, derived monitoringDerivedMetrics, pageReq MonitoringPageRequest, ) ([]MonitoringHotQuestion, MonitoringPage, error) { pageReq = normalizeMonitoringPageRequest(pageReq, defaultMonitoringListPageSize) var total int64 if err := s.businessPool.QueryRow(ctx, ` SELECT COUNT(*) FROM brand_questions q JOIN brand_keywords k ON k.id = q.keyword_id AND k.tenant_id = q.tenant_id AND k.brand_id = q.brand_id AND k.deleted_at IS NULL AND k.status = 'active' WHERE q.tenant_id = $1 AND q.brand_id = $2 AND q.deleted_at IS NULL AND q.status = 'active' AND ($3::bigint IS NULL OR q.keyword_id = $3) AND ($4::bigint IS NULL OR q.id = $4) AND ($5::text = '' OR q.question_text ILIKE '%' || $5 || '%') `, tenantID, brandID, nullableInt64(keywordID), nullableInt64(questionID), pageReq.Query).Scan(&total); err != nil { return nil, MonitoringPage{}, response.ErrInternal(50041, "query_failed", "failed to count hot questions") } page := monitoringPageFromRequest(pageReq, total) if total == 0 { return []MonitoringHotQuestion{}, page, nil } rows, err := s.businessPool.Query(ctx, ` SELECT q.id, q.keyword_id, q.question_text FROM brand_questions q JOIN brand_keywords k ON k.id = q.keyword_id AND k.tenant_id = q.tenant_id AND k.brand_id = q.brand_id AND k.deleted_at IS NULL AND k.status = 'active' WHERE q.tenant_id = $1 AND q.brand_id = $2 AND q.deleted_at IS NULL AND q.status = 'active' AND ($3::bigint IS NULL OR q.keyword_id = $3) AND ($4::bigint IS NULL OR q.id = $4) AND ($5::text = '' OR q.question_text ILIKE '%' || $5 || '%') ORDER BY q.keyword_id ASC, q.id ASC LIMIT $6 OFFSET $7 `, tenantID, brandID, nullableInt64(keywordID), nullableInt64(questionID), pageReq.Query, pageReq.PageSize, monitoringPageOffset(pageReq)) if err != nil { return nil, MonitoringPage{}, response.ErrInternal(50041, "query_failed", "failed to load hot questions") } defer rows.Close() questions := make([]monitoringConfiguredQuestion, 0, pageReq.PageSize) for rows.Next() { var item monitoringConfiguredQuestion if scanErr := rows.Scan(&item.ID, &item.KeywordID, &item.QuestionText); scanErr != nil { return nil, MonitoringPage{}, response.ErrInternal(50041, "scan_failed", "failed to parse hot questions") } item.QuestionHash = seededQuestionHash(item.QuestionText) questions = append(questions, item) } if err := rows.Err(); err != nil { return nil, MonitoringPage{}, response.ErrInternal(50041, "scan_failed", "failed to iterate hot questions") } if len(questions) == 0 { return []MonitoringHotQuestion{}, page, nil } items, err := s.loadHotQuestions(ctx, tenantID, brandID, questions, businessDate, aiPlatformID, derived) if err != nil { return nil, MonitoringPage{}, err } return items, page, nil } func (s *MonitoringService) loadCitationRanking( ctx context.Context, tenantID, brandID int64, questionIDs []int64, startDate, endDate time.Time, accessStates map[string]monitoringAccessState, aiPlatformID *string, ) ([]MonitoringCitationRanking, error) { if questionIDs != nil && len(questionIDs) == 0 { return []MonitoringCitationRanking{}, nil } platformQueryIDs := monitoringPlatformQueryIDs(aiPlatformID) type citationRankingAggregate struct { CitedAnswerCount int64 CitedArticleCount int64 CitationSourceCount int64 SaaSSourceCount int64 SampleCount int64 } aggregates := make(map[string]citationRankingAggregate) rows, err := s.monitoringPool.Query(ctx, ` WITH filtered_runs AS ( SELECT id, ai_platform_id FROM question_monitor_runs WHERE tenant_id = $1 AND ($2::bigint = 0 OR 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)) ), sample_counts AS ( SELECT ai_platform_id, COUNT(*)::bigint AS sample_count FROM filtered_runs GROUP BY ai_platform_id ), citation_counts AS ( SELECT fr.ai_platform_id, COUNT(cf.id)::bigint AS citation_source_count, COUNT(cf.id) FILTER ( WHERE cf.content_source_type IS NOT NULL AND cf.content_source_id IS NOT NULL )::bigint AS saas_source_count, COUNT(DISTINCT cf.run_id) FILTER ( WHERE cf.content_source_type IS NOT NULL AND cf.content_source_id IS NOT NULL )::bigint AS cited_answer_count, COUNT(DISTINCT cf.content_source_type || ':' || cf.content_source_id::text) FILTER ( WHERE cf.content_source_type IS NOT NULL AND cf.content_source_id IS NOT NULL )::bigint AS cited_article_count FROM filtered_runs fr JOIN monitoring_citation_facts cf ON cf.tenant_id = $1 AND cf.run_id = fr.id AND cf.business_date BETWEEN $4::date AND $5::date GROUP BY fr.ai_platform_id ) SELECT sc.ai_platform_id, sc.sample_count, COALESCE(cc.citation_source_count, 0)::bigint, COALESCE(cc.saas_source_count, 0)::bigint, COALESCE(cc.cited_answer_count, 0)::bigint, COALESCE(cc.cited_article_count, 0)::bigint FROM sample_counts sc LEFT JOIN citation_counts cc ON cc.ai_platform_id = sc.ai_platform_id `, tenantID, brandID, monitoringCollectorType, startDate.Format("2006-01-02"), endDate.Format("2006-01-02"), nullableInt64Array(questionIDs), nullableStringArray(platformQueryIDs)) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load citation ranking") } defer rows.Close() for rows.Next() { var platformID string var sampleCount int64 var citationSourceCount int64 var saasSourceCount int64 var citedAnswerCount int64 var citedArticleCount int64 if scanErr := rows.Scan( &platformID, &sampleCount, &citationSourceCount, &saasSourceCount, &citedAnswerCount, &citedArticleCount, ); scanErr != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to parse citation ranking") } platformID = normalizeMonitoringPlatformID(platformID) if platformID == "" { continue } item := aggregates[platformID] item.SampleCount += sampleCount item.CitationSourceCount += citationSourceCount item.SaaSSourceCount += saasSourceCount item.CitedAnswerCount += citedAnswerCount item.CitedArticleCount += citedArticleCount aggregates[platformID] = item } if err := rows.Err(); err != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate citation ranking") } items := make([]MonitoringCitationRanking, 0, len(aggregates)) for platformID, aggregate := range aggregates { if aggregate.SaaSSourceCount == 0 { continue } items = append(items, MonitoringCitationRanking{ AIPlatformID: platformID, PlatformName: platformDisplayName(platformID), SampleStatus: derivePlatformSampleStatus("", accessStates[platformID]), CitedAnswerCount: aggregate.CitedAnswerCount, CitedArticleCount: aggregate.CitedArticleCount, CitationSourceCount: aggregate.CitationSourceCount, SaaSSourceCount: aggregate.SaaSSourceCount, SaaSSourceRate: divideAsPointer(aggregate.SaaSSourceCount, aggregate.CitationSourceCount), CitationRate: divideAsPointer(aggregate.CitedAnswerCount, aggregate.SampleCount), }) } sort.Slice(items, func(left, right int) bool { if items[left].CitedAnswerCount != items[right].CitedAnswerCount { return items[left].CitedAnswerCount > items[right].CitedAnswerCount } if items[left].SaaSSourceCount != items[right].SaaSSourceCount { return items[left].SaaSSourceCount > items[right].SaaSSourceCount } if items[left].CitedArticleCount != items[right].CitedArticleCount { return items[left].CitedArticleCount > items[right].CitedArticleCount } return monitoringPlatformSortIndex(items[left].AIPlatformID) < monitoringPlatformSortIndex(items[right].AIPlatformID) }) return items, nil } func (s *MonitoringService) loadCitedArticles( ctx context.Context, tenantID, brandID int64, questionIDs []int64, startDate, endDate time.Time, aiPlatformID *string, ) ([]MonitoringCitedArticle, error) { pageReq := MonitoringPageRequest{Page: 1, PageSize: maxMonitoringListPageSize} items, _, err := s.loadCitedArticlesPage(ctx, tenantID, brandID, questionIDs, startDate, endDate, aiPlatformID, pageReq) return items, err } func (s *MonitoringService) loadCitedArticlesPage( ctx context.Context, tenantID, brandID int64, questionIDs []int64, startDate, endDate time.Time, aiPlatformID *string, pageReq MonitoringPageRequest, ) ([]MonitoringCitedArticle, MonitoringPage, error) { pageReq = normalizeMonitoringPageRequest(pageReq, defaultMonitoringListPageSize) if questionIDs != nil && len(questionIDs) == 0 { return []MonitoringCitedArticle{}, emptyMonitoringPage(pageReq), nil } platformQueryIDs := monitoringPlatformQueryIDs(aiPlatformID) startDateText := startDate.Format("2006-01-02") endDateText := endDate.Format("2006-01-02") var totalSampleCount int64 if err := s.monitoringPool.QueryRow(ctx, ` SELECT COUNT(*) FROM question_monitor_runs r WHERE r.tenant_id = $1 AND ($2::bigint = 0 OR 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, startDateText, endDateText, nullableInt64Array(questionIDs), nullableStringArray(platformQueryIDs)).Scan(&totalSampleCount); err != nil { return nil, MonitoringPage{}, response.ErrInternal(50041, "query_failed", "failed to count cited article denominator") } sourceRows, totalArticleCitationCount, totalSources, err := s.loadCitedArticleSourceRows(ctx, tenantID, brandID, questionIDs, startDateText, endDateText, platformQueryIDs, pageReq) if err != nil { return nil, MonitoringPage{}, err } page := monitoringPageFromRequest(pageReq, totalSources) if len(sourceRows) == 0 { return []MonitoringCitedArticle{}, page, nil } metaByKey, err := s.loadCitedArticleSourceMeta(ctx, tenantID, sourceRows) if err != nil { return nil, MonitoringPage{}, err } items := make([]MonitoringCitedArticle, 0, len(sourceRows)) for _, source := range sourceRows { item := metaByKey[monitoringContentSourceKey(source.SourceType, source.SourceID)] if item.SourceID == 0 { item = MonitoringCitedArticle{ ArticleTitle: "未命名文章", PublishPlatform: "外部文章", SourceType: source.SourceType, SourceID: source.SourceID, } } item.CitationCount = source.CitationCount item.CitationRate = divideAsPointer(source.CitationCount, totalSampleCount) item.SourceShare = divideAsPointer(source.CitationCount, totalArticleCitationCount) items = append(items, item) } return items, page, nil } type monitoringCitedArticleSourceRow struct { SourceType string SourceID int64 CitationCount int64 } func (s *MonitoringService) loadCitedArticleSourceRows( ctx context.Context, tenantID, brandID int64, questionIDs []int64, startDateText, endDateText string, platformQueryIDs []string, pageReq MonitoringPageRequest, ) ([]monitoringCitedArticleSourceRow, int64, int64, error) { rows, err := s.monitoringPool.Query(ctx, ` WITH filtered_sources AS ( SELECT cf.content_source_type, cf.content_source_id, COUNT(*)::bigint AS citation_count 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 ($2::bigint = 0 OR cf.brand_id = $2) AND cf.business_date BETWEEN $4::date AND $5::date AND cf.content_source_type IS NOT NULL AND cf.content_source_id IS NOT NULL 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)) GROUP BY cf.content_source_type, cf.content_source_id ), counts AS ( SELECT COALESCE(SUM(citation_count), 0)::bigint AS total_citation_count, COUNT(*)::bigint AS total_source_count FROM filtered_sources ), paged_sources AS ( SELECT content_source_type, content_source_id, citation_count FROM filtered_sources ORDER BY citation_count DESC, content_source_type ASC, content_source_id ASC LIMIT $8 OFFSET $9 ) SELECT ps.content_source_type, ps.content_source_id, ps.citation_count, counts.total_citation_count, counts.total_source_count FROM counts LEFT JOIN paged_sources ps ON TRUE `, tenantID, brandID, monitoringCollectorType, startDateText, endDateText, nullableInt64Array(questionIDs), nullableStringArray(platformQueryIDs), pageReq.PageSize, monitoringPageOffset(pageReq)) if err != nil { return nil, 0, 0, response.ErrInternal(50041, "query_failed", "failed to load cited articles") } defer rows.Close() items := make([]monitoringCitedArticleSourceRow, 0, pageReq.PageSize) var totalCitationCount int64 var totalSourceCount int64 for rows.Next() { var sourceType sql.NullString var sourceID sql.NullInt64 var citationCount sql.NullInt64 if scanErr := rows.Scan( &sourceType, &sourceID, &citationCount, &totalCitationCount, &totalSourceCount, ); scanErr != nil { return nil, 0, 0, response.ErrInternal(50041, "scan_failed", "failed to parse cited articles") } if !sourceType.Valid || !sourceID.Valid || !citationCount.Valid { continue } item := monitoringCitedArticleSourceRow{ SourceType: strings.TrimSpace(sourceType.String), SourceID: sourceID.Int64, CitationCount: citationCount.Int64, } if item.SourceType == "" || item.SourceID <= 0 { continue } items = append(items, item) } if err := rows.Err(); err != nil { return nil, 0, 0, response.ErrInternal(50041, "scan_failed", "failed to iterate cited articles") } return items, totalCitationCount, totalSourceCount, nil } func (s *MonitoringService) loadCitedArticleSourceMeta(ctx context.Context, tenantID int64, sources []monitoringCitedArticleSourceRow) (map[string]MonitoringCitedArticle, error) { result := make(map[string]MonitoringCitedArticle, len(sources)) publishedIDs := make([]int64, 0, len(sources)) manualIDs := make([]int64, 0, len(sources)) for _, source := range sources { switch source.SourceType { case monitoringContentSourcePublishedArticle: publishedIDs = append(publishedIDs, source.SourceID) case monitoringContentSourceManualMark: manualIDs = append(manualIDs, source.SourceID) } } if len(publishedIDs) > 0 { rows, err := s.monitoringPool.Query(ctx, ` SELECT DISTINCT ON (article_id) article_id, COALESCE(NULLIF(article_title_snapshot, ''), '未命名文章') AS article_title, COALESCE(NULLIF(publish_platform_name_snapshot, ''), '已发布内容') AS publish_platform, COALESCE(NULLIF(original_url, ''), normalized_url) AS source_url FROM monitoring_article_url_aliases WHERE tenant_id = $1 AND confidence = 'high' AND article_id = ANY($2) ORDER BY article_id, updated_at DESC, id DESC `, tenantID, publishedIDs) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load cited article metadata") } defer rows.Close() for rows.Next() { var articleID int64 var item MonitoringCitedArticle if scanErr := rows.Scan(&articleID, &item.ArticleTitle, &item.PublishPlatform, &item.SourceURL); scanErr != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to parse cited article metadata") } value := articleID item.ArticleID = &value item.ArticleTitle = repairMonitoringMojibake(item.ArticleTitle) item.PublishPlatform = repairMonitoringMojibake(item.PublishPlatform) item.SourceType = monitoringContentSourcePublishedArticle item.SourceID = articleID result[monitoringContentSourceKey(item.SourceType, item.SourceID)] = item } if err := rows.Err(); err != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate cited article metadata") } } if len(manualIDs) > 0 { rows, err := s.monitoringPool.Query(ctx, ` SELECT id, COALESCE(NULLIF(article_title, ''), '未命名文章') AS article_title, COALESCE(NULLIF(publish_platform, ''), '外部文章') AS publish_platform, COALESCE(NULLIF(original_url, ''), normalized_url) AS source_url FROM monitoring_user_marked_articles WHERE tenant_id = $1 AND id = ANY($2) `, tenantID, manualIDs) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load marked article metadata") } defer rows.Close() for rows.Next() { var item MonitoringCitedArticle if scanErr := rows.Scan(&item.SourceID, &item.ArticleTitle, &item.PublishPlatform, &item.SourceURL); scanErr != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to parse marked article metadata") } item.ArticleTitle = repairMonitoringMojibake(item.ArticleTitle) item.PublishPlatform = repairMonitoringMojibake(item.PublishPlatform) item.SourceType = monitoringContentSourceManualMark result[monitoringContentSourceKey(item.SourceType, item.SourceID)] = item } if err := rows.Err(); err != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate marked article metadata") } } mediaItems, err := s.loadMediaSupplyCitedArticleSourceMeta(ctx, tenantID, sources) if err != nil { return nil, err } for key, item := range mediaItems { result[key] = item } return result, nil } func (s *MonitoringService) loadMediaSupplyCitedArticleSourceMeta(ctx context.Context, tenantID int64, sources []monitoringCitedArticleSourceRow) (map[string]MonitoringCitedArticle, error) { result := make(map[string]MonitoringCitedArticle) mediaItemIDs := make([]int64, 0, len(sources)) for _, source := range sources { if source.SourceType == monitoringContentSourceMediaSupplyOrderItem { mediaItemIDs = append(mediaItemIDs, source.SourceID) } } if len(mediaItemIDs) == 0 { return result, nil } rows, err := s.businessPool.Query(ctx, ` SELECT i.id, o.article_id, COALESCE(NULLIF(o.title, ''), NULLIF(i.resource_name_snapshot, ''), '未命名文章') AS article_title, COALESCE(NULLIF(i.resource_name_snapshot, ''), $3) AS publish_platform, i.external_article_url FROM media_supply_order_items i JOIN media_supply_orders o ON o.id = i.order_id WHERE o.tenant_id = $1 AND i.id = ANY($2) AND o.deleted_at IS NULL `, tenantID, mediaItemIDs, monitoringMediaSupplySubmissionCitationLabel) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load media supply citation metadata") } defer rows.Close() for rows.Next() { var item MonitoringCitedArticle var articleID sql.NullInt64 if scanErr := rows.Scan(&item.SourceID, &articleID, &item.ArticleTitle, &item.PublishPlatform, &item.SourceURL); scanErr != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to parse media supply citation metadata") } item.ArticleID = nullableInt64Value(articleID) item.ArticleTitle = repairMonitoringMojibake(item.ArticleTitle) item.PublishPlatform = repairMonitoringMojibake(item.PublishPlatform) item.SourceType = monitoringContentSourceMediaSupplyOrderItem result[monitoringContentSourceKey(item.SourceType, item.SourceID)] = item } if err := rows.Err(); err != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate media supply citation metadata") } return result, nil } func monitoringContentSourceKey(sourceType string, sourceID int64) string { return strings.TrimSpace(sourceType) + ":" + strconv.FormatInt(sourceID, 10) } func monitoringAliasContentKey(alias monitoringPublishedAlias) string { sourceType := strings.TrimSpace(alias.SourceType) sourceID := alias.SourceID if sourceType == "" { sourceType = "published_article" } if sourceID == 0 && alias.ArticleID > 0 { sourceID = alias.ArticleID } if sourceID > 0 { return sourceType + ":" + strconv.FormatInt(sourceID, 10) } key := strings.TrimSpace(alias.NormalizedURL) if key == "" { key = strings.TrimSpace(alias.OriginalURL) } return sourceType + ":url:" + key } func (s *MonitoringService) loadQuestionText(ctx context.Context, tenantID, brandID, questionID int64) (string, error) { var text string if err := s.businessPool.QueryRow(ctx, ` SELECT question_text FROM brand_questions WHERE id = $1 AND tenant_id = $2 AND brand_id = $3 AND deleted_at IS NULL `, questionID, tenantID, brandID).Scan(&text); err != nil { if err == pgx.ErrNoRows { return "", response.ErrNotFound(40441, "question_not_found", "question not found") } return "", response.ErrInternal(50041, "query_failed", "failed to load question") } return text, nil } func (s *MonitoringService) loadLatestQuestionHash(ctx context.Context, tenantID, brandID, questionID int64, fromDate, toDate time.Time) ([]byte, error) { var hashBytes []byte if err := s.monitoringPool.QueryRow(ctx, ` SELECT question_hash FROM ( SELECT question_hash, business_date, COALESCE(completed_at, updated_at, created_at) AS event_at FROM question_monitor_runs WHERE tenant_id = $1 AND brand_id = $2 AND question_id = $3 AND collector_type = $4 AND business_date BETWEEN $5::date AND $6::date UNION ALL SELECT question_hash, business_date, COALESCE(completed_at, callback_received_at, leased_at, planned_at, updated_at, created_at) AS event_at FROM monitoring_collect_tasks WHERE tenant_id = $1 AND brand_id = $2 AND question_id = $3 AND collector_type = $4 AND business_date BETWEEN $5::date AND $6::date ) AS hashes ORDER BY business_date DESC, event_at DESC NULLS LAST LIMIT 1 `, tenantID, brandID, questionID, monitoringCollectorType, fromDate.Format("2006-01-02"), toDate.Format("2006-01-02")).Scan(&hashBytes); err != nil { if err == pgx.ErrNoRows { return nil, nil } return nil, response.ErrInternal(50041, "query_failed", "failed to resolve question hash") } return hashBytes, nil } func (s *MonitoringService) loadLatestPlatformRuns(ctx context.Context, tenantID, brandID int64, brandName string, questionID int64, hashBytes []byte, fromDate, toDate time.Time, aiPlatformID *string) (map[string]monitoringLatestPlatformRun, error) { platformQueryIDs := monitoringPlatformQueryIDs(aiPlatformID) rows, err := s.monitoringPool.Query(ctx, ` SELECT DISTINCT ON (r.ai_platform_id) r.ai_platform_id, r.id, r.completed_at, r.provider_model, r.raw_answer_text, r.raw_response_json, p.brand_mentioned, p.brand_mention_position 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.question_id = $3 AND r.question_hash = $4 AND r.collector_type = $5 AND r.status = 'succeeded' AND r.business_date BETWEEN $6::date AND $7::date AND ($8::text[] IS NULL OR r.ai_platform_id = ANY($8)) ORDER BY r.ai_platform_id ASC, r.completed_at DESC NULLS LAST, r.id DESC `, tenantID, brandID, questionID, hashBytes, monitoringCollectorType, fromDate.Format("2006-01-02"), toDate.Format("2006-01-02"), nullableStringArray(platformQueryIDs)) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load question runs") } defer rows.Close() items := make(map[string]monitoringLatestPlatformRun) for rows.Next() { var item monitoringLatestPlatformRun var rawResponseJSON []byte if scanErr := rows.Scan( &item.PlatformID, &item.RunID, &item.CompletedAt, &item.ProviderModel, &item.AnswerText, &rawResponseJSON, &item.BrandMentioned, &item.BrandMentionPosition, ); scanErr != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to parse latest question runs") } item.PlatformID = normalizeMonitoringPlatformID(item.PlatformID) if item.PlatformID == "" { continue } if item.AnswerText.Valid { item.AnswerText.String = repairMonitoringMojibake(item.AnswerText.String) } item.InlineCitations = extractMonitoringInlineCitationsFromRawResponse(rawResponseJSON, item.PlatformID) summary := resolveMonitoringAnswerParseSummary( item.AnswerText.String, brandName, monitoringStoredParseFields{ BrandMentioned: boolPointer(item.BrandMentioned), BrandMentionPosition: stringPointer(item.BrandMentionPosition), }, ) item.BrandMentioned = sql.NullBool{Bool: summary.BrandMentioned, Valid: true} item.BrandMentionPosition = pointerStringToNull(optionalString(summary.BrandMentionPosition)) existing, ok := items[item.PlatformID] if !ok || monitoringLatestPlatformRunIsNewer(item, existing) { items[item.PlatformID] = item } } return items, nil } func extractMonitoringInlineCitationsFromRawResponse(raw []byte, platformID string) []MonitoringSourceItem { if strings.TrimSpace(string(raw)) == "" || strings.TrimSpace(string(raw)) == "null" { return nil } var payload map[string]interface{} if err := json.Unmarshal(raw, &payload); err != nil { return nil } items := extractMonitoringSourceItems(payload["inline_citation_candidates"]) if len(items) == 0 && normalizeMonitoringPlatformID(platformID) == "deepseek" { items = extractMonitoringSourceItems(payload["inline_links"]) if len(items) == 0 { items = extractMonitoringSourceItems(payload["source_panel_links"]) } } if len(items) == 0 && normalizeMonitoringPlatformID(platformID) == "kimi" { items = extractMonitoringSourceItems(payload["search_results"]) } if len(items) == 0 { return nil } result := make([]MonitoringSourceItem, 0, len(items)) seen := make(map[string]struct{}, len(items)) for _, item := range items { normalizedURL := normalizeCitationURL(strings.TrimSpace(item.URL)) if normalizedURL == "" { normalizedURL = strings.TrimSpace(item.URL) } if normalizedURL == "" { continue } if _, exists := seen[normalizedURL]; exists { continue } seen[normalizedURL] = struct{}{} item.URL = normalizedURL if item.NormalizedURL == nil || strings.TrimSpace(*item.NormalizedURL) == "" { item.NormalizedURL = &normalizedURL } result = append(result, normalizeMonitoringSourceItemValue(item)) } if len(result) == 0 { return nil } return result } func (s *MonitoringService) loadCitationsForRuns(ctx context.Context, tenantID int64, runIDs []int64) (map[int64][]MonitoringQuestionDetailCitation, error) { result := make(map[int64][]MonitoringQuestionDetailCitation) if len(runIDs) == 0 { return result, nil } aliasIndex, err := loadMonitoringPublishedAliasIndex(ctx, s.monitoringPool, s.businessPool, tenantID) if err != nil { return nil, err } rows, err := s.monitoringPool.Query(ctx, ` WITH filtered_citations AS ( SELECT cf.id, cf.run_id, cf.cited_url, cf.cited_title, cf.normalized_url, cf.host, cf.registrable_domain, cf.site_key, cf.resolution_status, cf.resolution_confidence FROM monitoring_citation_facts cf WHERE cf.tenant_id = $1 AND cf.run_id = ANY($2) ), domain_keys AS ( SELECT DISTINCT host, registrable_domain, site_key FROM filtered_citations ), domain_mappings AS ( SELECT dk.host, dk.registrable_domain, dk.site_key, dm.registrable_domain AS mapped_domain, dm.site_key AS mapped_site_key, dm.site_name AS mapped_site_name FROM domain_keys dk LEFT JOIN LATERAL ( SELECT mapping.registrable_domain, mapping.site_key, mapping.site_name FROM site_domain_mappings mapping WHERE mapping.is_active = TRUE AND mapping.registrable_domain = ANY( ARRAY( SELECT DISTINCT candidate FROM ( VALUES (dk.registrable_domain), (dk.site_key), (dk.host) UNION ALL SELECT array_to_string(host_parts.value[idx.value:array_length(host_parts.value, 1)], '.') FROM (SELECT string_to_array(dk.host, '.') AS value) host_parts CROSS JOIN generate_subscripts(host_parts.value, 1) AS idx(value) WHERE idx.value > 1 ) candidates(candidate) WHERE candidate IS NOT NULL AND candidate <> '' ) ) ORDER BY LENGTH(mapping.registrable_domain) DESC, mapping.id DESC LIMIT 1 ) dm ON TRUE ) SELECT fc.run_id, fc.cited_url, fc.cited_title, fc.normalized_url, COALESCE(dm.mapped_site_name, fc.site_key, fc.registrable_domain) AS site_name, COALESCE(dm.mapped_site_key, fc.site_key) AS site_key, fc.resolution_status, fc.resolution_confidence FROM filtered_citations fc LEFT JOIN domain_mappings dm ON dm.host = fc.host AND dm.registrable_domain = fc.registrable_domain AND dm.site_key = fc.site_key ORDER BY fc.run_id ASC, fc.id ASC `, tenantID, runIDs) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load citations") } defer rows.Close() for rows.Next() { var runID int64 var citedURL string var citedTitle sql.NullString var normalizedURL string var siteName string var siteKey string var resolutionStatus string var resolutionConfidence string if scanErr := rows.Scan(&runID, &citedURL, &citedTitle, &normalizedURL, &siteName, &siteKey, &resolutionStatus, &resolutionConfidence); scanErr != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to parse citations") } siteName = repairMonitoringMojibake(siteName) if citedTitle.Valid { citedTitle.String = repairMonitoringMojibake(citedTitle.String) } var articleID *int64 var articleTitle *string var sourceType *string var sourceID *int64 var sourceURL *string if alias, ok := aliasIndex.Match(citedURL, normalizedURL); ok { if alias.ArticleID > 0 { value := alias.ArticleID articleID = &value } title := repairMonitoringMojibake(alias.ArticleTitle) articleTitle = &title aliasSourceType := strings.TrimSpace(alias.SourceType) if aliasSourceType == "" { aliasSourceType = monitoringContentSourcePublishedArticle } sourceType = &aliasSourceType aliasSourceID := alias.SourceID if aliasSourceID == 0 && alias.ArticleID > 0 { aliasSourceID = alias.ArticleID } if aliasSourceID > 0 { sourceID = &aliasSourceID } aliasSourceURL := strings.TrimSpace(alias.OriginalURL) if aliasSourceURL == "" { aliasSourceURL = strings.TrimSpace(alias.NormalizedURL) } if aliasSourceURL != "" { sourceURL = &aliasSourceURL } resolutionStatus = "resolved" resolutionConfidence = "high" } result[runID] = append(result[runID], MonitoringQuestionDetailCitation{ CitedURL: citedURL, CitedTitle: nullableStringValue(citedTitle), SiteName: siteName, SiteKey: siteKey, FaviconURL: faviconURL(siteKey), ArticleID: articleID, SourceType: sourceType, SourceID: sourceID, SourceURL: sourceURL, ArticleTitle: articleTitle, ResolutionStatus: resolutionStatus, ResolutionConfidence: resolutionConfidence, }) } if err := rows.Err(); err != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate citations") } return result, nil } func (s *MonitoringService) loadQuestionCitationAnalysis(ctx context.Context, tenantID int64, runIDs []int64) ([]MonitoringQuestionCitationStats, error) { if len(runIDs) == 0 { return []MonitoringQuestionCitationStats{}, nil } rows, err := s.monitoringPool.Query(ctx, ` WITH filtered_facts AS ( SELECT r.ai_platform_id, cf.id, cf.host, cf.registrable_domain, cf.site_key, cf.content_source_type, cf.content_source_id FROM question_monitor_runs r JOIN monitoring_citation_facts cf ON cf.tenant_id = r.tenant_id AND cf.run_id = r.id WHERE r.tenant_id = $1 AND r.status = 'succeeded' AND r.id = ANY($2) ), domain_keys AS ( SELECT DISTINCT host, registrable_domain, site_key FROM filtered_facts ), domain_mappings AS ( SELECT dk.host, dk.registrable_domain, dk.site_key, dm.registrable_domain AS mapped_domain, dm.site_key AS mapped_site_key, dm.site_name AS mapped_site_name FROM domain_keys dk LEFT JOIN LATERAL ( SELECT mapping.registrable_domain, mapping.site_key, mapping.site_name FROM site_domain_mappings mapping WHERE mapping.is_active = TRUE AND mapping.registrable_domain = ANY( ARRAY( SELECT DISTINCT candidate FROM ( VALUES (dk.registrable_domain), (dk.site_key), (dk.host) UNION ALL SELECT array_to_string(host_parts.value[idx.value:array_length(host_parts.value, 1)], '.') FROM (SELECT string_to_array(dk.host, '.') AS value) host_parts CROSS JOIN generate_subscripts(host_parts.value, 1) AS idx(value) WHERE idx.value > 1 ) candidates(candidate) WHERE candidate IS NOT NULL AND candidate <> '' ) ) ORDER BY LENGTH(mapping.registrable_domain) DESC, mapping.id DESC LIMIT 1 ) dm ON TRUE ) SELECT ff.ai_platform_id, COALESCE(dm.mapped_site_name, ff.site_key, ff.registrable_domain) AS site_name, COALESCE(dm.mapped_site_key, ff.site_key) AS site_key, COALESCE(dm.mapped_domain, ff.site_key, ff.registrable_domain) AS site_domain, ff.content_source_type, ff.content_source_id FROM filtered_facts ff LEFT JOIN domain_mappings dm ON dm.host = ff.host AND dm.registrable_domain = ff.registrable_domain AND dm.site_key = ff.site_key ORDER BY ff.ai_platform_id ASC, ff.id ASC `, tenantID, runIDs) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load question citation analysis") } defer rows.Close() type citationAnalysisKey struct { PlatformID string SiteKey string } aggregates := make(map[citationAnalysisKey]MonitoringQuestionCitationStats) totals := make(map[string]int64) for rows.Next() { var platformID string var siteName string var siteKey string var siteDomain string var sourceType sql.NullString var sourceID sql.NullInt64 if scanErr := rows.Scan(&platformID, &siteName, &siteKey, &siteDomain, &sourceType, &sourceID); scanErr != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to parse question citation analysis") } platformID = normalizeMonitoringPlatformID(platformID) if platformID == "" { continue } key := citationAnalysisKey{ PlatformID: platformID, SiteKey: siteKey, } item := aggregates[key] if item.SiteKey == "" { item = MonitoringQuestionCitationStats{ AIPlatformID: platformID, PlatformName: platformDisplayName(platformID), SiteName: siteName, SiteKey: siteKey, SiteDomain: siteDomain, } } item.CitationCount++ if sourceType.Valid && strings.TrimSpace(sourceType.String) != "" && sourceID.Valid { item.ContentCitationCount++ } aggregates[key] = item totals[platformID]++ } if err := rows.Err(); err != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate question citation analysis") } items := make([]MonitoringQuestionCitationStats, 0, len(aggregates)) for _, item := range aggregates { item.CitationRate = divideAsPointer(item.CitationCount, totals[item.AIPlatformID]) items = append(items, item) } sort.Slice(items, func(left, right int) bool { if items[left].AIPlatformID != items[right].AIPlatformID { return monitoringPlatformSortIndex(items[left].AIPlatformID) < monitoringPlatformSortIndex(items[right].AIPlatformID) } if items[left].CitationCount != items[right].CitationCount { return items[left].CitationCount > items[right].CitationCount } if items[left].SiteName != items[right].SiteName { return items[left].SiteName < items[right].SiteName } return items[left].SiteKey < items[right].SiteKey }) return items, nil } func (s *MonitoringService) loadQuestionContentCitations(ctx context.Context, tenantID int64, runIDs []int64) ([]MonitoringQuestionContentCitation, error) { if len(runIDs) == 0 { return []MonitoringQuestionContentCitation{}, nil } rows, err := s.monitoringPool.Query(ctx, ` SELECT r.ai_platform_id, cf.content_source_type, cf.content_source_id FROM question_monitor_runs r JOIN monitoring_citation_facts cf ON cf.tenant_id = r.tenant_id AND cf.run_id = r.id WHERE r.tenant_id = $1 AND r.status = 'succeeded' AND r.id = ANY($2) `, tenantID, runIDs) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load content citations") } defer rows.Close() type contentCitationKey struct { PlatformID string SourceType string SourceID int64 } aggregates := make(map[contentCitationKey]MonitoringQuestionContentCitation) totals := make(map[string]int64) for rows.Next() { var platformID string var sourceType sql.NullString var sourceID sql.NullInt64 if scanErr := rows.Scan(&platformID, &sourceType, &sourceID); scanErr != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to parse content citations") } platformID = normalizeMonitoringPlatformID(platformID) if platformID == "" { continue } totals[platformID]++ normalizedSourceType := strings.TrimSpace(sourceType.String) if !sourceType.Valid || normalizedSourceType == "" || !sourceID.Valid || sourceID.Int64 <= 0 { continue } key := contentCitationKey{ PlatformID: platformID, SourceType: normalizedSourceType, SourceID: sourceID.Int64, } item := aggregates[key] if item.SourceID == 0 { item = MonitoringQuestionContentCitation{ SourceType: normalizedSourceType, SourceID: sourceID.Int64, AIPlatformID: platformID, PlatformName: platformDisplayName(platformID), } } item.CitationCount++ aggregates[key] = item } if err := rows.Err(); err != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate content citations") } if len(aggregates) == 0 { return []MonitoringQuestionContentCitation{}, nil } sourceRows := make([]monitoringCitedArticleSourceRow, 0, len(aggregates)) for key := range aggregates { sourceRows = append(sourceRows, monitoringCitedArticleSourceRow{ SourceType: key.SourceType, SourceID: key.SourceID, }) } metaByKey, err := s.loadCitedArticleSourceMeta(ctx, tenantID, sourceRows) if err != nil { return nil, err } items := make([]MonitoringQuestionContentCitation, 0, len(aggregates)) for key, item := range aggregates { if meta := metaByKey[monitoringContentSourceKey(key.SourceType, key.SourceID)]; meta.SourceID > 0 { item.ArticleID = meta.ArticleID item.ArticleTitle = meta.ArticleTitle item.PublishPlatform = meta.PublishPlatform item.SourceURL = meta.SourceURL } if strings.TrimSpace(item.ArticleTitle) == "" { item.ArticleTitle = "未命名文章" } if strings.TrimSpace(item.PublishPlatform) == "" { item.PublishPlatform = "外部文章" } item.CitationRate = divideAsPointer(item.CitationCount, totals[item.AIPlatformID]) items = append(items, item) } sort.Slice(items, func(left, right int) bool { if items[left].AIPlatformID != items[right].AIPlatformID { return monitoringPlatformSortIndex(items[left].AIPlatformID) < monitoringPlatformSortIndex(items[right].AIPlatformID) } if items[left].CitationCount != items[right].CitationCount { return items[left].CitationCount > items[right].CitationCount } if items[left].SourceType != items[right].SourceType { return items[left].SourceType < items[right].SourceType } return items[left].SourceID < items[right].SourceID }) return items, nil } func (s *MonitoringService) ensureClientOnline(ctx context.Context, tenantID, workspaceID int64, clientID uuid.UUID) error { online, err := s.isClientOnline(ctx, tenantID, workspaceID, clientID) if err != nil { return err } if !online { return response.ErrConflict(40941, "desktop_client_offline", "current user's monitoring desktop client is offline") } return nil } func (s *MonitoringService) isClientOnline(ctx context.Context, tenantID, workspaceID int64, clientID uuid.UUID) (bool, error) { var lastSeenAt sql.NullTime err := s.businessPool.QueryRow(ctx, ` SELECT last_seen_at FROM desktop_clients WHERE id = $1 AND tenant_id = $2 AND workspace_id = $3 AND revoked_at IS NULL `, clientID, tenantID, workspaceID).Scan(&lastSeenAt) if err != nil { if err == pgx.ErrNoRows { return false, nil } return false, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring desktop client") } if !lastSeenAt.Valid || time.Since(lastSeenAt.Time) > desktopClientPresenceTTL { return false, nil } return true, nil } func parseDetailDateRange(dateFrom, dateTo string) (time.Time, time.Time, error) { return parseDetailDateRangeAt(dateFrom, dateTo, time.Now(), monitoringBusinessLocation()) } func parseDetailDateRangeAt(dateFrom, dateTo string, now time.Time, loc *time.Location) (time.Time, time.Time, error) { if strings.TrimSpace(dateFrom) == "" && strings.TrimSpace(dateTo) == "" { today := monitoringBusinessDayAt(now, loc) return today, today, nil } if strings.TrimSpace(dateFrom) == "" || strings.TrimSpace(dateTo) == "" { return time.Time{}, time.Time{}, response.ErrBadRequest(40031, "invalid_date_range", "date_from and date_to must be provided together") } fromDate, err := parseMonitoringBusinessDateInLocation(dateFrom, loc) if err != nil { return time.Time{}, time.Time{}, response.ErrBadRequest(40031, "invalid_date_range", "date_from must use YYYY-MM-DD") } toDate, err := parseMonitoringBusinessDateInLocation(dateTo, loc) if err != nil { return time.Time{}, time.Time{}, response.ErrBadRequest(40031, "invalid_date_range", "date_to must use YYYY-MM-DD") } if fromDate.After(toDate) { return time.Time{}, time.Time{}, response.ErrBadRequest(40031, "invalid_date_range", "date_from must not be later than date_to") } return fromDate, toDate, nil } func resolveDashboardDateWindow(days int, businessDate string) (time.Time, time.Time, error) { return resolveDashboardDateWindowAt(days, businessDate, time.Now(), monitoringBusinessLocation()) } func resolveDashboardDateWindowAt(days int, businessDate string, now time.Time, loc *time.Location) (time.Time, time.Time, error) { today := monitoringBusinessDayAt(now, loc) earliestAllowed := today.AddDate(0, 0, -(maxTrackingDays - 1)) endDate := today if strings.TrimSpace(businessDate) != "" { parsedDate, err := parseMonitoringBusinessDateInLocation(businessDate, loc) if err != nil { return time.Time{}, time.Time{}, response.ErrBadRequest(40031, "invalid_business_date", "business_date must use YYYY-MM-DD") } if parsedDate.After(today) { return time.Time{}, time.Time{}, response.ErrBadRequest(40031, "invalid_business_date", "business_date must not be later than today") } if parsedDate.Before(earliestAllowed) { return time.Time{}, time.Time{}, response.ErrBadRequest(40031, "invalid_business_date", "business_date must be within the latest 30 days") } endDate = parsedDate } if days <= 0 { days = maxTrackingDays } days = normalizeTrackingDays(days) startDate := endDate.AddDate(0, 0, -(days - 1)) if startDate.Before(earliestAllowed) { startDate = earliestAllowed } return startDate, endDate, nil } func trackingDateWindow(days int) (time.Time, time.Time) { endDate := monitoringBusinessToday() startDate := endDate.AddDate(0, 0, -(days - 1)) return startDate, endDate } func monitoringBusinessToday() time.Time { return monitoringBusinessDayAt(time.Now(), monitoringBusinessLocation()) } func monitoringBusinessDayAt(now time.Time, loc *time.Location) time.Time { if loc == nil { loc = time.Local } localNow := now.In(loc) return time.Date(localNow.Year(), localNow.Month(), localNow.Day(), 0, 0, 0, 0, loc) } func parseMonitoringBusinessDateInLocation(value string, loc *time.Location) (time.Time, error) { if loc == nil { loc = time.Local } return time.ParseInLocation("2006-01-02", strings.TrimSpace(value), loc) } func monitoringRecentMetricsStart(startDate, endDate time.Time) time.Time { recentStart := endDate.AddDate(0, 0, -1) if startDate.After(recentStart) { return startDate } return recentStart } func normalizeTrackingDays(days int) int { if days <= 0 { return defaultTrackingDays } if days > maxTrackingDays { return maxTrackingDays } return days } func normalizeCitationWindowDays(days int) int { if days == maxTrackingDays { return maxTrackingDays } return defaultTrackingDays } func resolveMonitoringPlatforms(enabled []string) []monitoringPlatformMetadata { if len(enabled) == 0 { return nil } platforms := make([]monitoringPlatformMetadata, 0, len(enabled)) seen := make(map[string]struct{}, len(enabled)) for _, item := range enabled { id := normalizeMonitoringPlatformID(item) if id == "" { continue } if _, supported := monitoringPlatformNames[id]; !supported { continue } if _, ok := seen[id]; ok { continue } seen[id] = struct{}{} platforms = append(platforms, monitoringPlatformMetadata{ ID: id, Name: platformDisplayName(id), }) } if len(platforms) == 0 { return nil } return platforms } func intersectMonitoringPlatforms(enabled []monitoringPlatformMetadata, requested []string) ([]monitoringPlatformMetadata, error) { if len(requested) == 0 { return enabled, nil } enabledIndex := make(map[string]monitoringPlatformMetadata, len(enabled)) for _, platform := range enabled { enabledIndex[platform.ID] = platform } result := make([]monitoringPlatformMetadata, 0, len(requested)) seen := make(map[string]struct{}, len(requested)) for _, raw := range requested { id := normalizeMonitoringPlatformID(raw) if id == "" { continue } if _, duplicate := seen[id]; duplicate { continue } seen[id] = struct{}{} platform, ok := enabledIndex[id] if !ok { return nil, response.ErrBadRequest(40031, "platform_not_enabled", fmt.Sprintf("platform %s is not enabled for this tenant", id)) } result = append(result, platform) } if len(result) == 0 { return nil, response.ErrBadRequest(40031, "invalid_platform_ids", "platform_ids contains no usable platform") } return result, nil } func filterMonitoringPlatforms(platforms []monitoringPlatformMetadata, aiPlatformID *string) []monitoringPlatformMetadata { platformID := normalizedOptionalMonitoringPlatformID(aiPlatformID) if platformID == "" { return platforms } filtered := make([]monitoringPlatformMetadata, 0, 1) for _, platform := range platforms { if platform.ID == platformID { filtered = append(filtered, platform) return filtered } } return nil } func platformDisplayName(platformID string) string { platformID = normalizeMonitoringPlatformID(platformID) if name, ok := monitoringPlatformNames[platformID]; ok { return name } if platformID == "" { return "未知平台" } return strings.ToUpper(platformID) } func normalizedOptionalString(value *string) string { if value == nil { return "" } return strings.TrimSpace(*value) } func mergeWeightedNullFloat(left sql.NullFloat64, leftWeight int64, right sql.NullFloat64, rightWeight int64) sql.NullFloat64 { switch { case left.Valid && right.Valid && leftWeight > 0 && rightWeight > 0: totalWeight := leftWeight + rightWeight if totalWeight <= 0 { return sql.NullFloat64{} } return sql.NullFloat64{ Float64: ((left.Float64 * float64(leftWeight)) + (right.Float64 * float64(rightWeight))) / float64(totalWeight), Valid: true, } case right.Valid: return right default: return left } } func maxNullTime(left, right sql.NullTime) sql.NullTime { switch { case !left.Valid: return right case !right.Valid: return left case right.Time.After(left.Time): return right default: return left } } func mergeMonitoringSampleStatus(left, right string) string { priorities := map[string]int{ "sampled": 4, "partial": 3, "pending": 2, "not_logged_in": 1, "unavailable": 0, "unsampled": 0, } if priorities[strings.TrimSpace(right)] > priorities[strings.TrimSpace(left)] { return right } return left } func monitoringLatestPlatformRunIsNewer(candidate, current monitoringLatestPlatformRun) bool { switch { case candidate.CompletedAt.Valid && !current.CompletedAt.Valid: return true case !candidate.CompletedAt.Valid && current.CompletedAt.Valid: return false case candidate.CompletedAt.Valid && current.CompletedAt.Valid: if candidate.CompletedAt.Time.After(current.CompletedAt.Time) { return true } if candidate.CompletedAt.Time.Before(current.CompletedAt.Time) { return false } } return candidate.RunID > current.RunID } func derivePlatformSampleStatus(rawStatus string, accessState monitoringAccessState) string { switch strings.TrimSpace(rawStatus) { case "sampled": return "sampled" case "pending": return "pending" case "not_logged_in": return "not_logged_in" case "unavailable": return "unavailable" } switch accessState.AccessStatus { case "not_logged_in": return "not_logged_in" case "unavailable": return "unavailable" case "accessible": return "pending" default: return "pending" } } func divideAsPointer(numerator, denominator int64) *float64 { if denominator <= 0 { return nil } value := float64(numerator) / float64(denominator) return &value } func pointerFloat64ToNull(value *float64) sql.NullFloat64 { if value == nil { return sql.NullFloat64{} } return sql.NullFloat64{Float64: *value, Valid: true} } func pointerStringToNull(value *string) sql.NullString { if value == nil { return sql.NullString{} } return sql.NullString{String: *value, Valid: true} } func floatPointer(value sql.NullFloat64) *float64 { if !value.Valid { return nil } v := value.Float64 return &v } func formatNullTime(value sql.NullTime) *string { if !value.Valid { return nil } text := value.Time.UTC().Format(time.RFC3339) return &text } func stringPointer(value sql.NullString) *string { if !value.Valid { return nil } v := value.String return &v } func nullableStringValue(value sql.NullString) *string { if !value.Valid { return nil } v := value.String return &v } func boolPointer(value sql.NullBool) *bool { if !value.Valid { return nil } v := value.Bool return &v } func nullableInt64Value(value sql.NullInt64) *int64 { if !value.Valid { return nil } v := value.Int64 return &v } func nullableInt64(value *int64) interface{} { if value == nil { return nil } return *value } func nullableInt64Array(values []int64) interface{} { if values == nil { return nil } return values } func nullableStringArray(values []string) interface{} { if len(values) == 0 { return nil } return values } func nullableString(value *string) interface{} { if value == nil { return nil } trimmed := strings.TrimSpace(*value) if trimmed == "" { return nil } return trimmed } func encodeQuestionHash(hashBytes []byte) string { if len(hashBytes) == 0 { return "" } return "v1:" + hex.EncodeToString(hashBytes) } func decodeQuestionHash(value string) ([]byte, error) { trimmed := strings.TrimSpace(value) trimmed = strings.TrimPrefix(trimmed, "v1:") if trimmed == "" { return nil, fmt.Errorf("empty hash") } return hex.DecodeString(trimmed) } func seededQuestionHash(text string) []byte { digest := md5.Sum([]byte(strings.TrimSpace(strings.ToLower(text)))) return digest[:] } func faviconURL(siteKey string) string { siteKey = strings.TrimSpace(siteKey) if siteKey == "" { return "" } return fmt.Sprintf("https://%s/favicon.ico", siteKey) }