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

package app

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgtype"
	"go.uber.org/zap"

	"github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq"
	"github.com/geo-platform/tenant-api/internal/shared/response"
	"github.com/geo-platform/tenant-api/internal/shared/stream"
	"github.com/geo-platform/tenant-api/internal/tenant/repository"
)

type monitoringCollectTaskQueryer interface {
	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
}

type monitorDesktopTaskSpec struct {
	MonitorTaskID       int64
	TenantID            int64
	WorkspaceID         int64
	TargetAccountID     uuid.UUID
	TargetClientID      uuid.UUID
	PlatformID          string
	BusinessDate        string
	TriggerSource       string
	RunMode             string
	QuestionID          int64
	QuestionHash        string
	QuestionText        string
	SchedulerGroupKey   string
	Priority            int
	Lane                string
	InterruptGeneration int
	RequestedByUserID   int64
}

type monitorDesktopAccountCandidate struct {
	PlatformID  string
	AccountID   uuid.UUID
	DBClientID  *uuid.UUID
	HealthRank  int
	OnlineRank  int
	RecordOrder int
}

type monitorDesktopTaskTarget struct {
	AccountID uuid.UUID
	ClientID  uuid.UUID
}

type monitorDesktopInterruptTarget struct {
	TaskID              string
	MonitorTaskID       int64
	TargetClientID      string
	WorkspaceID         int64
	InterruptGeneration int
}

func monitorDesktopTaskSpecIDs(specs []monitorDesktopTaskSpec) []int64 {
	if len(specs) == 0 {
		return nil
	}
	result := make([]int64, 0, len(specs))
	seen := make(map[int64]struct{}, len(specs))
	for _, spec := range specs {
		if spec.MonitorTaskID <= 0 {
			continue
		}
		if _, exists := seen[spec.MonitorTaskID]; exists {
			continue
		}
		seen[spec.MonitorTaskID] = struct{}{}
		result = append(result, spec.MonitorTaskID)
	}
	return result
}

func (s *MonitoringService) shouldUseDesktopTasksExecution(tenantID int64) bool {
	if s == nil {
		return false
	}
	rolloutPercentage, _ := s.currentConfig()
	if rolloutPercentage <= 0 {
		return false
	}
	if rolloutPercentage >= 100 {
		return true
	}
	mod := tenantID % 100
	if mod < 0 {
		mod = -mod
	}
	return int(mod) < rolloutPercentage
}

func (s *MonitoringService) loadMonitorDesktopTaskSpecs(
	ctx context.Context,
	q monitoringCollectTaskQueryer,
	tenantID, workspaceID int64,
	businessDate time.Time,
	questionIDs []int64,
	platformIDs []string,
	targetClientID uuid.UUID,
	requestedByUserID int64,
) ([]monitorDesktopTaskSpec, error) {
	if q == nil || len(questionIDs) == 0 || len(platformIDs) == 0 {
		return nil, nil
	}

	rows, err := q.Query(ctx, `
		SELECT
			t.id,
			t.tenant_id,
			t.ai_platform_id,
			t.business_date,
			t.trigger_source,
			t.run_mode,
			t.question_id,
			t.question_hash,
			t.dispatch_priority,
			t.dispatch_lane,
			t.interrupt_generation,
			COALESCE((
				SELECT question_text_snapshot
				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.deleted_at IS NULL
				ORDER BY s.projected_at DESC, s.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.business_date = $3::date
		  AND t.status = 'pending'
		  AND t.callback_received_at IS NULL
		  AND COALESCE(t.execution_owner, 'legacy') = 'desktop_tasks'
		  AND t.question_id = ANY($4)
		  AND t.ai_platform_id = ANY($5)
		ORDER BY t.dispatch_priority DESC, t.id ASC
	`, tenantID, monitoringCollectorType, businessDate.Format("2006-01-02"), questionIDs, platformIDs)
	if err != nil {
		return nil, response.ErrInternal(50041, "query_failed", "failed to load phase2 monitor desktop task specs")
	}
	defer rows.Close()

	result := make([]monitorDesktopTaskSpec, 0)
	for rows.Next() {
		var (
			spec         monitorDesktopTaskSpec
			questionHash []byte
			businessDay  time.Time
		)
		if scanErr := rows.Scan(
			&spec.MonitorTaskID,
			&spec.TenantID,
			&spec.PlatformID,
			&businessDay,
			&spec.TriggerSource,
			&spec.RunMode,
			&spec.QuestionID,
			&questionHash,
			&spec.Priority,
			&spec.Lane,
			&spec.InterruptGeneration,
			&spec.QuestionText,
		); scanErr != nil {
			return nil, response.ErrInternal(50041, "scan_failed", "failed to parse phase2 monitor desktop task specs")
		}
		spec.WorkspaceID = workspaceID
		spec.TargetClientID = targetClientID
		spec.RequestedByUserID = requestedByUserID
		spec.BusinessDate = businessDay.Format("2006-01-02")
		spec.QuestionHash = encodeQuestionHash(questionHash)
		spec.SchedulerGroupKey = monitoringSchedulerGroupKey(spec.QuestionHash, spec.QuestionID, spec.QuestionText)
		result = append(result, spec)
	}
	if err := rows.Err(); err != nil {
		return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate phase2 monitor desktop task specs")
	}
	return result, nil
}

