Files
geo/server/internal/tenant/app/monitoring_service.go
T
root 1eae6fb6d4
Deployment Config CI / Deployment Config (push) Successful in 29s
Frontend CI / Frontend (push) Successful in 4m4s
Backend CI / Backend (push) Failing after 7m12s
feat(questions): make brand questions the primary monitoring & template axis
- add /questions/combination-fill endpoint with AI-driven, IP-region-aware matrix fill
- extract ip2region resolver from ops/app into shared/ipregion for cross-service use
- thread question_id filter through dashboard composite, citation summary, and collect-now
- switch template wizard from keyword inputs to brand-question selection (primary + supplemental)
- pass brand_question and supplemental_questions through assist/title/outline prompts
- add AiWaitingModal + 45s client timeout and request_timeout error mapping for long AI flows

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-13 15:59:39 +08:00

4572 lines
149 KiB
Go

package app
import (
"context"
"crypto/md5"
crand "crypto/rand"
"database/sql"
"encoding/hex"
"encoding/json"
"fmt"
"math/big"
"sort"
"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
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"`
CitationRanking []MonitoringCitationRanking `json:"citation_ranking"`
CitedArticles []MonitoringCitedArticle `json:"cited_articles"`
}
type MonitoringCitationSummaryResponse struct {
CitationWindow MonitoringDashboardWindow `json:"citation_window"`
CitationRanking []MonitoringCitationRanking `json:"citation_ranking"`
CitedArticles []MonitoringCitedArticle `json:"cited_articles"`
}
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"`
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"`
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"`
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 monitoringPlatformMetadata struct {
ID string
Name string
}
type monitoringConfiguredQuestion struct {
ID int64
KeywordID int64
QuestionText string
QuestionHash []byte
}
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,
) (*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, err := s.loadHotQuestions(ctx, actor.TenantID, brand.ID, configuredQuestions, endDate, selectedPlatformID, derivedMetrics)
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,
CitationRanking: []MonitoringCitationRanking{},
CitedArticles: []MonitoringCitedArticle{},
}, nil
}
func (s *MonitoringService) CitationSummary(
ctx context.Context,
days int,
brandID int64,
keywordID *int64,
questionID *int64,
businessDate string,
aiPlatformID *string,
) (*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, err := s.loadCitedArticles(ctx, actor.TenantID, brandID, questionIDs, startDate, endDate, selectedPlatformID)
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,
}, 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, interruptTaskIDs, 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(len(interruptTaskIDs)), 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,
func() []int64 {
if !options.Preempt {
return nil
}
return collectNowInterruptTaskIDs(interruptTaskIDs)
}(),
); 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, fallbackLegacyTaskIDs, dispatchErr := s.dispatchMonitorDesktopTasks(ctx, phase2TaskSpecs)
if dispatchErr != nil {
if fallbackErr := s.fallbackMonitorDesktopTasksToLegacy(ctx, monitorDesktopTaskSpecIDs(phase2TaskSpecs)); fallbackErr != nil {
return nil, fallbackErr
}
phase2DispatchReady = false
phase2TaskSpecs = nil
if s.logger != nil {
s.logger.Warn("phase2 monitor desktop dispatch failed, falling back to legacy execution",
zap.Error(dispatchErr),
zap.Int("phase2_task_count", phase2TaskCount),
)
}
} else {
if len(fallbackLegacyTaskIDs) > 0 {
if fallbackErr := s.fallbackMonitorDesktopTasksToLegacy(ctx, fallbackLegacyTaskIDs); fallbackErr != nil {
return nil, fallbackErr
}
}
phase2PublishedTasks = publishedTasks
}
}
phase2InterruptTargets := make([]monitorDesktopInterruptTarget, 0)
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...)
interruptTargets, interruptErr := s.requestPhase2MonitorInterrupts(ctx, phase2TargetClientID, excludedMonitorTaskIDs, interruptGeneration)
if interruptErr != nil {
return nil, interruptErr
}
phase2InterruptTargets = append(phase2InterruptTargets, interruptTargets...)
}
if len(phase2DeferredTasks) > 0 {
abortedQueuedCount += int64(len(phase2DeferredTasks))
}
if len(phase2InterruptTargets) > 0 {
// Keep collect-now request counters in sync after per-platform dispatch
// fan-out chooses more than one desktop client.
if _, updateErr := s.monitoringPool.Exec(ctx, `
UPDATE monitoring_collect_requests
SET aborted_queued_count = aborted_queued_count + $2,
interrupt_requested_count = interrupt_requested_count + $3,
updated_at = NOW()
WHERE request_id = $1
`, requestID, len(phase2DeferredTasks), len(phase2InterruptTargets)); updateErr != nil {
return nil, response.ErrInternal(50041, "request_update_failed", "failed to persist phase2 preempt counts")
}
} else if len(phase2DeferredTasks) > 0 {
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),
)
}
}
for _, target := range phase2InterruptTargets {
publishPhase2MonitorControlEvent(context.Background(), s.rabbitMQ, s.logger, target)
}
hasEffectiveSnapshot := leasedCount > 0 || completedCount > 0
totalInterruptRequestedCount := int64(len(interruptTaskIDs) + len(phase2InterruptTargets))
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: totalInterruptRequestedCount,
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.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 (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 {
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, question.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, []int64, error) {
const runMode = "plugin_standard"
if strings.TrimSpace(executionOwner) == "" {
executionOwner = "legacy"
}
dateText := businessDate.Format("2006-01-02")
var refreshedCount int64
var createdCount int64
interruptTaskIDs := make([]int64, 0)
for _, question := range questions {
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,
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)
if err != nil {
return 0, 0, 0, 0, nil, response.ErrInternal(50041, "update_failed", "failed to refresh monitoring tasks")
}
refreshedCount += tag.RowsAffected()
rows, err := tx.Query(ctx, `
UPDATE monitoring_collect_tasks
SET question_hash = $7,
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'
RETURNING id
`, tenantID, brandID, question.ID, platformID, monitoringCollectorType, runMode, question.QuestionHash, dateText, platformTargetClientID, interruptGeneration)
if err != nil {
return 0, 0, 0, 0, nil, response.ErrInternal(50041, "update_failed", "failed to promote leased monitoring tasks")
}
for rows.Next() {
var taskID int64
if scanErr := rows.Scan(&taskID); scanErr != nil {
rows.Close()
return 0, 0, 0, 0, nil, response.ErrInternal(50041, "scan_failed", "failed to parse promoted leased monitoring task")
}
interruptTaskIDs = append(interruptTaskIDs, taskID)
}
if err := rows.Err(); err != nil {
rows.Close()
return 0, 0, 0, 0, nil, response.ErrInternal(50041, "scan_failed", "failed to iterate promoted leased monitoring tasks")
}
rows.Close()
tag, err = tx.Exec(ctx, `
INSERT INTO monitoring_collect_tasks (
tenant_id, brand_id, question_id, question_hash, 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, $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)
if err != nil {
return 0, 0, 0, 0, nil, 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, nil, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring tasks")
}
return refreshedCount, createdCount, leasedCount, completedCount, interruptTaskIDs, 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 collectNowInterruptTaskIDs(taskIDs []int64) []int64 {
if len(taskIDs) == 0 {
return nil
}
seen := make(map[int64]struct{}, len(taskIDs))
result := make([]int64, 0, len(taskIDs))
for _, taskID := range taskIDs {
if taskID <= 0 {
continue
}
if _, ok := seen[taskID]; ok {
continue
}
seen[taskID] = struct{}{}
result = append(result, taskID)
}
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.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 (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.business_date,
r.ai_platform_id,
r.question_id,
r.raw_answer_text,
p.brand_mentioned,
p.brand_mention_position,
p.first_recommended,
p.sentiment_label,
p.matched_brand_terms
FROM question_monitor_runs r
LEFT JOIN question_monitor_parse_results p
ON p.tenant_id = r.tenant_id AND p.run_id = r.id
WHERE r.tenant_id = $1
AND r.brand_id = $2
AND r.collector_type = $3
AND r.status = 'succeeded'
AND r.business_date BETWEEN $4::date AND $5::date
AND ($6::bigint[] IS NULL OR r.question_id = ANY($6))
AND ($7::text[] IS NULL OR r.ai_platform_id = ANY($7))
`, tenantID, brandID, monitoringCollectorType, startDate.Format("2006-01-02"), endDate.Format("2006-01-02"), nullableInt64Array(questionIDs), nullableStringArray(platformQueryIDs))
if err != nil {
return metrics, response.ErrInternal(50041, "query_failed", "failed to load monitoring derived parse metrics")
}
defer rows.Close()
for rows.Next() {
var businessDate time.Time
var platformID string
var questionID int64
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(
&businessDate,
&platformID,
&questionID,
&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),
},
)
dateKey := businessDate.Format("2006-01-02")
dateStats := metrics.ByDate[dateKey]
dateStats.add(summary)
metrics.ByDate[dateKey] = dateStats
platformStatsByDate, ok := metrics.ByDatePlatform[dateKey]
if !ok {
platformStatsByDate = make(map[string]monitoringDerivedRateStats)
metrics.ByDatePlatform[dateKey] = platformStatsByDate
}
platformStats := platformStatsByDate[platformID]
platformStats.add(summary)
platformStatsByDate[platformID] = platformStats
questionStatsByDate, ok := metrics.ByDateQuestion[dateKey]
if !ok {
questionStatsByDate = make(map[int64]monitoringDerivedRateStats)
metrics.ByDateQuestion[dateKey] = questionStatsByDate
}
questionStatsForDate := questionStatsByDate[questionID]
questionStatsForDate.add(summary)
questionStatsByDate[questionID] = questionStatsForDate
questionStats := metrics.ByQuestion[questionID]
questionStats.add(summary)
metrics.ByQuestion[questionID] = questionStats
}
if err := rows.Err(); err != nil {
return metrics, response.ErrInternal(50041, "scan_failed", "failed to iterate monitoring derived metrics")
}
return metrics, 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.sentiment_label = 'positive')::bigint AS positive_mentioned_count,
MAX(r.completed_at) AS last_sampled_at
FROM question_monitor_runs r
LEFT JOIN question_monitor_parse_results p
ON p.tenant_id = r.tenant_id AND p.run_id = r.id
WHERE r.tenant_id = $1
AND r.brand_id = $2
AND r.collector_type = $3
AND r.status = 'succeeded'
AND r.business_date BETWEEN $4::date AND $5::date
AND r.question_id = ANY($6)
AND ($7::text[] IS NULL OR r.ai_platform_id = ANY($7))
GROUP BY r.business_date
ORDER BY r.business_date DESC
`, tenantID, brandID, monitoringCollectorType, startDate.Format("2006-01-02"), endDate.Format("2006-01-02"), questionIDs, nullableStringArray(platformIDs))
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 question_monitor_runs r
LEFT JOIN question_monitor_parse_results p
ON p.tenant_id = r.tenant_id AND p.run_id = r.id
WHERE r.tenant_id = $1
AND r.brand_id = $2
AND r.collector_type = $3
AND r.business_date = $4::date
AND r.status = 'succeeded'
AND r.question_id = ANY($5)
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())
}
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 stats AS (
SELECT
r.question_id,
AVG(CASE WHEN p.brand_mentioned IS TRUE THEN 1 ELSE 0 END)::double precision AS mention_rate,
MAX(r.completed_at) AS last_sampled_at
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))
GROUP BY r.question_id
),
latest_hash AS (
SELECT DISTINCT ON (r.question_id)
r.question_id,
r.question_hash
FROM question_monitor_runs r
WHERE r.tenant_id = $1
AND r.brand_id = $2
AND r.collector_type = $3
AND r.status = 'succeeded'
AND r.business_date = $4::date
AND r.question_id = ANY($5)
AND ($6::text[] IS NULL OR r.ai_platform_id = ANY($6))
ORDER BY r.question_id, r.business_date DESC, COALESCE(r.completed_at, r.updated_at, r.created_at) DESC NULLS LAST, r.id DESC
)
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) 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)
aliasIndex, err := loadMonitoringPublishedAliasIndex(ctx, s.monitoringPool, tenantID)
if err != nil {
return nil, err
}
type citationRankingAggregate struct {
CitedAnswerCount int64
CitedArticleCount int64
CitationSourceCount int64
SaaSSourceCount int64
SampleCount int64
}
aggregates := make(map[string]citationRankingAggregate)
rows, err := s.monitoringPool.Query(ctx, `
SELECT
r.ai_platform_id,
COUNT(*) AS sample_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))
GROUP BY r.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
if scanErr := rows.Scan(&platformID, &sampleCount); 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
aggregates[platformID] = item
}
if err := rows.Err(); err != nil {
return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate citation ranking")
}
citedRuns := make(map[string]map[int64]struct{})
citedArticles := make(map[string]map[int64]struct{})
rows, err = s.monitoringPool.Query(ctx, `
SELECT
r.ai_platform_id,
cf.run_id,
cf.cited_url,
cf.normalized_url
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 ($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, 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 runID int64
var citedURL string
var normalizedURL string
if scanErr := rows.Scan(&platformID, &runID, &citedURL, &normalizedURL); scanErr != nil {
return nil, response.ErrInternal(50041, "scan_failed", "failed to parse citation ranking")
}
platformID = normalizeMonitoringPlatformID(platformID)
if platformID == "" {
continue
}
item := aggregates[platformID]
item.CitationSourceCount++
if platformID != "qwen" {
if alias, ok := aliasIndex.Match(citedURL, normalizedURL); ok {
item.SaaSSourceCount++
if _, exists := citedRuns[platformID]; !exists {
citedRuns[platformID] = make(map[int64]struct{})
}
citedRuns[platformID][runID] = struct{}{}
if _, exists := citedArticles[platformID]; !exists {
citedArticles[platformID] = make(map[int64]struct{})
}
citedArticles[platformID][alias.ArticleID] = struct{}{}
}
}
aggregates[platformID] = item
}
if err := rows.Err(); err != nil {
return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate citation ranking")
}
for platformID, item := range aggregates {
item.CitedAnswerCount = int64(len(citedRuns[platformID]))
item.CitedArticleCount = int64(len(citedArticles[platformID]))
aggregates[platformID] = item
}
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) {
if questionIDs != nil && len(questionIDs) == 0 {
return []MonitoringCitedArticle{}, nil
}
platformQueryIDs := monitoringPlatformQueryIDs(aiPlatformID)
aliasIndex, err := loadMonitoringPublishedAliasIndex(ctx, s.monitoringPool, tenantID)
if err != nil {
return nil, err
}
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, startDate.Format("2006-01-02"), endDate.Format("2006-01-02"), nullableInt64Array(questionIDs), nullableStringArray(platformQueryIDs)).Scan(&totalSampleCount); err != nil {
return nil, response.ErrInternal(50041, "query_failed", "failed to count cited article denominator")
}
rows, err := s.monitoringPool.Query(ctx, `
SELECT
cf.cited_url,
cf.normalized_url
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 r.ai_platform_id <> 'qwen'
AND r.collector_type = $3
AND r.status = 'succeeded'
AND r.business_date BETWEEN $4::date AND $5::date
AND ($6::bigint[] IS NULL OR r.question_id = ANY($6))
AND ($7::text[] IS NULL OR r.ai_platform_id = ANY($7))
`, tenantID, brandID, monitoringCollectorType, startDate.Format("2006-01-02"), endDate.Format("2006-01-02"), nullableInt64Array(questionIDs), nullableStringArray(platformQueryIDs))
if err != nil {
return nil, response.ErrInternal(50041, "query_failed", "failed to load cited articles")
}
defer rows.Close()
type citedArticleAggregate struct {
ArticleID int64
ArticleTitle string
PublishPlatform string
CitationCount int64
}
aggregates := make(map[int64]citedArticleAggregate)
var totalArticleCitationCount int64
for rows.Next() {
var citedURL string
var normalizedURL string
if scanErr := rows.Scan(&citedURL, &normalizedURL); scanErr != nil {
return nil, response.ErrInternal(50041, "scan_failed", "failed to parse cited articles")
}
alias, ok := aliasIndex.Match(citedURL, normalizedURL)
if !ok {
continue
}
item := aggregates[alias.ArticleID]
if item.ArticleID == 0 {
item.ArticleID = alias.ArticleID
item.ArticleTitle = alias.ArticleTitle
item.PublishPlatform = alias.PublishPlatform
}
item.CitationCount++
totalArticleCitationCount++
aggregates[alias.ArticleID] = item
}
if err := rows.Err(); err != nil {
return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate cited articles")
}
items := make([]MonitoringCitedArticle, 0, len(aggregates))
for _, item := range aggregates {
items = append(items, MonitoringCitedArticle{
ArticleID: item.ArticleID,
ArticleTitle: item.ArticleTitle,
PublishPlatform: item.PublishPlatform,
CitationCount: item.CitationCount,
CitationRate: divideAsPointer(item.CitationCount, totalSampleCount),
SourceShare: divideAsPointer(item.CitationCount, totalArticleCitationCount),
})
}
sort.Slice(items, func(left, right int) bool {
if items[left].CitationCount != items[right].CitationCount {
return items[left].CitationCount > items[right].CitationCount
}
return items[left].ArticleID < items[right].ArticleID
})
if len(items) > 10 {
items = items[:10]
}
return items, nil
}
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, 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
if alias, ok := aliasIndex.Match(citedURL, normalizedURL); ok {
articleID = &alias.ArticleID
title := repairMonitoringMojibake(alias.ArticleTitle)
articleTitle = &title
resolutionStatus = "resolved"
resolutionConfidence = "high"
}
result[runID] = append(result[runID], MonitoringQuestionDetailCitation{
CitedURL: citedURL,
CitedTitle: nullableStringValue(citedTitle),
SiteName: siteName,
SiteKey: siteKey,
FaviconURL: faviconURL(siteKey),
ArticleID: articleID,
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
}
aliasIndex, err := loadMonitoringPublishedAliasIndex(ctx, s.monitoringPool, tenantID)
if err != nil {
return nil, err
}
rows, err := s.monitoringPool.Query(ctx, `
WITH filtered_facts AS (
SELECT
r.ai_platform_id,
cf.id,
cf.cited_url,
cf.normalized_url,
cf.host,
cf.registrable_domain,
cf.site_key
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,
ff.cited_url,
ff.normalized_url,
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
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 citedURL string
var normalizedURL string
var siteName string
var siteKey string
var siteDomain string
if scanErr := rows.Scan(&platformID, &citedURL, &normalizedURL, &siteName, &siteKey, &siteDomain); 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 platformID != "qwen" {
if _, ok := aliasIndex.Match(citedURL, normalizedURL); ok {
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
}
aliasIndex, err := loadMonitoringPublishedAliasIndex(ctx, s.monitoringPool, tenantID)
if err != nil {
return nil, err
}
rows, err := s.monitoringPool.Query(ctx, `
SELECT
r.ai_platform_id,
cf.cited_url,
cf.normalized_url
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.ai_platform_id <> 'qwen'
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
ArticleID int64
}
aggregates := make(map[contentCitationKey]MonitoringQuestionContentCitation)
totals := make(map[string]int64)
for rows.Next() {
var platformID string
var citedURL string
var normalizedURL string
if scanErr := rows.Scan(&platformID, &citedURL, &normalizedURL); scanErr != nil {
return nil, response.ErrInternal(50041, "scan_failed", "failed to parse content citations")
}
platformID = normalizeMonitoringPlatformID(platformID)
if platformID == "" {
continue
}
totals[platformID]++
alias, ok := aliasIndex.Match(citedURL, normalizedURL)
if !ok {
continue
}
key := contentCitationKey{
PlatformID: platformID,
ArticleID: alias.ArticleID,
}
item := aggregates[key]
if item.ArticleID == 0 {
item = MonitoringQuestionContentCitation{
ArticleID: alias.ArticleID,
ArticleTitle: alias.ArticleTitle,
PublishPlatform: alias.PublishPlatform,
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")
}
items := make([]MonitoringQuestionContentCitation, 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
}
return items[left].ArticleID < items[right].ArticleID
})
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)
}