geo/server/internal/tenant/app/monitoring_callback_service.go

package app

import (
	"context"
	"database/sql"
	"encoding/json"
	"net/url"
	"strconv"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"
	"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"
)

const (
	defaultMonitoringLeaseLimit        = 3
	maxMonitoringLeaseLimit            = 50
	monitoringLeaseTTL                 = 15 * time.Minute
	monitoringTaskResultQueueIOTimeout = 5 * time.Second
)

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             int64  `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"`
	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 monitoringInstallationIdentity struct {
	ID       int64
	TenantID int64
}

type monitoringPlatformAccessSnapshotState struct {
	AccessStatus  string
	Connected     bool
	Accessible    bool
	ReasonCode    *string
	ReasonMessage *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
	LeaseTokenHash   sql.NullString
	LeasedToExecutor sql.NullString
	LeaseExpiresAt   sql.NullTime
	CompletedAt      sql.NullTime
	SkipReason       sql.NullString
	TriggerSource    string
	QuestionText     string
}

type monitoringLeasedTaskRow struct {
	ID            int64
	BrandID       int64
	QuestionID    int64
	QuestionHash  []byte
	AIPlatformID  string
	CollectorType string
	RunMode       string
	TriggerSource string
	BusinessDate  time.Time
	QuestionText  string
}

type monitoringAliasResolution struct {
	ArticleID        *int64
	PublishRecordID  *int64
	NormalizedURLKey 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        int64                       `json:"installation_id"`
	ReceivedAt            time.Time                   `json:"received_at"`
	QueueStatus           string                      `json:"queue_status"`
	QueueClaimedAt        *time.Time                  `json:"queue_claimed_at"`
	QueueEnqueuedAt       *time.Time                  `json:"queue_enqueued_at"`
	LastQueuePublishError *string                     `json:"last_queue_publish_error"`
	ReceivedAlertedAt     *time.Time                  `json:"received_alerted_at"`
	ReceivedAlertCount    int                         `json:"received_alert_count"`
	CallbackResult        MonitoringTaskResultRequest `json:"callback_result"`
}

type monitoringTaskResultQueueMetaPatch struct {
	QueueStatus           string     `json:"queue_status"`
	QueueClaimedAt        *time.Time `json:"queue_claimed_at"`
	QueueEnqueuedAt       *time.Time `json:"queue_enqueued_at"`
	LastQueuePublishError *string    `json:"last_queue_publish_error"`
}

type MonitoringTaskResultQueueMetaPatch = monitoringTaskResultQueueMetaPatch

func (s *MonitoringCallbackService) HandleTaskResult(ctx context.Context, installationID int64, installationToken string, taskID int64, req MonitoringTaskResultRequest) (*MonitoringTaskResultResponse, error) {
	installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
	if err != nil {
		return nil, err
	}

	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 task.LeasedToExecutor.Valid && strings.TrimSpace(task.LeasedToExecutor.String) != "" && task.LeasedToExecutor.String != strconv.FormatInt(installation.ID, 10) {
		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")
	}

	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
	}

	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")
	}

	if normalizeRunStatus(req.Status) == "" {
		return nil, response.ErrBadRequest(40041, "invalid_status", "status must be succeeded or failed")
	}

	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.ID, 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,
	})

	return &MonitoringTaskResultResponse{
		TaskID:               task.ID,
		TaskStatus:           "received",
		CitationSourceCount:  0,
		ContentCitationCount: 0,
	}, nil
}