func (s *MonitoringService) dispatchMonitorDesktopTasks(
	ctx context.Context,
	specs []monitorDesktopTaskSpec,
) ([]*repository.DesktopTask, []int64, error) {
	if s == nil || s.businessPool == nil || len(specs) == 0 {
		return nil, nil, nil
	}

	targets, err := s.loadMonitorDesktopTaskTargets(ctx, specs[0].TenantID, specs[0].WorkspaceID, specs[0].RequestedByUserID, specs)
	if err != nil {
		return nil, nil, err
	}

	tx, err := s.businessPool.Begin(ctx)
	if err != nil {
		return nil, nil, response.ErrInternal(50115, "desktop_task_begin_failed", "failed to start phase2 monitor desktop dispatch")
	}
	defer tx.Rollback(ctx)

	repo := repository.NewDesktopTaskRepository(tx)
	publishableTasks := make([]*repository.DesktopTask, 0, len(specs))
	fallbackLegacyTaskIDs := make([]int64, 0)
	for _, spec := range specs {
		target, ok := targets[normalizeMonitoringPlatformID(spec.PlatformID)]
		if !ok || target.AccountID == uuid.Nil || target.ClientID == uuid.Nil {
			fallbackLegacyTaskIDs = append(fallbackLegacyTaskIDs, spec.MonitorTaskID)
			continue
		}
		spec.TargetAccountID = target.AccountID
		spec.TargetClientID = target.ClientID

		task, shouldPublish, fallbackLegacy, upsertErr := s.upsertMonitorDesktopTask(ctx, tx, repo, spec)
		if upsertErr != nil {
			return nil, nil, upsertErr
		}
		if fallbackLegacy {
			fallbackLegacyTaskIDs = append(fallbackLegacyTaskIDs, spec.MonitorTaskID)
			continue
		}
		if shouldPublish && task != nil {
			publishableTasks = append(publishableTasks, task)
		}
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, nil, response.ErrInternal(50116, "desktop_task_commit_failed", "failed to commit phase2 monitor desktop dispatch")
	}
	return publishableTasks, fallbackLegacyTaskIDs, nil
}

