From a48f450de609e803e2433bb97b6188b1053715c6 Mon Sep 17 00:00:00 2001 From: liangxu Date: Fri, 19 Jun 2026 20:02:59 +0800 Subject: [PATCH] feat: enhance desktop client presence management and add monitoring for active platforms --- .../internal/tenant/app/desktop_presence.go | 148 ++++++++++++++++-- .../tenant/app/desktop_presence_test.go | 57 +++++++ .../app/monitoring_daily_task_worker.go | 96 ++++++++++-- .../app/monitoring_daily_task_worker_test.go | 80 ++++++++++ .../app/monitoring_phase2_desktop_tasks.go | 104 ++++++++++-- .../monitoring_phase2_desktop_tasks_test.go | 93 ++++++++--- 6 files changed, 503 insertions(+), 75 deletions(-) diff --git a/server/internal/tenant/app/desktop_presence.go b/server/internal/tenant/app/desktop_presence.go index 54321a9..55f4267 100644 --- a/server/internal/tenant/app/desktop_presence.go +++ b/server/internal/tenant/app/desktop_presence.go @@ -16,6 +16,11 @@ const desktopClientPresenceTTL = 30 * time.Second const desktopAccountPresenceTTL = 30 * time.Second const desktopTaskConsumerPresenceTTL = 75 * time.Second +type desktopAccountPresenceClient struct { + ClientID uuid.UUID + OnlineSinceUnixMilli int64 +} + func desktopClientPresenceKey(clientID uuid.UUID) string { return "desktop:presence:client:" + clientID.String() } @@ -44,7 +49,13 @@ func markDesktopClientPresent(ctx context.Context, redis *goredis.Client, client return } - _ = redis.Set(ctx, desktopClientPresenceKey(clientID), time.Now().UTC().Format(time.RFC3339), desktopClientPresenceTTL).Err() + key := desktopClientPresenceKey(clientID) + now := strconv.FormatInt(time.Now().UTC().UnixMilli(), 10) + set, err := redis.SetNX(ctx, key, now, desktopClientPresenceTTL).Result() + if err != nil || set { + return + } + _ = redis.Expire(ctx, key, desktopClientPresenceTTL).Err() } func markDesktopTaskConsumerPresent(ctx context.Context, redis *goredis.Client, clientID uuid.UUID) { @@ -198,8 +209,11 @@ func markDesktopAccountsPresent(ctx context.Context, redis *goredis.Client, clie nextAccountSet[accountID] = struct{}{} } - now := float64(time.Now().UTC().UnixMilli()) - staleBefore := strconv.FormatInt(time.Now().UTC().Add(-desktopAccountPresenceTTL).UnixMilli(), 10) + now := time.Now().UTC() + onlineSinceUnixMilli := now.UnixMilli() + if value, err := redis.Get(ctx, desktopClientPresenceKey(clientID)).Result(); err == nil { + onlineSinceUnixMilli = parseDesktopPresenceUnixMilli(value, now) + } clientAccountsKey := desktopClientAccountsPresenceKey(clientID) clientMember := clientID.String() @@ -222,9 +236,8 @@ func markDesktopAccountsPresent(ctx context.Context, redis *goredis.Client, clie } for _, accountID := range uniqueIDs { key := desktopAccountPresenceKey(accountID) - pipe.ZRemRangeByScore(ctx, key, "-inf", staleBefore) pipe.ZAdd(ctx, key, goredis.Z{ - Score: now, + Score: float64(onlineSinceUnixMilli), Member: clientMember, }) pipe.Expire(ctx, key, desktopAccountPresenceTTL*4) @@ -259,44 +272,145 @@ func clearDesktopClientPresenceState(ctx context.Context, redis *goredis.Client, } func loadDesktopAccountPresence(ctx context.Context, redis *goredis.Client, accountIDs []uuid.UUID) map[uuid.UUID]uuid.UUID { + clientsByAccount := loadDesktopAccountPresenceClients(ctx, redis, accountIDs) + if len(clientsByAccount) == 0 { + return nil + } + + result := make(map[uuid.UUID]uuid.UUID, len(clientsByAccount)) + for accountID, clients := range clientsByAccount { + if len(clients) == 0 { + continue + } + result[accountID] = clients[len(clients)-1].ClientID + } + return result +} + +func loadDesktopAccountPresenceClients(ctx context.Context, redis *goredis.Client, accountIDs []uuid.UUID) map[uuid.UUID][]desktopAccountPresenceClient { if redis == nil || len(accountIDs) == 0 { return nil } uniqueIDs := uniqueUUIDs(accountIDs) - staleBefore := strconv.FormatInt(time.Now().UTC().Add(-desktopAccountPresenceTTL).UnixMilli(), 10) pipe := redis.Pipeline() commands := make(map[uuid.UUID]*goredis.ZSliceCmd, len(uniqueIDs)) for _, accountID := range uniqueIDs { key := desktopAccountPresenceKey(accountID) - pipe.ZRemRangeByScore(ctx, key, "-inf", staleBefore) - commands[accountID] = pipe.ZRevRangeWithScores(ctx, key, 0, 0) + commands[accountID] = pipe.ZRangeWithScores(ctx, key, 0, -1) } if _, err := pipe.Exec(ctx); err != nil && err != goredis.Nil { return nil } - result := make(map[uuid.UUID]uuid.UUID, len(commands)) + result := make(map[uuid.UUID][]desktopAccountPresenceClient, len(commands)) + clientIDs := make([]uuid.UUID, 0) + staleMembersByAccount := make(map[uuid.UUID][]interface{}) for accountID, cmd := range commands { values := cmd.Val() if len(values) == 0 { continue } - clientIDText, ok := values[0].Member.(string) - if !ok || clientIDText == "" { - continue + seenClients := make(map[uuid.UUID]struct{}, len(values)) + for _, value := range values { + clientIDText, ok := value.Member.(string) + if !ok || clientIDText == "" { + staleMembersByAccount[accountID] = append(staleMembersByAccount[accountID], value.Member) + continue + } + clientID, err := uuid.Parse(clientIDText) + if err != nil { + staleMembersByAccount[accountID] = append(staleMembersByAccount[accountID], clientIDText) + continue + } + if _, exists := seenClients[clientID]; exists { + continue + } + seenClients[clientID] = struct{}{} + result[accountID] = append(result[accountID], desktopAccountPresenceClient{ + ClientID: clientID, + OnlineSinceUnixMilli: int64(value.Score), + }) + clientIDs = append(clientIDs, clientID) } - clientID, err := uuid.Parse(clientIDText) - if err != nil { - continue - } - result[accountID] = clientID } + + clientIDs = uniqueUUIDs(clientIDs) + if len(clientIDs) == 0 { + pruneDesktopAccountPresenceMembers(ctx, redis, staleMembersByAccount) + cleanupDesktopAccountPresenceKeys(ctx, redis, uniqueIDs) + return nil + } + + presencePipe := redis.Pipeline() + presenceCommands := make(map[uuid.UUID]*goredis.IntCmd, len(clientIDs)) + for _, clientID := range clientIDs { + presenceCommands[clientID] = presencePipe.Exists(ctx, desktopClientPresenceKey(clientID)) + } + if _, err := presencePipe.Exec(ctx); err != nil && err != goredis.Nil { + return nil + } + + for accountID, clients := range result { + kept := clients[:0] + for _, client := range clients { + cmd := presenceCommands[client.ClientID] + if cmd != nil && cmd.Val() > 0 { + kept = append(kept, client) + continue + } + staleMembersByAccount[accountID] = append(staleMembersByAccount[accountID], client.ClientID.String()) + } + if len(kept) == 0 { + delete(result, accountID) + continue + } + result[accountID] = kept + } + + pruneDesktopAccountPresenceMembers(ctx, redis, staleMembersByAccount) + cleanupDesktopAccountPresenceKeys(ctx, redis, uniqueIDs) return result } +func parseDesktopPresenceUnixMilli(value string, fallback time.Time) int64 { + return parseDesktopPresenceTime(value, fallback).UnixMilli() +} + +func parseDesktopPresenceTime(value string, fallback time.Time) time.Time { + if millis, err := strconv.ParseInt(value, 10, 64); err == nil && millis > 0 { + return time.UnixMilli(millis).UTC() + } + if parsed, err := time.Parse(time.RFC3339Nano, value); err == nil { + return parsed.UTC() + } + if parsed, err := time.Parse(time.RFC3339, value); err == nil { + return parsed.UTC() + } + return fallback.UTC() +} + +func pruneDesktopAccountPresenceMembers(ctx context.Context, redis *goredis.Client, membersByAccount map[uuid.UUID][]interface{}) { + if redis == nil || len(membersByAccount) == 0 { + return + } + + pipe := redis.Pipeline() + shouldExec := false + for accountID, members := range membersByAccount { + if len(members) == 0 { + continue + } + pipe.ZRem(ctx, desktopAccountPresenceKey(accountID), members...) + shouldExec = true + } + if shouldExec { + _, _ = pipe.Exec(ctx) + } +} + func uniqueUUIDs(values []uuid.UUID) []uuid.UUID { if len(values) == 0 { return nil diff --git a/server/internal/tenant/app/desktop_presence_test.go b/server/internal/tenant/app/desktop_presence_test.go index b1a45a5..41f3021 100644 --- a/server/internal/tenant/app/desktop_presence_test.go +++ b/server/internal/tenant/app/desktop_presence_test.go @@ -2,6 +2,7 @@ package app import ( "context" + "strconv" "testing" "time" @@ -111,3 +112,59 @@ func TestDesktopTaskConsumerPresenceIsSeparateFromHeartbeatPresence(t *testing.T require.NotNil(t, taskConsumerPresence) assert.True(t, taskConsumerPresence[clientID]) } + +func TestDesktopClientPresenceKeepsOriginalOnlineSinceAcrossHeartbeats(t *testing.T) { + ctx := context.Background() + server := miniredis.RunT(t) + redis := goredis.NewClient(&goredis.Options{Addr: server.Addr()}) + t.Cleanup(func() { + require.NoError(t, redis.Close()) + }) + + clientID := uuid.New() + + markDesktopClientPresent(ctx, redis, clientID) + firstOnlineSince := redis.Get(ctx, desktopClientPresenceKey(clientID)).Val() + require.NotEmpty(t, firstOnlineSince) + + time.Sleep(time.Millisecond) + markDesktopClientPresent(ctx, redis, clientID) + secondOnlineSince := redis.Get(ctx, desktopClientPresenceKey(clientID)).Val() + + assert.Equal(t, firstOnlineSince, secondOnlineSince) + assert.Positive(t, redis.TTL(ctx, desktopClientPresenceKey(clientID)).Val()) +} + +func TestDesktopAccountPresenceClientsKeepLongRunningOnlineClientAndOrderByOnlineSince(t *testing.T) { + ctx := context.Background() + server := miniredis.RunT(t) + redis := goredis.NewClient(&goredis.Options{Addr: server.Addr()}) + t.Cleanup(func() { + require.NoError(t, redis.Close()) + }) + + accountID := uuid.New() + earlierClientID := uuid.New() + laterClientID := uuid.New() + offlineClientID := uuid.New() + now := time.Date(2026, 6, 19, 10, 0, 0, 0, time.UTC) + earlierOnlineSince := now.Add(-time.Hour).UnixMilli() + laterOnlineSince := now.UnixMilli() + + require.NoError(t, redis.Set(ctx, desktopClientPresenceKey(earlierClientID), strconv.FormatInt(earlierOnlineSince, 10), desktopClientPresenceTTL).Err()) + require.NoError(t, redis.Set(ctx, desktopClientPresenceKey(laterClientID), strconv.FormatInt(laterOnlineSince, 10), desktopClientPresenceTTL).Err()) + require.NoError(t, redis.ZAdd(ctx, desktopAccountPresenceKey(accountID), + goredis.Z{Score: float64(laterOnlineSince), Member: laterClientID.String()}, + goredis.Z{Score: float64(earlierOnlineSince), Member: earlierClientID.String()}, + goredis.Z{Score: float64(now.Add(-2 * time.Hour).UnixMilli()), Member: offlineClientID.String()}, + ).Err()) + + clientsByAccount := loadDesktopAccountPresenceClients(ctx, redis, []uuid.UUID{accountID}) + + require.Len(t, clientsByAccount[accountID], 2) + assert.Equal(t, earlierClientID, clientsByAccount[accountID][0].ClientID) + assert.Equal(t, earlierOnlineSince, clientsByAccount[accountID][0].OnlineSinceUnixMilli) + assert.Equal(t, laterClientID, clientsByAccount[accountID][1].ClientID) + assert.Equal(t, laterOnlineSince, clientsByAccount[accountID][1].OnlineSinceUnixMilli) + assert.False(t, redis.ZScore(ctx, desktopAccountPresenceKey(accountID), offlineClientID.String()).Val() > 0) +} diff --git a/server/internal/tenant/app/monitoring_daily_task_worker.go b/server/internal/tenant/app/monitoring_daily_task_worker.go index 01a04ca..bd81a6b 100644 --- a/server/internal/tenant/app/monitoring_daily_task_worker.go +++ b/server/internal/tenant/app/monitoring_daily_task_worker.go @@ -31,6 +31,7 @@ const ( defaultMonitoringDailyMaxMaterializePerRun = 64 defaultMonitoringDailyMaxMaterializePerPlan = 12 defaultMonitoringDailyMaxMaterializePerBrand = 4 + monitoringDailyPlatformActiveWindow = 24 * time.Hour monitoringDailyMaterializeStartOffset = 5 * time.Minute ) @@ -501,6 +502,13 @@ func (s *MonitoringService) generateDailyMonitoringTasksForBrand( if dispatchLimit <= 0 { return createdCount, dueCount, 0, 0, nil } + dispatchablePlatformIDs, err := s.loadDailyMonitorDispatchablePlatformIDs(ctx, plan.TenantID, plan.WorkspaceID, candidates) + if err != nil { + return createdCount, dueCount, 0, 0, err + } + if len(dispatchablePlatformIDs) == 0 { + return createdCount, dueCount, 0, 0, nil + } specs, err := s.loadDueDailyMonitorDesktopTaskSpecs( ctx, s.monitoringPool, @@ -510,6 +518,7 @@ func (s *MonitoringService) generateDailyMonitoringTasksForBrand( businessDay, now.Add(defaultMonitoringDailyLookahead), now.Add(-defaultMonitoringDailyDesktopRetryCooldown), + dispatchablePlatformIDs, dispatchLimit, ) if err != nil { @@ -606,6 +615,8 @@ func (s *MonitoringService) loadDailyMonitoringPrimaryClientPlans( AND p.status = 'active' AND p.deleted_at IS NULL WHERE dc.revoked_at IS NULL + AND dc.last_seen_at IS NOT NULL + AND dc.last_seen_at >= NOW() - $5::interval AND EXISTS ( SELECT 1 FROM platform_accounts pa @@ -650,7 +661,7 @@ func (s *MonitoringService) loadDailyMonitoringPrimaryClientPlans( $3::int AS max_questions_per_brand FROM selected_clients c ORDER BY c.tenant_id ASC, c.workspace_id ASC - `, freeBrandLimit, paidBrandLimit, questionLimit, supportedPlatforms) + `, freeBrandLimit, paidBrandLimit, questionLimit, supportedPlatforms, formatPgInterval(monitoringDailyPlatformActiveWindow)) if err != nil { appErr := response.ErrInternal(50041, "query_failed", "failed to load daily monitoring plans") appErr.Cause = err @@ -716,27 +727,32 @@ func (s *MonitoringService) loadDailyMonitoringPlanPlatformJSON( ) AS input(tenant_id, workspace_id, client_id) ) SELECT - p.tenant_id, - p.workspace_id, - p.client_id, + sc.tenant_id, + sc.workspace_id, + sc.client_id, jsonb_agg(p.platform_id ORDER BY p.platform_id) AS enabled_platforms - FROM ( + FROM selected_clients sc + JOIN ( SELECT DISTINCT - sc.tenant_id, - sc.workspace_id, - sc.client_id, + dc.tenant_id, + dc.workspace_id, pa.platform_id - FROM selected_clients sc + FROM desktop_clients dc JOIN platform_accounts pa - ON pa.tenant_id = sc.tenant_id - AND pa.workspace_id = sc.workspace_id - AND pa.client_id = sc.client_id + ON pa.tenant_id = dc.tenant_id + AND pa.workspace_id = dc.workspace_id + AND pa.client_id = dc.id WHERE pa.deleted_at IS NULL AND pa.platform_id = ANY($4::text[]) + AND dc.revoked_at IS NULL + AND dc.last_seen_at IS NOT NULL + AND dc.last_seen_at >= NOW() - $5::interval ) p - GROUP BY p.tenant_id, p.workspace_id, p.client_id - ORDER BY p.tenant_id ASC, p.workspace_id ASC, p.client_id ASC - `, tenantIDs, workspaceIDs, clientIDs, supportedPlatforms) + ON p.tenant_id = sc.tenant_id + AND p.workspace_id = sc.workspace_id + GROUP BY sc.tenant_id, sc.workspace_id, sc.client_id + ORDER BY sc.tenant_id ASC, sc.workspace_id ASC, sc.client_id ASC + `, tenantIDs, workspaceIDs, clientIDs, supportedPlatforms, formatPgInterval(monitoringDailyPlatformActiveWindow)) if err != nil { appErr := response.ErrInternal(50041, "query_failed", "failed to load daily monitoring plan platforms") appErr.Cause = err @@ -888,6 +904,46 @@ func (s *MonitoringService) countDailyMonitoringMaterializedTasks(ctx context.Co return count, nil } +func (s *MonitoringService) loadDailyMonitorDispatchablePlatformIDs( + ctx context.Context, + tenantID, workspaceID int64, + candidates []monitoringDailyTaskCandidate, +) ([]string, error) { + if len(candidates) == 0 { + return nil, nil + } + specs := make([]monitorDesktopTaskSpec, 0, len(candidates)) + for _, candidate := range candidates { + platformID := normalizeMonitoringPlatformID(candidate.Platform.ID) + if platformID == "" { + continue + } + specs = append(specs, monitorDesktopTaskSpec{ + TenantID: tenantID, + WorkspaceID: workspaceID, + PlatformID: platformID, + QuestionID: candidate.Question.ID, + QuestionText: candidate.Question.QuestionText, + }) + } + if len(specs) == 0 { + return nil, nil + } + + targets, err := s.loadMonitorDesktopTaskTargets(ctx, tenantID, workspaceID, 0, specs) + if err != nil { + return nil, err + } + if len(targets) == 0 { + return nil, nil + } + platformIDs := make([]string, 0, len(targets)) + for platformID := range targets { + platformIDs = append(platformIDs, platformID) + } + return reconcileEnabledMonitoringPlatforms(platformIDs), nil +} + func (s *MonitoringService) ensureDailyMonitoringCollectTasks( ctx context.Context, tx pgx.Tx, @@ -984,11 +1040,16 @@ func (s *MonitoringService) loadDueDailyMonitorDesktopTaskSpecs( businessDate time.Time, cutoff time.Time, retryBefore time.Time, + platformIDs []string, limit int, ) ([]monitorDesktopTaskSpec, error) { if q == nil || limit <= 0 { return nil, nil } + platformIDs = reconcileEnabledMonitoringPlatforms(platformIDs) + if len(platformIDs) == 0 { + return nil, nil + } rows, err := q.Query(ctx, ` WITH candidate_tasks AS ( @@ -1003,10 +1064,11 @@ func (s *MonitoringService) loadDueDailyMonitorDesktopTaskSpecs( AND t.trigger_source = 'automatic' AND t.run_mode = 'desktop_standard' AND COALESCE(t.execution_owner, 'legacy') = 'desktop_tasks' + AND t.ai_platform_id = ANY($7::text[]) AND (t.dispatch_after IS NULL OR t.dispatch_after <= $5) AND (t.last_dispatched_at IS NULL OR t.last_dispatched_at <= $6) ORDER BY t.dispatch_after ASC NULLS FIRST, t.dispatch_priority DESC, t.id ASC - LIMIT $7 + LIMIT $8 FOR UPDATE SKIP LOCKED ), claimed_tasks AS ( @@ -1043,7 +1105,7 @@ func (s *MonitoringService) loadDueDailyMonitorDesktopTaskSpecs( ), '') AS question_text_snapshot FROM claimed_tasks t ORDER BY t.dispatch_after ASC NULLS FIRST, t.dispatch_priority DESC, t.id ASC - `, tenantID, brandID, monitoringCollectorType, monitoringBusinessDateText(businessDate), cutoff.UTC(), retryBefore.UTC(), limit) + `, tenantID, brandID, monitoringCollectorType, monitoringBusinessDateText(businessDate), cutoff.UTC(), retryBefore.UTC(), platformIDs, limit) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load due daily monitor desktop task specs") } diff --git a/server/internal/tenant/app/monitoring_daily_task_worker_test.go b/server/internal/tenant/app/monitoring_daily_task_worker_test.go index 0b37229..00d3dff 100644 --- a/server/internal/tenant/app/monitoring_daily_task_worker_test.go +++ b/server/internal/tenant/app/monitoring_daily_task_worker_test.go @@ -10,6 +10,8 @@ import ( "time" "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -329,9 +331,87 @@ func TestMonitoringSchedulerMetricsPrometheusHandlerIncludesRuntimeAndProcessCol assert.True(t, strings.HasPrefix(recorder.Header().Get("Content-Type"), "text/plain")) } +func TestLoadDueDailyMonitorDesktopTaskSpecsFiltersToDispatchablePlatforms(t *testing.T) { + ctx := context.Background() + queryer := &captureDailyMonitorTaskQueryer{} + businessDay := time.Date(2026, 6, 19, 0, 0, 0, 0, time.UTC) + + specs, err := (&MonitoringService{}).loadDueDailyMonitorDesktopTaskSpecs( + ctx, + queryer, + 4, + 8, + 2, + businessDay, + businessDay.Add(time.Hour), + businessDay.Add(-time.Hour), + []string{"qwen", "unknown", "doubao"}, + 7, + ) + + require.NoError(t, err) + assert.Empty(t, specs) + assert.True(t, queryer.called) + assert.Contains(t, normalizeSQLWhitespace(queryer.sql), "AND t.ai_platform_id = ANY($7::text[])") + assert.Contains(t, normalizeSQLWhitespace(queryer.sql), "LIMIT $8") + require.Len(t, queryer.args, 8) + assert.Equal(t, int64(4), queryer.args[0]) + assert.Equal(t, int64(2), queryer.args[1]) + assert.Equal(t, []string{"doubao", "qwen"}, queryer.args[6]) + assert.Equal(t, 7, queryer.args[7]) +} + +func TestLoadDueDailyMonitorDesktopTaskSpecsSkipsWithoutDispatchablePlatforms(t *testing.T) { + ctx := context.Background() + queryer := &captureDailyMonitorTaskQueryer{} + businessDay := time.Date(2026, 6, 19, 0, 0, 0, 0, time.UTC) + + specs, err := (&MonitoringService{}).loadDueDailyMonitorDesktopTaskSpecs( + ctx, + queryer, + 4, + 8, + 2, + businessDay, + businessDay.Add(time.Hour), + businessDay.Add(-time.Hour), + []string{"unknown"}, + 7, + ) + + require.NoError(t, err) + assert.Empty(t, specs) + assert.False(t, queryer.called) +} + func mustParseDailyMonitoringUUID(t *testing.T, value string) uuid.UUID { t.Helper() id, err := uuid.Parse(value) require.NoError(t, err) return id } + +type captureDailyMonitorTaskQueryer struct { + called bool + sql string + args []any +} + +func (q *captureDailyMonitorTaskQueryer) Query(_ context.Context, sql string, args ...any) (pgx.Rows, error) { + q.called = true + q.sql = sql + q.args = args + return emptyDailyMonitorRows{}, nil +} + +type emptyDailyMonitorRows struct{} + +func (r emptyDailyMonitorRows) Close() {} +func (r emptyDailyMonitorRows) Err() error { return nil } +func (r emptyDailyMonitorRows) CommandTag() pgconn.CommandTag { return pgconn.CommandTag{} } +func (r emptyDailyMonitorRows) FieldDescriptions() []pgconn.FieldDescription { return nil } +func (r emptyDailyMonitorRows) Next() bool { return false } +func (r emptyDailyMonitorRows) Scan(...any) error { return nil } +func (r emptyDailyMonitorRows) Values() ([]any, error) { return nil, nil } +func (r emptyDailyMonitorRows) RawValues() [][]byte { return nil } +func (r emptyDailyMonitorRows) Conn() *pgx.Conn { return nil } diff --git a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go index 88e5db5..d9cf26d 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go @@ -56,6 +56,16 @@ type monitorDesktopTaskTarget struct { ClientID uuid.UUID } +type monitorDesktopTaskTargetCandidate struct { + Target monitorDesktopTaskTarget + HealthRank int + OnlineRank int + SourceRank int + OnlineSinceUnixMilli int64 + RecordOrder int + ClientOrder int +} + type monitorDesktopInterruptTarget struct { TaskID string MonitorTaskID int64 @@ -334,7 +344,7 @@ func (s *MonitoringService) loadMonitorDesktopTaskTargets( } accountIDs := monitorDesktopAccountCandidateAccountIDs(candidates) - accountPresence := loadDesktopAccountPresence(ctx, s.redis, accountIDs) + accountPresence := loadDesktopAccountPresenceClients(ctx, s.redis, accountIDs) clientIDs := monitorDesktopAccountCandidateClientIDs(candidates, accountPresence) onlineClients, err := s.loadOnlineMonitorDesktopClients(ctx, workspaceID, clientIDs) if err != nil { @@ -373,15 +383,17 @@ func monitorDesktopAccountCandidateAccountIDs(candidates []monitorDesktopAccount return uniqueUUIDs(result) } -func monitorDesktopAccountCandidateClientIDs(candidates []monitorDesktopAccountCandidate, accountPresence map[uuid.UUID]uuid.UUID) []uuid.UUID { +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 { - if clientID, ok := accountPresence[candidate.AccountID]; ok && clientID != uuid.Nil { - result = append(result, clientID) + for _, client := range accountPresence[candidate.AccountID] { + if client.ClientID != uuid.Nil { + result = append(result, client.ClientID) + } } } } @@ -430,35 +442,95 @@ func (s *MonitoringService) loadOnlineMonitorDesktopClients(ctx context.Context, func selectMonitorDesktopTaskTargets( candidates []monitorDesktopAccountCandidate, - accountPresence map[uuid.UUID]uuid.UUID, + accountPresence map[uuid.UUID][]desktopAccountPresenceClient, onlineClients map[uuid.UUID]bool, ) map[string]monitorDesktopTaskTarget { result := make(map[string]monitorDesktopTaskTarget) - targetsByPlatform := make(map[string][]monitorDesktopTaskTarget) + bestByPlatform := make(map[string]monitorDesktopTaskTargetCandidate) 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}) + 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, + } + if current, ok := bestByPlatform[platformID]; !ok || item.betterThan(current) { + bestByPlatform[platformID] = item + } + } + } + if candidate.DBClientID == nil || *candidate.DBClientID == uuid.Nil || !onlineClients[*candidate.DBClientID] { continue } - if candidate.DBClientID != nil && *candidate.DBClientID != uuid.Nil && onlineClients[*candidate.DBClientID] { - targetsByPlatform[platformID] = append(targetsByPlatform[platformID], monitorDesktopTaskTarget{AccountID: candidate.AccountID, ClientID: *candidate.DBClientID}) + 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, + } + if current, ok := bestByPlatform[platformID]; !ok || item.betterThan(current) { + bestByPlatform[platformID] = item } } - for platformID, targets := range targetsByPlatform { - if len(targets) == 0 { - continue - } - index := randomIndex(len(targets)) - result[platformID] = targets[index] + for platformID, candidate := range bestByPlatform { + result[platformID] = candidate.Target } return result } +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, diff --git a/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go b/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go index 51a55b9..8fc350e 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go @@ -20,8 +20,10 @@ func TestSelectMonitorDesktopTaskTargetsPrefersPresentAccountClient(t *testing.T DBClientID: &staleClientID, }, }, - map[uuid.UUID]uuid.UUID{ - accountID: liveClientID, + map[uuid.UUID][]desktopAccountPresenceClient{ + accountID: { + {ClientID: liveClientID, OnlineSinceUnixMilli: 100}, + }, }, map[uuid.UUID]bool{ liveClientID: true, @@ -77,30 +79,71 @@ func TestSelectMonitorDesktopTaskTargetsSkipsOfflineClients(t *testing.T) { assert.Empty(t, targets) } -func TestSelectMonitorDesktopTaskTargetsRandomizesAmongOnlineCandidates(t *testing.T) { - clientIDs := []uuid.UUID{ - mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000301"), - mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000302"), - } - accountIDs := []uuid.UUID{ - mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000401"), - mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000402"), - } +func TestSelectMonitorDesktopTaskTargetsPrefersHealthierOnlineCandidate(t *testing.T) { + riskClientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000301") + liveClientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000302") + riskAccountID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000401") + liveAccountID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000402") - for i := 0; i < 40; i++ { - targets := selectMonitorDesktopTaskTargets( - []monitorDesktopAccountCandidate{ - {PlatformID: "qwen", AccountID: accountIDs[0], DBClientID: &clientIDs[0]}, - {PlatformID: "qwen", AccountID: accountIDs[1], DBClientID: &clientIDs[1]}, + targets := selectMonitorDesktopTaskTargets( + []monitorDesktopAccountCandidate{ + { + PlatformID: "qwen", + AccountID: riskAccountID, + DBClientID: &riskClientID, + HealthRank: 1, + RecordOrder: 0, }, - nil, - map[uuid.UUID]bool{ - clientIDs[0]: true, - clientIDs[1]: true, + { + PlatformID: "qwen", + AccountID: liveAccountID, + DBClientID: &liveClientID, + HealthRank: 0, + RecordOrder: 1, }, - ) - target := targets["qwen"] - assert.Contains(t, clientIDs, target.ClientID) - assert.Contains(t, accountIDs, target.AccountID) - } + }, + nil, + map[uuid.UUID]bool{ + riskClientID: true, + liveClientID: true, + }, + ) + + assert.Equal(t, monitorDesktopTaskTarget{ + AccountID: liveAccountID, + ClientID: liveClientID, + }, targets["qwen"]) +} + +func TestSelectMonitorDesktopTaskTargetsPrefersEarlierOnlineClientForSameAccount(t *testing.T) { + accountID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000501") + earlierClientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000601") + laterClientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000602") + + targets := selectMonitorDesktopTaskTargets( + []monitorDesktopAccountCandidate{ + { + PlatformID: "doubao", + AccountID: accountID, + DBClientID: &laterClientID, + HealthRank: 0, + RecordOrder: 0, + }, + }, + map[uuid.UUID][]desktopAccountPresenceClient{ + accountID: { + {ClientID: laterClientID, OnlineSinceUnixMilli: 200}, + {ClientID: earlierClientID, OnlineSinceUnixMilli: 100}, + }, + }, + map[uuid.UUID]bool{ + earlierClientID: true, + laterClientID: true, + }, + ) + + assert.Equal(t, monitorDesktopTaskTarget{ + AccountID: accountID, + ClientID: earlierClientID, + }, targets["doubao"]) }