Files
geo/server/internal/tenant/app/monitoring_service.go
T
root 7605890a14 refactor(server/monitoring): route monitor desktop tasks by account presence
Daily monitoring tasks were pinned to a single primary client. If that
client went offline the task wedged even when another desktop client of
the same workspace was online and bound to the same account. Drop the
primary-client constraint on the materialized collect rows, and instead
pick a target per-platform from live account/client presence in Redis,
falling back to the DB client_id when the desktop client is still flagged
online. Desktop task lease for kind=monitor now matches by account
ownership in addition to target_client_id, so any client owning the
account can drain the queue.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-28 09:14:24 +08:00

3598 lines
118 KiB
Go

package app
import (
"context"
"crypto/md5"
"database/sql"
"encoding/hex"
"encoding/json"
"fmt"
"sort"
"strings"
"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
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
}
type MonitoringDashboardCompositeResponse struct {
Overview MonitoringOverview `json:"overview"`
Runtime MonitoringDashboardRuntime `json:"runtime"`
PlatformBreakdown []MonitoringPlatformDaily `json:"platform_breakdown"`
HotQuestions []MonitoringHotQuestion `json:"hot_questions"`
CitationRanking []MonitoringCitationRanking `json:"citation_ranking"`
CitedArticles []MonitoringCitedArticle `json:"cited_articles"`
}
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"`
}
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"`
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"`
}
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,
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
}
questionIDs := configuredQuestionIDs(configuredQuestions)
platforms := defaultMonitoringPlatformMetadata()
selectedPlatformID := normalizedOptionalMonitoringPlatformPointer(aiPlatformID)
filteredPlatforms := filterMonitoringPlatforms(platforms, selectedPlatformID)
startDate, endDate, err := resolveDashboardDateWindow(days, businessDate)
if err != nil {
return nil, err
}
derivedMetrics, err := s.loadDerivedParseMetrics(ctx, actor.TenantID, brand.ID, brand.Name, startDate, endDate, selectedPlatformID)
if err != nil {
return nil, err
}
overview, _, err := s.loadOverview(ctx, actor.TenantID, brand.ID, 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, 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
}
citationRanking, err := s.loadCitationRanking(ctx, actor.TenantID, brand.ID, questionIDs, startDate, endDate, accessStates, selectedPlatformID)
if err != nil {
return nil, err
}
citedArticles, err := s.loadCitedArticles(ctx, actor.TenantID, brand.ID, questionIDs, startDate, endDate, selectedPlatformID)
if err != nil {
return nil, err
}
return &MonitoringDashboardCompositeResponse{
Overview: overview,
Runtime: runtime,
PlatformBreakdown: platformBreakdown,
HotQuestions: hotQuestions,
CitationRanking: citationRanking,
CitedArticles: citedArticles,
}, nil
}
func (s *MonitoringService) loadDashboardRuntime(ctx context.Context, actor auth.Actor, workspaceID int64, quota monitoringQuotaConfig) (MonitoringDashboardRuntime, error) {
online := false
if quota.PrimaryClientID != nil {
var err error
online, err = s.isClientOnline(ctx, actor.TenantID, workspaceID, *quota.PrimaryClientID)
if err != nil {
return MonitoringDashboardRuntime{}, err
}
}
return MonitoringDashboardRuntime{
CurrentUserClientOnline: online,
PlatformAuthorizationStatus: monitoringPlatformAuthorizationStatus(quota),
AuthorizedMonitoringPlatformIDs: append([]string(nil), quota.EnabledPlatforms...),
AuthorizedMonitoringPlatformCount: len(quota.EnabledPlatforms),
}, 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, brand.ID, questionID, hashBytes, fromDate, toDate, selectedPlatformID)
if err != nil {
return nil, err
}
contentCitations, err := s.loadQuestionContentCitations(ctx, actor.TenantID, brand.ID, questionID, hashBytes, fromDate, toDate, selectedPlatformID)
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,
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
}
limitApplied := false
if len(configuredQuestions) > monitoringCollectNowQuestionLimit {
configuredQuestions = configuredQuestions[:monitoringCollectNowQuestionLimit]
limitApplied = true
}
if len(configuredQuestions) == 0 {
return &MonitoringCollectNowResponse{
CollectionMode: quota.CollectionMode,
RefreshedTaskCount: 0,
CreatedTaskCount: 0,
LeasedTaskCount: 0,
CompletedTaskCount: 0,
HasEffectiveSnapshot: true,
Message: "当前关键词下暂无品牌库问题,未创建采集任务",
}, nil
}
clientID, err := s.resolveCollectNowClientID(ctx, actor.TenantID, workspaceID, actor.UserID, quota.PrimaryClientID, options.TargetClientID)
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")
}
if err := s.ensureClientOnline(ctx, actor.TenantID, workspaceID, *clientID); err != nil {
return nil, err
}
targetPlatformIDs, err := s.loadMonitoringClientPlatformIDs(ctx, actor.TenantID, workspaceID, *clientID)
if err != nil {
return nil, err
}
platforms := resolveMonitoringPlatforms(targetPlatformIDs)
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)
executionOwner := "legacy"
if s.shouldUseDesktopTasksExecution(actor.TenantID) {
executionOwner = "desktop_tasks"
}
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 {
if strings.TrimSpace(existingRequest.TargetClientID) == clientID.String() {
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); err != nil {
return nil, err
}
interruptGeneration, err := s.nextCollectNowInterruptGeneration(ctx, tx, *clientID, executionOwner)
if err != nil {
return nil, err
}
refreshedCount, createdCount, leasedCount, completedCount, interruptTaskIDs, err := s.ensureCollectNowTasks(
ctx,
tx,
actor.TenantID,
brand.ID,
configuredQuestions,
platforms,
todayTime,
*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,
)
if err != nil {
return nil, err
}
}
abortedQueuedCount := int64(0)
if options.Preempt {
abortedQueuedCount, err = s.deferQueuedNormalTasks(
ctx,
tx,
actor.TenantID,
todayTime,
brand.ID,
configuredQuestionIDs(configuredQuestions),
monitoringPlatformIDs(platforms),
*clientID,
requestID,
ttlExpiresAt,
)
if err != nil {
return nil, err
}
}
requestScopeJSON, err := json.Marshal(map[string]any{
"brand_id": brand.ID,
"keyword_id": keywordID,
"question_ids": questionIDs,
"platform_ids": platformIDs,
"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,
*clientID,
affectedTaskCount,
interruptGeneration,
func() []int64 {
if !options.Preempt {
return nil
}
return 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)
}
phase2DeferredTasks, err = s.deferQueuedPhase2MonitorTasks(ctx, *clientID, excludedMonitorTaskIDs)
if err != nil {
return nil, err
}
if len(phase2DeferredTasks) > 0 {
abortedQueuedCount += int64(len(phase2DeferredTasks))
if _, updateErr := s.monitoringPool.Exec(ctx, `
UPDATE monitoring_collect_requests
SET aborted_queued_count = aborted_queued_count + $2,
updated_at = NOW()
WHERE request_id = $1
`, requestID, len(phase2DeferredTasks)); updateErr != nil {
return nil, response.ErrInternal(50041, "request_update_failed", "failed to persist phase2 deferred queue count")
}
}
phase2InterruptTargets, err = s.requestPhase2MonitorInterrupts(ctx, *clientID, excludedMonitorTaskIDs, interruptGeneration)
if err != nil {
return nil, err
}
if len(phase2InterruptTargets) > 0 {
if _, updateErr := s.monitoringPool.Exec(ctx, `
UPDATE monitoring_collect_requests
SET interrupt_requested_count = interrupt_requested_count + $2,
updated_at = NOW()
WHERE request_id = $1
`, requestID, len(phase2InterruptTargets)); updateErr != nil {
return nil, response.ErrInternal(50041, "request_update_failed", "failed to persist phase2 interrupt request 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,
) (*uuid.UUID, error) {
if preferredClientID != nil {
if err := s.ensureClientBelongsToUser(ctx, tenantID, workspaceID, userID, *preferredClientID); err != nil {
return nil, err
}
return preferredClientID, nil
}
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) ensureClientBelongsToUser(
ctx context.Context,
tenantID, workspaceID, userID int64,
clientID uuid.UUID,
) 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 user_id = $4
AND revoked_at IS NULL
AND last_seen_at IS NOT NULL
AND last_seen_at >= NOW() - $5::interval
)
`, clientID, tenantID, workspaceID, userID, formatPgInterval(desktopClientPresenceTTL)).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,
clientID uuid.UUID,
executionOwner string,
) (int, error) {
var next int
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 = $1
`, clientID).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 = $1
`, clientID).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 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 (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,
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 {
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, platform.ID, monitoringCollectorType, runMode, question.QuestionHash, dateText, targetClientID, 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, platform.ID, monitoringCollectorType, runMode, question.QuestionHash, dateText, targetClientID, 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, platform.ID, monitoringCollectorType, runMode, dateText, targetClientID, 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.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.loadMonitoringClientPlatformIDs(ctx, actor.TenantID, workspaceID, *clientID)
if err != nil {
return cfg, err
}
cfg.EnabledPlatforms = platforms
return cfg, 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) 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 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,
brandName string,
startDate, endDate time.Time,
aiPlatformID *string,
) (monitoringDerivedMetrics, error) {
metrics := newMonitoringDerivedMetrics()
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::text[] IS NULL OR r.ai_platform_id = ANY($6))
`, tenantID, brandID, monitoringCollectorType, startDate.Format("2006-01-02"), endDate.Format("2006-01-02"), 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,
startDate, endDate time.Time,
collectionMode string,
aiPlatformID *string,
derived monitoringDerivedMetrics,
) (MonitoringOverview, time.Time, error) {
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) 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) 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, businessDate time.Time, collectionMode string, platforms []monitoringPlatformMetadata, accessStates map[string]monitoringAccessState, derived monitoringDerivedMetrics) ([]MonitoringPlatformDaily, error) {
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) 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)
rows, err := s.monitoringPool.Query(ctx, `
WITH run_counts AS (
SELECT
r.ai_platform_id,
COUNT(*) AS sample_count
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 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
),
citation_counts AS (
SELECT
r.ai_platform_id,
COUNT(DISTINCT cf.run_id) FILTER (WHERE cf.article_id IS NOT NULL AND r.ai_platform_id <> 'qwen') AS cited_answer_count,
COUNT(DISTINCT cf.article_id) FILTER (WHERE cf.article_id IS NOT NULL AND r.ai_platform_id <> 'qwen') AS cited_article_count
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.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
)
SELECT
rc.ai_platform_id,
rc.sample_count,
COALESCE(cc.cited_answer_count, 0) AS cited_answer_count,
COALESCE(cc.cited_article_count, 0) AS cited_article_count,
CASE
WHEN rc.sample_count > 0 THEN COALESCE(cc.cited_answer_count, 0)::double precision / rc.sample_count
ELSE NULL
END AS citation_rate
FROM run_counts rc
LEFT JOIN citation_counts cc ON cc.ai_platform_id = rc.ai_platform_id
WHERE COALESCE(cc.cited_answer_count, 0) > 0
ORDER BY cited_answer_count DESC, cited_article_count DESC, rc.ai_platform_id ASC
`, 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()
type citationRankingAggregate struct {
CitedAnswerCount int64
CitedArticleCount int64
SampleCount int64
}
aggregates := make(map[string]citationRankingAggregate)
for rows.Next() {
var platformID string
var sampleCount int64
var citedAnswerCount int64
var citedArticleCount int64
if scanErr := rows.Scan(&platformID, &sampleCount, &citedAnswerCount, &citedArticleCount, new(sql.NullFloat64)); scanErr != nil {
return nil, response.ErrInternal(50041, "scan_failed", "failed to parse citation ranking")
}
platformID = normalizeMonitoringPlatformID(platformID)
if platformID == "" {
continue
}
item := aggregates[platformID]
item.CitedAnswerCount += citedAnswerCount
item.CitedArticleCount += citedArticleCount
item.SampleCount += sampleCount
aggregates[platformID] = item
}
items := make([]MonitoringCitationRanking, 0, len(aggregates))
for platformID, aggregate := range aggregates {
items = append(items, MonitoringCitationRanking{
AIPlatformID: platformID,
PlatformName: platformDisplayName(platformID),
SampleStatus: derivePlatformSampleStatus("", accessStates[platformID]),
CitedAnswerCount: aggregate.CitedAnswerCount,
CitedArticleCount: aggregate.CitedArticleCount,
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].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)
var totalSampleCount int64
if err := s.monitoringPool.QueryRow(ctx, `
SELECT COUNT(*)
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 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.article_id,
COALESCE(MAX(alias.article_title_snapshot), '未命名文章') AS article_title,
COALESCE(MAX(alias.publish_platform_name_snapshot), '已发布内容') AS publish_platform,
COUNT(*) AS citation_count
FROM monitoring_citation_facts cf
JOIN question_monitor_runs r
ON r.tenant_id = cf.tenant_id AND r.id = cf.run_id
LEFT JOIN monitoring_article_url_aliases alias
ON alias.tenant_id = cf.tenant_id AND alias.article_id = cf.article_id
WHERE cf.tenant_id = $1
AND cf.brand_id = $2
AND cf.article_id IS NOT NULL
AND r.ai_platform_id <> 'qwen'
AND r.collector_type = $3
AND r.status = 'succeeded'
AND r.business_date BETWEEN $4::date AND $5::date
AND ($6::bigint[] IS NULL OR r.question_id = ANY($6))
AND ($7::text[] IS NULL OR r.ai_platform_id = ANY($7))
GROUP BY cf.article_id
ORDER BY citation_count DESC, cf.article_id ASC
LIMIT 10
`, 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()
items := make([]MonitoringCitedArticle, 0)
for rows.Next() {
var articleID int64
var title string
var publishPlatform string
var citationCount int64
if scanErr := rows.Scan(&articleID, &title, &publishPlatform, &citationCount); scanErr != nil {
return nil, response.ErrInternal(50041, "scan_failed", "failed to parse cited articles")
}
items = append(items, MonitoringCitedArticle{
ArticleID: articleID,
ArticleTitle: title,
PublishPlatform: publishPlatform,
CitationCount: citationCount,
CitationRate: divideAsPointer(citationCount, totalSampleCount),
})
}
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
}
rows, err := s.monitoringPool.Query(ctx, `
SELECT
cf.run_id,
cf.cited_url,
cf.cited_title,
COALESCE(tm.site_name, gm.site_name, cf.site_key, cf.registrable_domain) AS site_name,
COALESCE(tm.site_key, gm.site_key, cf.site_key) AS site_key,
cf.article_id,
alias.article_title_snapshot AS article_title,
cf.resolution_status,
cf.resolution_confidence
FROM monitoring_citation_facts cf
LEFT JOIN LATERAL (
SELECT site_key, site_name
FROM site_domain_mappings
WHERE tenant_id = cf.tenant_id
AND is_active = TRUE
AND (
registrable_domain = cf.registrable_domain
OR registrable_domain = cf.site_key
OR cf.host = registrable_domain
OR cf.host LIKE '%.' || registrable_domain
)
ORDER BY LENGTH(registrable_domain) DESC, id DESC
LIMIT 1
) tm ON TRUE
LEFT JOIN LATERAL (
SELECT site_key, site_name
FROM site_domain_mappings
WHERE tenant_id IS NULL
AND is_active = TRUE
AND (
registrable_domain = cf.registrable_domain
OR registrable_domain = cf.site_key
OR cf.host = registrable_domain
OR cf.host LIKE '%.' || registrable_domain
)
ORDER BY LENGTH(registrable_domain) DESC, id DESC
LIMIT 1
) gm ON tm.site_key IS NULL
LEFT JOIN monitoring_article_url_aliases alias
ON alias.tenant_id = cf.tenant_id AND alias.article_id = cf.article_id
WHERE cf.tenant_id = $1
AND cf.run_id = ANY($2)
ORDER BY cf.run_id ASC, cf.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 siteName string
var siteKey string
var articleID sql.NullInt64
var articleTitle sql.NullString
var resolutionStatus string
var resolutionConfidence string
if scanErr := rows.Scan(&runID, &citedURL, &citedTitle, &siteName, &siteKey, &articleID, &articleTitle, &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)
}
if articleTitle.Valid {
articleTitle.String = repairMonitoringMojibake(articleTitle.String)
}
result[runID] = append(result[runID], MonitoringQuestionDetailCitation{
CitedURL: citedURL,
CitedTitle: nullableStringValue(citedTitle),
SiteName: siteName,
SiteKey: siteKey,
FaviconURL: faviconURL(siteKey),
ArticleID: nullableInt64Value(articleID),
ArticleTitle: nullableStringValue(articleTitle),
ResolutionStatus: resolutionStatus,
ResolutionConfidence: resolutionConfidence,
})
}
return result, nil
}
func (s *MonitoringService) loadQuestionCitationAnalysis(ctx context.Context, tenantID, brandID, questionID int64, hashBytes []byte, fromDate, toDate time.Time, aiPlatformID *string) ([]MonitoringQuestionCitationStats, error) {
platformQueryIDs := monitoringPlatformQueryIDs(aiPlatformID)
rows, err := s.monitoringPool.Query(ctx, `
WITH grouped AS (
SELECT
r.ai_platform_id,
COALESCE(tm.site_name, gm.site_name, cf.site_key, cf.registrable_domain) AS site_name,
COALESCE(tm.site_key, gm.site_key, cf.site_key) AS site_key,
COALESCE(tm.registrable_domain, gm.registrable_domain, cf.site_key, cf.registrable_domain) AS site_domain,
COUNT(cf.id) AS citation_count,
COUNT(*) FILTER (WHERE cf.article_id IS NOT NULL AND r.ai_platform_id <> 'qwen') AS content_citation_count
FROM question_monitor_runs r
JOIN monitoring_citation_facts cf
ON cf.tenant_id = r.tenant_id AND cf.run_id = r.id
LEFT JOIN LATERAL (
SELECT registrable_domain, site_key, site_name
FROM site_domain_mappings
WHERE tenant_id = cf.tenant_id
AND is_active = TRUE
AND (
registrable_domain = cf.registrable_domain
OR registrable_domain = cf.site_key
OR cf.host = registrable_domain
OR cf.host LIKE '%.' || registrable_domain
)
ORDER BY LENGTH(registrable_domain) DESC, id DESC
LIMIT 1
) tm ON TRUE
LEFT JOIN LATERAL (
SELECT registrable_domain, site_key, site_name
FROM site_domain_mappings
WHERE tenant_id IS NULL
AND is_active = TRUE
AND (
registrable_domain = cf.registrable_domain
OR registrable_domain = cf.site_key
OR cf.host = registrable_domain
OR cf.host LIKE '%.' || registrable_domain
)
ORDER BY LENGTH(registrable_domain) DESC, id DESC
LIMIT 1
) gm ON tm.site_key IS NULL
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))
GROUP BY
r.ai_platform_id,
COALESCE(tm.site_name, gm.site_name, cf.site_key, cf.registrable_domain),
COALESCE(tm.site_key, gm.site_key, cf.site_key),
COALESCE(tm.registrable_domain, gm.registrable_domain, cf.site_key, cf.registrable_domain)
),
totals AS (
SELECT
ai_platform_id,
SUM(citation_count) AS total_citation_count
FROM grouped
GROUP BY ai_platform_id
)
SELECT
g.ai_platform_id,
g.site_name,
g.site_key,
g.site_domain,
g.citation_count,
CASE
WHEN t.total_citation_count > 0
THEN g.citation_count::double precision / t.total_citation_count::double precision
ELSE NULL
END AS citation_rate,
g.content_citation_count
FROM grouped g
JOIN totals t
ON t.ai_platform_id = g.ai_platform_id
ORDER BY g.ai_platform_id ASC, g.citation_count DESC, g.site_name ASC
`, 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 citation analysis")
}
defer rows.Close()
type citationAnalysisKey struct {
PlatformID string
SiteKey string
}
aggregates := make(map[citationAnalysisKey]MonitoringQuestionCitationStats)
totals := make(map[string]int64)
for rows.Next() {
var platformID string
var siteName string
var siteKey string
var siteDomain string
var citationCount int64
var contentCitationCount int64
if scanErr := rows.Scan(&platformID, &siteName, &siteKey, &siteDomain, &citationCount, new(sql.NullFloat64), &contentCitationCount); 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 += citationCount
item.ContentCitationCount += contentCitationCount
aggregates[key] = item
totals[platformID] += citationCount
}
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, brandID, questionID int64, hashBytes []byte, fromDate, toDate time.Time, aiPlatformID *string) ([]MonitoringQuestionContentCitation, error) {
platformQueryIDs := monitoringPlatformQueryIDs(aiPlatformID)
rows, err := s.monitoringPool.Query(ctx, `
WITH platform_totals AS (
SELECT
r.ai_platform_id,
COUNT(cf.id) AS total_citation_count
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.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))
GROUP BY r.ai_platform_id
)
SELECT
cf.article_id,
COALESCE(MAX(alias.article_title_snapshot), '未命名文章') AS article_title,
COALESCE(MAX(alias.publish_platform_name_snapshot), '已发布内容') AS publish_platform,
r.ai_platform_id,
COUNT(*) AS citation_count,
MAX(pt.total_citation_count) AS total_citation_count
FROM question_monitor_runs r
JOIN monitoring_citation_facts cf
ON cf.tenant_id = r.tenant_id AND cf.run_id = r.id
JOIN platform_totals pt
ON pt.ai_platform_id = r.ai_platform_id
LEFT JOIN monitoring_article_url_aliases alias
ON alias.tenant_id = cf.tenant_id AND alias.article_id = cf.article_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 cf.article_id IS NOT NULL
AND r.ai_platform_id <> 'qwen'
AND ($8::text[] IS NULL OR r.ai_platform_id = ANY($8))
GROUP BY cf.article_id, r.ai_platform_id
ORDER BY citation_count DESC, cf.article_id ASC
`, 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 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 articleID int64
var articleTitle string
var publishPlatform string
var platformID string
var citationCount int64
var totalCitationCount int64
if scanErr := rows.Scan(&articleID, &articleTitle, &publishPlatform, &platformID, &citationCount, &totalCitationCount); scanErr != nil {
return nil, response.ErrInternal(50041, "scan_failed", "failed to parse content citations")
}
platformID = normalizeMonitoringPlatformID(platformID)
if platformID == "" {
continue
}
key := contentCitationKey{
PlatformID: platformID,
ArticleID: articleID,
}
item := aggregates[key]
if item.ArticleID == 0 {
item = MonitoringQuestionContentCitation{
ArticleID: articleID,
ArticleTitle: articleTitle,
PublishPlatform: publishPlatform,
AIPlatformID: platformID,
PlatformName: platformDisplayName(platformID),
}
}
item.CitationCount += citationCount
aggregates[key] = item
totals[platformID] += citationCount
_ = totalCitationCount
}
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 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)
}