func (s *MonitoringCallbackService) Heartbeat(ctx context.Context, installationID int64, installationToken string, req MonitoringHeartbeatRequest) (*MonitoringHeartbeatResponse, error) {
	installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
	if err != nil {
		return nil, err
	}

	isPrimaryInstallation, err := s.resolveHeartbeatPrimaryInstallation(ctx, installation.TenantID, installation.ID)
	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
	}

	tx, err := s.monitoringPool.Begin(ctx)
	if err != nil {
		return nil, response.ErrInternal(50041, "begin_failed", "failed to start monitoring heartbeat")
	}
	defer tx.Rollback(ctx)

	var reconciledSkippedTaskCount int64
	affectedBrandIDs := make(map[int64]struct{})
	for _, platform := range platforms {
		accessChanged, err := s.upsertPlatformAccessSnapshot(ctx, tx, installation, businessDate, platform)
		if err != nil {
			return nil, err
		}

		if !isPrimaryInstallation {
			continue
		}

		needsProjectionRefresh := accessChanged
		if monitoringStringValue(platform.AccessStatus) != "accessible" {
			count, reconcileErr := s.reconcilePendingTasksForPlatform(ctx, tx, installation.TenantID, businessDate, 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, 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
			}
		}
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring heartbeat")
	}

	_ = 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 int64, installationToken string, req MonitoringLeaseTasksRequest) (*MonitoringLeaseTasksResponse, error) {
	installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
	if err != nil {
		return nil, err
	}

	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
	}

	rows, err := s.selectPendingLeaseTasks(ctx, tx, installation.TenantID, 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"),
			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 int64, installationToken string, req MonitoringResumeTasksRequest) (*MonitoringResumeTasksResponse, error) {
	installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
	if err != nil {
		return nil, err
	}

	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
	}

	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"),
			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 int64, installationToken string, taskID int64, req MonitoringSkipTaskRequest) (*MonitoringSkipTaskResponse, error) {
	installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
	if err != nil {
		return nil, err
	}

	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 := validateTaskLease(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) 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 task.Status != "received" {
		return nil, nil
	}

	return s.processTaskResult(ctx, task, envelope.CallbackResult)
}

func (s *MonitoringCallbackService) authenticateInstallation(ctx context.Context, installationID int64, installationToken string) (*monitoringInstallationIdentity, error) {
	tokenHash := sharedauth.HashToken(strings.TrimSpace(installationToken))

	row := &monitoringInstallationIdentity{}
	if err := s.businessPool.QueryRow(ctx, `
		SELECT id, tenant_id
		FROM plugin_installations
		WHERE id = $1
		  AND installation_token_hash = $2
		  AND status = 'active'
		  AND deleted_at IS NULL
	`, installationID, tokenHash).Scan(&row.ID, &row.TenantID); err != nil {
		if err == pgx.ErrNoRows {
			return nil, response.ErrUnauthorized(40141, "plugin_installation_unauthorized", "plugin installation is invalid or expired")
		}
		return nil, response.ErrInternal(50041, "plugin_installation_lookup_failed", "failed to validate plugin installation")
	}

	return row, nil
}

func (s *MonitoringCallbackService) resolveHeartbeatPrimaryInstallation(ctx context.Context, tenantID, installationID int64) (bool, error) {
	var preferred sql.NullInt64
	err := s.businessPool.QueryRow(ctx, `
		SELECT primary_installation_id
		FROM tenant_monitoring_quotas
		WHERE tenant_id = $1
	`, tenantID).Scan(&preferred)
	if err != nil && err != pgx.ErrNoRows {
		return false, response.ErrInternal(50041, "query_failed", "failed to inspect primary monitoring installation")
	}

	if !preferred.Valid || preferred.Int64 == installationID {
		if !preferred.Valid {
			if err := s.persistHeartbeatPrimaryInstallationID(ctx, tenantID, installationID); err != nil {
				return false, err
			}
		}
		return true, nil
	}

	online, err := s.isMonitoringInstallationRecentlyOnline(ctx, tenantID, preferred.Int64)
	if err != nil {
		return false, err
	}
	if online {
		return false, nil
	}

	if err := s.persistHeartbeatPrimaryInstallationID(ctx, tenantID, installationID); err != nil {
		return false, err
	}
	return true, nil
}

func (s *MonitoringCallbackService) isMonitoringInstallationRecentlyOnline(ctx context.Context, tenantID, installationID int64) (bool, error) {
	var lastSeenAt sql.NullTime
	err := s.businessPool.QueryRow(ctx, `
		SELECT last_seen_at
		FROM plugin_installations
		WHERE id = $1
		  AND tenant_id = $2
		  AND status = 'active'
		  AND deleted_at IS NULL
	`, installationID, tenantID).Scan(&lastSeenAt)
	if err != nil {
		if err == pgx.ErrNoRows {
			return false, nil
		}
		return false, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring installation heartbeat")
	}

	if !lastSeenAt.Valid {
		return false, nil
	}
	return time.Since(lastSeenAt.Time) <= monitoringOnlineDuration, nil
}

