package app import ( "context" "crypto/sha256" "encoding/hex" "encoding/json" "errors" "fmt" "strings" "time" "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" goredis "github.com/redis/go-redis/v9" "github.com/geo-platform/tenant-api/internal/shared/auth" sharedcache "github.com/geo-platform/tenant-api/internal/shared/cache" "github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq" "github.com/geo-platform/tenant-api/internal/shared/response" "github.com/geo-platform/tenant-api/internal/tenant/repository" ) const desktopAccountHealthReportTTL = 24 * time.Hour type DesktopAccountService struct { pool *pgxpool.Pool repo repository.DesktopAccountRepository clientRepo repository.DesktopClientRepository cache sharedcache.Cache redis *goredis.Client rabbitMQ *rabbitmq.Client now func() time.Time } func NewDesktopAccountService( pool *pgxpool.Pool, repo repository.DesktopAccountRepository, clientRepo repository.DesktopClientRepository, redis *goredis.Client, rabbitMQClient *rabbitmq.Client, ) *DesktopAccountService { return &DesktopAccountService{ pool: pool, repo: repo, clientRepo: clientRepo, redis: redis, rabbitMQ: rabbitMQClient, now: time.Now, } } func (s *DesktopAccountService) WithCache(c sharedcache.Cache) *DesktopAccountService { s.cache = c return s } type DesktopAccountView struct { ID string `json:"id"` Platform string `json:"platform"` PlatformUID string `json:"platform_uid"` DisplayName string `json:"display_name"` AvatarURL *string `json:"avatar_url"` Health string `json:"health"` RuntimeHealth *string `json:"runtime_health"` RuntimeVerifiedAt *time.Time `json:"runtime_verified_at"` RuntimeCheckedAt *time.Time `json:"runtime_checked_at"` RuntimeAuthState *string `json:"runtime_auth_state"` RuntimeProbeState *string `json:"runtime_probe_state"` RuntimeAuthReason *string `json:"runtime_auth_reason"` HealthSource string `json:"health_source"` ClientID *string `json:"client_id"` AccountFingerprint *string `json:"account_fingerprint"` VerifiedAt *time.Time `json:"verified_at"` Tags []string `json:"tags"` SyncVersion int64 `json:"sync_version"` DeletedAt *time.Time `json:"deleted_at"` DeleteRequestedAt *time.Time `json:"delete_requested_at"` ClientOnline *bool `json:"client_online"` AccountSessionOnline *bool `json:"account_session_online"` TaskConsumerOnline *bool `json:"task_consumer_online"` ClientLastSeenAt *time.Time `json:"client_last_seen_at"` ClientDeviceName *string `json:"client_device_name"` QueuedPublishTaskCount int `json:"queued_publish_task_count"` OldestQueuedPublishTaskAt *time.Time `json:"oldest_queued_publish_task_at"` } type DesktopAccountDeleteResult struct { Account DesktopAccountView `json:"account"` CancelledQueuedPublishTasks int `json:"cancelled_queued_publish_task_count"` CompletedServerSideUnbinding bool `json:"completed_server_side_unbinding"` } type UpsertDesktopAccountRequest struct { Platform string `json:"platform" binding:"required"` PlatformUID string `json:"platform_uid" binding:"required"` AccountFingerprint *string `json:"account_fingerprint"` DisplayName string `json:"display_name" binding:"required"` AvatarURL *string `json:"avatar_url"` Health string `json:"health" binding:"required,oneof=live expired captcha risk"` VerifiedAt *time.Time `json:"verified_at"` Tags []string `json:"tags"` IfSyncVersion *int64 `json:"if_sync_version"` } type PatchDesktopAccountRequest struct { DisplayName *string `json:"display_name"` Health *string `json:"health" binding:"omitempty,oneof=live expired captcha risk"` VerifiedAt *time.Time `json:"verified_at"` Tags *[]string `json:"tags"` IfSyncVersion int64 `json:"if_sync_version" binding:"required"` } type DesktopAccountHealthReportRequest struct { AccountID string `json:"account_id" binding:"required"` Platform string `json:"platform" binding:"required"` PlatformUID string `json:"platform_uid" binding:"required"` Health string `json:"health" binding:"required,oneof=live expired captcha risk"` AuthState string `json:"auth_state" binding:"required"` ProbeState string `json:"probe_state" binding:"required"` AuthReason *string `json:"auth_reason"` DisplayName *string `json:"display_name"` AvatarURL *string `json:"avatar_url"` VerifiedAt *time.Time `json:"verified_at"` CheckedAt time.Time `json:"checked_at" binding:"required"` } type ReportDesktopAccountHealthRequest struct { Reports []DesktopAccountHealthReportRequest `json:"reports" binding:"required,min=1,max=100"` } type ReportDesktopAccountHealthResponse struct { AcceptedCount int `json:"accepted_count"` BufferedCount int `json:"buffered_count"` } type bufferedDesktopAccountHealthReport struct { TenantID int64 `json:"tenant_id"` WorkspaceID int64 `json:"workspace_id"` UserID int64 `json:"user_id"` ClientID string `json:"client_id"` AccountID string `json:"account_id"` Platform string `json:"platform"` PlatformUID string `json:"platform_uid"` Health string `json:"health"` AuthState string `json:"auth_state"` ProbeState string `json:"probe_state"` AuthReason *string `json:"auth_reason"` DisplayName *string `json:"display_name"` AvatarURL *string `json:"avatar_url"` VerifiedAt *time.Time `json:"verified_at"` CheckedAt time.Time `json:"checked_at"` ReceivedAt time.Time `json:"received_at"` } func (s *DesktopAccountService) ListByClient(ctx context.Context, client *repository.DesktopClient) ([]DesktopAccountView, error) { if client == nil { return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required") } rows, err := s.repo.ListByClient(ctx, client.WorkspaceID, client.ID) if err != nil { return nil, response.ErrInternal(50081, "desktop_accounts_query_failed", "failed to list desktop accounts") } clientMap, accountPresenceMap, clientPresenceMap, taskConsumerPresenceMap := s.loadClientState(ctx, client.WorkspaceID, rows) items := make([]DesktopAccountView, 0, len(rows)) for _, item := range rows { items = append(items, s.buildDesktopAccountView(&item, clientMap, accountPresenceMap, clientPresenceMap, taskConsumerPresenceMap)) } s.applyRuntimeHealth(ctx, client.WorkspaceID, items) s.applyPublishQueueSummaries(ctx, client.WorkspaceID, items) return items, nil } func (s *DesktopAccountService) ListForActor(ctx context.Context, actor auth.Actor) ([]DesktopAccountView, error) { if actor.PrimaryWorkspaceID == 0 || actor.UserID == 0 { return nil, response.ErrUnauthorized(40132, "workspace_scope_missing", "workspace context is required") } rows, err := s.repo.ListByUser(ctx, actor.PrimaryWorkspaceID, actor.UserID) if err != nil { return nil, response.ErrInternal(50081, "desktop_accounts_query_failed", "failed to list desktop accounts") } clientMap, accountPresenceMap, clientPresenceMap, taskConsumerPresenceMap := s.loadClientState(ctx, actor.PrimaryWorkspaceID, rows) items := make([]DesktopAccountView, 0, len(rows)) for _, item := range rows { items = append(items, s.buildDesktopAccountView(&item, clientMap, accountPresenceMap, clientPresenceMap, taskConsumerPresenceMap)) } s.applyRuntimeHealth(ctx, actor.PrimaryWorkspaceID, items) s.applyPublishQueueSummaries(ctx, actor.PrimaryWorkspaceID, items) return items, nil } func (s *DesktopAccountService) Upsert(ctx context.Context, client *repository.DesktopClient, req UpsertDesktopAccountRequest) (*DesktopAccountView, error) { if client == nil { return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required") } if req.IfSyncVersion != nil { existing, err := s.repo.GetByIdentity(ctx, client.WorkspaceID, client.ID, req.Platform, req.PlatformUID) if err != nil && !errors.Is(err, pgx.ErrNoRows) { return nil, response.ErrInternal(50082, "desktop_account_lookup_failed", "failed to inspect desktop account before upsert") } if existing == nil || errors.Is(err, pgx.ErrNoRows) { return nil, response.ErrConflict(40981, "desktop_account_sync_version_required_existing", "if_sync_version is only valid when account already exists") } } account, err := s.repo.Upsert(ctx, repository.UpsertDesktopAccountParams{ DesktopID: uuid.New(), TenantID: client.TenantID, WorkspaceID: client.WorkspaceID, UserID: client.UserID, ClientID: client.ID, Platform: req.Platform, PlatformUID: req.PlatformUID, AccountFingerprint: req.AccountFingerprint, DisplayName: req.DisplayName, AvatarURL: req.AvatarURL, Health: req.Health, VerifiedAt: req.VerifiedAt, Tags: req.Tags, IfSyncVersion: req.IfSyncVersion, }) if err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, response.ErrConflict(40982, "desktop_account_sync_conflict", "desktop account sync version conflict") } appErr := response.ErrInternal(50083, "desktop_account_upsert_failed", "failed to upsert desktop account") appErr.Cause = err return nil, appErr } view := s.buildDesktopAccountView(account, nil, nil, nil, nil) return &view, nil } func (s *DesktopAccountService) ReportHealth(ctx context.Context, client *repository.DesktopClient, req ReportDesktopAccountHealthRequest) (*ReportDesktopAccountHealthResponse, error) { if client == nil { return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required") } if len(req.Reports) == 0 { return nil, response.ErrBadRequest(40087, "empty_account_health_reports", "at least one account health report is required") } if len(req.Reports) > 100 { return nil, response.ErrBadRequest(40087, "too_many_account_health_reports", "at most 100 account health reports are allowed") } accepted := 0 buffered := 0 now := s.now().UTC() type preparedHealthReport struct { accountID uuid.UUID latestKey string signature string payload []byte report bufferedDesktopAccountHealthReport } prepared := make([]preparedHealthReport, 0, len(req.Reports)) for _, item := range req.Reports { accountID, err := uuid.Parse(strings.TrimSpace(item.AccountID)) if err != nil { return nil, response.ErrBadRequest(40087, "invalid_account_health_report_account_id", "account_id must be a uuid") } platform := strings.TrimSpace(item.Platform) platformUID := strings.TrimSpace(item.PlatformUID) if platform == "" || platformUID == "" { return nil, response.ErrBadRequest(40087, "invalid_account_health_report_identity", "platform and platform_uid are required") } report := bufferedDesktopAccountHealthReport{ TenantID: client.TenantID, WorkspaceID: client.WorkspaceID, UserID: client.UserID, ClientID: client.ID.String(), AccountID: accountID.String(), Platform: platform, PlatformUID: platformUID, Health: item.Health, AuthState: strings.TrimSpace(item.AuthState), ProbeState: strings.TrimSpace(item.ProbeState), AuthReason: trimOptionalString(item.AuthReason), DisplayName: trimOptionalString(item.DisplayName), AvatarURL: trimOptionalString(item.AvatarURL), VerifiedAt: item.VerifiedAt, CheckedAt: item.CheckedAt.UTC(), ReceivedAt: now, } if report.AuthState == "" || report.ProbeState == "" || report.CheckedAt.IsZero() { return nil, response.ErrBadRequest(40087, "invalid_account_health_report_state", "auth_state, probe_state and checked_at are required") } payload, err := json.Marshal(report) if err != nil { return nil, response.ErrInternal(50087, "account_health_report_encode_failed", "failed to encode account health report") } latestKey := desktopAccountHealthLatestKey(client.WorkspaceID, accountID) prepared = append(prepared, preparedHealthReport{ accountID: accountID, latestKey: latestKey, signature: desktopAccountHealthSignature(report), payload: payload, report: report, }) accepted += 1 } changed := make([]preparedHealthReport, 0, len(prepared)) if s.redis != nil { signaturePipe := s.redis.Pipeline() signatureCommands := make(map[uuid.UUID]*goredis.StringCmd, len(prepared)) for _, item := range prepared { signatureCommands[item.accountID] = signaturePipe.Get(ctx, item.latestKey+":signature") } if _, err := signaturePipe.Exec(ctx); err != nil && !errors.Is(err, goredis.Nil) { return nil, response.ErrInternal(50087, "account_health_report_cache_failed", "failed to inspect account health report cache") } pipe := s.redis.Pipeline() for _, item := range prepared { previousSignature, signatureErr := signatureCommands[item.accountID].Result() if signatureErr != nil && !errors.Is(signatureErr, goredis.Nil) { return nil, response.ErrInternal(50087, "account_health_report_cache_failed", "failed to inspect account health report cache") } pipe.Set(ctx, item.latestKey, item.payload, desktopAccountHealthReportTTL) if previousSignature != item.signature { changed = append(changed, item) } } if _, err := pipe.Exec(ctx); err != nil { return nil, response.ErrInternal(50087, "account_health_report_buffer_failed", "failed to buffer account health reports") } } else { changed = append(changed, prepared...) } if s.rabbitMQ == nil { return &ReportDesktopAccountHealthResponse{AcceptedCount: accepted, BufferedCount: 0}, nil } for _, item := range changed { if err := s.rabbitMQ.PublishDesktopAccountHealth(ctx, item.payload); err != nil { return nil, response.ErrInternal(50087, "account_health_report_queue_failed", "failed to enqueue account health report") } buffered += 1 if s.redis != nil { if err := s.redis.Set(ctx, item.latestKey+":signature", item.signature, desktopAccountHealthReportTTL).Err(); err != nil { return nil, response.ErrInternal(50087, "account_health_report_signature_failed", "failed to record account health report signature") } } } return &ReportDesktopAccountHealthResponse{AcceptedCount: accepted, BufferedCount: buffered}, nil } func (s *DesktopAccountService) Patch(ctx context.Context, client *repository.DesktopClient, desktopID uuid.UUID, req PatchDesktopAccountRequest) (*DesktopAccountView, error) { if client == nil { return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required") } account, err := s.repo.Patch(ctx, repository.PatchDesktopAccountParams{ DesktopID: desktopID, WorkspaceID: client.WorkspaceID, DisplayName: req.DisplayName, Health: req.Health, VerifiedAt: req.VerifiedAt, Tags: req.Tags, IfSyncVersion: req.IfSyncVersion, ClientID: client.ID, }) if err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, response.ErrConflict(40982, "desktop_account_sync_conflict", "desktop account sync version conflict") } return nil, response.ErrInternal(50084, "desktop_account_patch_failed", "failed to patch desktop account") } view := s.buildDesktopAccountView(account, nil, nil, nil, nil) return &view, nil } func (s *DesktopAccountService) Tombstone(ctx context.Context, client *repository.DesktopClient, desktopID uuid.UUID, ifSyncVersion int64) (*DesktopAccountView, error) { if client == nil { return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required") } account, err := s.repo.Tombstone(ctx, desktopID, client.WorkspaceID, client.ID, ifSyncVersion) if err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, response.ErrConflict(40982, "desktop_account_sync_conflict", "desktop account sync version conflict") } return nil, response.ErrInternal(50085, "desktop_account_delete_failed", "failed to delete desktop account") } view := s.buildDesktopAccountView(account, nil, nil, nil, nil) invalidateScheduleTaskCaches(ctx, s.cache, client.TenantID, nil) return &view, nil } func (s *DesktopAccountService) RequestDelete(ctx context.Context, actor auth.Actor, desktopID uuid.UUID, undo bool) (*DesktopAccountView, error) { if actor.PrimaryWorkspaceID == 0 || actor.UserID == 0 { return nil, response.ErrUnauthorized(40132, "workspace_scope_missing", "workspace context is required") } var ( account *repository.DesktopAccount err error ) if undo { account, err = s.repo.ClearDeleteRequested(ctx, desktopID, actor.PrimaryWorkspaceID, actor.UserID) } else { account, err = s.repo.MarkDeleteRequested(ctx, desktopID, actor.PrimaryWorkspaceID, actor.UserID) } if err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, response.ErrNotFound(40481, "desktop_account_not_found", "desktop account not found") } return nil, response.ErrInternal(50086, "desktop_account_request_delete_failed", "failed to update desktop account delete request") } view := s.buildDesktopAccountView(account, nil, nil, nil, nil) return &view, nil } func (s *DesktopAccountService) DeleteForActor(ctx context.Context, actor auth.Actor, desktopID uuid.UUID) (*DesktopAccountDeleteResult, error) { if actor.PrimaryWorkspaceID == 0 || actor.UserID == 0 { return nil, response.ErrUnauthorized(40132, "workspace_scope_missing", "workspace context is required") } if s.pool == nil { return nil, response.ErrServiceUnavailable(50342, "desktop_publish_job_store_unavailable", "desktop publish job store is unavailable") } tx, err := s.pool.Begin(ctx) if err != nil { return nil, response.ErrInternal(50086, "desktop_account_delete_begin_failed", "failed to start desktop account delete") } defer tx.Rollback(ctx) accountRepo := repository.NewDesktopAccountRepository(tx) account, err := accountRepo.TombstoneForActor(ctx, desktopID, actor.PrimaryWorkspaceID, actor.UserID) if err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, response.ErrNotFound(40481, "desktop_account_not_found", "desktop account not found") } return nil, response.ErrInternal(50085, "desktop_account_delete_failed", "failed to delete desktop account") } cancelled, err := cancelQueuedPublishTasksForDesktopAccountUnbind(ctx, tx, account.TenantID, account.WorkspaceID, account.DesktopID) if err != nil { return nil, err } if err := tx.Commit(ctx); err != nil { return nil, response.ErrInternal(50086, "desktop_account_delete_commit_failed", "failed to commit desktop account delete") } view := s.buildDesktopAccountView(account, nil, nil, nil, nil) invalidateScheduleTaskCaches(ctx, s.cache, actor.TenantID, nil) return &DesktopAccountDeleteResult{ Account: view, CancelledQueuedPublishTasks: cancelled, CompletedServerSideUnbinding: true, }, nil } func desktopAccountHealthLatestKey(workspaceID int64, accountID uuid.UUID) string { return fmt.Sprintf("desktop:account-health:latest:%d:%s", workspaceID, accountID.String()) } func desktopAccountHealthSignature(report bufferedDesktopAccountHealthReport) string { payload, _ := json.Marshal(struct { AccountID string `json:"account_id"` Platform string `json:"platform"` PlatformUID string `json:"platform_uid"` Health string `json:"health"` AuthState string `json:"auth_state"` ProbeState string `json:"probe_state"` AuthReason *string `json:"auth_reason"` DisplayName *string `json:"display_name"` AvatarURL *string `json:"avatar_url"` }{ AccountID: report.AccountID, Platform: report.Platform, PlatformUID: report.PlatformUID, Health: report.Health, AuthState: report.AuthState, ProbeState: report.ProbeState, AuthReason: report.AuthReason, DisplayName: report.DisplayName, AvatarURL: report.AvatarURL, }) sum := sha256.Sum256(payload) return hex.EncodeToString(sum[:]) } func trimOptionalString(value *string) *string { if value == nil { return nil } trimmed := strings.TrimSpace(*value) if trimmed == "" { return nil } return &trimmed } func loadDesktopAccountRuntimeHealth( ctx context.Context, redis *goredis.Client, workspaceID int64, accountIDs []uuid.UUID, ) map[uuid.UUID]bufferedDesktopAccountHealthReport { if redis == nil || len(accountIDs) == 0 { return nil } uniqueIDs := uniqueUUIDs(accountIDs) pipe := redis.Pipeline() commands := make(map[uuid.UUID]*goredis.StringCmd, len(uniqueIDs)) for _, accountID := range uniqueIDs { commands[accountID] = pipe.Get(ctx, desktopAccountHealthLatestKey(workspaceID, accountID)) } if _, err := pipe.Exec(ctx); err != nil && !errors.Is(err, goredis.Nil) { return nil } result := make(map[uuid.UUID]bufferedDesktopAccountHealthReport, len(commands)) for accountID, cmd := range commands { payload, err := cmd.Bytes() if err != nil { continue } var report bufferedDesktopAccountHealthReport if err := json.Unmarshal(payload, &report); err != nil { continue } if report.WorkspaceID != workspaceID || report.AccountID != accountID.String() || report.CheckedAt.IsZero() { continue } result[accountID] = report } return result } func (s *DesktopAccountService) applyRuntimeHealth(ctx context.Context, workspaceID int64, items []DesktopAccountView) { if s.redis == nil || len(items) == 0 { return } accountIDs := make([]uuid.UUID, 0, len(items)) accountIndex := make(map[uuid.UUID]int, len(items)) for index, item := range items { accountID, err := uuid.Parse(item.ID) if err != nil { continue } accountIDs = append(accountIDs, accountID) accountIndex[accountID] = index } reports := loadDesktopAccountRuntimeHealth(ctx, s.redis, workspaceID, accountIDs) for accountID, report := range reports { index, ok := accountIndex[accountID] if !ok { continue } if !shouldApplyDesktopAccountRuntimeHealth(items[index], report) { continue } health := effectiveDesktopAccountRuntimeHealth(report) authState := report.AuthState probeState := report.ProbeState checkedAt := report.CheckedAt items[index].Health = health items[index].RuntimeHealth = &health items[index].RuntimeVerifiedAt = report.VerifiedAt items[index].RuntimeCheckedAt = &checkedAt items[index].RuntimeAuthState = &authState items[index].RuntimeProbeState = &probeState items[index].RuntimeAuthReason = report.AuthReason items[index].HealthSource = "runtime" if report.VerifiedAt != nil { items[index].VerifiedAt = report.VerifiedAt } } } type desktopAccountPublishQueueSummary struct { Count int OldestAt *time.Time } func (s *DesktopAccountService) applyPublishQueueSummaries(ctx context.Context, workspaceID int64, items []DesktopAccountView) { if s == nil || s.pool == nil || len(items) == 0 { return } accountIDs := make([]uuid.UUID, 0, len(items)) accountIndex := make(map[uuid.UUID]int, len(items)) for index, item := range items { accountID, err := uuid.Parse(item.ID) if err != nil { continue } accountIDs = append(accountIDs, accountID) accountIndex[accountID] = index } if len(accountIDs) == 0 { return } summaries, err := s.loadQueuedPublishTaskSummaries(ctx, workspaceID, accountIDs) if err != nil { return } for accountID, summary := range summaries { index, ok := accountIndex[accountID] if !ok { continue } items[index].QueuedPublishTaskCount = summary.Count items[index].OldestQueuedPublishTaskAt = summary.OldestAt } } func (s *DesktopAccountService) loadQueuedPublishTaskSummaries( ctx context.Context, workspaceID int64, accountIDs []uuid.UUID, ) (map[uuid.UUID]desktopAccountPublishQueueSummary, error) { if s == nil || s.pool == nil || len(accountIDs) == 0 { return nil, nil } rows, err := s.pool.Query(ctx, ` SELECT dt.target_account_id, COUNT(*)::int, MIN(COALESCE(dt.enqueued_at, dt.created_at)) FROM desktop_tasks AS dt WHERE dt.workspace_id = $1 AND dt.kind = 'publish' AND dt.status = 'queued' AND dt.target_account_id = ANY($2::uuid[]) AND dt.attempts < $3 AND NOT EXISTS ( SELECT 1 FROM desktop_publish_jobs AS j WHERE j.desktop_id = dt.job_id AND j.status <> 'queued' ) GROUP BY dt.target_account_id `, workspaceID, uniqueUUIDs(accountIDs), desktopPublishMaxAttempts) if err != nil { return nil, err } defer rows.Close() summaries := make(map[uuid.UUID]desktopAccountPublishQueueSummary) for rows.Next() { var ( accountID uuid.UUID count int oldestAt time.Time ) if err := rows.Scan(&accountID, &count, &oldestAt); err != nil { return nil, err } oldestAt = oldestAt.UTC() summaries[accountID] = desktopAccountPublishQueueSummary{ Count: count, OldestAt: &oldestAt, } } if err := rows.Err(); err != nil { return nil, err } return summaries, nil } func shouldApplyDesktopAccountRuntimeHealth(item DesktopAccountView, report bufferedDesktopAccountHealthReport) bool { if item.VerifiedAt == nil { return true } return !report.CheckedAt.Before(*item.VerifiedAt) } func effectiveDesktopAccountRuntimeHealth(report bufferedDesktopAccountHealthReport) string { authState := strings.TrimSpace(report.AuthState) probeState := strings.TrimSpace(report.ProbeState) switch authState { case "active": return "live" case "expiring_soon": if probeState != "network_error" { return "live" } } return report.Health } func (s *DesktopAccountService) buildDesktopAccountView( account *repository.DesktopAccount, clientMap map[uuid.UUID]*repository.DesktopClient, accountPresenceMap map[uuid.UUID]uuid.UUID, clientPresenceMap map[uuid.UUID]bool, taskConsumerPresenceMap map[uuid.UUID]bool, ) DesktopAccountView { var clientID *string var clientOnline *bool var accountSessionOnline *bool var taskConsumerOnline *bool var clientLastSeenAt *time.Time var clientDeviceName *string if account.ClientID != nil { value := account.ClientID.String() clientID = &value clientOnlineValue := false clientOnline = &clientOnlineValue accountSessionOnlineValue := false accountSessionOnline = &accountSessionOnlineValue taskConsumerOnlineValue := false taskConsumerOnline = &taskConsumerOnlineValue if clientMap != nil { if client, ok := clientMap[*account.ClientID]; ok && client != nil { clientLastSeenAt = client.LastSeenAt clientDeviceName = client.DeviceName clientOnlineValue = resolveDesktopClientOnline( client, clientPresenceMap[*account.ClientID], clientPresenceMap != nil, s.nowUTC(), ) } } if accountPresenceMap != nil { if activeClientID, ok := accountPresenceMap[account.DesktopID]; ok && activeClientID == *account.ClientID { accountSessionOnlineValue = true } } if taskConsumerPresenceMap != nil && taskConsumerPresenceMap[*account.ClientID] { taskConsumerOnlineValue = true } } return DesktopAccountView{ ID: account.DesktopID.String(), Platform: account.Platform, PlatformUID: normalizePlatformUID(account.PlatformUID), DisplayName: account.DisplayName, AvatarURL: account.AvatarURL, Health: account.Health, HealthSource: "database", ClientID: clientID, AccountFingerprint: account.AccountFingerprint, VerifiedAt: account.VerifiedAt, Tags: account.Tags, SyncVersion: account.SyncVersion, DeletedAt: account.DeletedAt, DeleteRequestedAt: account.DeleteRequestedAt, ClientOnline: clientOnline, AccountSessionOnline: accountSessionOnline, TaskConsumerOnline: taskConsumerOnline, ClientLastSeenAt: clientLastSeenAt, ClientDeviceName: clientDeviceName, } } func (s *DesktopAccountService) loadClientState( ctx context.Context, workspaceID int64, accounts []repository.DesktopAccount, ) (map[uuid.UUID]*repository.DesktopClient, map[uuid.UUID]uuid.UUID, map[uuid.UUID]bool, map[uuid.UUID]bool) { clientIDs := make([]uuid.UUID, 0, len(accounts)*2) accountIDs := make([]uuid.UUID, 0, len(accounts)) seen := make(map[uuid.UUID]struct{}, len(accounts)*2) for _, account := range accounts { accountIDs = append(accountIDs, account.DesktopID) if account.ClientID == nil { continue } if _, ok := seen[*account.ClientID]; ok { continue } seen[*account.ClientID] = struct{}{} clientIDs = append(clientIDs, *account.ClientID) } accountPresenceMap := loadDesktopAccountPresence(ctx, s.redis, accountIDs) for _, clientID := range accountPresenceMap { if _, ok := seen[clientID]; ok { continue } seen[clientID] = struct{}{} clientIDs = append(clientIDs, clientID) } clientMap := make(map[uuid.UUID]*repository.DesktopClient, len(clientIDs)) if s.clientRepo != nil { for _, clientID := range clientIDs { client, err := s.clientRepo.GetByID(ctx, clientID, workspaceID) if err != nil { if errors.Is(err, pgx.ErrNoRows) { continue } continue } clientMap[clientID] = client } } clientPresenceMap := loadDesktopClientPresence(ctx, s.redis, clientIDs) taskConsumerPresenceMap := loadDesktopTaskConsumerPresence(ctx, s.redis, clientIDs) return clientMap, accountPresenceMap, clientPresenceMap, taskConsumerPresenceMap } func (s *DesktopAccountService) nowUTC() time.Time { if s != nil && s.now != nil { return s.now().UTC() } return time.Now().UTC() } func resolveDesktopClientOnline( client *repository.DesktopClient, presence bool, presenceKnown bool, now time.Time, ) bool { if client == nil || client.RevokedAt != nil { return false } if presenceKnown { return presence } if client.LastSeenAt == nil { return false } return now.UTC().Sub(client.LastSeenAt.UTC()) <= desktopClientPresenceTTL } func normalizePlatformUID(value string) string { normalized := strings.TrimSpace(value) for strings.HasPrefix(strings.ToLower(normalized), "platform:") { normalized = strings.TrimSpace(normalized[len("platform:"):]) } return normalized }