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

package app

import (
	"context"
	"encoding/json"
	"fmt"
	"sort"
	"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/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 monitorDesktopTaskTargetCandidate struct {
	Target               monitorDesktopTaskTarget
	HealthRank           int
	OnlineRank           int
	SourceRank           int
	OnlineSinceUnixMilli int64
	RecordOrder          int
	ClientOrder          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(NULLIF(t.question_text_snapshot, ''), (
				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 btrim(s.question_text_snapshot) <> ''
				ORDER BY
					CASE WHEN s.deleted_at IS NULL THEN 0 ELSE 1 END,
					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.QuestionText = strings.TrimSpace(spec.QuestionText)
		if spec.QuestionText == "" {
			continue
		}
		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,
	maxClientBacklog ...int,
) ([]*repository.DesktopTask, []int64, error) {
	if s == nil || s.businessPool == nil || len(specs) == 0 {
		return nil, nil, nil
	}

	backlogLimit := 0
	if len(maxClientBacklog) > 0 && maxClientBacklog[0] > 0 {
		backlogLimit = maxClientBacklog[0]
	}
	var targets map[string]monitorDesktopTaskTarget
	if monitorDesktopTaskSpecsNeedTargetLookup(specs) {
		var err error
		targets, err = s.loadMonitorDesktopTaskTargets(ctx, specs[0].TenantID, specs[0].WorkspaceID, specs[0].RequestedByUserID, specs, backlogLimit)
		if err != nil {
			return nil, nil, err
		}
	}
	clientBacklog := make(map[uuid.UUID]int)
	if backlogLimit > 0 {
		var err error
		clientBacklog, err = s.loadMonitorDesktopClientBacklog(
			ctx,
			specs[0].TenantID,
			specs[0].WorkspaceID,
			monitorDesktopTaskDispatchClientIDs(specs, targets),
			backlogLimit,
		)
		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))
	deferredTaskIDs := make([]int64, 0)
	for _, spec := range specs {
		if spec.TargetAccountID == uuid.Nil || spec.TargetClientID == uuid.Nil {
			target, ok := targets[normalizeMonitoringPlatformID(spec.PlatformID)]
			if !ok || target.AccountID == uuid.Nil || target.ClientID == uuid.Nil {
				deferredTaskIDs = append(deferredTaskIDs, spec.MonitorTaskID)
				continue
			}
			spec.TargetAccountID = target.AccountID
			spec.TargetClientID = target.ClientID
		}

		task, shouldPublish, shouldDefer, upsertErr := s.upsertMonitorDesktopTask(
			ctx,
			tx,
			repo,
			spec,
			func(oldClientID, newClientID uuid.UUID) bool {
				return reserveMonitorDesktopClientBacklog(oldClientID, newClientID, clientBacklog, backlogLimit)
			},
			func(oldClientID, newClientID uuid.UUID) (bool, error) {
				return s.reserveMonitorDesktopClientBacklogTx(ctx, tx, spec.TenantID, spec.WorkspaceID, oldClientID, newClientID, backlogLimit)
			},
		)
		if upsertErr != nil {
			return nil, nil, upsertErr
		}
		if shouldDefer {
			deferredTaskIDs = append(deferredTaskIDs, spec.MonitorTaskID)
			continue
		}
		if shouldPublish && task != nil {
			publishableTasks = append(publishableTasks, task)
		}
	}

	if len(deferredTaskIDs) > 0 {
		if err := deferPhase2MonitorCollectTasks(ctx, tx, deferredTaskIDs, defaultMonitoringDailyDesktopRetryCooldown); err != nil {
			return nil, nil, err
		}
	}

	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, deferredTaskIDs, nil
}

func monitorDesktopTaskDispatchClientIDs(specs []monitorDesktopTaskSpec, targets map[string]monitorDesktopTaskTarget) []uuid.UUID {
	if len(specs) == 0 {
		return nil
	}
	result := make([]uuid.UUID, 0, len(specs)*2)
	for _, spec := range specs {
		if spec.TargetClientID != uuid.Nil {
			result = append(result, spec.TargetClientID)
		}
		if targets == nil {
			continue
		}
		target, ok := targets[normalizeMonitoringPlatformID(spec.PlatformID)]
		if ok && target.ClientID != uuid.Nil {
			result = append(result, target.ClientID)
		}
	}
	return uniqueUUIDs(result)
}

func monitorDesktopTaskSpecsNeedTargetLookup(specs []monitorDesktopTaskSpec) bool {
	for _, spec := range specs {
		if spec.TargetAccountID == uuid.Nil || spec.TargetClientID == uuid.Nil {
			return true
		}
	}
	return false
}

func monitorDesktopTaskClientSlotLockKey(clientID uuid.UUID) int64 {
	return int64(monitoringDailyStableHash("monitor_desktop_client_slot", clientID.String()))
}

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

	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 := loadDesktopAccountPresenceClients(ctx, s.redis, accountIDs)
	clientIDs := monitorDesktopAccountCandidateClientIDs(candidates, accountPresence)
	onlineClients, err := s.loadOnlineMonitorDesktopClients(ctx, workspaceID, clientIDs)
	if err != nil {
		return nil, err
	}
	activeBacklog, err := s.loadMonitorDesktopClientBacklog(ctx, tenantID, workspaceID, clientIDs, backlogLimit)
	if err != nil {
		return nil, err
	}
	return selectMonitorDesktopTaskTargets(candidates, accountPresence, onlineClients, activeBacklog, backlogLimit), 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][]desktopAccountPresenceClient) []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 {
			for _, client := range accountPresence[candidate.AccountID] {
				if client.ClientID != uuid.Nil {
					result = append(result, client.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 (s *MonitoringService) loadMonitorDesktopClientBacklog(
	ctx context.Context,
	tenantID, workspaceID int64,
	clientIDs []uuid.UUID,
	limit int,
) (map[uuid.UUID]int, error) {
	result := make(map[uuid.UUID]int)
	clientIDs = uniqueUUIDs(clientIDs)
	if s == nil || s.businessPool == nil || limit <= 0 || len(clientIDs) == 0 {
		return result, nil
	}

	rows, err := s.businessPool.Query(ctx, `
		SELECT target_client_id, COUNT(*)::int
		FROM desktop_tasks
		WHERE tenant_id = $1
		  AND workspace_id = $2
		  AND target_client_id = ANY($3::uuid[])
		  AND kind = 'monitor'
		  AND status IN ('queued', 'in_progress')
		GROUP BY target_client_id
	`, tenantID, workspaceID, clientIDs)
	if err != nil {
		return nil, response.ErrInternal(50117, "desktop_client_backlog_lookup_failed", "failed to resolve monitor desktop client backlog")
	}
	defer rows.Close()

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

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

		seenClientIDs := make(map[uuid.UUID]struct{})
		if accountPresence != nil {
			for index, client := range accountPresence[candidate.AccountID] {
				if client.ClientID == uuid.Nil || !onlineClients[client.ClientID] {
					continue
				}
				seenClientIDs[client.ClientID] = struct{}{}
				item := monitorDesktopTaskTargetCandidate{
					Target: monitorDesktopTaskTarget{
						AccountID: candidate.AccountID,
						ClientID:  client.ClientID,
					},
					HealthRank:           candidate.HealthRank,
					OnlineRank:           0,
					SourceRank:           0,
					OnlineSinceUnixMilli: client.OnlineSinceUnixMilli,
					RecordOrder:          candidate.RecordOrder,
					ClientOrder:          index,
				}
				candidatesByPlatform[platformID] = append(candidatesByPlatform[platformID], item)
			}
		}
		if candidate.DBClientID == nil || *candidate.DBClientID == uuid.Nil || !onlineClients[*candidate.DBClientID] {
			continue
		}
		if _, exists := seenClientIDs[*candidate.DBClientID]; exists {
			continue
		}
		item := monitorDesktopTaskTargetCandidate{
			Target: monitorDesktopTaskTarget{
				AccountID: candidate.AccountID,
				ClientID:  *candidate.DBClientID,
			},
			HealthRank:  candidate.HealthRank,
			OnlineRank:  0,
			SourceRank:  1,
			RecordOrder: candidate.RecordOrder,
		}
		candidatesByPlatform[platformID] = append(candidatesByPlatform[platformID], item)
	}

	platformIDs := make([]string, 0, len(candidatesByPlatform))
	for platformID := range candidatesByPlatform {
		platformIDs = append(platformIDs, platformID)
	}
	sort.Strings(platformIDs)

	for _, platformID := range platformIDs {
		platformCandidates := candidatesByPlatform[platformID]
		sort.SliceStable(platformCandidates, func(i, j int) bool {
			return platformCandidates[i].betterThan(platformCandidates[j])
		})
		for _, candidate := range platformCandidates {
			if monitorDesktopClientBacklogFull(candidate.Target.ClientID, clientBacklog, maxClientBacklog) {
				continue
			}
			result[platformID] = candidate.Target
			if maxClientBacklog > 0 {
				clientBacklog[candidate.Target.ClientID]++
			}
			break
		}
	}
	return result
}

func monitorDesktopClientBacklogFull(clientID uuid.UUID, clientBacklog map[uuid.UUID]int, maxClientBacklog int) bool {
	return maxClientBacklog > 0 && clientID != uuid.Nil && clientBacklog[clientID] >= maxClientBacklog
}

func reserveMonitorDesktopClientBacklog(oldClientID, newClientID uuid.UUID, clientBacklog map[uuid.UUID]int, maxClientBacklog int) bool {
	if maxClientBacklog <= 0 || newClientID == uuid.Nil || oldClientID == newClientID {
		return true
	}
	if clientBacklog == nil {
		return true
	}
	if monitorDesktopClientBacklogFull(newClientID, clientBacklog, maxClientBacklog) {
		return false
	}
	clientBacklog[newClientID]++
	if oldClientID != uuid.Nil && clientBacklog[oldClientID] > 0 {
		clientBacklog[oldClientID]--
	}
	return true
}

func (s *MonitoringService) reserveMonitorDesktopClientBacklogTx(
	ctx context.Context,
	tx pgx.Tx,
	tenantID, workspaceID int64,
	oldClientID, newClientID uuid.UUID,
	maxClientBacklog int,
) (bool, error) {
	if tx == nil || maxClientBacklog <= 0 || newClientID == uuid.Nil || oldClientID == newClientID {
		return true, nil
	}
	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, monitorDesktopTaskClientSlotLockKey(newClientID)); err != nil {
		return false, response.ErrInternal(50123, "desktop_client_backlog_lock_failed", "failed to lock monitor desktop client backlog")
	}
	var backlog int
	if err := tx.QueryRow(ctx, `
		SELECT COUNT(*)::int
		FROM desktop_tasks
		WHERE tenant_id = $1
		  AND workspace_id = $2
		  AND target_client_id = $3
		  AND kind = 'monitor'
		  AND status IN ('queued', 'in_progress')
	`, tenantID, workspaceID, newClientID).Scan(&backlog); err != nil {
		return false, response.ErrInternal(50124, "desktop_client_backlog_check_failed", "failed to check monitor desktop client backlog")
	}
	return backlog < maxClientBacklog, nil
}

func (candidate monitorDesktopTaskTargetCandidate) betterThan(current monitorDesktopTaskTargetCandidate) bool {
	if candidate.HealthRank != current.HealthRank {
		return candidate.HealthRank < current.HealthRank
	}
	if candidate.SourceRank != current.SourceRank {
		return candidate.SourceRank < current.SourceRank
	}
	if candidate.OnlineSinceUnixMilli != current.OnlineSinceUnixMilli {
		if candidate.OnlineSinceUnixMilli == 0 {
			return false
		}
		if current.OnlineSinceUnixMilli == 0 {
			return true
		}
		return candidate.OnlineSinceUnixMilli < current.OnlineSinceUnixMilli
	}
	if candidate.OnlineRank != current.OnlineRank {
		return candidate.OnlineRank < current.OnlineRank
	}
	if candidate.RecordOrder != current.RecordOrder {
		return candidate.RecordOrder < current.RecordOrder
	}
	if candidate.ClientOrder != current.ClientOrder {
		return candidate.ClientOrder < current.ClientOrder
	}
	return candidate.Target.ClientID.String() < current.Target.ClientID.String()
}

func (s *MonitoringService) upsertMonitorDesktopTask(
	ctx context.Context,
	tx pgx.Tx,
	repo repository.DesktopTaskRepository,
	spec monitorDesktopTaskSpec,
	reserveClientBacklog func(oldClientID, newClientID uuid.UUID) bool,
	reserveClientBacklogTx func(oldClientID, newClientID uuid.UUID) (bool, error),
) (*repository.DesktopTask, bool, bool, error) {
	var (
		existingDesktopID      uuid.UUID
		existingTargetClientID uuid.UUID
		existingStatus         string
		existingLeaseExpiresAt pgtype.Timestamptz
		existingActiveAttempt  pgtype.UUID
	)
	err := tx.QueryRow(ctx, upsertMonitorDesktopTaskActiveLookupSQL(), spec.MonitorTaskID).Scan(&existingDesktopID, &existingTargetClientID, &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
		}
		if reserveClientBacklog != nil && !reserveClientBacklog(existingTargetClientID, spec.TargetClientID) {
			return nil, false, true, nil
		}
		if reserveClientBacklogTx != nil {
			allowed, reserveErr := reserveClientBacklogTx(existingTargetClientID, spec.TargetClientID)
			if reserveErr != nil {
				return nil, false, false, reserveErr
			}
			if !allowed {
				return nil, false, true, nil
			}
		}
		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
	}
	if reserveClientBacklog != nil && !reserveClientBacklog(uuid.Nil, spec.TargetClientID) {
		return nil, false, true, nil
	}
	if reserveClientBacklogTx != nil {
		allowed, reserveErr := reserveClientBacklogTx(uuid.Nil, spec.TargetClientID)
		if reserveErr != nil {
			return nil, false, false, reserveErr
		}
		if !allowed {
			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 upsertMonitorDesktopTaskActiveLookupSQL() string {
	return `
		SELECT desktop_id, target_client_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')
		ORDER BY created_at DESC, id DESC
		LIMIT 1
		FOR UPDATE
	`
}

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 deferPhase2MonitorCollectTasks(ctx context.Context, tx pgx.Tx, taskIDs []int64, delay time.Duration) error {
	if tx == nil || len(taskIDs) == 0 {
		return nil
	}
	if delay <= 0 {
		delay = defaultMonitoringDailyDesktopRetryCooldown
	}
	if _, err := tx.Exec(ctx, `
		UPDATE monitoring_collect_tasks
		SET status = 'pending',
		    lease_token_hash = NULL,
		    leased_to_executor = NULL,
		    leased_at = NULL,
		    lease_expires_at = NULL,
		    callback_received_at = NULL,
		    completed_at = NULL,
		    skip_reason = NULL,
		    dispatch_after = NOW() + $2::interval,
		    error_message = NULL,
		    updated_at = NOW()
		WHERE id = ANY($1::bigint[])
		  AND COALESCE(execution_owner, 'legacy') = 'desktop_tasks'
		  AND callback_received_at IS NULL
	`, uniqueInt64s(taskIDs), formatPgInterval(delay)); err != nil {
		return response.ErrInternal(50041, "task_defer_failed", "failed to defer phase2 monitor tasks until a desktop client is online")
	}
	return 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 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) {
	questionText := strings.TrimSpace(spec.QuestionText)
	if questionText == "" {
		return nil, response.ErrInternal(50041, "missing_question_text_snapshot", "monitor desktop task question text is empty")
	}
	title := questionText
	platformName := platformDisplayName(spec.PlatformID)
	title = fmt.Sprintf("%s · %s", platformName, title)
	payload, err := json.Marshal(map[string]any{
		"title":                  title,
		"business_date":          spec.BusinessDate,
		"scheduler_group_key":    spec.SchedulerGroupKey,
		"question_text":          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
}