func (s *MonitoringService) loadMonitorDesktopTaskTargets(
	ctx context.Context,
	tenantID, workspaceID, userID int64,
	specs []monitorDesktopTaskSpec,
) (map[string]monitorDesktopTaskTarget, error) {
	platformIDs := uniqueMonitorDesktopTaskPlatformIDs(specs)
	if len(platformIDs) == 0 {
		return nil, nil
	}

	rows, err := s.businessPool.Query(ctx, `
		SELECT
			platform_id,
			desktop_id,
			COALESCE(client_id::text, '') AS client_id,
			CASE
				WHEN client_id IS NOT NULL
				 AND EXISTS (
					SELECT 1
					FROM desktop_clients dc
					WHERE dc.id = platform_accounts.client_id
					  AND dc.workspace_id = platform_accounts.workspace_id
					  AND dc.revoked_at IS NULL
					  AND dc.last_seen_at IS NOT NULL
					  AND dc.last_seen_at >= NOW() - $5::interval
				 )
				THEN 0
				ELSE 1
			END AS online_rank,
			CASE health
				WHEN 'live' THEN 0
				WHEN 'risk' THEN 1
				WHEN 'captcha' THEN 2
				ELSE 3
			END AS health_rank
		FROM platform_accounts
		WHERE tenant_id = $1
		  AND workspace_id = $2
		  AND deleted_at IS NULL
		  AND platform_id = ANY($3::text[])
		  AND ($4::bigint = 0 OR user_id = $4)
		ORDER BY platform_id ASC,
			CASE
				WHEN client_id IS NOT NULL
				 AND EXISTS (
					SELECT 1
					FROM desktop_clients dc
					WHERE dc.id = platform_accounts.client_id
					  AND dc.workspace_id = platform_accounts.workspace_id
					  AND dc.revoked_at IS NULL
					  AND dc.last_seen_at IS NOT NULL
					  AND dc.last_seen_at >= NOW() - $5::interval
				 )
				THEN 0
				ELSE 1
			END ASC,
			CASE health
				WHEN 'live' THEN 0
				WHEN 'risk' THEN 1
				WHEN 'captcha' THEN 2
				ELSE 3
			END ASC,
			verified_at DESC NULLS LAST,
			updated_at DESC,
			created_at DESC
	`, tenantID, workspaceID, platformIDs, userID, formatPgInterval(desktopClientPresenceTTL))
	if err != nil {
		return nil, response.ErrInternal(50117, "desktop_account_lookup_failed", "failed to resolve monitor desktop accounts")
	}
	defer rows.Close()

	candidates := make([]monitorDesktopAccountCandidate, 0)
	for rows.Next() {
		var (
			item         monitorDesktopAccountCandidate
			clientIDText string
		)
		if scanErr := rows.Scan(&item.PlatformID, &item.AccountID, &clientIDText, &item.OnlineRank, &item.HealthRank); scanErr != nil {
			return nil, response.ErrInternal(50117, "desktop_account_lookup_failed", "failed to parse monitor desktop accounts")
		}
		if parsedClientID, parseErr := uuid.Parse(strings.TrimSpace(clientIDText)); parseErr == nil {
			item.DBClientID = &parsedClientID
		}
		item.PlatformID = normalizeMonitoringPlatformID(item.PlatformID)
		item.RecordOrder = len(candidates)
		candidates = append(candidates, item)
	}
	if err := rows.Err(); err != nil {
		return nil, response.ErrInternal(50117, "desktop_account_lookup_failed", "failed to iterate monitor desktop accounts")
	}

	accountIDs := monitorDesktopAccountCandidateAccountIDs(candidates)
	accountPresence := loadDesktopAccountPresence(ctx, s.redis, accountIDs)
	clientIDs := monitorDesktopAccountCandidateClientIDs(candidates, accountPresence)
	onlineClients, err := s.loadOnlineMonitorDesktopClients(ctx, workspaceID, clientIDs)
	if err != nil {
		return nil, err
	}
	return selectMonitorDesktopTaskTargets(candidates, accountPresence, onlineClients), nil
}

func uniqueMonitorDesktopTaskPlatformIDs(specs []monitorDesktopTaskSpec) []string {
	if len(specs) == 0 {
		return nil
	}
	seen := make(map[string]struct{}, len(specs))
	result := make([]string, 0, len(specs))
	for _, spec := range specs {
		platformID := normalizeMonitoringPlatformID(spec.PlatformID)
		if platformID == "" {
			continue
		}
		if _, ok := seen[platformID]; ok {
			continue
		}
		seen[platformID] = struct{}{}
		result = append(result, platformID)
	}
	return result
}

func monitorDesktopAccountCandidateAccountIDs(candidates []monitorDesktopAccountCandidate) []uuid.UUID {
	result := make([]uuid.UUID, 0, len(candidates))
	for _, candidate := range candidates {
		if candidate.AccountID != uuid.Nil {
			result = append(result, candidate.AccountID)
		}
	}
	return uniqueUUIDs(result)
}

func monitorDesktopAccountCandidateClientIDs(candidates []monitorDesktopAccountCandidate, accountPresence map[uuid.UUID]uuid.UUID) []uuid.UUID {
	result := make([]uuid.UUID, 0, len(candidates)*2)
	for _, candidate := range candidates {
		if candidate.DBClientID != nil && *candidate.DBClientID != uuid.Nil {
			result = append(result, *candidate.DBClientID)
		}
		if accountPresence != nil {
			if clientID, ok := accountPresence[candidate.AccountID]; ok && clientID != uuid.Nil {
				result = append(result, clientID)
			}
		}
	}
	return uniqueUUIDs(result)
}

