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