func (s *MonitoringCallbackService) persistHeartbeatPrimaryInstallationID(ctx context.Context, tenantID, installationID int64) error {
	if _, err := s.businessPool.Exec(ctx, `
		INSERT INTO tenant_monitoring_quotas (
			tenant_id, primary_installation_id
		)
		VALUES ($1, $2)
		ON CONFLICT (tenant_id) DO UPDATE
		SET primary_installation_id = EXCLUDED.primary_installation_id,
			updated_at = NOW()
	`, tenantID, installationID); err != nil {
		return response.ErrInternal(50041, "update_failed", "failed to persist primary monitoring installation")
	}
	return 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,
			lease_token_hash,
			leased_to_executor,
			lease_expires_at,
			completed_at,
			skip_reason,
			trigger_source,
			COALESCE((
				SELECT question_text_snapshot
				FROM monitoring_question_config_snapshots
				WHERE tenant_id = t.tenant_id
				  AND brand_id = t.brand_id
				  AND question_id = t.question_id
				  AND question_hash = t.question_hash
				  AND deleted_at IS NULL
				ORDER BY projected_at DESC, id DESC
				LIMIT 1
			), '') AS question_text_snapshot
		FROM monitoring_collect_tasks t
		WHERE id = $1
		  AND tenant_id = $2
	`, taskID, tenantID).Scan(
		&item.ID,
		&item.TenantID,
		&item.BrandID,
		&item.QuestionID,
		&item.QuestionHash,
		&item.AIPlatformID,
		&item.CollectorType,
		&item.RunMode,
		&item.BusinessDate,
		&item.Status,
		&item.LeaseTokenHash,
		&item.LeasedToExecutor,
		&item.LeaseExpiresAt,
		&item.CompletedAt,
		&item.SkipReason,
		&item.TriggerSource,
		&item.QuestionText,
	); err != nil {
		if err == pgx.ErrNoRows {
			return nil, response.ErrNotFound(40441, "monitoring_task_not_found", "monitoring task not found")
		}
		return nil, response.ErrInternal(50041, "query_failed", "failed to load monitoring task")
	}

	return item, nil
}

func (s *MonitoringCallbackService) loadCollectTaskByID(ctx context.Context, taskID int64) (*monitoringCollectTask, error) {
	item := &monitoringCollectTask{}
	if err := s.monitoringPool.QueryRow(ctx, `
		SELECT
			id,
			tenant_id,
			brand_id,
			question_id,
			question_hash,
			ai_platform_id,
			collector_type,
			run_mode,
			business_date,
			status,
			lease_token_hash,
			leased_to_executor,
			lease_expires_at,
			completed_at,
			skip_reason,
			trigger_source,
			COALESCE((
				SELECT question_text_snapshot
				FROM monitoring_question_config_snapshots
				WHERE tenant_id = t.tenant_id
				  AND brand_id = t.brand_id
				  AND question_id = t.question_id
				  AND question_hash = t.question_hash
				  AND deleted_at IS NULL
				ORDER BY projected_at DESC, id DESC
				LIMIT 1
			), '') AS question_text_snapshot
		FROM monitoring_collect_tasks t
		WHERE id = $1
	`, taskID).Scan(
		&item.ID,
		&item.TenantID,
		&item.BrandID,
		&item.QuestionID,
		&item.QuestionHash,
		&item.AIPlatformID,
		&item.CollectorType,
		&item.RunMode,
		&item.BusinessDate,
		&item.Status,
		&item.LeaseTokenHash,
		&item.LeasedToExecutor,
		&item.LeaseExpiresAt,
		&item.CompletedAt,
		&item.SkipReason,
		&item.TriggerSource,
		&item.QuestionText,
	); err != nil {
		if err == pgx.ErrNoRows {
			return nil, response.ErrNotFound(40441, "monitoring_task_not_found", "monitoring task not found")
		}
		return nil, response.ErrInternal(50041, "query_failed", "failed to load monitoring task")
	}

	return item, nil
}

func (s *MonitoringCallbackService) selectPendingLeaseTasks(
	ctx context.Context,
	tx pgx.Tx,
	tenantID int64,
	businessDate string,
	limit int,
	eligiblePlatformIDs []string,
) ([]monitoringLeasedTaskRow, error) {
	queryArgs := []interface{}{tenantID, monitoringCollectorType, businessDate, limit}
	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((
				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 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.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.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, installationID int64,
	businessDate string,
	eligiblePlatformIDs []string,
) ([]monitoringLeasedTaskRow, error) {
	queryArgs := []interface{}{tenantID, monitoringCollectorType, businessDate, strconv.FormatInt(installationID, 10)}
	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((
				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 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.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, 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,
		    skip_reason = NULL,
		    error_message = NULL,
		    updated_at = NOW()
		WHERE id = $1
	`, taskID, sharedauth.HashToken(strings.TrimSpace(leaseToken)), strconv.FormatInt(installationID, 10), 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, taskID int64, receivedAt time.Time, requestPayloadJSON []byte) error {
	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
	`, taskID, 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) (bool, error) {
	previous, err := s.loadPlatformAccessSnapshotState(ctx, tx, installation.TenantID, installation.ID, businessDate, platform.AIPlatformID)
	if err != nil {
		return false, err
	}

	nextState := monitoringPlatformAccessSnapshotState{
		AccessStatus:  monitoringStringValue(platform.AccessStatus),
		Connected:     derefBool(platform.Connected),
		Accessible:    derefBool(platform.Accessible),
		ReasonCode:    optionalString(monitoringStringValue(platform.ReasonCode)),
		ReasonMessage: optionalString(monitoringStringValue(platform.ReasonMessage)),
	}

	if _, err := tx.Exec(ctx, `
		INSERT INTO monitoring_platform_access_snapshots (
			tenant_id, installation_id, ai_platform_id, business_date, access_status,
			connected, accessible, detected_at, reason_code, reason_message
		)
		VALUES ($1, $2, $3, $4::date, $5, $6, $7, $8, $9, $10)
		ON CONFLICT (tenant_id, installation_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,
		platform.AccessStatus,
		derefBool(platform.Connected),
		derefBool(platform.Accessible),
		resolveHeartbeatDetectedAt(platform.DetectedAt),
		nullableString(platform.ReasonCode),
		nullableString(platform.ReasonMessage),
	); err != nil {
		return false, response.ErrInternal(50041, "access_snapshot_upsert_failed", "failed to persist monitoring platform access snapshot")
	}
	return hasPlatformAccessSnapshotChanged(previous, nextState), nil
}