func (s *MonitoringService) loadOnlineMonitorDesktopClients(ctx context.Context, workspaceID int64, clientIDs []uuid.UUID) (map[uuid.UUID]bool, error) {
	result := make(map[uuid.UUID]bool)
	clientIDs = uniqueUUIDs(clientIDs)
	if len(clientIDs) == 0 {
		return result, nil
	}

	for clientID, online := range loadDesktopClientPresence(ctx, s.redis, clientIDs) {
		if online {
			result[clientID] = true
		}
	}

	rows, err := s.businessPool.Query(ctx, `
		SELECT id
		FROM desktop_clients
		WHERE workspace_id = $1
		  AND id = ANY($2::uuid[])
		  AND revoked_at IS NULL
		  AND last_seen_at IS NOT NULL
		  AND last_seen_at >= NOW() - $3::interval
	`, workspaceID, clientIDs, formatPgInterval(desktopClientPresenceTTL))
	if err != nil {
		return nil, response.ErrInternal(50117, "desktop_client_lookup_failed", "failed to resolve online monitor desktop clients")
	}
	defer rows.Close()

	for rows.Next() {
		var clientID uuid.UUID
		if scanErr := rows.Scan(&clientID); scanErr != nil {
			return nil, response.ErrInternal(50117, "desktop_client_lookup_failed", "failed to parse online monitor desktop clients")
		}
		result[clientID] = true
	}
	if err := rows.Err(); err != nil {
		return nil, response.ErrInternal(50117, "desktop_client_lookup_failed", "failed to iterate online monitor desktop clients")
	}
	return result, nil
}

func selectMonitorDesktopTaskTargets(
	candidates []monitorDesktopAccountCandidate,
	accountPresence map[uuid.UUID]uuid.UUID,
	onlineClients map[uuid.UUID]bool,
) map[string]monitorDesktopTaskTarget {
	result := make(map[string]monitorDesktopTaskTarget)
	targetsByPlatform := make(map[string][]monitorDesktopTaskTarget)
	for _, candidate := range candidates {
		platformID := normalizeMonitoringPlatformID(candidate.PlatformID)
		if platformID == "" || candidate.AccountID == uuid.Nil {
			continue
		}

		if clientID, ok := accountPresence[candidate.AccountID]; ok && clientID != uuid.Nil && onlineClients[clientID] {
			targetsByPlatform[platformID] = append(targetsByPlatform[platformID], monitorDesktopTaskTarget{AccountID: candidate.AccountID, ClientID: clientID})
			continue
		}
		if candidate.DBClientID != nil && *candidate.DBClientID != uuid.Nil && onlineClients[*candidate.DBClientID] {
			targetsByPlatform[platformID] = append(targetsByPlatform[platformID], monitorDesktopTaskTarget{AccountID: candidate.AccountID, ClientID: *candidate.DBClientID})
		}
	}
	for platformID, targets := range targetsByPlatform {
		if len(targets) == 0 {
			continue
		}
		index := randomIndex(len(targets))
		result[platformID] = targets[index]
	}
	return result
}

