Files
geo/server/internal/tenant/app/monitoring_callback_service.go
T
root b345ee26e4 feat(monitoring): match citations by canonical link and scope summary by brand/keyword/platform/date
Replace the domain+title fuzzy alias scoring with a canonical
candidate-key index built from published-link aliases — only exact
matches now resolve to a SaaS source, eliminating false positives
across articles sharing a host (e.g. m.163.com vs www.163.com).

Surface the summary scoping that already existed on loadCitationRanking
/ loadCitedArticles by accepting brand_id, keyword_id, ai_platform_id
and business_date on CitationSummary, plumbing them through the handler
and admin tracking view.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 11:46:17 +08:00

3513 lines
114 KiB
Go

package app
import (
"bytes"
"context"
"database/sql"
"encoding/json"
"errors"
"net/url"
"regexp"
"strconv"
"strings"
"time"
"unicode"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgtype"
"github.com/jackc/pgx/v5/pgxpool"
"go.uber.org/zap"
"golang.org/x/net/publicsuffix"
sharedauth "github.com/geo-platform/tenant-api/internal/shared/auth"
sharedllm "github.com/geo-platform/tenant-api/internal/shared/llm"
"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/tenant/repository"
)
const (
defaultMonitoringLeaseLimit = 3
maxMonitoringLeaseLimit = 50
monitoringLeaseTTL = 15 * time.Minute
monitoringTaskResultQueueIOTimeout = 5 * time.Second
// The current primary renews the DB lease only after this interval. With the
// desktop client's 25s heartbeat and a 90s lease TTL, this keeps normal
// heartbeats read-only while still tolerating one missed heartbeat.
monitoringHeartbeatPrimaryRenewAfter = desktopClientPresenceTTL / 3
// Access snapshots are semantic state, not a liveness clock. Stable heartbeat
// reports skip writes and only refresh the row at this interval so dashboards
// do not depend on per-heartbeat database churn.
monitoringHeartbeatAccessSnapshotRefreshAfter = 10 * time.Minute
)
var monitoringCitationMarkerPattern = regexp.MustCompile(`(?i)\[\s*citation\s*:\s*\d+\s*\]`)
type MonitoringCallbackService struct {
businessPool *pgxpool.Pool
monitoringPool *pgxpool.Pool
rabbitMQ *rabbitmq.Client
llm sharedllm.Client
logger *zap.Logger
}
func NewMonitoringCallbackService(
businessPool, monitoringPool *pgxpool.Pool,
rabbitMQClient *rabbitmq.Client,
llmClient sharedllm.Client,
logger *zap.Logger,
) *MonitoringCallbackService {
return &MonitoringCallbackService{
businessPool: businessPool,
monitoringPool: monitoringPool,
rabbitMQ: rabbitMQClient,
llm: llmClient,
logger: logger,
}
}
type MonitoringTaskResultRequest struct {
LeaseToken string `json:"lease_token" binding:"required"`
Status string `json:"status"`
ProviderModel *string `json:"provider_model"`
ProviderRequestID *string `json:"provider_request_id"`
RequestID *string `json:"request_id"`
Answer *string `json:"answer"`
RawResponseJSON map[string]interface{} `json:"raw_response_json"`
Citations []MonitoringSourceItem `json:"citations"`
SearchResults []MonitoringSourceItem `json:"search_results"`
BrandMentioned *bool `json:"brand_mentioned"`
BrandMentionPosition *string `json:"brand_mention_position"`
FirstRecommended *bool `json:"first_recommended"`
SentimentLabel *string `json:"sentiment_label"`
MatchedBrandTerms []string `json:"matched_brand_terms"`
ErrorMessage *string `json:"error_message"`
CompletedAt *time.Time `json:"completed_at"`
}
type MonitoringSourceItem struct {
URL string `json:"url"`
Title *string `json:"title"`
SiteName *string `json:"site_name"`
SiteKey *string `json:"site_key"`
NormalizedURL *string `json:"normalized_url"`
Host *string `json:"host"`
RegistrableDomain *string `json:"registrable_domain"`
Subdomain *string `json:"subdomain"`
Suffix *string `json:"suffix"`
ArticleID *int64 `json:"article_id"`
PublishRecordID *int64 `json:"publish_record_id"`
ResolutionStatus *string `json:"resolution_status"`
ResolutionConfidence *string `json:"resolution_confidence"`
}
type MonitoringHeartbeatRequest struct {
Platforms []MonitoringHeartbeatPlatform `json:"platforms"`
}
type MonitoringHeartbeatPlatform struct {
AIPlatformID string `json:"ai_platform_id"`
AccessStatus *string `json:"access_status"`
Connected *bool `json:"connected"`
Accessible *bool `json:"accessible"`
DetectedAt *time.Time `json:"detected_at"`
ReasonCode *string `json:"reason_code"`
ReasonMessage *string `json:"reason_message"`
}
type MonitoringHeartbeatResponse struct {
InstallationID string `json:"installation_id"`
BusinessDate string `json:"business_date"`
UpdatedPlatformCount int `json:"updated_platform_count"`
ReconciledSkippedTaskCount int64 `json:"reconciled_skipped_task_count"`
}
type MonitoringLeaseTasksRequest struct {
Limit int `json:"limit"`
AIPlatformIDs []string `json:"ai_platform_ids"`
}
type MonitoringResumeTasksRequest struct {
AIPlatformIDs []string `json:"ai_platform_ids"`
}
type MonitoringLeaseTasksResponse struct {
BusinessDate string `json:"business_date"`
LeaseCount int `json:"lease_count"`
Tasks []MonitoringLeaseTask `json:"tasks"`
}
type MonitoringResumeTasksResponse struct {
BusinessDate string `json:"business_date"`
ResumeCount int `json:"resume_count"`
Tasks []MonitoringLeaseTask `json:"tasks"`
}
type MonitoringLeaseTask struct {
TaskID int64 `json:"task_id"`
BrandID int64 `json:"brand_id"`
QuestionID int64 `json:"question_id"`
QuestionHash string `json:"question_hash"`
QuestionText string `json:"question_text"`
AIPlatformID string `json:"ai_platform_id"`
PlatformName string `json:"platform_name"`
CollectorType string `json:"collector_type"`
RunMode string `json:"run_mode"`
TriggerSource string `json:"trigger_source"`
BusinessDate string `json:"business_date"`
Priority int `json:"priority,omitempty"`
Lane string `json:"lane,omitempty"`
LeaseToken string `json:"lease_token"`
LeaseExpiresAt string `json:"lease_expires_at"`
}
type MonitoringTaskResultResponse struct {
TaskID int64 `json:"task_id"`
RunID *int64 `json:"run_id"`
TaskStatus string `json:"task_status"`
CitationSourceCount int `json:"citation_source_count"`
ContentCitationCount int `json:"content_citation_count"`
}
type MonitoringSkipTaskRequest struct {
LeaseToken string `json:"lease_token" binding:"required"`
SkipReason *string `json:"skip_reason"`
ErrorMessage *string `json:"error_message"`
CompletedAt *time.Time `json:"completed_at"`
}
type MonitoringSkipTaskResponse struct {
TaskID int64 `json:"task_id"`
TaskStatus string `json:"task_status"`
SkipReason string `json:"skip_reason"`
CompletedAt *string `json:"completed_at"`
}
type MonitoringCancelTaskRequest struct {
LeaseToken string `json:"lease_token" binding:"required"`
Reason *string `json:"reason"`
}
type MonitoringCancelTaskResponse struct {
TaskID int64 `json:"task_id"`
TaskStatus string `json:"task_status"`
}
type monitoringInstallationIdentity struct {
ID string
TenantID int64
WorkspaceID int64
}
func monitoringInstallationFromDesktopClient(client *repository.DesktopClient) *monitoringInstallationIdentity {
if client == nil {
return nil
}
return &monitoringInstallationIdentity{
ID: client.ID.String(),
TenantID: client.TenantID,
WorkspaceID: client.WorkspaceID,
}
}
type monitoringPlatformAccessSnapshotState struct {
AccessStatus string
Connected bool
Accessible bool
ReasonCode *string
ReasonMessage *string
UpdatedAt time.Time
}
type monitoringHeartbeatSnapshotDecision struct {
WriteSnapshot bool
StateChanged bool
RefreshDue bool
PendingReconcileCandidate bool
ReconcilePending bool
SnapshotOperation string
}
type monitoringCollectTask struct {
ID int64
TenantID int64
BrandID int64
QuestionID int64
QuestionHash []byte
AIPlatformID string
CollectorType string
RunMode string
BusinessDate time.Time
Status string
ExecutionOwner string
LeaseTokenHash sql.NullString
LeasedToExecutor sql.NullString
LeaseExpiresAt sql.NullTime
CallbackReceivedAt sql.NullTime
CompletedAt sql.NullTime
SkipReason sql.NullString
TriggerSource string
QuestionText string
}
type monitoringDesktopExecutionLease struct {
DesktopTaskID uuid.UUID
Status string
TargetClientID uuid.UUID
LeaseTokenHash []byte
InterruptGeneration int
}
type monitoringLeasedTaskRow struct {
ID int64
BrandID int64
QuestionID int64
QuestionHash []byte
AIPlatformID string
CollectorType string
RunMode string
TriggerSource string
BusinessDate time.Time
Priority int
Lane string
QuestionText string
}
type monitoringAliasResolution struct {
ArticleID *int64
PublishRecordID *int64
NormalizedURLKey string
ResolutionStatus string
ResolutionConfidence string
}
type monitoringAliasLookupInput struct {
NormalizedURL string
MatchKeys []string
RegistrableDomain string
SiteKey string
Host string
LastPathSegment *string
TitleKey string
}
type monitoringAliasCandidate struct {
NormalizedURL string
MatchKeys []string
ArticleID *int64
PublishRecordID *int64
RegistrableDomain string
SiteKey string
Host string
LastPathSegment string
TitleKey string
}
type monitoringCitationFact struct {
CitedURL string
CitedTitle *string
NormalizedURL string
Host string
RegistrableDomain string
Subdomain *string
Suffix *string
SiteKey string
ArticleID *int64
PublishRecordID *int64
ResolutionStatus string
ResolutionConfidence string
}
type monitoringTaskResultEnvelope struct {
Version string `json:"version"`
TaskID int64 `json:"task_id"`
TenantID int64 `json:"tenant_id"`
InstallationID string `json:"installation_id"`
ReceivedAt time.Time `json:"received_at"`
QueueStatus string `json:"queue_status"`
QueueClaimedAt *time.Time `json:"queue_claimed_at"`
QueueEnqueuedAt *time.Time `json:"queue_enqueued_at"`
LastQueuePublishError *string `json:"last_queue_publish_error"`
ReceivedAlertedAt *time.Time `json:"received_alerted_at"`
ReceivedAlertCount int `json:"received_alert_count"`
CallbackResult MonitoringTaskResultRequest `json:"callback_result"`
}
type monitoringTaskResultQueueMetaPatch struct {
QueueStatus string `json:"queue_status"`
QueueClaimedAt *time.Time `json:"queue_claimed_at"`
QueueEnqueuedAt *time.Time `json:"queue_enqueued_at"`
LastQueuePublishError *string `json:"last_queue_publish_error"`
}
type MonitoringTaskResultQueueMetaPatch = monitoringTaskResultQueueMetaPatch
func (s *MonitoringCallbackService) HandleTaskResult(ctx context.Context, installationID string, installationToken string, taskID int64, req MonitoringTaskResultRequest) (*MonitoringTaskResultResponse, error) {
installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
if err != nil {
return nil, err
}
return s.handleTaskResultForInstallation(ctx, installation, taskID, req)
}
func (s *MonitoringCallbackService) HandleTaskResultForDesktopClient(
ctx context.Context,
client *repository.DesktopClient,
taskID int64,
req MonitoringTaskResultRequest,
) (*MonitoringTaskResultResponse, error) {
if client == nil {
return nil, response.ErrUnauthorized(40141, "desktop_client_missing", "desktop client context is required")
}
return s.handleTaskResultForInstallation(ctx, monitoringInstallationFromDesktopClient(client), taskID, req)
}
func (s *MonitoringCallbackService) handleTaskResultForInstallation(
ctx context.Context,
installation *monitoringInstallationIdentity,
taskID int64,
req MonitoringTaskResultRequest,
) (*MonitoringTaskResultResponse, error) {
task, err := s.loadCollectTask(ctx, installation.TenantID, taskID)
if err != nil {
return nil, err
}
if task.CollectorType != monitoringCollectorType {
return nil, response.ErrConflict(40941, "collector_type_mismatch", "task collector type is not desktop")
}
if task.Status == "completed" {
runID, citationSourceCount, contentCitationCount, loadErr := s.loadExistingRunSummary(ctx, task)
if loadErr != nil {
return nil, loadErr
}
_ = s.touchInstallation(ctx, installation.ID)
return &MonitoringTaskResultResponse{
TaskID: task.ID,
RunID: runID,
TaskStatus: task.Status,
CitationSourceCount: citationSourceCount,
ContentCitationCount: contentCitationCount,
}, nil
}
if task.ExecutionOwner == "desktop_tasks" {
if err := s.validateDesktopMonitorExecutionLease(ctx, task, installation.ID, req.LeaseToken); err != nil {
return nil, err
}
} else {
if task.LeasedToExecutor.Valid && strings.TrimSpace(task.LeasedToExecutor.String) != "" && task.LeasedToExecutor.String != installation.ID {
return nil, response.ErrUnauthorized(40141, "lease_owner_mismatch", "task is leased to another installation")
}
if strings.TrimSpace(task.LeaseTokenHash.String) == "" {
return nil, response.ErrConflict(40941, "lease_token_missing", "task does not have an active lease token")
}
if sharedauth.HashToken(strings.TrimSpace(req.LeaseToken)) != task.LeaseTokenHash.String {
return nil, response.ErrUnauthorized(40141, "invalid_lease_token", "lease token is invalid")
}
}
switch task.Status {
case "leased", "received", "failed", "expired", "pending":
default:
return nil, response.ErrConflict(40941, "task_status_invalid", "task status does not allow result submission")
}
runStatus := normalizeRunStatus(req.Status)
if runStatus == "" {
return nil, response.ErrBadRequest(40041, "invalid_status", "status must be succeeded or failed")
}
if err := validateMonitoringTaskResultSubmission(task.AIPlatformID, runStatus, req); err != nil {
return nil, err
}
receivedAt := time.Now().UTC().Round(0)
envelope := newMonitoringTaskResultEnvelope(task, installation.ID, receivedAt, req)
requestPayloadJSON, err := json.Marshal(envelope)
if err != nil {
return nil, response.ErrInternal(50041, "payload_encode_failed", "failed to encode monitoring callback payload")
}
tx, err := s.monitoringPool.Begin(ctx)
if err != nil {
return nil, response.ErrInternal(50041, "begin_failed", "failed to start monitoring callback receive")
}
defer tx.Rollback(ctx)
if err := s.markTaskResultReceived(ctx, tx, task, receivedAt, requestPayloadJSON); err != nil {
return nil, err
}
if err := tx.Commit(ctx); err != nil {
return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring callback receive")
}
_ = s.touchInstallation(ctx, installation.ID)
publishCtx, cancel := context.WithTimeout(context.Background(), monitoringTaskResultQueueIOTimeout)
defer cancel()
if err := s.publishTaskResultEnvelope(publishCtx, requestPayloadJSON); err != nil {
s.tryPatchTaskResultQueueMeta(task.ID, receivedAt, monitoringTaskResultQueueMetaPatch{
QueueStatus: "publish_failed",
LastQueuePublishError: optionalString(err.Error()),
})
return nil, response.ErrServiceUnavailable(50341, "monitoring_result_queue_unavailable", "failed to queue monitoring result")
}
enqueuedAt := time.Now().UTC().Round(0)
s.tryPatchTaskResultQueueMeta(task.ID, receivedAt, monitoringTaskResultQueueMetaPatch{
QueueStatus: "queued",
QueueEnqueuedAt: &enqueuedAt,
})
if task.ExecutionOwner == "desktop_tasks" {
if finalizeErr := s.completeLinkedDesktopMonitorTask(ctx, task, req); finalizeErr != nil && s.logger != nil {
s.logger.Warn("phase2 monitor desktop task finalize failed",
zap.Int64("monitor_task_id", task.ID),
zap.String("ai_platform_id", task.AIPlatformID),
zap.Error(finalizeErr),
)
}
}
return &MonitoringTaskResultResponse{
TaskID: task.ID,
TaskStatus: "received",
CitationSourceCount: 0,
ContentCitationCount: 0,
}, nil
}
func (s *MonitoringCallbackService) Heartbeat(ctx context.Context, installationID string, installationToken string, req MonitoringHeartbeatRequest) (_ *MonitoringHeartbeatResponse, err error) {
startedAt := time.Now()
snapshotMetrics := monitoringHeartbeatSnapshotMetricsEvent{}
defer func() {
if !snapshotMetrics.Active {
return
}
snapshotMetrics.Duration = time.Since(startedAt)
snapshotMetrics.Err = err
recordMonitoringHeartbeatSnapshotMetrics(snapshotMetrics)
}()
installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
if err != nil {
return nil, err
}
isPrimaryInstallation, err := s.resolveHeartbeatPrimaryInstallation(ctx, installation.TenantID, installation)
if err != nil {
return nil, err
}
now := time.Now().UTC()
businessDay, businessDate := monitoringBusinessDayAndDateAt(now)
platforms, err := normalizeHeartbeatPlatforms(req.Platforms, now)
if err != nil {
return nil, err
}
snapshotMetrics.Active = true
snapshotMetrics.PlatformCount = len(platforms)
if len(platforms) == 0 {
_ = s.touchInstallation(ctx, installation.ID)
return &MonitoringHeartbeatResponse{
InstallationID: installation.ID,
BusinessDate: businessDate,
UpdatedPlatformCount: 0,
ReconciledSkippedTaskCount: 0,
}, nil
}
platformIDs := monitoringHeartbeatPlatformIDs(platforms)
previousStates, err := s.loadPlatformAccessSnapshotStates(ctx, installation.TenantID, installation.ID, businessDate, platformIDs)
if err != nil {
return nil, err
}
type heartbeatPlatformAction struct {
platform MonitoringHeartbeatPlatform
decision monitoringHeartbeatSnapshotDecision
}
actions := make([]heartbeatPlatformAction, 0, len(platforms))
inaccessiblePlatformIDs := make([]string, 0, len(platforms))
needsTransaction := false
for _, platform := range platforms {
nextState := monitoringPlatformAccessSnapshotStateFromHeartbeat(platform)
decision := decideMonitoringHeartbeatSnapshot(previousStates[platform.AIPlatformID], nextState, now, isPrimaryInstallation)
if decision.WriteSnapshot {
needsTransaction = true
} else {
snapshotMetrics.SnapshotSkippedCount++
}
if decision.PendingReconcileCandidate {
inaccessiblePlatformIDs = append(inaccessiblePlatformIDs, platform.AIPlatformID)
}
actions = append(actions, heartbeatPlatformAction{
platform: platform,
decision: decision,
})
}
if isPrimaryInstallation && len(inaccessiblePlatformIDs) > 0 {
pendingPlatforms, pendingErr := s.loadPendingMonitoringTaskPlatformIDs(ctx, installation.TenantID, businessDate, inaccessiblePlatformIDs)
if pendingErr != nil {
return nil, pendingErr
}
for idx := range actions {
if !actions[idx].decision.PendingReconcileCandidate {
continue
}
if _, ok := pendingPlatforms[actions[idx].platform.AIPlatformID]; ok {
actions[idx].decision.ReconcilePending = true
needsTransaction = true
}
}
}
if !needsTransaction {
_ = s.touchInstallation(ctx, installation.ID)
return &MonitoringHeartbeatResponse{
InstallationID: installation.ID,
BusinessDate: businessDate,
UpdatedPlatformCount: len(platforms),
ReconciledSkippedTaskCount: 0,
}, nil
}
tx, err := s.monitoringPool.Begin(ctx)
if err != nil {
return nil, response.ErrInternal(50041, "begin_failed", "failed to start monitoring heartbeat")
}
snapshotMetrics.TransactionOpened = true
defer tx.Rollback(ctx)
var reconciledSkippedTaskCount int64
var snapshotInsertedCount int64
var snapshotUpdatedCount int64
var snapshotRefreshedCount int64
var projectionRebuildCount int64
affectedBrandIDs := make(map[int64]struct{})
for _, action := range actions {
if action.decision.WriteSnapshot {
if err := s.upsertPlatformAccessSnapshot(ctx, tx, installation, businessDate, action.platform); err != nil {
return nil, err
}
switch action.decision.SnapshotOperation {
case "insert":
snapshotInsertedCount++
case "update":
snapshotUpdatedCount++
case "refresh":
snapshotRefreshedCount++
}
}
if !isPrimaryInstallation {
continue
}
needsProjectionRefresh := action.decision.StateChanged
if action.decision.ReconcilePending {
count, reconcileErr := s.reconcilePendingTasksForPlatform(ctx, tx, installation.TenantID, businessDate, action.platform.AIPlatformID)
if reconcileErr != nil {
return nil, reconcileErr
}
reconciledSkippedTaskCount += count
if count > 0 {
needsProjectionRefresh = true
}
}
if !needsProjectionRefresh {
continue
}
brandIDs, loadErr := s.loadMonitoringAffectedBrandIDs(ctx, tx, installation.TenantID, businessDay, action.platform.AIPlatformID)
if loadErr != nil {
return nil, loadErr
}
for _, brandID := range brandIDs {
affectedBrandIDs[brandID] = struct{}{}
}
}
if isPrimaryInstallation {
for brandID := range affectedBrandIDs {
if err := s.rebuildBrandDailyProjection(ctx, tx, installation.TenantID, brandID, businessDay); err != nil {
return nil, err
}
projectionRebuildCount++
}
}
if err := tx.Commit(ctx); err != nil {
return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring heartbeat")
}
snapshotMetrics.SnapshotInsertedCount = snapshotInsertedCount
snapshotMetrics.SnapshotUpdatedCount = snapshotUpdatedCount
snapshotMetrics.SnapshotRefreshedCount = snapshotRefreshedCount
snapshotMetrics.ReconciledSkippedTaskCount = reconciledSkippedTaskCount
snapshotMetrics.ProjectionRebuildCount = projectionRebuildCount
_ = s.touchInstallation(ctx, installation.ID)
if isPrimaryInstallation && len(affectedBrandIDs) > 0 {
s.dispatchMonitoringProjectionRebuilds(installation.TenantID, businessDay, affectedBrandIDs, "heartbeat")
}
return &MonitoringHeartbeatResponse{
InstallationID: installation.ID,
BusinessDate: businessDate,
UpdatedPlatformCount: len(platforms),
ReconciledSkippedTaskCount: reconciledSkippedTaskCount,
}, nil
}
func (s *MonitoringCallbackService) LeaseTasks(ctx context.Context, installationID string, installationToken string, req MonitoringLeaseTasksRequest) (*MonitoringLeaseTasksResponse, error) {
installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
if err != nil {
return nil, err
}
return s.leaseTasksForInstallation(ctx, installation, req)
}
func (s *MonitoringCallbackService) LeaseTasksForDesktopClient(ctx context.Context, client *repository.DesktopClient, req MonitoringLeaseTasksRequest) (*MonitoringLeaseTasksResponse, error) {
if client == nil {
return nil, response.ErrUnauthorized(40141, "desktop_client_missing", "desktop client context is required")
}
return s.leaseTasksForInstallation(ctx, monitoringInstallationFromDesktopClient(client), req)
}
func (s *MonitoringCallbackService) leaseTasksForInstallation(
ctx context.Context,
installation *monitoringInstallationIdentity,
req MonitoringLeaseTasksRequest,
) (*MonitoringLeaseTasksResponse, error) {
limit := normalizeMonitoringLeaseLimit(req.Limit)
eligiblePlatformIDs := normalizeMonitoringLeasePlatformIDs(req.AIPlatformIDs)
now := time.Now().UTC()
businessDay, businessDate := monitoringBusinessDayAndDateAt(now)
leaseExpiresAt := now.Add(monitoringLeaseTTL)
tx, err := s.monitoringPool.Begin(ctx)
if err != nil {
return nil, response.ErrInternal(50041, "begin_failed", "failed to start monitoring task lease")
}
defer tx.Rollback(ctx)
if _, err := expireMonitoringLeasedTasks(ctx, tx, &installation.TenantID, now); err != nil {
return nil, err
}
if _, err := requeueMonitoringExpiredTasks(ctx, tx, installation.TenantID, businessDay); err != nil {
return nil, err
}
if _, err := skipMonitoringTasksBeforeBusinessDate(ctx, tx, installation.TenantID, businessDay, "legacy"); err != nil {
return nil, err
}
rows, err := s.selectPendingLeaseTasks(ctx, tx, installation, businessDate, limit, eligiblePlatformIDs)
if err != nil {
return nil, err
}
if len(rows) == 0 {
if err := tx.Commit(ctx); err != nil {
return nil, response.ErrInternal(50041, "commit_failed", "failed to finish empty monitoring lease")
}
_ = s.touchInstallation(ctx, installation.ID)
return &MonitoringLeaseTasksResponse{
BusinessDate: businessDate,
LeaseCount: 0,
Tasks: []MonitoringLeaseTask{},
}, nil
}
tasks := make([]MonitoringLeaseTask, 0, len(rows))
for _, row := range rows {
leaseToken, tokenErr := newSessionToken()
if tokenErr != nil {
return nil, response.ErrInternal(50041, "lease_token_failed", "failed to generate monitoring lease token")
}
if err := s.markTaskLeased(ctx, tx, installation.ID, row.ID, leaseToken, now, leaseExpiresAt); err != nil {
return nil, err
}
tasks = append(tasks, MonitoringLeaseTask{
TaskID: row.ID,
BrandID: row.BrandID,
QuestionID: row.QuestionID,
QuestionHash: encodeQuestionHash(row.QuestionHash),
QuestionText: row.QuestionText,
AIPlatformID: row.AIPlatformID,
PlatformName: platformDisplayName(row.AIPlatformID),
CollectorType: row.CollectorType,
RunMode: row.RunMode,
TriggerSource: row.TriggerSource,
BusinessDate: row.BusinessDate.Format("2006-01-02"),
Priority: row.Priority,
Lane: row.Lane,
LeaseToken: leaseToken,
LeaseExpiresAt: leaseExpiresAt.Format(time.RFC3339),
})
}
if err := tx.Commit(ctx); err != nil {
return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring task lease")
}
_ = s.touchInstallation(ctx, installation.ID)
return &MonitoringLeaseTasksResponse{
BusinessDate: businessDate,
LeaseCount: len(tasks),
Tasks: tasks,
}, nil
}
func (s *MonitoringCallbackService) ResumeTasks(ctx context.Context, installationID string, installationToken string, req MonitoringResumeTasksRequest) (*MonitoringResumeTasksResponse, error) {
installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
if err != nil {
return nil, err
}
return s.resumeTasksForInstallation(ctx, installation, req)
}
func (s *MonitoringCallbackService) ResumeTasksForDesktopClient(ctx context.Context, client *repository.DesktopClient, req MonitoringResumeTasksRequest) (*MonitoringResumeTasksResponse, error) {
if client == nil {
return nil, response.ErrUnauthorized(40141, "desktop_client_missing", "desktop client context is required")
}
return s.resumeTasksForInstallation(ctx, monitoringInstallationFromDesktopClient(client), req)
}
func (s *MonitoringCallbackService) resumeTasksForInstallation(
ctx context.Context,
installation *monitoringInstallationIdentity,
req MonitoringResumeTasksRequest,
) (*MonitoringResumeTasksResponse, error) {
eligiblePlatformIDs := normalizeMonitoringLeasePlatformIDs(req.AIPlatformIDs)
now := time.Now().UTC()
businessDay, businessDate := monitoringBusinessDayAndDateAt(now)
leaseExpiresAt := now.Add(monitoringLeaseTTL)
tx, err := s.monitoringPool.Begin(ctx)
if err != nil {
return nil, response.ErrInternal(50041, "begin_failed", "failed to start monitoring task resume")
}
defer tx.Rollback(ctx)
if _, err := expireMonitoringLeasedTasks(ctx, tx, &installation.TenantID, now); err != nil {
return nil, err
}
if _, err := requeueMonitoringExpiredTasks(ctx, tx, installation.TenantID, businessDay); err != nil {
return nil, err
}
if _, err := skipMonitoringTasksBeforeBusinessDate(ctx, tx, installation.TenantID, businessDay, "legacy"); err != nil {
return nil, err
}
rows, err := s.selectLeasedTasksForInstallation(ctx, tx, installation.TenantID, installation.ID, businessDate, eligiblePlatformIDs)
if err != nil {
return nil, err
}
if len(rows) == 0 {
if err := tx.Commit(ctx); err != nil {
return nil, response.ErrInternal(50041, "commit_failed", "failed to finish empty monitoring resume")
}
_ = s.touchInstallation(ctx, installation.ID)
return &MonitoringResumeTasksResponse{
BusinessDate: businessDate,
ResumeCount: 0,
Tasks: []MonitoringLeaseTask{},
}, nil
}
tasks := make([]MonitoringLeaseTask, 0, len(rows))
for _, row := range rows {
leaseToken, tokenErr := newSessionToken()
if tokenErr != nil {
return nil, response.ErrInternal(50041, "lease_token_failed", "failed to generate monitoring resume token")
}
if err := s.markTaskLeased(ctx, tx, installation.ID, row.ID, leaseToken, now, leaseExpiresAt); err != nil {
return nil, err
}
tasks = append(tasks, MonitoringLeaseTask{
TaskID: row.ID,
BrandID: row.BrandID,
QuestionID: row.QuestionID,
QuestionHash: encodeQuestionHash(row.QuestionHash),
QuestionText: row.QuestionText,
AIPlatformID: row.AIPlatformID,
PlatformName: platformDisplayName(row.AIPlatformID),
CollectorType: row.CollectorType,
RunMode: row.RunMode,
TriggerSource: row.TriggerSource,
BusinessDate: row.BusinessDate.Format("2006-01-02"),
Priority: row.Priority,
Lane: row.Lane,
LeaseToken: leaseToken,
LeaseExpiresAt: leaseExpiresAt.Format(time.RFC3339),
})
}
if err := tx.Commit(ctx); err != nil {
return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring task resume")
}
_ = s.touchInstallation(ctx, installation.ID)
return &MonitoringResumeTasksResponse{
BusinessDate: businessDate,
ResumeCount: len(tasks),
Tasks: tasks,
}, nil
}
func (s *MonitoringCallbackService) SkipTask(ctx context.Context, installationID string, installationToken string, taskID int64, req MonitoringSkipTaskRequest) (*MonitoringSkipTaskResponse, error) {
installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
if err != nil {
return nil, err
}
return s.skipTaskForInstallation(ctx, installation, taskID, req)
}
func (s *MonitoringCallbackService) SkipTaskForDesktopClient(
ctx context.Context,
client *repository.DesktopClient,
taskID int64,
req MonitoringSkipTaskRequest,
) (*MonitoringSkipTaskResponse, error) {
if client == nil {
return nil, response.ErrUnauthorized(40141, "desktop_client_missing", "desktop client context is required")
}
return s.skipTaskForInstallation(ctx, monitoringInstallationFromDesktopClient(client), taskID, req)
}
func (s *MonitoringCallbackService) skipTaskForInstallation(
ctx context.Context,
installation *monitoringInstallationIdentity,
taskID int64,
req MonitoringSkipTaskRequest,
) (*MonitoringSkipTaskResponse, error) {
task, err := s.loadCollectTask(ctx, installation.TenantID, taskID)
if err != nil {
return nil, err
}
if task.CollectorType != monitoringCollectorType {
return nil, response.ErrConflict(40941, "collector_type_mismatch", "task collector type is not plugin")
}
if err := s.validateCollectTaskExecution(ctx, task, installation.ID, req.LeaseToken); err != nil {
return nil, err
}
if task.Status == "skipped" {
_ = s.touchInstallation(ctx, installation.ID)
return &MonitoringSkipTaskResponse{
TaskID: task.ID,
TaskStatus: task.Status,
SkipReason: strings.TrimSpace(task.SkipReason.String),
CompletedAt: formatNullTime(task.CompletedAt),
}, nil
}
switch task.Status {
case "leased", "received", "failed", "expired", "pending":
default:
return nil, response.ErrConflict(40941, "task_status_invalid", "task status does not allow skip submission")
}
completedAt := time.Now().UTC()
if req.CompletedAt != nil && !req.CompletedAt.IsZero() {
completedAt = req.CompletedAt.UTC()
}
skipReason := normalizeSkipReason(req.SkipReason)
tx, err := s.monitoringPool.Begin(ctx)
if err != nil {
return nil, response.ErrInternal(50041, "begin_failed", "failed to start monitoring task skip")
}
defer tx.Rollback(ctx)
if _, err := tx.Exec(ctx, `
UPDATE monitoring_collect_tasks
SET status = 'skipped',
callback_received_at = NOW(),
completed_at = $2,
skip_reason = $3,
error_message = $4,
updated_at = NOW()
WHERE id = $1
`, task.ID, completedAt, skipReason, nullableString(req.ErrorMessage)); err != nil {
return nil, response.ErrInternal(50041, "task_update_failed", "failed to update monitoring task skip status")
}
if err := s.rebuildBrandDailyProjection(ctx, tx, task.TenantID, task.BrandID, task.BusinessDate); err != nil {
return nil, err
}
if err := tx.Commit(ctx); err != nil {
return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring task skip")
}
_ = s.touchInstallation(ctx, installation.ID)
completedAtText := completedAt.Format(time.RFC3339)
return &MonitoringSkipTaskResponse{
TaskID: task.ID,
TaskStatus: "skipped",
SkipReason: skipReason,
CompletedAt: &completedAtText,
}, nil
}
func (s *MonitoringCallbackService) CancelTaskForDesktopClient(
ctx context.Context,
client *repository.DesktopClient,
taskID int64,
req MonitoringCancelTaskRequest,
) (*MonitoringCancelTaskResponse, error) {
if client == nil {
return nil, response.ErrUnauthorized(40141, "desktop_client_missing", "desktop client context is required")
}
installation := monitoringInstallationFromDesktopClient(client)
task, err := s.loadCollectTask(ctx, installation.TenantID, taskID)
if err != nil {
return nil, err
}
if task.CollectorType != monitoringCollectorType {
return nil, response.ErrConflict(40941, "collector_type_mismatch", "task collector type is not desktop")
}
if err := s.validateCollectTaskExecution(ctx, task, installation.ID, req.LeaseToken); err != nil {
return nil, err
}
var desktopTaskCanceled *repository.DesktopTask
if task.ExecutionOwner == "desktop_tasks" {
desktopExecution, leaseErr := s.loadActiveDesktopMonitorExecutionLease(ctx, task.ID)
if leaseErr != nil {
return nil, leaseErr
}
if desktopExecution == nil || desktopExecution.Status != "in_progress" {
return nil, response.ErrConflict(40941, "task_status_invalid", "task status does not allow lease cancel")
}
errorJSON, marshalErr := json.Marshal(map[string]any{
"reason": optionalStringValue(req.Reason),
"source": "legacy_monitor_cancel",
})
if marshalErr != nil {
return nil, response.ErrInternal(50041, "payload_encode_failed", "failed to encode phase2 desktop cancel payload")
}
taskRepo := repository.NewDesktopTaskRepository(s.businessPool)
canceledTask, cancelErr := taskRepo.CancelByLease(ctx, desktopExecution.DesktopTaskID, desktopExecution.LeaseTokenHash, errorJSON)
if cancelErr != nil {
if errors.Is(cancelErr, pgx.ErrNoRows) {
return nil, response.ErrConflict(40941, "task_status_invalid", "phase2 monitor desktop task is no longer active")
}
return nil, response.ErrInternal(50041, "desktop_task_cancel_failed", "failed to cancel phase2 monitor desktop task lease")
}
desktopTaskCanceled = canceledTask
} else if task.Status != "leased" {
return nil, response.ErrConflict(40941, "task_status_invalid", "task status does not allow lease cancel")
}
if _, err := s.monitoringPool.Exec(ctx, `
UPDATE monitoring_collect_tasks
SET status = 'pending',
lease_token_hash = NULL,
leased_to_executor = NULL,
leased_at = NULL,
lease_expires_at = NULL,
callback_received_at = NULL,
dispatch_after = NOW(),
error_message = $2,
updated_at = NOW()
WHERE id = $1
`, task.ID, nullableString(req.Reason)); err != nil {
return nil, response.ErrInternal(50041, "task_update_failed", "failed to cancel monitoring task lease")
}
if desktopTaskCanceled != nil {
title, businessDate, schedulerGroupKey, questionText := deriveDesktopTaskEventMetadataFromPayload(
desktopTaskCanceled.Kind,
desktopTaskCanceled.Payload,
)
publishDesktopTaskEvent(ctx, s.rabbitMQ, DesktopTaskEvent{
Type: "task_canceled",
TaskID: desktopTaskCanceled.DesktopID.String(),
JobID: desktopTaskCanceled.JobID.String(),
WorkspaceID: desktopTaskCanceled.WorkspaceID,
TargetAccountID: desktopTaskCanceled.TargetAccountID.String(),
TargetClientID: desktopTaskCanceled.TargetClientID.String(),
Platform: desktopTaskCanceled.Platform,
Title: title,
BusinessDate: businessDate,
SchedulerGroupKey: schedulerGroupKey,
QuestionText: questionText,
Status: desktopTaskCanceled.Status,
Kind: desktopTaskCanceled.Kind,
Priority: desktopTaskCanceled.Priority,
Lane: desktopTaskCanceled.Lane,
InterruptGeneration: desktopTaskCanceled.InterruptGeneration,
UpdatedAt: desktopTaskCanceled.UpdatedAt,
})
}
return &MonitoringCancelTaskResponse{
TaskID: task.ID,
TaskStatus: "pending",
}, nil
}
func (s *MonitoringCallbackService) ProcessQueuedTaskResult(ctx context.Context, envelope monitoringTaskResultEnvelope) (*MonitoringTaskResultResponse, error) {
if envelope.TaskID <= 0 {
return nil, response.ErrBadRequest(40041, "invalid_task_id", "task id is required")
}
task, err := s.loadCollectTaskByID(ctx, envelope.TaskID)
if err != nil {
if appErr, ok := err.(*response.AppError); ok && appErr.HTTPStatus == 404 {
return nil, nil
}
return nil, err
}
if envelope.TenantID > 0 && envelope.TenantID != task.TenantID {
return nil, response.ErrConflict(40941, "task_tenant_mismatch", "task tenant does not match queued monitoring result")
}
if !monitoringTaskReadyForQueuedResult(task) {
return nil, nil
}
return s.processTaskResult(ctx, task, envelope.CallbackResult)
}
func (s *MonitoringCallbackService) authenticateInstallation(ctx context.Context, installationID string, installationToken string) (*monitoringInstallationIdentity, error) {
tokenHash := HashDesktopClientToken(strings.TrimSpace(installationToken))
row := &monitoringInstallationIdentity{}
if err := s.businessPool.QueryRow(ctx, `
SELECT id::text, tenant_id, workspace_id
FROM desktop_clients
WHERE id = $1::uuid
AND token_hash = $2
AND revoked_at IS NULL
`, strings.TrimSpace(installationID), tokenHash).Scan(&row.ID, &row.TenantID, &row.WorkspaceID); err != nil {
if err == pgx.ErrNoRows {
return nil, response.ErrUnauthorized(40141, "desktop_client_unauthorized", "desktop client is invalid or expired")
}
return nil, response.ErrInternal(50041, "desktop_client_lookup_failed", "failed to validate desktop client")
}
return row, nil
}
func (s *MonitoringCallbackService) resolveHeartbeatPrimaryInstallation(ctx context.Context, tenantID int64, installation *monitoringInstallationIdentity) (bool, error) {
if installation == nil || strings.TrimSpace(installation.ID) == "" {
return false, nil
}
isPrimary, err := s.claimHeartbeatPrimaryLease(ctx, tenantID, installation.WorkspaceID, installation.ID)
if err != nil {
return false, err
}
return isPrimary, nil
}
func (s *MonitoringCallbackService) claimHeartbeatPrimaryLease(ctx context.Context, tenantID, workspaceID int64, installationID string) (bool, error) {
startedAt := time.Now()
installationID = strings.TrimSpace(installationID)
leaseTTL := formatPgInterval(desktopClientPresenceTTL)
state, err := s.loadHeartbeatPrimaryLeaseState(ctx, tenantID, workspaceID, installationID)
if err != nil {
recordHeartbeatPrimaryLeaseMetric("error", time.Since(startedAt))
return false, err
}
if !state.Found {
inserted, insertErr := s.tryInsertHeartbeatPrimaryLease(ctx, tenantID, workspaceID, installationID, leaseTTL)
if insertErr != nil {
recordHeartbeatPrimaryLeaseMetric("error", time.Since(startedAt))
return false, insertErr
}
if inserted {
recordHeartbeatPrimaryLeaseMetric("inserted", time.Since(startedAt))
recordHeartbeatPrimaryLeaseWriteMetric("insert")
return true, nil
}
recordHeartbeatPrimaryLeaseMetric("contention", time.Since(startedAt))
return s.resolveHeartbeatPrimaryAfterContention(ctx, tenantID, workspaceID, installationID)
}
decision := decideHeartbeatPrimaryLease(state)
switch decision.Action {
case heartbeatPrimaryLeaseActionPrimaryHit:
recordHeartbeatPrimaryLeaseMetric("primary_hit", time.Since(startedAt))
return true, nil
case heartbeatPrimaryLeaseActionNonPrimaryLive:
recordHeartbeatPrimaryLeaseMetric("non_primary_live", time.Since(startedAt))
return false, nil
case heartbeatPrimaryLeaseActionRenew:
updated, updateErr := s.renewHeartbeatPrimaryLease(ctx, tenantID, workspaceID, installationID, leaseTTL)
if updateErr != nil {
recordHeartbeatPrimaryLeaseMetric("error", time.Since(startedAt))
return false, updateErr
}
if updated {
recordHeartbeatPrimaryLeaseMetric("renewed", time.Since(startedAt))
recordHeartbeatPrimaryLeaseWriteMetric("renew")
return true, nil
}
case heartbeatPrimaryLeaseActionClaimExpired, heartbeatPrimaryLeaseActionClaimUnavailable:
updated, updateErr := s.claimExpiredHeartbeatPrimaryLease(ctx, tenantID, workspaceID, installationID, state.PrimaryInstallationID, leaseTTL, decision.Action)
if updateErr != nil {
recordHeartbeatPrimaryLeaseMetric("error", time.Since(startedAt))
return false, updateErr
}
if updated {
if decision.Action == heartbeatPrimaryLeaseActionClaimExpired {
recordHeartbeatPrimaryLeaseMetric("claimed_expired", time.Since(startedAt))
} else {
recordHeartbeatPrimaryLeaseMetric("claimed_unavailable", time.Since(startedAt))
}
recordHeartbeatPrimaryLeaseWriteMetric("claim")
return true, nil
}
}
recordHeartbeatPrimaryLeaseMetric("contention", time.Since(startedAt))
return s.resolveHeartbeatPrimaryAfterContention(ctx, tenantID, workspaceID, installationID)
}
func (s *MonitoringCallbackService) tryInsertHeartbeatPrimaryLease(ctx context.Context, tenantID, workspaceID int64, installationID, leaseTTL string) (bool, error) {
tag, err := s.businessPool.Exec(ctx, `
INSERT INTO desktop_client_primary_leases (
tenant_id,
workspace_id,
primary_client_id,
lease_expires_at
)
VALUES ($1, $2, $3::uuid, NOW() + $4::interval)
ON CONFLICT (tenant_id, workspace_id) DO NOTHING
`, tenantID, workspaceID, installationID, leaseTTL)
if err != nil {
return false, response.ErrInternal(50041, "primary_lease_insert_failed", "failed to initialize monitoring desktop primary lease")
}
return tag.RowsAffected() > 0, nil
}
type heartbeatPrimaryLeaseState struct {
Found bool
PrimaryInstallationID string
CurrentIsPrimary bool
LeaseLive bool
RenewDue bool
PrimaryAvailable bool
}
func (s *MonitoringCallbackService) loadHeartbeatPrimaryLeaseState(ctx context.Context, tenantID, workspaceID int64, installationID string) (heartbeatPrimaryLeaseState, error) {
var state heartbeatPrimaryLeaseState
renewAfter := formatPgInterval(monitoringHeartbeatPrimaryRenewAfter)
if err := s.businessPool.QueryRow(ctx, `
SELECT
l.primary_client_id::text,
l.primary_client_id = $3::uuid AS current_is_primary,
l.lease_expires_at > NOW() AS lease_live,
l.updated_at <= NOW() - $4::interval AS renew_due,
dc.id IS NOT NULL AND dc.revoked_at IS NULL AS primary_available
FROM desktop_client_primary_leases l
LEFT JOIN desktop_clients dc ON dc.id = l.primary_client_id
WHERE l.tenant_id = $1
AND l.workspace_id = $2
`, tenantID, workspaceID, installationID, renewAfter).Scan(
&state.PrimaryInstallationID,
&state.CurrentIsPrimary,
&state.LeaseLive,
&state.RenewDue,
&state.PrimaryAvailable,
); err != nil {
if err == pgx.ErrNoRows {
return state, nil
}
return state, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring desktop primary lease")
}
state.Found = true
return state, nil
}
type heartbeatPrimaryLeaseAction string
const (
heartbeatPrimaryLeaseActionPrimaryHit heartbeatPrimaryLeaseAction = "primary_hit"
heartbeatPrimaryLeaseActionNonPrimaryLive heartbeatPrimaryLeaseAction = "non_primary_live"
heartbeatPrimaryLeaseActionRenew heartbeatPrimaryLeaseAction = "renew"
heartbeatPrimaryLeaseActionClaimExpired heartbeatPrimaryLeaseAction = "claim_expired"
heartbeatPrimaryLeaseActionClaimUnavailable heartbeatPrimaryLeaseAction = "claim_unavailable"
)
type heartbeatPrimaryLeaseDecision struct {
Action heartbeatPrimaryLeaseAction
}
func decideHeartbeatPrimaryLease(state heartbeatPrimaryLeaseState) heartbeatPrimaryLeaseDecision {
if !state.Found {
return heartbeatPrimaryLeaseDecision{Action: heartbeatPrimaryLeaseActionNonPrimaryLive}
}
if state.CurrentIsPrimary {
if !state.LeaseLive || !state.PrimaryAvailable || state.RenewDue {
return heartbeatPrimaryLeaseDecision{Action: heartbeatPrimaryLeaseActionRenew}
}
return heartbeatPrimaryLeaseDecision{Action: heartbeatPrimaryLeaseActionPrimaryHit}
}
if !state.LeaseLive {
return heartbeatPrimaryLeaseDecision{Action: heartbeatPrimaryLeaseActionClaimExpired}
}
if !state.PrimaryAvailable {
return heartbeatPrimaryLeaseDecision{Action: heartbeatPrimaryLeaseActionClaimUnavailable}
}
return heartbeatPrimaryLeaseDecision{Action: heartbeatPrimaryLeaseActionNonPrimaryLive}
}
func (s *MonitoringCallbackService) renewHeartbeatPrimaryLease(ctx context.Context, tenantID, workspaceID int64, installationID, leaseTTL string) (bool, error) {
tag, err := s.businessPool.Exec(ctx, `
UPDATE desktop_client_primary_leases
SET lease_expires_at = NOW() + $4::interval,
updated_at = NOW()
WHERE tenant_id = $1
AND workspace_id = $2
AND primary_client_id = $3::uuid
`, tenantID, workspaceID, installationID, leaseTTL)
if err != nil {
return false, response.ErrInternal(50041, "primary_lease_update_failed", "failed to renew monitoring desktop primary lease")
}
return tag.RowsAffected() > 0, nil
}
func (s *MonitoringCallbackService) claimExpiredHeartbeatPrimaryLease(
ctx context.Context,
tenantID, workspaceID int64,
installationID string,
previousPrimaryInstallationID string,
leaseTTL string,
action heartbeatPrimaryLeaseAction,
) (bool, error) {
claimPredicate := "lease_expires_at <= NOW()"
if action == heartbeatPrimaryLeaseActionClaimUnavailable {
claimPredicate = `NOT EXISTS (
SELECT 1
FROM desktop_clients dc
WHERE dc.id = desktop_client_primary_leases.primary_client_id
AND dc.revoked_at IS NULL
)`
}
query := `
UPDATE desktop_client_primary_leases
SET primary_client_id = $3::uuid,
lease_expires_at = NOW() + $5::interval,
updated_at = NOW()
WHERE tenant_id = $1
AND workspace_id = $2
AND primary_client_id = $4::uuid
AND ` + claimPredicate
tag, err := s.businessPool.Exec(ctx, query, tenantID, workspaceID, installationID, previousPrimaryInstallationID, leaseTTL)
if err != nil {
return false, response.ErrInternal(50041, "primary_lease_update_failed", "failed to claim monitoring desktop primary lease")
}
return tag.RowsAffected() > 0, nil
}
func (s *MonitoringCallbackService) resolveHeartbeatPrimaryAfterContention(ctx context.Context, tenantID, workspaceID int64, installationID string) (bool, error) {
state, err := s.loadHeartbeatPrimaryLeaseState(ctx, tenantID, workspaceID, installationID)
if err != nil {
return false, err
}
return state.CurrentIsPrimary && state.LeaseLive && state.PrimaryAvailable, nil
}
func (s *MonitoringCallbackService) loadCollectTask(ctx context.Context, tenantID, taskID int64) (*monitoringCollectTask, error) {
item := &monitoringCollectTask{}
if err := s.monitoringPool.QueryRow(ctx, `
SELECT
id,
tenant_id,
brand_id,
question_id,
question_hash,
ai_platform_id,
collector_type,
run_mode,
business_date,
status,
COALESCE(execution_owner, 'legacy') AS execution_owner,
lease_token_hash,
leased_to_executor,
lease_expires_at,
callback_received_at,
completed_at,
skip_reason,
trigger_source,
COALESCE((
SELECT question_text_snapshot
FROM monitoring_question_config_snapshots
WHERE tenant_id = t.tenant_id
AND brand_id = t.brand_id
AND question_id = t.question_id
AND question_hash = t.question_hash
AND deleted_at IS NULL
ORDER BY projected_at DESC, id DESC
LIMIT 1
), '') AS question_text_snapshot
FROM monitoring_collect_tasks t
WHERE id = $1
AND tenant_id = $2
`, taskID, tenantID).Scan(
&item.ID,
&item.TenantID,
&item.BrandID,
&item.QuestionID,
&item.QuestionHash,
&item.AIPlatformID,
&item.CollectorType,
&item.RunMode,
&item.BusinessDate,
&item.Status,
&item.ExecutionOwner,
&item.LeaseTokenHash,
&item.LeasedToExecutor,
&item.LeaseExpiresAt,
&item.CallbackReceivedAt,
&item.CompletedAt,
&item.SkipReason,
&item.TriggerSource,
&item.QuestionText,
); err != nil {
if err == pgx.ErrNoRows {
return nil, response.ErrNotFound(40441, "monitoring_task_not_found", "monitoring task not found")
}
return nil, response.ErrInternal(50041, "query_failed", "failed to load monitoring task")
}
return item, nil
}
func (s *MonitoringCallbackService) loadCollectTaskByID(ctx context.Context, taskID int64) (*monitoringCollectTask, error) {
item := &monitoringCollectTask{}
if err := s.monitoringPool.QueryRow(ctx, `
SELECT
id,
tenant_id,
brand_id,
question_id,
question_hash,
ai_platform_id,
collector_type,
run_mode,
business_date,
status,
COALESCE(execution_owner, 'legacy') AS execution_owner,
lease_token_hash,
leased_to_executor,
lease_expires_at,
callback_received_at,
completed_at,
skip_reason,
trigger_source,
COALESCE((
SELECT question_text_snapshot
FROM monitoring_question_config_snapshots
WHERE tenant_id = t.tenant_id
AND brand_id = t.brand_id
AND question_id = t.question_id
AND question_hash = t.question_hash
AND deleted_at IS NULL
ORDER BY projected_at DESC, id DESC
LIMIT 1
), '') AS question_text_snapshot
FROM monitoring_collect_tasks t
WHERE id = $1
`, taskID).Scan(
&item.ID,
&item.TenantID,
&item.BrandID,
&item.QuestionID,
&item.QuestionHash,
&item.AIPlatformID,
&item.CollectorType,
&item.RunMode,
&item.BusinessDate,
&item.Status,
&item.ExecutionOwner,
&item.LeaseTokenHash,
&item.LeasedToExecutor,
&item.LeaseExpiresAt,
&item.CallbackReceivedAt,
&item.CompletedAt,
&item.SkipReason,
&item.TriggerSource,
&item.QuestionText,
); err != nil {
if err == pgx.ErrNoRows {
return nil, response.ErrNotFound(40441, "monitoring_task_not_found", "monitoring task not found")
}
return nil, response.ErrInternal(50041, "query_failed", "failed to load monitoring task")
}
return item, nil
}
func (s *MonitoringCallbackService) loadActiveDesktopMonitorExecutionLease(
ctx context.Context,
monitorTaskID int64,
) (*monitoringDesktopExecutionLease, error) {
item := &monitoringDesktopExecutionLease{}
if err := s.businessPool.QueryRow(ctx, `
SELECT
desktop_id,
status,
target_client_id,
lease_token_hash,
interrupt_generation
FROM desktop_tasks
WHERE kind = 'monitor'
AND monitor_task_id = $1
AND status IN ('queued', 'in_progress', 'unknown')
ORDER BY created_at DESC, id DESC
LIMIT 1
`, monitorTaskID).Scan(
&item.DesktopTaskID,
&item.Status,
&item.TargetClientID,
&item.LeaseTokenHash,
&item.InterruptGeneration,
); err != nil {
if err == pgx.ErrNoRows {
return nil, nil
}
return nil, response.ErrInternal(50041, "query_failed", "failed to inspect phase2 monitor desktop execution lease")
}
return item, nil
}
func (s *MonitoringCallbackService) selectPendingLeaseTasks(
ctx context.Context,
tx pgx.Tx,
installation *monitoringInstallationIdentity,
businessDate string,
limit int,
eligiblePlatformIDs []string,
) ([]monitoringLeasedTaskRow, error) {
queryArgs := []interface{}{installation.TenantID, monitoringCollectorType, businessDate, limit, installation.ID}
query := `
SELECT
t.id,
t.brand_id,
t.question_id,
t.question_hash,
t.ai_platform_id,
t.collector_type,
t.run_mode,
t.trigger_source,
t.business_date,
COALESCE(t.dispatch_priority, 100) AS dispatch_priority,
COALESCE(t.dispatch_lane, 'normal') AS dispatch_lane,
COALESCE((
SELECT question_text_snapshot
FROM monitoring_question_config_snapshots
WHERE tenant_id = t.tenant_id
AND brand_id = t.brand_id
AND question_id = t.question_id
AND question_hash = t.question_hash
AND deleted_at IS NULL
ORDER BY projected_at DESC, id DESC
LIMIT 1
), '') AS question_text_snapshot
FROM monitoring_collect_tasks t
WHERE t.tenant_id = $1
AND t.collector_type = $2
AND t.status = 'pending'
AND t.business_date = $3::date
AND COALESCE(t.execution_owner, 'legacy') = 'legacy'
AND (t.dispatch_after IS NULL OR t.dispatch_after <= NOW())
AND (
t.target_client_id IS NULL
OR t.target_client_id = $5::uuid
)
AND EXISTS (
SELECT 1
FROM monitoring_question_config_snapshots s
WHERE s.tenant_id = t.tenant_id
AND s.brand_id = t.brand_id
AND s.question_id = t.question_id
AND s.question_hash = t.question_hash
AND s.monitor_enabled = TRUE
AND s.deleted_at IS NULL
AND s.superseded_at IS NULL
)
`
if len(eligiblePlatformIDs) > 0 {
query += `
AND t.ai_platform_id = ANY($6::text[])
`
queryArgs = append(queryArgs, eligiblePlatformIDs)
}
query += `
ORDER BY CASE t.dispatch_lane
WHEN 'high' THEN 70
WHEN 'normal_boosted' THEN 50
WHEN 'normal' THEN 40
WHEN 'retry' THEN 20
ELSE 0
END DESC,
t.dispatch_priority DESC,
t.dispatch_after ASC NULLS FIRST,
t.planned_at ASC,
t.id ASC
FOR UPDATE SKIP LOCKED
LIMIT $4
`
rows, err := tx.Query(ctx, query, queryArgs...)
if err != nil {
return nil, response.ErrInternal(50041, "lease_query_failed", "failed to select monitoring lease tasks")
}
defer rows.Close()
result := make([]monitoringLeasedTaskRow, 0, limit)
for rows.Next() {
var item monitoringLeasedTaskRow
if scanErr := rows.Scan(
&item.ID,
&item.BrandID,
&item.QuestionID,
&item.QuestionHash,
&item.AIPlatformID,
&item.CollectorType,
&item.RunMode,
&item.TriggerSource,
&item.BusinessDate,
&item.Priority,
&item.Lane,
&item.QuestionText,
); scanErr != nil {
return nil, response.ErrInternal(50041, "lease_scan_failed", "failed to parse monitoring lease task")
}
result = append(result, item)
}
if err := rows.Err(); err != nil {
return nil, response.ErrInternal(50041, "lease_scan_failed", "failed to iterate monitoring lease tasks")
}
return result, nil
}
func (s *MonitoringCallbackService) selectLeasedTasksForInstallation(
ctx context.Context,
tx pgx.Tx,
tenantID int64,
installationID string,
businessDate string,
eligiblePlatformIDs []string,
) ([]monitoringLeasedTaskRow, error) {
queryArgs := []interface{}{tenantID, monitoringCollectorType, businessDate, installationID}
query := `
SELECT
t.id,
t.brand_id,
t.question_id,
t.question_hash,
t.ai_platform_id,
t.collector_type,
t.run_mode,
t.trigger_source,
t.business_date,
COALESCE(t.dispatch_priority, 100) AS dispatch_priority,
COALESCE(t.dispatch_lane, 'normal') AS dispatch_lane,
COALESCE((
SELECT question_text_snapshot
FROM monitoring_question_config_snapshots
WHERE tenant_id = t.tenant_id
AND brand_id = t.brand_id
AND question_id = t.question_id
AND question_hash = t.question_hash
AND deleted_at IS NULL
ORDER BY projected_at DESC, id DESC
LIMIT 1
), '') AS question_text_snapshot
FROM monitoring_collect_tasks t
WHERE t.tenant_id = $1
AND t.collector_type = $2
AND t.status = 'leased'
AND t.business_date = $3::date
AND COALESCE(t.execution_owner, 'legacy') = 'legacy'
AND t.leased_to_executor = $4
AND EXISTS (
SELECT 1
FROM monitoring_question_config_snapshots s
WHERE s.tenant_id = t.tenant_id
AND s.brand_id = t.brand_id
AND s.question_id = t.question_id
AND s.question_hash = t.question_hash
AND s.monitor_enabled = TRUE
AND s.deleted_at IS NULL
AND s.superseded_at IS NULL
)
`
if len(eligiblePlatformIDs) > 0 {
query += `
AND t.ai_platform_id = ANY($5::text[])
`
queryArgs = append(queryArgs, eligiblePlatformIDs)
}
query += `
ORDER BY t.leased_at ASC NULLS LAST, t.planned_at ASC, t.id ASC
FOR UPDATE SKIP LOCKED
`
rows, err := tx.Query(ctx, query, queryArgs...)
if err != nil {
return nil, response.ErrInternal(50041, "resume_query_failed", "failed to select monitoring resumed tasks")
}
defer rows.Close()
result := make([]monitoringLeasedTaskRow, 0)
for rows.Next() {
var item monitoringLeasedTaskRow
if scanErr := rows.Scan(
&item.ID,
&item.BrandID,
&item.QuestionID,
&item.QuestionHash,
&item.AIPlatformID,
&item.CollectorType,
&item.RunMode,
&item.TriggerSource,
&item.BusinessDate,
&item.Priority,
&item.Lane,
&item.QuestionText,
); scanErr != nil {
return nil, response.ErrInternal(50041, "resume_scan_failed", "failed to parse monitoring resumed task")
}
result = append(result, item)
}
if err := rows.Err(); err != nil {
return nil, response.ErrInternal(50041, "resume_scan_failed", "failed to iterate monitoring resumed tasks")
}
return result, nil
}
func (s *MonitoringCallbackService) markTaskLeased(ctx context.Context, tx pgx.Tx, installationID string, taskID int64, leaseToken string, leasedAt, leaseExpiresAt time.Time) error {
if _, err := tx.Exec(ctx, `
UPDATE monitoring_collect_tasks
SET status = 'leased',
lease_token_hash = $2,
leased_to_executor = $3,
leased_at = $4,
lease_expires_at = $5,
last_dispatched_at = NOW(),
dispatch_attempts = COALESCE(dispatch_attempts, 0) + 1,
skip_reason = NULL,
error_message = NULL,
updated_at = NOW()
WHERE id = $1
`, taskID, sharedauth.HashToken(strings.TrimSpace(leaseToken)), installationID, leasedAt, leaseExpiresAt); err != nil {
return response.ErrInternal(50041, "lease_update_failed", "failed to update monitoring lease task")
}
return nil
}
func (s *MonitoringCallbackService) markTaskResultReceived(ctx context.Context, tx pgx.Tx, task *monitoringCollectTask, receivedAt time.Time, requestPayloadJSON []byte) error {
if task == nil {
return response.ErrBadRequest(40041, "invalid_task_id", "monitoring task is required")
}
nextStatus := "received"
if task.ExecutionOwner == "desktop_tasks" {
nextStatus = "pending"
}
if _, err := tx.Exec(ctx, `
UPDATE monitoring_collect_tasks
SET status = $2,
callback_received_at = $3,
completed_at = NULL,
skip_reason = NULL,
error_message = NULL,
request_payload_json = (
CASE
WHEN request_payload_json IS NULL OR jsonb_typeof(request_payload_json) <> 'object'
THEN '{}'::jsonb
ELSE request_payload_json
END
) || $4::jsonb,
updated_at = NOW()
WHERE id = $1
`, task.ID, nextStatus, receivedAt, requestPayloadJSON); err != nil {
return response.ErrInternal(50041, "task_receive_failed", "failed to persist monitoring callback payload")
}
return nil
}
func (s *MonitoringCallbackService) upsertPlatformAccessSnapshot(ctx context.Context, tx pgx.Tx, installation *monitoringInstallationIdentity, businessDate string, platform MonitoringHeartbeatPlatform) error {
if _, err := tx.Exec(ctx, `
INSERT INTO monitoring_platform_access_snapshots (
tenant_id, client_id, ai_platform_id, business_date, access_status,
connected, accessible, detected_at, reason_code, reason_message
)
VALUES ($1, $2::uuid, $3, $4::date, $5, $6, $7, $8, $9, $10)
ON CONFLICT (tenant_id, client_id, ai_platform_id, business_date)
DO UPDATE SET
access_status = EXCLUDED.access_status,
connected = EXCLUDED.connected,
accessible = EXCLUDED.accessible,
detected_at = EXCLUDED.detected_at,
reason_code = EXCLUDED.reason_code,
reason_message = EXCLUDED.reason_message,
updated_at = NOW()
`,
installation.TenantID,
installation.ID,
platform.AIPlatformID,
businessDate,
monitoringStringValue(platform.AccessStatus),
derefBool(platform.Connected),
derefBool(platform.Accessible),
resolveHeartbeatDetectedAt(platform.DetectedAt),
nullableString(platform.ReasonCode),
nullableString(platform.ReasonMessage),
); err != nil {
return response.ErrInternal(50041, "access_snapshot_upsert_failed", "failed to persist monitoring platform access snapshot")
}
return nil
}
func (s *MonitoringCallbackService) loadPlatformAccessSnapshotStates(ctx context.Context, tenantID int64, installationID string, businessDate string, aiPlatformIDs []string) (map[string]*monitoringPlatformAccessSnapshotState, error) {
states := make(map[string]*monitoringPlatformAccessSnapshotState, len(aiPlatformIDs))
if len(aiPlatformIDs) == 0 {
return states, nil
}
rows, err := s.monitoringPool.Query(ctx, `
SELECT ai_platform_id, access_status, connected, accessible, reason_code, reason_message, updated_at
FROM monitoring_platform_access_snapshots
WHERE tenant_id = $1
AND client_id = $2::uuid
AND business_date = $3::date
AND ai_platform_id = ANY($4::text[])
`, tenantID, installationID, businessDate, aiPlatformIDs)
if err != nil {
return nil, response.ErrInternal(50041, "access_snapshot_lookup_failed", "failed to inspect monitoring platform access snapshots")
}
defer rows.Close()
for rows.Next() {
item := &monitoringPlatformAccessSnapshotState{}
var platformID string
var reasonCode sql.NullString
var reasonMessage sql.NullString
if scanErr := rows.Scan(
&platformID,
&item.AccessStatus,
&item.Connected,
&item.Accessible,
&reasonCode,
&reasonMessage,
&item.UpdatedAt,
); scanErr != nil {
return nil, response.ErrInternal(50041, "access_snapshot_scan_failed", "failed to parse monitoring platform access snapshots")
}
if reasonCode.Valid {
item.ReasonCode = optionalString(reasonCode.String)
}
if reasonMessage.Valid {
item.ReasonMessage = optionalString(reasonMessage.String)
}
states[platformID] = item
}
if err := rows.Err(); err != nil {
return nil, response.ErrInternal(50041, "access_snapshot_scan_failed", "failed to iterate monitoring platform access snapshots")
}
return states, nil
}
func monitoringPlatformAccessSnapshotStateFromHeartbeat(platform MonitoringHeartbeatPlatform) monitoringPlatformAccessSnapshotState {
return monitoringPlatformAccessSnapshotState{
AccessStatus: monitoringStringValue(platform.AccessStatus),
Connected: derefBool(platform.Connected),
Accessible: derefBool(platform.Accessible),
ReasonCode: optionalString(monitoringStringValue(platform.ReasonCode)),
ReasonMessage: optionalString(monitoringStringValue(platform.ReasonMessage)),
}
}
func decideMonitoringHeartbeatSnapshot(previous *monitoringPlatformAccessSnapshotState, next monitoringPlatformAccessSnapshotState, now time.Time, isPrimary bool) monitoringHeartbeatSnapshotDecision {
stateChanged := hasPlatformAccessSnapshotChanged(previous, next)
refreshDue := false
if previous != nil && !stateChanged {
refreshDue = previous.UpdatedAt.IsZero() || !previous.UpdatedAt.After(now.Add(-monitoringHeartbeatAccessSnapshotRefreshAfter))
}
operation := "skip"
if previous == nil {
operation = "insert"
} else if stateChanged {
operation = "update"
} else if refreshDue {
operation = "refresh"
}
return monitoringHeartbeatSnapshotDecision{
WriteSnapshot: stateChanged || refreshDue,
StateChanged: stateChanged,
RefreshDue: refreshDue,
PendingReconcileCandidate: isPrimary && next.AccessStatus != "accessible",
SnapshotOperation: operation,
}
}
func hasPlatformAccessSnapshotChanged(previous *monitoringPlatformAccessSnapshotState, next monitoringPlatformAccessSnapshotState) bool {
if previous == nil {
return true
}
if previous.AccessStatus != next.AccessStatus ||
previous.Connected != next.Connected ||
previous.Accessible != next.Accessible {
return true
}
if monitoringStringValue(previous.ReasonCode) != monitoringStringValue(next.ReasonCode) {
return true
}
if monitoringStringValue(previous.ReasonMessage) != monitoringStringValue(next.ReasonMessage) {
return true
}
return false
}
func (s *MonitoringCallbackService) loadPendingMonitoringTaskPlatformIDs(ctx context.Context, tenantID int64, businessDate string, aiPlatformIDs []string) (map[string]struct{}, error) {
result := make(map[string]struct{}, len(aiPlatformIDs))
if len(aiPlatformIDs) == 0 {
return result, nil
}
rows, err := s.monitoringPool.Query(ctx, `
SELECT ai_platform_id
FROM monitoring_collect_tasks
WHERE tenant_id = $1
AND ai_platform_id = ANY($2::text[])
AND collector_type = $3
AND business_date = $4::date
AND status = 'pending'
GROUP BY ai_platform_id
`, tenantID, aiPlatformIDs, monitoringCollectorType, businessDate)
if err != nil {
return nil, response.ErrInternal(50041, "pending_task_platform_lookup_failed", "failed to inspect pending monitoring task platforms")
}
defer rows.Close()
for rows.Next() {
var platformID string
if scanErr := rows.Scan(&platformID); scanErr != nil {
return nil, response.ErrInternal(50041, "pending_task_platform_scan_failed", "failed to parse pending monitoring task platforms")
}
result[platformID] = struct{}{}
}
if err := rows.Err(); err != nil {
return nil, response.ErrInternal(50041, "pending_task_platform_scan_failed", "failed to iterate pending monitoring task platforms")
}
return result, nil
}
func (s *MonitoringCallbackService) reconcilePendingTasksForPlatform(ctx context.Context, tx pgx.Tx, tenantID int64, businessDate, aiPlatformID string) (int64, error) {
tag, err := tx.Exec(ctx, `
UPDATE monitoring_collect_tasks
SET status = 'skipped',
skip_reason = 'platform_not_accessible',
error_message = 'platform heartbeat reported not accessible',
completed_at = NOW(),
updated_at = NOW()
WHERE tenant_id = $1
AND ai_platform_id = $2
AND collector_type = $3
AND business_date = $4::date
AND status = 'pending'
`, tenantID, aiPlatformID, monitoringCollectorType, businessDate)
if err != nil {
return 0, response.ErrInternal(50041, "task_reconcile_failed", "failed to reconcile pending monitoring tasks")
}
return tag.RowsAffected(), nil
}
func (s *MonitoringCallbackService) upsertRun(ctx context.Context, tx pgx.Tx, task *monitoringCollectTask, req MonitoringTaskResultRequest, runStatus string, completedAt time.Time, rawPayload []byte) (int64, error) {
var runID int64
if err := tx.QueryRow(ctx, `
INSERT INTO question_monitor_runs (
tenant_id, brand_id, question_id, question_text_snapshot, question_hash,
ai_platform_id, collector_type, trigger_source, run_mode, business_date,
provider_model, provider_request_id, request_id, raw_answer_text, raw_response_json,
status, completed_at
)
VALUES (
$1, $2, $3, $4, $5,
$6, $7, $8, $9, $10::date,
$11, $12, $13, $14, $15,
$16, $17
)
ON CONFLICT (
tenant_id, brand_id, question_id, ai_platform_id, collector_type, run_mode, business_date
)
DO UPDATE SET
question_text_snapshot = EXCLUDED.question_text_snapshot,
question_hash = EXCLUDED.question_hash,
provider_model = EXCLUDED.provider_model,
provider_request_id = EXCLUDED.provider_request_id,
request_id = EXCLUDED.request_id,
raw_answer_text = EXCLUDED.raw_answer_text,
raw_response_json = EXCLUDED.raw_response_json,
status = EXCLUDED.status,
completed_at = EXCLUDED.completed_at,
updated_at = NOW()
RETURNING id
`,
task.TenantID,
task.BrandID,
task.QuestionID,
task.QuestionText,
task.QuestionHash,
task.AIPlatformID,
task.CollectorType,
task.TriggerSource,
task.RunMode,
task.BusinessDate.Format("2006-01-02"),
nullableString(req.ProviderModel),
nullableString(req.ProviderRequestID),
nullableString(req.RequestID),
nullableString(req.Answer),
rawPayload,
runStatus,
completedAt,
).Scan(&runID); err != nil {
return 0, response.ErrInternal(50041, "run_upsert_failed", "failed to persist monitoring run")
}
return runID, nil
}
func (s *MonitoringCallbackService) upsertParseResult(ctx context.Context, tx pgx.Tx, task *monitoringCollectTask, runID int64, req MonitoringTaskResultRequest, parseVersion string) error {
matchedTermsJSON, err := json.Marshal(req.MatchedBrandTerms)
if err != nil {
return response.ErrInternal(50041, "parse_payload_failed", "failed to encode matched brand terms")
}
if strings.TrimSpace(parseVersion) == "" {
parseVersion = "server_heuristic_v1"
}
if _, err := tx.Exec(ctx, `
INSERT INTO question_monitor_parse_results (
tenant_id, run_id, brand_id, question_id, question_hash, ai_platform_id, business_date,
brand_mentioned, brand_mention_position, first_recommended, sentiment_label, matched_brand_terms,
parse_version, parse_status
)
VALUES (
$1, $2, $3, $4, $5, $6, $7::date,
$8, $9, $10, $11, $12,
$13, 'succeeded'
)
ON CONFLICT (tenant_id, run_id)
DO UPDATE SET
brand_mentioned = EXCLUDED.brand_mentioned,
brand_mention_position = EXCLUDED.brand_mention_position,
first_recommended = EXCLUDED.first_recommended,
sentiment_label = EXCLUDED.sentiment_label,
matched_brand_terms = EXCLUDED.matched_brand_terms,
parse_version = EXCLUDED.parse_version,
parse_status = EXCLUDED.parse_status,
updated_at = NOW()
`,
task.TenantID,
runID,
task.BrandID,
task.QuestionID,
task.QuestionHash,
task.AIPlatformID,
task.BusinessDate.Format("2006-01-02"),
nullableBool(req.BrandMentioned),
nullableString(req.BrandMentionPosition),
nullableBool(req.FirstRecommended),
nullableString(req.SentimentLabel),
matchedTermsJSON,
parseVersion,
); err != nil {
return response.ErrInternal(50041, "parse_result_upsert_failed", "failed to persist parse result")
}
return nil
}
func (s *MonitoringCallbackService) processTaskResult(ctx context.Context, task *monitoringCollectTask, req MonitoringTaskResultRequest) (*MonitoringTaskResultResponse, error) {
req = normalizeMonitoringTaskResultRequestValue(req)
runStatus := normalizeRunStatus(req.Status)
if runStatus == "" {
return nil, response.ErrBadRequest(40041, "invalid_status", "status must be succeeded or failed")
}
parseReq := req
parseVersion := ""
if runStatus == "succeeded" {
var enrichErr error
parseReq, parseVersion, enrichErr = s.enrichMonitoringParsePayload(ctx, task, req)
if enrichErr != nil {
return nil, enrichErr
}
}
completedAt := time.Now().UTC().Round(0)
if req.CompletedAt != nil && !req.CompletedAt.IsZero() {
completedAt = req.CompletedAt.UTC().Round(0)
}
rawPayload, sourceInputs := buildMonitoringRawPayload(task.AIPlatformID, req)
tx, err := s.monitoringPool.Begin(ctx)
if err != nil {
return nil, response.ErrInternal(50041, "begin_failed", "failed to start monitoring result ingest")
}
defer tx.Rollback(ctx)
runID, err := s.upsertRun(ctx, tx, task, req, runStatus, completedAt, rawPayload)
if err != nil {
return nil, err
}
if runStatus == "succeeded" {
if err := s.upsertParseResult(ctx, tx, task, runID, parseReq, parseVersion); err != nil {
return nil, err
}
}
citationSourceCount := 0
contentCitationCount := 0
if err := s.replaceCitationFacts(ctx, tx, task.TenantID, runID); err != nil {
return nil, err
}
if runStatus == "succeeded" {
facts, buildErr := s.buildCitationFacts(ctx, tx, task.TenantID, task.AIPlatformID, sourceInputs)
if buildErr != nil {
return nil, buildErr
}
for _, fact := range facts {
if err := s.insertCitationFact(ctx, tx, task, runID, fact); err != nil {
return nil, err
}
citationSourceCount++
if fact.ArticleID != nil {
contentCitationCount++
}
}
}
taskStatus := "completed"
if runStatus == "failed" {
taskStatus = "failed"
}
if err := s.updateCollectTaskStatus(ctx, tx, task.ID, taskStatus, completedAt, req.ErrorMessage); err != nil {
return nil, err
}
if err := s.rebuildBrandDailyProjection(ctx, tx, task.TenantID, task.BrandID, task.BusinessDate); err != nil {
return nil, err
}
if err := tx.Commit(ctx); err != nil {
return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring result ingest")
}
return &MonitoringTaskResultResponse{
TaskID: task.ID,
RunID: &runID,
TaskStatus: taskStatus,
CitationSourceCount: citationSourceCount,
ContentCitationCount: contentCitationCount,
}, nil
}
func buildDesktopMonitorTaskResultValue(task *monitoringCollectTask, req MonitoringTaskResultRequest, runStatus string) map[string]any {
result := map[string]any{
"monitoring_task_id": task.ID,
"status": runStatus,
}
if task != nil {
if strings.TrimSpace(task.AIPlatformID) != "" {
result["platform"] = strings.TrimSpace(task.AIPlatformID)
}
if strings.TrimSpace(task.QuestionText) != "" {
result["question_text"] = strings.TrimSpace(task.QuestionText)
}
}
if value := strings.TrimSpace(optionalStringValue(req.ProviderModel)); value != "" {
result["provider_model"] = value
}
if value := strings.TrimSpace(optionalStringValue(req.ProviderRequestID)); value != "" {
result["provider_request_id"] = value
}
if value := strings.TrimSpace(optionalStringValue(req.RequestID)); value != "" {
result["request_id"] = value
}
if value := strings.TrimSpace(optionalStringValue(req.Answer)); value != "" {
result["answer"] = value
}
if len(req.Citations) > 0 {
result["citations"] = req.Citations
}
if len(req.SearchResults) > 0 {
result["search_results"] = req.SearchResults
}
if len(req.RawResponseJSON) > 0 {
result["raw_response_json"] = req.RawResponseJSON
}
return result
}
func buildDesktopMonitorTaskErrorValue(task *monitoringCollectTask, req MonitoringTaskResultRequest) map[string]any {
message := strings.TrimSpace(optionalStringValue(req.ErrorMessage))
if message == "" {
message = "monitoring task failed"
}
result := map[string]any{
"code": "monitoring_task_failed",
"message": message,
"monitoring_task_id": task.ID,
}
if task != nil && strings.TrimSpace(task.AIPlatformID) != "" {
result["platform"] = strings.TrimSpace(task.AIPlatformID)
}
if len(req.RawResponseJSON) > 0 {
result["raw_response_json"] = req.RawResponseJSON
}
return result
}
func (s *MonitoringCallbackService) completeLinkedDesktopMonitorTask(
ctx context.Context,
task *monitoringCollectTask,
req MonitoringTaskResultRequest,
) error {
if s == nil || s.businessPool == nil || task == nil || task.ExecutionOwner != "desktop_tasks" {
return nil
}
runStatus := normalizeRunStatus(req.Status)
if runStatus == "" {
return nil
}
tx, err := s.businessPool.Begin(ctx)
if err != nil {
return response.ErrInternal(50041, "desktop_task_complete_begin_failed", "failed to start phase2 desktop task completion")
}
defer tx.Rollback(ctx)
var (
desktopID uuid.UUID
workspaceID int64
activeAttemptID pgtype.UUID
)
if err := tx.QueryRow(ctx, `
SELECT desktop_id, workspace_id, active_attempt_id
FROM desktop_tasks
WHERE kind = 'monitor'
AND monitor_task_id = $1
AND status IN ('in_progress', 'unknown')
ORDER BY created_at DESC, id DESC
LIMIT 1
FOR UPDATE
`, task.ID).Scan(&desktopID, &workspaceID, &activeAttemptID); err != nil {
if err == pgx.ErrNoRows {
return nil
}
return response.ErrInternal(50041, "desktop_task_lookup_failed", "failed to inspect phase2 desktop task state")
}
resultJSON, err := marshalOptionalJSON(buildDesktopMonitorTaskResultValue(task, req, runStatus))
if err != nil {
return response.ErrInternal(50041, "desktop_task_payload_encode_failed", "failed to encode phase2 desktop task result payload")
}
var errorJSON []byte
if runStatus == "failed" {
errorJSON, err = marshalOptionalJSON(buildDesktopMonitorTaskErrorValue(task, req))
if err != nil {
return response.ErrInternal(50041, "desktop_task_payload_encode_failed", "failed to encode phase2 desktop task error payload")
}
}
if _, err := tx.Exec(ctx, `
UPDATE desktop_tasks
SET status = $2,
result = $3,
error = $4,
active_attempt_id = NULL,
lease_token_hash = NULL,
lease_expires_at = NULL,
updated_at = NOW()
WHERE desktop_id = $1
`, desktopID, runStatus, resultJSON, errorJSON); err != nil {
return response.ErrInternal(50041, "desktop_task_complete_failed", "failed to finalize phase2 desktop task")
}
taskRepo := repository.NewDesktopTaskRepository(tx)
if activeAttemptID.Valid {
resolvedAttemptID, convErr := uuid.FromBytes(activeAttemptID.Bytes[:])
if convErr != nil {
return response.ErrInternal(50041, "desktop_task_attempt_decode_failed", "failed to decode phase2 desktop task attempt id")
}
if finishErr := taskRepo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{
TaskID: desktopID,
AttemptID: resolvedAttemptID,
FinalStatus: &runStatus,
Error: errorJSON,
}); finishErr != nil {
return response.ErrInternal(50041, "desktop_task_attempt_finish_failed", "failed to finalize phase2 desktop task attempt")
}
}
finalizedTask, err := taskRepo.GetByDesktopID(ctx, desktopID, workspaceID)
if err != nil {
return response.ErrInternal(50041, "desktop_task_reload_failed", "failed to reload finalized phase2 desktop task")
}
if err := tx.Commit(ctx); err != nil {
return response.ErrInternal(50041, "desktop_task_complete_commit_failed", "failed to commit phase2 desktop task completion")
}
title, businessDate, schedulerGroupKey, questionText := deriveDesktopTaskEventMetadataFromPayload(
finalizedTask.Kind,
finalizedTask.Payload,
)
if publishErr := publishDesktopTaskEvent(ctx, s.rabbitMQ, DesktopTaskEvent{
Type: "task_completed",
TaskID: finalizedTask.DesktopID.String(),
JobID: finalizedTask.JobID.String(),
WorkspaceID: finalizedTask.WorkspaceID,
TargetAccountID: finalizedTask.TargetAccountID.String(),
TargetClientID: finalizedTask.TargetClientID.String(),
Platform: finalizedTask.Platform,
Title: title,
BusinessDate: businessDate,
SchedulerGroupKey: schedulerGroupKey,
QuestionText: questionText,
Status: finalizedTask.Status,
Kind: finalizedTask.Kind,
Priority: finalizedTask.Priority,
Lane: finalizedTask.Lane,
InterruptGeneration: finalizedTask.InterruptGeneration,
UpdatedAt: finalizedTask.UpdatedAt,
}); publishErr != nil && s.logger != nil {
s.logger.Warn("phase2 desktop task completion event publish failed",
zap.Int64("monitor_task_id", task.ID),
zap.String("desktop_task_id", finalizedTask.DesktopID.String()),
zap.Error(publishErr),
)
}
return nil
}
func (s *MonitoringCallbackService) replaceCitationFacts(ctx context.Context, tx pgx.Tx, tenantID, runID int64) error {
if _, err := tx.Exec(ctx, `
DELETE FROM monitoring_citation_facts
WHERE tenant_id = $1 AND run_id = $2
`, tenantID, runID); err != nil {
return response.ErrInternal(50041, "citation_cleanup_failed", "failed to reset existing citation facts")
}
return nil
}
func (s *MonitoringCallbackService) loadMonitoringBrandName(ctx context.Context, tenantID, brandID int64) (string, error) {
var brandName string
if err := s.businessPool.QueryRow(ctx, `
SELECT name
FROM brands
WHERE id = $1
AND tenant_id = $2
AND deleted_at IS NULL
`, brandID, tenantID).Scan(&brandName); err != nil {
if err == pgx.ErrNoRows {
return "", response.ErrInternal(50041, "brand_lookup_failed", "failed to resolve monitoring brand")
}
return "", response.ErrInternal(50041, "brand_lookup_failed", "failed to resolve monitoring brand")
}
return strings.TrimSpace(brandName), nil
}
func (s *MonitoringCallbackService) enrichMonitoringParsePayload(ctx context.Context, task *monitoringCollectTask, req MonitoringTaskResultRequest) (MonitoringTaskResultRequest, string, error) {
brandName, err := s.loadMonitoringBrandName(ctx, task.TenantID, task.BrandID)
if err != nil {
return req, "", err
}
answer := monitoringStringValue(req.Answer)
stored := buildMonitoringStoredParseFields(req)
summary := resolveMonitoringAnswerParseSummary(answer, brandName, stored)
parseVersion := resolveMonitoringParseVersion(req)
if shouldUseMonitoringLLMParse(req) && strings.TrimSpace(answer) != "" && s.llm != nil && s.llm.Validate() == nil {
llmSummary, llmErr := parseMonitoringAnswerWithLLM(ctx, s.llm, answer, brandName)
if llmErr != nil {
if s.logger != nil {
s.logger.Warn("monitoring answer llm parse failed",
zap.Int64("task_id", task.ID),
zap.Int64("tenant_id", task.TenantID),
zap.Int64("brand_id", task.BrandID),
zap.Int64("question_id", task.QuestionID),
zap.String("ai_platform_id", task.AIPlatformID),
zap.Error(llmErr),
)
}
} else {
summary = overlayMonitoringStoredParseSummary(llmSummary, stored)
if !hasMonitoringParsePayload(req) {
parseVersion = "server_llm_v1"
}
}
}
enriched := req
enriched.BrandMentioned = monitoringBoolPointer(summary.BrandMentioned)
enriched.BrandMentionPosition = optionalString(summary.BrandMentionPosition)
enriched.FirstRecommended = monitoringBoolPointer(summary.FirstRecommended)
enriched.SentimentLabel = optionalString(summary.SentimentLabel)
enriched.MatchedBrandTerms = append([]string(nil), summary.MatchedBrandTerms...)
return enriched, parseVersion, nil
}
func buildMonitoringStoredParseFields(req MonitoringTaskResultRequest) monitoringStoredParseFields {
return monitoringStoredParseFields{
BrandMentioned: cloneBool(req.BrandMentioned),
BrandMentionPosition: cloneString(req.BrandMentionPosition),
FirstRecommended: cloneBool(req.FirstRecommended),
SentimentLabel: cloneString(req.SentimentLabel),
MatchedBrandTerms: append([]string(nil), req.MatchedBrandTerms...),
}
}
func resolveMonitoringParseVersion(req MonitoringTaskResultRequest) string {
if hasMonitoringParsePayload(req) {
return "plugin_v1"
}
return "server_heuristic_v1"
}
func shouldUseMonitoringLLMParse(req MonitoringTaskResultRequest) bool {
return !hasCompleteMonitoringParsePayload(req)
}
func (s *MonitoringCallbackService) buildCitationFacts(ctx context.Context, tx pgx.Tx, tenantID int64, platformID string, inputs []MonitoringSourceItem) ([]monitoringCitationFact, error) {
if len(inputs) == 0 {
return nil, nil
}
resolveContentCitations := monitoringPlatformResolvesContentCitations(platformID)
aliasMap := map[string]monitoringAliasResolution{}
if resolveContentCitations {
lookupInputs := make([]monitoringAliasLookupInput, 0, len(inputs))
seenURLs := make(map[string]struct{}, len(inputs))
for _, item := range inputs {
key := strings.TrimSpace(monitoringStringValue(item.NormalizedURL))
if key == "" {
key = normalizeCitationURL(item.URL)
}
if key == "" {
continue
}
if _, ok := seenURLs[key]; ok {
continue
}
seenURLs[key] = struct{}{}
host, registrableDomain, _, _ := deriveCitationURLParts(item.URL, key)
siteKey := strings.TrimSpace(monitoringStringValue(item.SiteKey))
if siteKey == "" {
siteKey = registrableDomain
}
lookupInputs = append(lookupInputs, monitoringAliasLookupInput{
NormalizedURL: key,
MatchKeys: monitoringCitationCandidateKeys(item.URL, key),
RegistrableDomain: strings.ToLower(strings.TrimSpace(registrableDomain)),
SiteKey: strings.ToLower(strings.TrimSpace(siteKey)),
Host: strings.ToLower(strings.TrimSpace(host)),
LastPathSegment: monitoringAliasLastPathSegment(key),
TitleKey: normalizeMonitoringAliasTitle(monitoringStringValue(item.Title)),
})
}
var err error
aliasMap, err = s.loadAliasResolutions(ctx, tx, tenantID, lookupInputs)
if err != nil {
return nil, err
}
}
result := make([]monitoringCitationFact, 0, len(inputs))
seenFacts := make(map[string]struct{}, len(inputs))
for _, input := range inputs {
fact, ok := buildMonitoringCitationFact(input, aliasMap, resolveContentCitations)
if !ok {
continue
}
dedupKey := fact.NormalizedURL + "|" + strconv.FormatInt(derefInt64(fact.ArticleID), 10) + "|" + strconv.FormatInt(derefInt64(fact.PublishRecordID), 10)
if _, exists := seenFacts[dedupKey]; exists {
continue
}
seenFacts[dedupKey] = struct{}{}
result = append(result, fact)
}
return result, nil
}
func (s *MonitoringCallbackService) loadAliasResolutions(ctx context.Context, tx pgx.Tx, tenantID int64, lookupInputs []monitoringAliasLookupInput) (map[string]monitoringAliasResolution, error) {
result := make(map[string]monitoringAliasResolution)
if len(lookupInputs) == 0 {
return result, nil
}
normalizedURLs := make([]string, 0, len(lookupInputs))
domainKeys := make([]string, 0, len(lookupInputs)*3)
seenDomainKeys := map[string]struct{}{}
for _, input := range lookupInputs {
if input.NormalizedURL != "" {
normalizedURLs = append(normalizedURLs, input.NormalizedURL)
}
for _, key := range []string{input.RegistrableDomain, input.SiteKey, input.Host} {
key = strings.ToLower(strings.TrimSpace(key))
if key == "" {
continue
}
if _, ok := seenDomainKeys[key]; ok {
continue
}
seenDomainKeys[key] = struct{}{}
domainKeys = append(domainKeys, key)
}
}
rows, err := tx.Query(ctx, `
SELECT
normalized_url,
article_id,
publish_record_id,
registrable_domain,
site_key,
host,
last_path_segment,
article_title_snapshot
FROM monitoring_article_url_aliases
WHERE tenant_id = $1
AND (
normalized_url = ANY($2::text[])
OR NULLIF(LOWER(TRIM(registrable_domain)), '') = ANY($3::text[])
OR NULLIF(LOWER(TRIM(site_key)), '') = ANY($3::text[])
OR NULLIF(LOWER(TRIM(host)), '') = ANY($3::text[])
)
`, tenantID, normalizedURLs, domainKeys)
if err != nil {
return nil, response.ErrInternal(50041, "alias_lookup_failed", "failed to resolve article aliases")
}
defer rows.Close()
candidates := make([]monitoringAliasCandidate, 0)
for rows.Next() {
var candidate monitoringAliasCandidate
var articleID sql.NullInt64
var publishRecordID sql.NullInt64
var registrableDomain sql.NullString
var siteKey sql.NullString
var host sql.NullString
var lastPathSegment sql.NullString
var articleTitle sql.NullString
if scanErr := rows.Scan(
&candidate.NormalizedURL,
&articleID,
&publishRecordID,
&registrableDomain,
&siteKey,
&host,
&lastPathSegment,
&articleTitle,
); scanErr != nil {
return nil, response.ErrInternal(50041, "alias_scan_failed", "failed to parse article alias resolution")
}
candidate.ArticleID = nullableInt64Value(articleID)
candidate.PublishRecordID = nullableInt64Value(publishRecordID)
candidate.RegistrableDomain = strings.ToLower(strings.TrimSpace(nullableStringText(registrableDomain)))
candidate.SiteKey = strings.ToLower(strings.TrimSpace(nullableStringText(siteKey)))
candidate.Host = strings.ToLower(strings.TrimSpace(nullableStringText(host)))
candidate.LastPathSegment = strings.TrimSpace(nullableStringText(lastPathSegment))
candidate.TitleKey = normalizeMonitoringAliasTitle(nullableStringText(articleTitle))
candidate.MatchKeys = monitoringCitationCandidateKeys(candidate.NormalizedURL, candidate.NormalizedURL)
candidates = append(candidates, candidate)
}
if err := rows.Err(); err != nil {
return nil, response.ErrInternal(50041, "alias_scan_failed", "failed to iterate article aliases")
}
for _, input := range lookupInputs {
bestScore := -1
bestAmbiguous := false
var best monitoringAliasCandidate
for _, candidate := range candidates {
score := scoreMonitoringAliasCandidate(input, candidate)
if score < 0 {
continue
}
if score > bestScore {
bestScore = score
bestAmbiguous = false
best = candidate
continue
}
if score == bestScore && !sameInt64Pointer(candidate.ArticleID, best.ArticleID) {
bestAmbiguous = true
}
}
if bestScore < 100 || bestAmbiguous {
continue
}
result[input.NormalizedURL] = monitoringAliasResolution{
ArticleID: cloneInt64(best.ArticleID),
PublishRecordID: cloneInt64(best.PublishRecordID),
NormalizedURLKey: best.NormalizedURL,
ResolutionStatus: "resolved",
ResolutionConfidence: "high",
}
}
return result, nil
}
func scoreMonitoringAliasCandidate(input monitoringAliasLookupInput, candidate monitoringAliasCandidate) int {
if input.NormalizedURL != "" && candidate.NormalizedURL == input.NormalizedURL {
return 100
}
for _, inputKey := range input.MatchKeys {
inputKey = strings.TrimSpace(inputKey)
if inputKey == "" {
continue
}
for _, candidateKey := range candidate.MatchKeys {
if inputKey == strings.TrimSpace(candidateKey) && candidateKey != "" {
return 100
}
}
}
return -1
}
func monitoringAliasDomainMatches(input monitoringAliasLookupInput, candidate monitoringAliasCandidate) bool {
inputKeys := []string{input.RegistrableDomain, input.SiteKey, input.Host}
candidateKeys := []string{candidate.RegistrableDomain, candidate.SiteKey, candidate.Host}
for _, inputKey := range inputKeys {
inputKey = strings.ToLower(strings.TrimSpace(inputKey))
if inputKey == "" {
continue
}
for _, candidateKey := range candidateKeys {
if inputKey == strings.ToLower(strings.TrimSpace(candidateKey)) {
return true
}
}
}
return false
}
func normalizeMonitoringAliasTitle(value string) string {
return strings.Map(func(r rune) rune {
if unicode.IsLetter(r) || unicode.IsNumber(r) {
return unicode.ToLower(r)
}
return -1
}, value)
}
func nullableStringText(value sql.NullString) string {
if !value.Valid {
return ""
}
return strings.TrimSpace(value.String)
}
func sameInt64Pointer(left, right *int64) bool {
if left == nil || right == nil {
return left == right
}
return *left == *right
}
func (s *MonitoringCallbackService) insertCitationFact(ctx context.Context, tx pgx.Tx, task *monitoringCollectTask, runID int64, fact monitoringCitationFact) error {
if _, err := tx.Exec(ctx, `
INSERT INTO monitoring_citation_facts (
tenant_id, run_id, brand_id, ai_platform_id, business_date,
cited_url, cited_title, normalized_url, host, registrable_domain,
subdomain, suffix, site_key, article_id, publish_record_id,
resolution_status, resolution_confidence
)
VALUES (
$1, $2, $3, $4, $5::date,
$6, $7, $8, $9, $10,
$11, $12, $13, $14, $15,
$16, $17
)
`,
task.TenantID,
runID,
task.BrandID,
task.AIPlatformID,
task.BusinessDate.Format("2006-01-02"),
fact.CitedURL,
nullableString(fact.CitedTitle),
fact.NormalizedURL,
fact.Host,
fact.RegistrableDomain,
nullableString(fact.Subdomain),
nullableString(fact.Suffix),
fact.SiteKey,
nullableInt64(fact.ArticleID),
nullableInt64(fact.PublishRecordID),
fact.ResolutionStatus,
fact.ResolutionConfidence,
); err != nil {
return response.ErrInternal(50041, "citation_insert_failed", "failed to persist citation facts")
}
return nil
}
func (s *MonitoringCallbackService) updateCollectTaskStatus(ctx context.Context, tx pgx.Tx, taskID int64, status string, completedAt time.Time, errorMessage *string) error {
if _, err := tx.Exec(ctx, `
UPDATE monitoring_collect_tasks
SET status = $2,
callback_received_at = NOW(),
completed_at = $3,
error_message = $4,
updated_at = NOW()
WHERE id = $1
`, taskID, status, completedAt, nullableString(errorMessage)); err != nil {
return response.ErrInternal(50041, "task_update_failed", "failed to update monitoring task status")
}
return nil
}
func (s *MonitoringCallbackService) loadExistingRunSummary(ctx context.Context, task *monitoringCollectTask) (*int64, int, int, error) {
var runID sql.NullInt64
if err := s.monitoringPool.QueryRow(ctx, `
SELECT id
FROM question_monitor_runs
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 = $7::date
LIMIT 1
`,
task.TenantID,
task.BrandID,
task.QuestionID,
task.AIPlatformID,
task.CollectorType,
task.RunMode,
task.BusinessDate.Format("2006-01-02"),
).Scan(&runID); err != nil && err != pgx.ErrNoRows {
return nil, 0, 0, response.ErrInternal(50041, "run_lookup_failed", "failed to inspect existing monitoring run")
}
if !runID.Valid {
return nil, 0, 0, nil
}
var citationSourceCount int
var contentCitationCount int
if err := s.monitoringPool.QueryRow(ctx, `
SELECT
COUNT(*)::int AS citation_source_count,
COUNT(*) FILTER (WHERE article_id IS NOT NULL AND $3::text <> 'qwen')::int AS content_citation_count
FROM monitoring_citation_facts
WHERE tenant_id = $1
AND run_id = $2
`, task.TenantID, runID.Int64, normalizeMonitoringPlatformID(task.AIPlatformID)).Scan(&citationSourceCount, &contentCitationCount); err != nil {
return nil, 0, 0, response.ErrInternal(50041, "citation_lookup_failed", "failed to inspect existing citation facts")
}
value := runID.Int64
return &value, citationSourceCount, contentCitationCount, nil
}
func (s *MonitoringCallbackService) touchInstallation(ctx context.Context, installationID string) error {
_, err := s.businessPool.Exec(ctx, `
UPDATE desktop_clients
SET last_seen_at = NOW()
WHERE id = $1::uuid
`, installationID)
return err
}
func (s *MonitoringCallbackService) publishTaskResultEnvelope(ctx context.Context, payload []byte) error {
if s == nil || s.rabbitMQ == nil {
return response.ErrServiceUnavailable(50341, "monitoring_result_queue_unavailable", "rabbitmq client is not configured")
}
if err := s.rabbitMQ.PublishMonitorResultIngest(ctx, payload); err != nil {
return response.ErrServiceUnavailable(50341, "monitoring_result_queue_unavailable", err.Error())
}
return nil
}
func (s *MonitoringCallbackService) PublishTaskResultEnvelope(ctx context.Context, payload []byte) error {
return s.publishTaskResultEnvelope(ctx, payload)
}
func (s *MonitoringCallbackService) tryPatchTaskResultQueueMeta(taskID int64, receivedAt time.Time, patch monitoringTaskResultQueueMetaPatch) {
if s == nil || s.monitoringPool == nil {
return
}
patchJSON, err := json.Marshal(patch)
if err != nil {
if s.logger != nil {
s.logger.Warn("monitoring queue metadata patch encode failed", zap.Error(err), zap.Int64("task_id", taskID))
}
return
}
ctx, cancel := context.WithTimeout(context.Background(), monitoringTaskResultQueueIOTimeout)
defer cancel()
if _, err := s.monitoringPool.Exec(ctx, `
UPDATE monitoring_collect_tasks
SET request_payload_json = (
CASE
WHEN request_payload_json IS NULL OR jsonb_typeof(request_payload_json) <> 'object'
THEN '{}'::jsonb
ELSE request_payload_json
END
) || $3::jsonb,
updated_at = NOW()
WHERE id = $1
AND callback_received_at = $2
`, taskID, receivedAt, patchJSON); err != nil && s.logger != nil {
s.logger.Warn("monitoring queue metadata patch failed", zap.Error(err), zap.Int64("task_id", taskID))
}
}
func (s *MonitoringCallbackService) PatchTaskResultQueueMeta(taskID int64, receivedAt time.Time, patch MonitoringTaskResultQueueMetaPatch) {
s.tryPatchTaskResultQueueMeta(taskID, receivedAt, patch)
}
func buildMonitoringRawPayload(platformID string, req MonitoringTaskResultRequest) ([]byte, []MonitoringSourceItem) {
payload := make(map[string]interface{}, len(req.RawResponseJSON)+4)
for key, value := range req.RawResponseJSON {
payload[key] = value
}
if _, ok := payload["answer"]; !ok && req.Answer != nil {
payload["answer"] = strings.TrimSpace(*req.Answer)
}
if _, ok := payload["citations"]; !ok && len(req.Citations) > 0 {
payload["citations"] = req.Citations
}
if _, ok := payload["search_results"]; !ok && len(req.SearchResults) > 0 {
payload["search_results"] = req.SearchResults
}
if _, ok := payload["status"]; !ok && strings.TrimSpace(req.Status) != "" {
payload["status"] = strings.TrimSpace(req.Status)
}
sourceInputs := buildMonitoringCitationSourceInputs(platformID, req, payload)
if len(payload) == 0 {
return nil, sourceInputs
}
data, err := json.Marshal(payload)
if err != nil {
return nil, sourceInputs
}
return data, sourceInputs
}
func buildMonitoringCitationSourceInputs(platformID string, req MonitoringTaskResultRequest, payload map[string]interface{}) []MonitoringSourceItem {
includeSearchResults := !monitoringPlatformUsesDedicatedCitationPanel(platformID)
sourceInputs := append([]MonitoringSourceItem{}, req.Citations...)
if includeSearchResults {
sourceInputs = append(sourceInputs, req.SearchResults...)
}
sourceInputs = append(sourceInputs, extractMonitoringSourceItems(payload["citations"])...)
sourceInputs = append(sourceInputs, extractMonitoringSourceItems(payload["references"])...)
sourceInputs = append(sourceInputs, extractMonitoringSourceItems(payload["sources"])...)
sourceInputs = append(sourceInputs, extractMonitoringSourceItems(payload["thought_references"])...)
sourceInputs = append(sourceInputs, extractMonitoringSourceItems(payload["thinking_references"])...)
if includeSearchResults {
sourceInputs = append(sourceInputs, extractMonitoringSourceItems(payload["search_results"])...)
sourceInputs = append(sourceInputs, extractMonitoringSourceItems(payload["searchResults"])...)
}
return sourceInputs
}
func monitoringPlatformUsesDedicatedCitationPanel(platformID string) bool {
return false
}
func monitoringPlatformResolvesContentCitations(platformID string) bool {
switch normalizeMonitoringPlatformID(platformID) {
case "qwen":
return false
default:
return true
}
}
func newMonitoringTaskResultEnvelope(task *monitoringCollectTask, installationID string, receivedAt time.Time, req MonitoringTaskResultRequest) monitoringTaskResultEnvelope {
return monitoringTaskResultEnvelope{
Version: "v1",
TaskID: task.ID,
TenantID: task.TenantID,
InstallationID: installationID,
ReceivedAt: receivedAt,
QueueStatus: "pending",
QueueClaimedAt: nil,
QueueEnqueuedAt: nil,
LastQueuePublishError: nil,
ReceivedAlertedAt: nil,
ReceivedAlertCount: 0,
CallbackResult: req,
}
}
func extractMonitoringSourceItems(value interface{}) []MonitoringSourceItem {
items, ok := value.([]interface{})
if !ok {
return nil
}
result := make([]MonitoringSourceItem, 0, len(items))
for _, item := range items {
switch typed := item.(type) {
case string:
if strings.TrimSpace(typed) == "" {
continue
}
result = append(result, MonitoringSourceItem{URL: typed})
case map[string]interface{}:
entry := MonitoringSourceItem{
URL: firstString(typed, "url", "cited_url", "link"),
Title: optionalString(firstString(typed, "title", "cited_title", "name", "text")),
SiteName: optionalString(firstString(typed, "site_name", "siteName")),
SiteKey: optionalString(firstString(typed, "site_key")),
NormalizedURL: optionalString(firstString(typed, "normalized_url")),
Host: optionalString(firstString(typed, "host")),
RegistrableDomain: optionalString(firstString(typed, "registrable_domain")),
Subdomain: optionalString(firstString(typed, "subdomain")),
Suffix: optionalString(firstString(typed, "suffix")),
ResolutionStatus: optionalString(firstString(typed, "resolution_status")),
ResolutionConfidence: optionalString(firstString(typed, "resolution_confidence")),
}
if articleID := firstInt64(typed, "article_id"); articleID != nil {
entry.ArticleID = articleID
}
if publishRecordID := firstInt64(typed, "publish_record_id"); publishRecordID != nil {
entry.PublishRecordID = publishRecordID
}
result = append(result, entry)
}
}
return result
}
func buildMonitoringCitationFact(input MonitoringSourceItem, aliasMap map[string]monitoringAliasResolution, resolveContentCitations bool) (monitoringCitationFact, bool) {
rawURL := strings.TrimSpace(input.URL)
normalizedURL := strings.TrimSpace(monitoringStringValue(input.NormalizedURL))
if normalizedURL == "" {
normalizedURL = normalizeCitationURL(rawURL)
}
host := strings.TrimSpace(monitoringStringValue(input.Host))
registrableDomain := strings.TrimSpace(monitoringStringValue(input.RegistrableDomain))
subdomain := cloneString(input.Subdomain)
suffix := cloneString(input.Suffix)
if parsedHost, parsedRegistrableDomain, parsedSubdomain, parsedSuffix := deriveCitationURLParts(rawURL, normalizedURL); host == "" {
host = parsedHost
if registrableDomain == "" {
registrableDomain = parsedRegistrableDomain
}
if subdomain == nil && parsedSubdomain != nil {
subdomain = parsedSubdomain
}
if suffix == nil && parsedSuffix != nil {
suffix = parsedSuffix
}
}
if host == "" {
host = strings.TrimSpace(monitoringStringValue(input.SiteKey))
}
if host == "" {
host = registrableDomain
}
if registrableDomain == "" {
registrableDomain = host
}
if normalizedURL == "" {
normalizedURL = strings.TrimSpace(rawURL)
}
if rawURL == "" {
rawURL = normalizedURL
}
if rawURL == "" || normalizedURL == "" || host == "" || registrableDomain == "" {
return monitoringCitationFact{}, false
}
siteKey := strings.TrimSpace(monitoringStringValue(input.SiteKey))
if siteKey == "" {
siteKey = registrableDomain
}
var articleID *int64
var publishRecordID *int64
var aliasResolution *monitoringAliasResolution
if resolveContentCitations {
articleID = cloneInt64(input.ArticleID)
publishRecordID = cloneInt64(input.PublishRecordID)
if alias, ok := aliasMap[normalizedURL]; ok {
aliasCopy := alias
aliasResolution = &aliasCopy
if articleID == nil {
articleID = cloneInt64(alias.ArticleID)
}
if publishRecordID == nil {
publishRecordID = cloneInt64(alias.PublishRecordID)
}
}
}
resolutionStatus := strings.TrimSpace(monitoringStringValue(input.ResolutionStatus))
if resolutionStatus == "" {
if aliasResolution != nil && strings.TrimSpace(aliasResolution.ResolutionStatus) != "" {
resolutionStatus = strings.TrimSpace(aliasResolution.ResolutionStatus)
} else {
resolutionStatus = "resolved"
}
}
resolutionConfidence := strings.TrimSpace(monitoringStringValue(input.ResolutionConfidence))
if resolutionConfidence == "" {
if aliasResolution != nil && strings.TrimSpace(aliasResolution.ResolutionConfidence) != "" {
resolutionConfidence = strings.TrimSpace(aliasResolution.ResolutionConfidence)
} else {
resolutionConfidence = "high"
}
}
return monitoringCitationFact{
CitedURL: rawURL,
CitedTitle: cloneString(input.Title),
NormalizedURL: normalizedURL,
Host: host,
RegistrableDomain: registrableDomain,
Subdomain: subdomain,
Suffix: suffix,
SiteKey: siteKey,
ArticleID: articleID,
PublishRecordID: publishRecordID,
ResolutionStatus: resolutionStatus,
ResolutionConfidence: resolutionConfidence,
}, true
}
func normalizeRunStatus(value string) string {
switch strings.ToLower(strings.TrimSpace(value)) {
case "", "succeeded", "success", "completed":
return "succeeded"
case "failed", "error":
return "failed"
default:
return ""
}
}
func validateMonitoringTaskResultSubmission(platformID string, runStatus string, req MonitoringTaskResultRequest) error {
if runStatus != "succeeded" {
return nil
}
answer := strings.TrimSpace(monitoringStringValue(req.Answer))
if answer == "" {
return response.ErrBadRequest(40041, "invalid_monitoring_result_answer", "succeeded monitoring result requires a non-empty answer")
}
if monitoringCitationMarkerPattern.MatchString(answer) {
_, sourceInputs := buildMonitoringRawPayload(platformID, req)
if len(sourceInputs) == 0 {
return response.ErrBadRequest(40041, "invalid_monitoring_result_citations", "succeeded monitoring result with citation markers requires citation sources")
}
}
return nil
}
func normalizeMonitoringLeaseLimit(value int) int {
if value <= 0 {
return defaultMonitoringLeaseLimit
}
if value > maxMonitoringLeaseLimit {
return maxMonitoringLeaseLimit
}
return value
}
func normalizeMonitoringLeasePlatformIDs(values []string) []string {
if len(values) == 0 {
return nil
}
result := make([]string, 0, len(values))
seen := make(map[string]struct{}, len(values))
for _, value := range values {
platformID := strings.TrimSpace(value)
if platformID == "" {
continue
}
if _, ok := seen[platformID]; ok {
continue
}
seen[platformID] = struct{}{}
result = append(result, platformID)
}
if len(result) == 0 {
return nil
}
return result
}
func monitoringHeartbeatPlatformIDs(platforms []MonitoringHeartbeatPlatform) []string {
if len(platforms) == 0 {
return nil
}
result := make([]string, 0, len(platforms))
for _, platform := range platforms {
platformID := strings.TrimSpace(platform.AIPlatformID)
if platformID != "" {
result = append(result, platformID)
}
}
return result
}
func normalizeHeartbeatPlatforms(items []MonitoringHeartbeatPlatform, fallbackDetectedAt time.Time) ([]MonitoringHeartbeatPlatform, error) {
if len(items) == 0 {
return nil, nil
}
result := make([]MonitoringHeartbeatPlatform, 0, len(items))
seen := make(map[string]struct{}, len(items))
for _, item := range items {
platformID := strings.TrimSpace(item.AIPlatformID)
if platformID == "" {
return nil, response.ErrBadRequest(40041, "invalid_platform_status", "ai_platform_id is required")
}
status, connected, accessible, err := normalizeMonitoringAccessStatus(item.AccessStatus, item.Connected, item.Accessible)
if err != nil {
return nil, err
}
item.AIPlatformID = platformID
item.AccessStatus = &status
item.Connected = &connected
item.Accessible = &accessible
if item.DetectedAt == nil || item.DetectedAt.IsZero() {
value := fallbackDetectedAt
item.DetectedAt = &value
} else {
value := item.DetectedAt.UTC()
item.DetectedAt = &value
}
if item.ReasonCode != nil {
item.ReasonCode = optionalString(*item.ReasonCode)
}
if item.ReasonMessage != nil {
item.ReasonMessage = optionalString(*item.ReasonMessage)
}
if _, ok := seen[platformID]; ok {
continue
}
seen[platformID] = struct{}{}
result = append(result, item)
}
return result, nil
}
func normalizeMonitoringAccessStatus(accessStatus *string, connected *bool, accessible *bool) (string, bool, bool, error) {
if accessStatus != nil {
switch strings.ToLower(strings.TrimSpace(*accessStatus)) {
case "accessible":
return "accessible", true, true, nil
case "not_logged_in":
return "not_logged_in", false, false, nil
case "unavailable":
return "unavailable", derefBool(connected), false, nil
case "":
default:
return "", false, false, response.ErrBadRequest(40041, "invalid_access_status", "access_status must be accessible, not_logged_in, or unavailable")
}
}
if accessible != nil {
if *accessible {
return "accessible", true, true, nil
}
if connected != nil && !*connected {
return "not_logged_in", false, false, nil
}
return "unavailable", derefBool(connected), false, nil
}
if connected != nil {
if *connected {
return "accessible", true, true, nil
}
return "not_logged_in", false, false, nil
}
return "", false, false, response.ErrBadRequest(40041, "invalid_access_status", "access status requires access_status, connected, or accessible")
}
func (s *MonitoringCallbackService) validateCollectTaskExecution(
ctx context.Context,
task *monitoringCollectTask,
installationID string,
leaseToken string,
) error {
if task == nil {
return response.ErrBadRequest(40041, "invalid_task_id", "monitoring task is required")
}
if task.ExecutionOwner == "desktop_tasks" {
return s.validateDesktopMonitorExecutionLease(ctx, task, installationID, leaseToken)
}
return validateTaskLease(task, installationID, leaseToken)
}
func (s *MonitoringCallbackService) validateDesktopMonitorExecutionLease(
ctx context.Context,
task *monitoringCollectTask,
installationID string,
leaseToken string,
) error {
desktopExecution, err := s.loadActiveDesktopMonitorExecutionLease(ctx, task.ID)
if err != nil {
return err
}
if desktopExecution == nil {
return response.ErrConflict(40941, "desktop_task_missing", "phase2 monitor desktop task is not active")
}
if desktopExecution.TargetClientID.String() != installationID {
return response.ErrUnauthorized(40141, "lease_owner_mismatch", "task is leased to another installation")
}
if desktopExecution.Status != "in_progress" {
return response.ErrConflict(40941, "task_status_invalid", "phase2 monitor desktop task is not currently in progress")
}
if len(desktopExecution.LeaseTokenHash) == 0 {
return response.ErrConflict(40941, "lease_token_missing", "phase2 monitor desktop task does not have an active lease token")
}
if !bytes.Equal(HashDesktopClientToken(strings.TrimSpace(leaseToken)), desktopExecution.LeaseTokenHash) {
return response.ErrUnauthorized(40141, "invalid_lease_token", "lease token is invalid")
}
return nil
}
func validateTaskLease(task *monitoringCollectTask, installationID string, leaseToken string) error {
if task.LeasedToExecutor.Valid && strings.TrimSpace(task.LeasedToExecutor.String) != "" && task.LeasedToExecutor.String != installationID {
return response.ErrUnauthorized(40141, "lease_owner_mismatch", "task is leased to another installation")
}
if strings.TrimSpace(task.LeaseTokenHash.String) == "" {
return response.ErrConflict(40941, "lease_token_missing", "task does not have an active lease token")
}
if sharedauth.HashToken(strings.TrimSpace(leaseToken)) != task.LeaseTokenHash.String {
return response.ErrUnauthorized(40141, "invalid_lease_token", "lease token is invalid")
}
return nil
}
func monitoringTaskReadyForQueuedResult(task *monitoringCollectTask) bool {
if task == nil {
return false
}
if task.ExecutionOwner == "desktop_tasks" {
return task.Status == "pending" && task.CallbackReceivedAt.Valid
}
return task.Status == "received"
}
func normalizeSkipReason(value *string) string {
if value == nil {
return "plugin_skipped"
}
trimmed := strings.TrimSpace(*value)
if trimmed == "" {
return "plugin_skipped"
}
return trimmed
}
func hasMonitoringParsePayload(req MonitoringTaskResultRequest) bool {
return req.BrandMentioned != nil ||
strings.TrimSpace(monitoringStringValue(req.BrandMentionPosition)) != "" ||
req.FirstRecommended != nil ||
strings.TrimSpace(monitoringStringValue(req.SentimentLabel)) != "" ||
len(req.MatchedBrandTerms) > 0
}
func hasCompleteMonitoringParsePayload(req MonitoringTaskResultRequest) bool {
return req.BrandMentioned != nil &&
strings.TrimSpace(monitoringStringValue(req.BrandMentionPosition)) != "" &&
req.FirstRecommended != nil &&
strings.TrimSpace(monitoringStringValue(req.SentimentLabel)) != "" &&
req.MatchedBrandTerms != nil
}
func normalizeCitationURL(raw string) string {
trimmed := strings.TrimSpace(raw)
if trimmed == "" {
return ""
}
parsed, err := url.Parse(trimmed)
if err != nil {
return trimmed
}
parsed.Host = strings.ToLower(parsed.Host)
parsed.Fragment = ""
parsed.RawFragment = ""
parsed.RawQuery = normalizeCitationQuery(parsed)
if parsed.Path != "/" {
parsed.Path = strings.TrimRight(parsed.Path, "/")
}
return parsed.String()
}
func normalizeCitationQuery(parsed *url.URL) string {
if parsed == nil {
return ""
}
host := strings.ToLower(strings.TrimSpace(parsed.Hostname()))
allowedParams := []string{}
switch {
case host == "baijiahao.baidu.com":
allowedParams = []string{"id"}
}
if len(allowedParams) == 0 {
return ""
}
query := parsed.Query()
normalized := url.Values{}
for _, key := range allowedParams {
if value := strings.TrimSpace(query.Get(key)); value != "" {
normalized.Set(key, value)
}
}
return normalized.Encode()
}
func deriveCitationURLParts(rawURL, normalizedURL string) (string, string, *string, *string) {
candidate := strings.TrimSpace(normalizedURL)
if candidate == "" {
candidate = strings.TrimSpace(rawURL)
}
if candidate == "" {
return "", "", nil, nil
}
parsed, err := url.Parse(candidate)
if err != nil {
return "", "", nil, nil
}
host := strings.ToLower(strings.TrimSpace(parsed.Hostname()))
if host == "" {
return "", "", nil, nil
}
registrableDomain, err := publicsuffix.EffectiveTLDPlusOne(host)
if err != nil {
registrableDomain = host
}
suffixValue, _ := publicsuffix.PublicSuffix(host)
var suffix *string
if suffixValue != "" {
suffix = &suffixValue
}
var subdomain *string
if registrableDomain != "" && host != registrableDomain {
value := strings.TrimSuffix(host, "."+registrableDomain)
value = strings.TrimSuffix(value, ".")
if value != "" {
subdomain = &value
}
}
return host, registrableDomain, subdomain, suffix
}
func firstString(values map[string]interface{}, keys ...string) string {
for _, key := range keys {
raw, ok := values[key]
if !ok {
continue
}
switch typed := raw.(type) {
case string:
if trimmed := strings.TrimSpace(typed); trimmed != "" {
return trimmed
}
}
}
return ""
}
func firstInt64(values map[string]interface{}, keys ...string) *int64 {
for _, key := range keys {
raw, ok := values[key]
if !ok {
continue
}
switch typed := raw.(type) {
case float64:
value := int64(typed)
return &value
case int64:
value := typed
return &value
case int:
value := int64(typed)
return &value
case string:
trimmed := strings.TrimSpace(typed)
if trimmed == "" {
continue
}
value, err := strconv.ParseInt(trimmed, 10, 64)
if err == nil {
return &value
}
}
}
return nil
}
func optionalString(value string) *string {
if strings.TrimSpace(value) == "" {
return nil
}
text := strings.TrimSpace(value)
return &text
}
func ValidateMonitoringTaskResultEnvelope(payload []byte) error {
_, err := decodeMonitoringTaskResultEnvelope(payload)
return err
}
func monitoringStringValue(value *string) string {
if value == nil {
return ""
}
return strings.TrimSpace(*value)
}
func resolveHeartbeatDetectedAt(value *time.Time) time.Time {
if value == nil || value.IsZero() {
return time.Now().UTC()
}
return value.UTC()
}
func cloneString(value *string) *string {
if value == nil {
return nil
}
trimmed := strings.TrimSpace(*value)
if trimmed == "" {
return nil
}
copyValue := trimmed
return &copyValue
}
func cloneBool(value *bool) *bool {
if value == nil {
return nil
}
copyValue := *value
return &copyValue
}
func cloneInt64(value *int64) *int64 {
if value == nil {
return nil
}
copyValue := *value
return &copyValue
}
func nullableBool(value *bool) interface{} {
if value == nil {
return nil
}
return *value
}
func monitoringBoolPointer(value bool) *bool {
copyValue := value
return &copyValue
}
func derefBool(value *bool) bool {
if value == nil {
return false
}
return *value
}
func derefInt64(value *int64) int64 {
if value == nil {
return 0
}
return *value
}