func (s *MonitoringCallbackService) loadPlatformAccessSnapshotState(ctx context.Context, tx pgx.Tx, tenantID, installationID int64, businessDate, aiPlatformID string) (*monitoringPlatformAccessSnapshotState, error) {
	item := &monitoringPlatformAccessSnapshotState{}
	var reasonCode sql.NullString
	var reasonMessage sql.NullString
	if err := tx.QueryRow(ctx, `
		SELECT access_status, connected, accessible, reason_code, reason_message
		FROM monitoring_platform_access_snapshots
		WHERE tenant_id = $1
		  AND installation_id = $2
		  AND ai_platform_id = $3
		  AND business_date = $4::date
	`, tenantID, installationID, aiPlatformID, businessDate).Scan(
		&item.AccessStatus,
		&item.Connected,
		&item.Accessible,
		&reasonCode,
		&reasonMessage,
	); err != nil {
		if err == pgx.ErrNoRows {
			return nil, nil
		}
		return nil, response.ErrInternal(50041, "access_snapshot_lookup_failed", "failed to inspect monitoring platform access snapshot")
	}
	item.ReasonCode = optionalString(reasonCode.String)
	item.ReasonMessage = optionalString(reasonMessage.String)
	return item, nil
}

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) 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) {
	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(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, sourceInputs)
		if buildErr != nil {
			return nil, buildErr
		}
		for _, fact := range facts {
			if err := s.insertCitationFact(ctx, tx, task, runID, fact); err != nil {
				return nil, err
			}
			citationSourceCount++
			if fact.ArticleID != nil {
				contentCitationCount++
			}
		}
	}

	taskStatus := "completed"
	if runStatus == "failed" {
		taskStatus = "failed"
	}
	if err := s.updateCollectTaskStatus(ctx, tx, task.ID, taskStatus, completedAt, req.ErrorMessage); err != nil {
		return nil, err
	}

	if err := s.rebuildBrandDailyProjection(ctx, tx, task.TenantID, task.BrandID, task.BusinessDate); err != nil {
		return nil, err
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring result ingest")
	}

	return &MonitoringTaskResultResponse{
		TaskID:               task.ID,
		RunID:                &runID,
		TaskStatus:           taskStatus,
		CitationSourceCount:  citationSourceCount,
		ContentCitationCount: contentCitationCount,
	}, nil
}