func (s *MonitoringService) upsertMonitorDesktopTask(
	ctx context.Context,
	tx pgx.Tx,
	repo repository.DesktopTaskRepository,
	spec monitorDesktopTaskSpec,
) (*repository.DesktopTask, bool, bool, error) {
	var (
		existingDesktopID      uuid.UUID
		existingStatus         string
		existingLeaseExpiresAt pgtype.Timestamptz
		existingActiveAttempt  pgtype.UUID
	)
	err := tx.QueryRow(ctx, `
		SELECT desktop_id, status, lease_expires_at, active_attempt_id
		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
		FOR UPDATE
	`, spec.MonitorTaskID).Scan(&existingDesktopID, &existingStatus, &existingLeaseExpiresAt, &existingActiveAttempt)
	switch err {
	case nil:
		task, getErr := repo.GetByDesktopID(ctx, existingDesktopID, spec.WorkspaceID)
		if getErr != nil {
			return nil, false, false, response.ErrInternal(50118, "desktop_task_lookup_failed", "failed to reload monitor desktop task")
		}
		expiredInProgress := existingStatus == "in_progress" &&
			existingLeaseExpiresAt.Valid &&
			existingLeaseExpiresAt.Time.Before(time.Now().UTC())
		if existingStatus != "queued" && !expiredInProgress {
			return task, false, false, nil
		}
		targetAccountID := spec.TargetAccountID
		if targetAccountID == uuid.Nil {
			targetAccountID = task.TargetAccountID
		}
		payloadJSON, payloadErr := marshalMonitorDesktopTaskPayload(spec)
		if payloadErr != nil {
			return nil, false, false, payloadErr
		}

		if expiredInProgress && existingActiveAttempt.Valid {
			expiredAttemptID, convErr := uuid.FromBytes(existingActiveAttempt.Bytes[:])
			if convErr == nil {
				expiredErrorJSON, marshalErr := marshalOptionalJSON(map[string]any{
					"reason":  "lease_expired_requeue",
					"message": "phase2 monitor desktop task lease expired before completion; task has been re-queued",
					"source":  "monitoring_collect_now",
				})
				if marshalErr != nil {
					return nil, false, false, response.ErrInternal(50119, "desktop_task_update_failed", "failed to encode expired phase2 monitor desktop attempt payload")
				}
				if finishErr := repo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{
					TaskID:      existingDesktopID,
					AttemptID:   expiredAttemptID,
					FinalStatus: literalStringPtr("aborted"),
					Error:       expiredErrorJSON,
				}); finishErr != nil && s.logger != nil {
					s.logger.Warn("phase2 expired monitor desktop attempt finish failed",
						zap.String("desktop_id", existingDesktopID.String()),
						zap.Int64("monitor_task_id", spec.MonitorTaskID),
						zap.Error(finishErr),
					)
				}
			}
		}

		if _, execErr := tx.Exec(ctx, `
			UPDATE desktop_tasks
			SET target_account_id = $2,
			    target_client_id = $3,
			    platform_id = $4,
			    payload = $5,
			    status = 'queued',
			    active_attempt_id = NULL,
			    lease_token_hash = NULL,
			    lease_expires_at = NULL,
			    priority = $6,
			    lane = $7,
			    lane_weight = $8,
			    source = $9,
			    scheduler_group_key = $10,
			    interrupt_generation = GREATEST(interrupt_generation, $11),
			    interrupted_at = CASE
			        WHEN $12 THEN COALESCE(interrupted_at, NOW())
			        ELSE interrupted_at
			    END,
			    interrupt_reason = CASE
			        WHEN $12 THEN 'lease_expired_requeue'
			        ELSE interrupt_reason
			    END,
			    enqueued_at = NOW(),
			    updated_at = NOW()
			WHERE desktop_id = $1
		`, existingDesktopID, targetAccountID, spec.TargetClientID, spec.PlatformID, payloadJSON, spec.Priority, spec.Lane, monitoringLaneWeight(spec.Lane), monitoringDesktopTaskSource(spec.TriggerSource), nullableString(optionalNonEmptyString(spec.SchedulerGroupKey)), spec.InterruptGeneration, expiredInProgress); execErr != nil {
			return nil, false, false, response.ErrInternal(50119, "desktop_task_update_failed", "failed to promote queued phase2 monitor desktop task")
		}
		task, getErr = repo.GetByDesktopID(ctx, existingDesktopID, spec.WorkspaceID)
		if getErr != nil {
			return nil, false, false, response.ErrInternal(50118, "desktop_task_lookup_failed", "failed to reload promoted monitor desktop task")
		}
		return task, true, false, nil
	case pgx.ErrNoRows:
	default:
		return nil, false, false, response.ErrInternal(50118, "desktop_task_lookup_failed", "failed to inspect active phase2 monitor desktop task")
	}

	targetAccountID := spec.TargetAccountID
	if targetAccountID == uuid.Nil || spec.TargetClientID == uuid.Nil {
		return nil, false, true, nil
	}
	payloadJSON, payloadErr := marshalMonitorDesktopTaskPayload(spec)
	if payloadErr != nil {
		return nil, false, false, payloadErr
	}
	desktopID := uuid.New()
	if _, execErr := tx.Exec(ctx, `
		INSERT INTO desktop_tasks (
			desktop_id,
			job_id,
			tenant_id,
			workspace_id,
			target_account_id,
			target_client_id,
			platform_id,
			kind,
			payload,
			status,
			priority,
			lane,
			lane_weight,
			source,
			scheduler_group_key,
			monitor_task_id,
			control_flags,
			interrupt_generation,
			enqueued_at
		)
		VALUES (
			$1,
			$2,
			$3,
			$4,
			$5,
			$6,
			$7,
			'monitor',
			$8,
			'queued',
			$9,
			$10,
			$11,
			$12,
			$13,
			$14,
			'{}'::jsonb,
			$15,
			NOW()
		)
	`, desktopID, uuid.New(), spec.TenantID, spec.WorkspaceID, targetAccountID, spec.TargetClientID, spec.PlatformID, payloadJSON, spec.Priority, spec.Lane, monitoringLaneWeight(spec.Lane), monitoringDesktopTaskSource(spec.TriggerSource), nullableString(optionalNonEmptyString(spec.SchedulerGroupKey)), spec.MonitorTaskID, spec.InterruptGeneration); execErr != nil {
		return nil, false, false, response.ErrInternal(50120, "desktop_task_create_failed", "failed to create phase2 monitor desktop task")
	}
	task, getErr := repo.GetByDesktopID(ctx, desktopID, spec.WorkspaceID)
	if getErr != nil {
		return nil, false, false, response.ErrInternal(50118, "desktop_task_lookup_failed", "failed to reload created phase2 monitor desktop task")
	}
	return task, true, false, nil
}

