Files
geo/server/internal/tenant/app/monitoring_callback_service.go
T
root 28633cf570
Desktop Client Build / Resolve Build Metadata (push) Successful in 44s
Backend CI / Backend (push) Successful in 17m5s
Desktop Client Build / Build Desktop Client (push) Successful in 25m13s
Desktop Client Build / Publish Client Artifacts to NAS (push) Successful in 42s
feat(monitoring): fail same-client queued tasks on desktop authorization failure
When a desktop monitoring task fails with a non-retryable authorization failure
(login expired, challenge required, risk control), all pending same-client/same-platform
tasks for the same business day are immediately bulk-failed as non-retryable, avoiding
wasted execution attempts. Adds preflight authorization checks before task execution,
enriches failure payloads with disposition metadata (code, category, retryable flags),
and triggers account health reports on auth state degradation.

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
2026-05-15 23:50:50 +08:00

3692 lines
119 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
TargetClientID sql.NullString
}
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
type monitoringTaskFailureDisposition struct {
NonRetryable bool
Category string
Code string
}
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(target_client_id::text, '') AS target_client_id,
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.TargetClientID,
&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(target_client_id::text, '') AS target_client_id,
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.TargetClientID,
&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")
}
if _, err := tx.Exec(ctx, `
UPDATE monitoring_collect_tasks
SET status = 'received',
callback_received_at = $2,
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
) || $3::jsonb,
updated_at = NOW()
WHERE id = $1
`, task.ID, 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 runStatus == "failed" {
if _, err := s.failRemainingMonitoringTasksForAuthorizationFailure(ctx, tx, task, req, completedAt); 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"
}
disposition := monitoringTaskFailureDispositionFromRequest(req)
result := map[string]any{
"code": "monitoring_task_failed",
"message": message,
"monitoring_task_id": task.ID,
}
if disposition.Code != "" {
result["code"] = disposition.Code
}
if disposition.NonRetryable {
result["retryable"] = false
result["non_retryable"] = true
}
if disposition.Category != "" {
result["failure_category"] = disposition.Category
}
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 monitoringTaskFailureDispositionFromRequest(req MonitoringTaskResultRequest) monitoringTaskFailureDisposition {
runtimeError, _ := req.RawResponseJSON["runtime_error"].(map[string]interface{})
if len(runtimeError) == 0 {
if nested, ok := req.RawResponseJSON["error"].(map[string]interface{}); ok {
runtimeError = nested
}
}
disposition := monitoringTaskFailureDisposition{
Category: monitoringFailureString(runtimeError["failure_category"]),
Code: monitoringFailureString(runtimeError["code"]),
}
message := monitoringFailureString(runtimeError["message"])
if message == "" {
message = strings.TrimSpace(optionalStringValue(req.ErrorMessage))
}
if monitoringFailureBoolEquals(runtimeError["retryable"], false) ||
monitoringFailureBoolEquals(runtimeError["non_retryable"], true) ||
monitoringAuthorizationFailureCode(disposition.Code) ||
monitoringAuthorizationFailureMessage(message) {
disposition.NonRetryable = true
}
if disposition.NonRetryable && disposition.Category == "" {
disposition.Category = "ai_platform_authorization"
}
return disposition
}
func monitoringFailureString(value interface{}) string {
text, ok := value.(string)
if !ok {
return ""
}
return strings.TrimSpace(text)
}
func monitoringFailureBoolEquals(value interface{}, expected bool) bool {
boolean, ok := value.(bool)
if !ok {
return false
}
return boolean == expected
}
func monitoringAuthorizationFailureCode(code string) bool {
normalized := strings.ToLower(strings.TrimSpace(code))
if normalized == "" {
return false
}
if normalized == "desktop_account_auth_expired" ||
normalized == "desktop_account_challenge_required" ||
normalized == "desktop_account_risk_control" {
return true
}
return strings.HasSuffix(normalized, "_login_required") ||
strings.HasSuffix(normalized, "_login_expired") ||
strings.HasSuffix(normalized, "_not_logged_in") ||
strings.HasSuffix(normalized, "_challenge_required")
}
func monitoringAuthorizationFailureMessage(message string) bool {
normalized := strings.ToLower(strings.TrimSpace(message))
if normalized == "" {
return false
}
return strings.Contains(normalized, "login required") ||
strings.Contains(normalized, "login expired") ||
strings.Contains(normalized, "not logged in") ||
strings.Contains(normalized, "unauthorized") ||
strings.Contains(normalized, "登录态失效") ||
strings.Contains(normalized, "登录态已失效") ||
strings.Contains(normalized, "登录已失效") ||
strings.Contains(normalized, "请重新登录") ||
strings.Contains(normalized, "请先登录") ||
strings.Contains(normalized, "未登录")
}
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")
}
bulkFailure, err := bulkFailQueuedDesktopTasksForAuthorizationFailure(ctx, tx, finalizedTask, errorJSON)
if err != nil {
return err
}
if err := tx.Commit(ctx); err != nil {
return response.ErrInternal(50041, "desktop_task_complete_commit_failed", "failed to commit phase2 desktop task completion")
}
if err := failMonitoringCollectTasksForDesktopAuthorizationFailure(ctx, s.monitoringPool, bulkFailure.MonitorTaskIDs, bulkFailure.ErrorPayload); err != nil {
return err
}
publishDesktopTaskLifecycleEvent(ctx, s.rabbitMQ, s.logger, finalizedTask, "task_completed")
for _, failedTask := range bulkFailure.Tasks {
publishDesktopTaskLifecycleEvent(ctx, s.rabbitMQ, s.logger, failedTask, "task_completed")
}
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) failRemainingMonitoringTasksForAuthorizationFailure(
ctx context.Context,
tx pgx.Tx,
task *monitoringCollectTask,
req MonitoringTaskResultRequest,
completedAt time.Time,
) (int64, error) {
if task == nil {
return 0, nil
}
targetClientID := monitoringAuthorizationFailureTargetClientID(task)
if targetClientID == "" {
return 0, nil
}
disposition := monitoringTaskFailureDispositionFromRequest(req)
if !disposition.NonRetryable || disposition.Category != "ai_platform_authorization" {
return 0, nil
}
runtimeError := buildDesktopMonitorTaskErrorValue(task, req)
runtimeError["blocked_by_authorization_failure_task_id"] = task.ID
runtimeError["blocked_by_authorization_failure_platform"] = task.AIPlatformID
requestPayloadJSON, err := json.Marshal(map[string]any{
"runtime_error": runtimeError,
"blocked_by_authorization_failure": true,
})
if err != nil {
return 0, response.ErrInternal(50041, "task_payload_encode_failed", "failed to encode monitoring authorization failure payload")
}
message := strings.TrimSpace(optionalStringValue(req.ErrorMessage))
if message == "" {
message = monitoringFailureString(runtimeError["message"])
}
if message == "" {
message = "登录态已失效,任务未执行。"
}
executionOwner := strings.TrimSpace(task.ExecutionOwner)
if executionOwner == "" {
executionOwner = "legacy"
}
tag, err := tx.Exec(ctx, `
UPDATE monitoring_collect_tasks
SET status = 'failed',
lease_token_hash = NULL,
leased_to_executor = NULL,
leased_at = NULL,
lease_expires_at = NULL,
callback_received_at = NOW(),
completed_at = $7,
skip_reason = NULL,
error_message = $8,
request_payload_json = (
CASE
WHEN request_payload_json IS NULL OR jsonb_typeof(request_payload_json) <> 'object'
THEN '{}'::jsonb
ELSE request_payload_json
END
) || $9::jsonb,
updated_at = NOW()
WHERE tenant_id = $1
AND collector_type = $2
AND business_date = $3::date
AND ai_platform_id = $4
AND COALESCE(execution_owner, 'legacy') = $10
AND id <> $5
AND status IN ('pending', 'leased', 'expired')
AND callback_received_at IS NULL
AND (
target_client_id::text = $6
OR leased_to_executor = $6
)
`, task.TenantID, task.CollectorType, task.BusinessDate.Format("2006-01-02"), task.AIPlatformID, task.ID, targetClientID, completedAt, message, requestPayloadJSON, executionOwner)
if err != nil {
return 0, response.ErrInternal(50041, "task_update_failed", "failed to fail remaining monitoring tasks after authorization failure")
}
return tag.RowsAffected(), nil
}
func monitoringAuthorizationFailureTargetClientID(task *monitoringCollectTask) string {
if task == nil {
return ""
}
if task.TargetClientID.Valid {
if trimmed := strings.TrimSpace(task.TargetClientID.String); trimmed != "" {
return trimmed
}
}
if task.LeasedToExecutor.Valid {
return strings.TrimSpace(task.LeasedToExecutor.String)
}
return ""
}
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 == "received" || 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
}