func (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, inputs []MonitoringSourceItem) ([]monitoringCitationFact, error) {
	if len(inputs) == 0 {
		return nil, nil
	}

	normalizedURLs := make([]string, 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{}{}
		normalizedURLs = append(normalizedURLs, key)
	}

	aliasMap, err := s.loadAliasResolutions(ctx, tx, tenantID, normalizedURLs)
	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)
		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, normalizedURLs []string) (map[string]monitoringAliasResolution, error) {
	result := make(map[string]monitoringAliasResolution)
	if len(normalizedURLs) == 0 {
		return result, nil
	}

	rows, err := tx.Query(ctx, `
		SELECT normalized_url, article_id, publish_record_id
		FROM monitoring_article_url_aliases
		WHERE tenant_id = $1
		  AND normalized_url = ANY($2)
	`, tenantID, normalizedURLs)
	if err != nil {
		return nil, response.ErrInternal(50041, "alias_lookup_failed", "failed to resolve article aliases")
	}
	defer rows.Close()

	for rows.Next() {
		var normalizedURL string
		var articleID sql.NullInt64
		var publishRecordID sql.NullInt64
		if scanErr := rows.Scan(&normalizedURL, &articleID, &publishRecordID); scanErr != nil {
			return nil, response.ErrInternal(50041, "alias_scan_failed", "failed to parse article alias resolution")
		}
		result[normalizedURL] = monitoringAliasResolution{
			ArticleID:        nullableInt64Value(articleID),
			PublishRecordID:  nullableInt64Value(publishRecordID),
			NormalizedURLKey: normalizedURL,
		}
	}

	return result, nil
}

func (s *MonitoringCallbackService) insertCitationFact(ctx context.Context, tx pgx.Tx, task *monitoringCollectTask, runID int64, fact monitoringCitationFact) error {
	if _, err := tx.Exec(ctx, `
		INSERT INTO monitoring_citation_facts (
			tenant_id, run_id, brand_id, ai_platform_id, business_date,
			cited_url, cited_title, normalized_url, host, registrable_domain,
			subdomain, suffix, site_key, article_id, publish_record_id,
			resolution_status, resolution_confidence
		)
		VALUES (
			$1, $2, $3, $4, $5::date,
			$6, $7, $8, $9, $10,
			$11, $12, $13, $14, $15,
			$16, $17
		)
	`,
		task.TenantID,
		runID,
		task.BrandID,
		task.AIPlatformID,
		task.BusinessDate.Format("2006-01-02"),
		fact.CitedURL,
		nullableString(fact.CitedTitle),
		fact.NormalizedURL,
		fact.Host,
		fact.RegistrableDomain,
		nullableString(fact.Subdomain),
		nullableString(fact.Suffix),
		fact.SiteKey,
		nullableInt64(fact.ArticleID),
		nullableInt64(fact.PublishRecordID),
		fact.ResolutionStatus,
		fact.ResolutionConfidence,
	); err != nil {
		return response.ErrInternal(50041, "citation_insert_failed", "failed to persist citation facts")
	}
	return nil
}