func (s *MonitoringService) fallbackMonitorDesktopTasksToLegacy(ctx context.Context, taskIDs []int64) error {
	if s == nil || s.monitoringPool == nil || len(taskIDs) == 0 {
		return nil
	}
	if _, err := s.monitoringPool.Exec(ctx, `
		UPDATE monitoring_collect_tasks
		SET execution_owner = 'legacy',
		    updated_at = NOW()
		WHERE id = ANY($1)
	`, taskIDs); err != nil {
		return response.ErrInternal(50041, "update_failed", "failed to fallback phase2 monitor tasks to legacy execution")
	}
	return nil
}

func (s *MonitoringService) requestPhase2MonitorInterrupts(
	ctx context.Context,
	targetClientID uuid.UUID,
	excludedMonitorTaskIDs []int64,
	interruptGeneration int,
) ([]monitorDesktopInterruptTarget, error) {
	if s == nil || s.businessPool == nil {
		return nil, nil
	}

	queryArgs := []any{targetClientID, interruptGeneration}
	query := `
		UPDATE desktop_tasks
		SET control_flags = (
				CASE
					WHEN control_flags IS NULL OR jsonb_typeof(control_flags) <> 'object'
						THEN '{}'::jsonb
					ELSE control_flags
				END
			) || jsonb_build_object(
				'interrupt_requested', true,
				'interrupt_generation', $2::integer,
				'interrupt_reason', 'collect_now_preempt'
			),
		    interrupt_generation = GREATEST(interrupt_generation, $2::integer),
		    interrupted_at = COALESCE(interrupted_at, NOW()),
		    interrupt_reason = 'collect_now_preempt',
		    updated_at = NOW()
		WHERE kind = 'monitor'
		  AND target_client_id = $1
		  AND status = 'in_progress'
		  AND lane IN ('normal', 'normal_boosted', 'retry')
		  AND monitor_task_id IS NOT NULL
	`
	if len(excludedMonitorTaskIDs) > 0 {
		query += ` AND NOT (monitor_task_id = ANY($3::bigint[]))`
		queryArgs = append(queryArgs, excludedMonitorTaskIDs)
	}
	query += `
		RETURNING desktop_id::text, monitor_task_id, target_client_id::text, workspace_id, interrupt_generation
	`

	rows, err := s.businessPool.Query(ctx, query, queryArgs...)
	if err != nil {
		if s.logger != nil {
			s.logger.Error("phase2 monitor interrupt query failed",
				zap.Error(err),
				zap.String("target_client_id", targetClientID.String()),
				zap.Int("interrupt_generation", interruptGeneration),
				zap.Int("excluded_monitor_task_count", len(excludedMonitorTaskIDs)),
			)
		}
		return nil, response.ErrInternal(50121, "desktop_task_interrupt_failed", "failed to mark active phase2 monitor desktop tasks for interrupt")
	}
	defer rows.Close()

	result := make([]monitorDesktopInterruptTarget, 0)
	for rows.Next() {
		var item monitorDesktopInterruptTarget
		if scanErr := rows.Scan(&item.TaskID, &item.MonitorTaskID, &item.TargetClientID, &item.WorkspaceID, &item.InterruptGeneration); scanErr != nil {
			return nil, response.ErrInternal(50121, "desktop_task_interrupt_failed", "failed to parse active phase2 monitor interrupt targets")
		}
		result = append(result, item)
	}
	if err := rows.Err(); err != nil {
		return nil, response.ErrInternal(50121, "desktop_task_interrupt_failed", "failed to iterate active phase2 monitor interrupt targets")
	}
	return result, nil
}

func (s *MonitoringService) deferQueuedPhase2MonitorTasks(
	ctx context.Context,
	targetClientID uuid.UUID,
	excludedMonitorTaskIDs []int64,
) ([]*repository.DesktopTask, error) {
	if s == nil || s.businessPool == nil {
		return nil, nil
	}

	queryArgs := []any{targetClientID}
	query := `
		UPDATE desktop_tasks
		SET priority = 50,
		    lane = 'retry',
		    lane_weight = 20,
		    source = 'retry_recovery',
		    control_flags = (
				CASE
					WHEN control_flags IS NULL OR jsonb_typeof(control_flags) <> 'object'
						THEN '{}'::jsonb
					ELSE control_flags
				END
			) || jsonb_build_object(
				'deferred_by_collect_now', true,
				'deferred_at', NOW()::text
			),
		    updated_at = NOW()
		WHERE kind = 'monitor'
		  AND target_client_id = $1
		  AND status = 'queued'
		  AND lane IN ('normal', 'normal_boosted')
		  AND monitor_task_id IS NOT NULL
	`
	if len(excludedMonitorTaskIDs) > 0 {
		query += ` AND NOT (monitor_task_id = ANY($2::bigint[]))`
		queryArgs = append(queryArgs, excludedMonitorTaskIDs)
	}
	query += ` RETURNING desktop_id, workspace_id`

	rows, err := s.businessPool.Query(ctx, query, queryArgs...)
	if err != nil {
		return nil, response.ErrInternal(50122, "desktop_task_defer_failed", "failed to defer queued phase2 monitor desktop tasks")
	}
	defer rows.Close()

	repo := repository.NewDesktopTaskRepository(s.businessPool)
	result := make([]*repository.DesktopTask, 0)
	for rows.Next() {
		var (
			desktopID   uuid.UUID
			workspaceID int64
		)
		if scanErr := rows.Scan(&desktopID, &workspaceID); scanErr != nil {
			return nil, response.ErrInternal(50122, "desktop_task_defer_failed", "failed to parse deferred phase2 monitor desktop tasks")
		}
		task, getErr := repo.GetByDesktopID(ctx, desktopID, workspaceID)
		if getErr != nil {
			return nil, response.ErrInternal(50118, "desktop_task_lookup_failed", "failed to reload deferred phase2 monitor desktop task")
		}
		result = append(result, task)
	}
	if err := rows.Err(); err != nil {
		return nil, response.ErrInternal(50122, "desktop_task_defer_failed", "failed to iterate deferred phase2 monitor desktop tasks")
	}
	return result, nil
}

func publishMonitorDesktopTaskAvailable(
	ctx context.Context,
	clientID string,
	task *repository.DesktopTask,
	rabbitMQPublisher func(context.Context, DesktopTaskEvent) error,
	dispatchPublisher func(context.Context, stream.DesktopDispatchEvent) error,
) error {
	if task == nil || strings.TrimSpace(clientID) == "" {
		return nil
	}
	title, businessDate, schedulerGroupKey, questionText := deriveDesktopTaskEventMetadataFromPayload(task.Kind, task.Payload)
	event := DesktopTaskEvent{
		Type:                "task_available",
		TaskID:              task.DesktopID.String(),
		JobID:               task.JobID.String(),
		WorkspaceID:         task.WorkspaceID,
		TargetAccountID:     task.TargetAccountID.String(),
		TargetClientID:      clientID,
		Platform:            task.Platform,
		Title:               title,
		BusinessDate:        businessDate,
		SchedulerGroupKey:   schedulerGroupKey,
		QuestionText:        questionText,
		Status:              task.Status,
		Kind:                task.Kind,
		Priority:            task.Priority,
		Lane:                task.Lane,
		InterruptGeneration: task.InterruptGeneration,
		UpdatedAt:           task.UpdatedAt,
	}
	if rabbitMQPublisher != nil {
		if err := rabbitMQPublisher(ctx, event); err != nil {
			return err
		}
	}
	if dispatchPublisher != nil {
		if err := dispatchPublisher(ctx, stream.DesktopDispatchEvent{
			Type:                event.Type,
			TaskID:              event.TaskID,
			JobID:               event.JobID,
			WorkspaceID:         event.WorkspaceID,
			TargetAccountID:     event.TargetAccountID,
			TargetClientID:      event.TargetClientID,
			Platform:            event.Platform,
			Title:               event.Title,
			BusinessDate:        event.BusinessDate,
			SchedulerGroupKey:   event.SchedulerGroupKey,
			QuestionText:        event.QuestionText,
			Status:              event.Status,
			Kind:                event.Kind,
			Priority:            event.Priority,
			Lane:                event.Lane,
			InterruptGeneration: event.InterruptGeneration,
			UpdatedAt:           event.UpdatedAt,
		}); err != nil {
			return err
		}
	}
	return nil
}