func (s *MonitoringCallbackService) updateCollectTaskStatus(ctx context.Context, tx pgx.Tx, taskID int64, status string, completedAt time.Time, errorMessage *string) error {
	if _, err := tx.Exec(ctx, `
		UPDATE monitoring_collect_tasks
		SET status = $2,
		    callback_received_at = NOW(),
		    completed_at = $3,
		    error_message = $4,
		    updated_at = NOW()
		WHERE id = $1
	`, taskID, status, completedAt, nullableString(errorMessage)); err != nil {
		return response.ErrInternal(50041, "task_update_failed", "failed to update monitoring task status")
	}
	return nil
}

func (s *MonitoringCallbackService) loadExistingRunSummary(ctx context.Context, task *monitoringCollectTask) (*int64, int, int, error) {
	var runID sql.NullInt64
	if err := s.monitoringPool.QueryRow(ctx, `
		SELECT id
		FROM question_monitor_runs
		WHERE tenant_id = $1
		  AND brand_id = $2
		  AND question_id = $3
		  AND ai_platform_id = $4
		  AND collector_type = $5
		  AND run_mode = $6
		  AND business_date = $7::date
		LIMIT 1
	`,
		task.TenantID,
		task.BrandID,
		task.QuestionID,
		task.AIPlatformID,
		task.CollectorType,
		task.RunMode,
		task.BusinessDate.Format("2006-01-02"),
	).Scan(&runID); err != nil && err != pgx.ErrNoRows {
		return nil, 0, 0, response.ErrInternal(50041, "run_lookup_failed", "failed to inspect existing monitoring run")
	}

	if !runID.Valid {
		return nil, 0, 0, nil
	}

	var citationSourceCount int
	var contentCitationCount int
	if err := s.monitoringPool.QueryRow(ctx, `
		SELECT
			COUNT(*)::int AS citation_source_count,
			COUNT(*) FILTER (WHERE article_id IS NOT NULL)::int AS content_citation_count
		FROM monitoring_citation_facts
		WHERE tenant_id = $1
		  AND run_id = $2
	`, task.TenantID, runID.Int64).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 int64) error {
	_, err := s.businessPool.Exec(ctx, `
		UPDATE plugin_installations
		SET last_seen_at = NOW(), updated_at = NOW()
		WHERE id = $1
	`, 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(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 := append([]MonitoringSourceItem{}, req.Citations...)
	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"])...)
	sourceInputs = append(sourceInputs, extractMonitoringSourceItems(payload["search_results"])...)
	sourceInputs = append(sourceInputs, extractMonitoringSourceItems(payload["searchResults"])...)

	if len(payload) == 0 {
		return nil, sourceInputs
	}

	data, err := json.Marshal(payload)
	if err != nil {
		return nil, sourceInputs
	}
	return data, sourceInputs
}

func newMonitoringTaskResultEnvelope(task *monitoringCollectTask, installationID int64, 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")),
				SiteName:             optionalString(firstString(typed, "site_name")),
				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) (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
	}

	articleID := cloneInt64(input.ArticleID)
	publishRecordID := cloneInt64(input.PublishRecordID)
	if alias, ok := aliasMap[normalizedURL]; ok {
		if articleID == nil {
			articleID = cloneInt64(alias.ArticleID)
		}
		if publishRecordID == nil {
			publishRecordID = cloneInt64(alias.PublishRecordID)
		}
	}

	resolutionStatus := strings.TrimSpace(monitoringStringValue(input.ResolutionStatus))
	if resolutionStatus == "" {
		resolutionStatus = "resolved"
	}
	resolutionConfidence := strings.TrimSpace(monitoringStringValue(input.ResolutionConfidence))
	if resolutionConfidence == "" {
		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 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 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 validateTaskLease(task *monitoringCollectTask, installationID int64, leaseToken string) error {
	if task.LeasedToExecutor.Valid && strings.TrimSpace(task.LeasedToExecutor.String) != "" && task.LeasedToExecutor.String != strconv.FormatInt(installationID, 10) {
		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 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 = ""

	if parsed.Path != "/" {
		parsed.Path = strings.TrimRight(parsed.Path, "/")
	}

	return parsed.String()
}

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
}