func publishPhase2MonitorControlEvent(
	ctx context.Context,
	client *rabbitmq.Client,
	logger *zap.Logger,
	target monitorDesktopInterruptTarget,
) {
	if client == nil || strings.TrimSpace(target.TargetClientID) == "" || strings.TrimSpace(target.TaskID) == "" {
		return
	}
	if err := publishDesktopDispatchEvent(ctx, client, logger, stream.DesktopDispatchEvent{
		Type:                "task_control",
		TaskID:              target.TaskID,
		WorkspaceID:         target.WorkspaceID,
		TargetClientID:      target.TargetClientID,
		Status:              "in_progress",
		Kind:                "monitor",
		Lane:                "high",
		InterruptGeneration: target.InterruptGeneration,
		Control:             "interrupt_requested",
		Reason:              "collect_now_preempt",
		UpdatedAt:           time.Now().UTC(),
	}); err != nil && logger != nil {
		logger.Warn("phase2 monitor control event publish failed",
			zap.String("task_id", target.TaskID),
			zap.Int64("monitor_task_id", target.MonitorTaskID),
			zap.Error(err),
		)
	}
}

func monitoringLaneWeight(lane string) int {
	switch strings.TrimSpace(lane) {
	case "high":
		return 70
	case "normal_boosted":
		return 50
	case "normal":
		return 40
	case "retry":
		return 20
	default:
		return 0
	}
}

func monitoringDesktopTaskSource(triggerSource string) string {
	if strings.EqualFold(strings.TrimSpace(triggerSource), "manual") {
		return "collect_now"
	}
	return "daily_schedule"
}

func monitoringSchedulerGroupKey(questionHash string, questionID int64, questionText string) string {
	if trimmed := strings.TrimSpace(questionHash); trimmed != "" {
		return trimmed
	}
	if questionID > 0 {
		return fmt.Sprintf("qid:%d", questionID)
	}
	if trimmed := strings.TrimSpace(questionText); trimmed != "" {
		return trimmed
	}
	return "monitor"
}

func marshalMonitorDesktopTaskPayload(spec monitorDesktopTaskSpec) ([]byte, error) {
	title := strings.TrimSpace(spec.QuestionText)
	platformName := platformDisplayName(spec.PlatformID)
	if title != "" {
		title = fmt.Sprintf("%s · %s", platformName, title)
	} else {
		title = fmt.Sprintf("监控任务 · %s", platformName)
	}
	payload, err := json.Marshal(map[string]any{
		"title":                  title,
		"business_date":          spec.BusinessDate,
		"scheduler_group_key":    spec.SchedulerGroupKey,
		"question_text":          spec.QuestionText,
		"question_id":            spec.QuestionID,
		"question_hash":          spec.QuestionHash,
		"ai_platform_id":         spec.PlatformID,
		"platform_name":          platformName,
		"collector_type":         monitoringCollectorType,
		"run_mode":               spec.RunMode,
		"trigger_source":         spec.TriggerSource,
		"monitoring_task_id":     spec.MonitorTaskID,
		"priority":               spec.Priority,
		"lane":                   spec.Lane,
		"interrupt_generation":   spec.InterruptGeneration,
		"legacy_monitoring_task": false,
	})
	if err != nil {
		return nil, response.ErrInternal(50041, "payload_encode_failed", "failed to encode phase2 monitor desktop task payload")
	}
	return payload, nil
}

func optionalNonEmptyString(value string) *string {
	trimmed := strings.TrimSpace(value)
	if trimmed == "" {
		return nil
	}
	return &trimmed
}