From 67be43319e366d9f0773f92e0f1ca44173bd2453 Mon Sep 17 00:00:00 2001 From: liangxu Date: Sun, 7 Jun 2026 19:15:20 +0800 Subject: [PATCH] feat(desktop&tenant): Enhance desktop account and task consumer presence management - Added new fields to DesktopAccountInfo for tracking account session and task consumer online status, along with queued publish task count and oldest queued publish task timestamp. - Updated DesktopDispatchEvent to include server time for better synchronization. - Implemented separate presence management for task consumers, including marking presence and loading presence state. - Enhanced DesktopAccountService to apply publish queue summaries and manage task consumer presence. - Introduced new methods for replaying queued publish tasks in DesktopDispatchHandler. - Updated tests to cover new presence management logic and ensure correct behavior of account session and task consumer online status. --- .../src/lib/publish-account-cards.ts | 34 ++- apps/admin-web/src/views/MediaView.vue | 31 ++- .../src/main/runtime-controller.ts | 17 +- packages/shared-types/src/index.ts | 4 + .../shared/stream/desktop_dispatch.go | 43 +-- .../internal/shared/swagger/descriptions.go | 10 + .../tenant/app/desktop_account_service.go | 251 +++++++++++++----- .../app/desktop_account_service_test.go | 73 ++++- .../internal/tenant/app/desktop_presence.go | 57 +++- .../tenant/app/desktop_presence_test.go | 27 ++ .../tenant/app/desktop_task_events.go | 28 ++ .../tenant/app/desktop_task_service.go | 1 + .../tenant/app/publish_job_service.go | 27 +- .../transport/desktop_account_handler.go | 1 + .../transport/desktop_dispatch_handler.go | 161 ++++++++++- 15 files changed, 636 insertions(+), 129 deletions(-) diff --git a/apps/admin-web/src/lib/publish-account-cards.ts b/apps/admin-web/src/lib/publish-account-cards.ts index 2fde296..1929506 100644 --- a/apps/admin-web/src/lib/publish-account-cards.ts +++ b/apps/admin-web/src/lib/publish-account-cards.ts @@ -25,8 +25,12 @@ export interface PublishAccountCard { verifiedAt: string | null clientId: string | null clientOnline: boolean | null + accountSessionOnline: boolean | null + taskConsumerOnline: boolean | null clientDeviceName: string | null clientLastSeenAt: string | null + queuedPublishTaskCount: number + oldestQueuedPublishTaskAt: string | null publishState: PublishState selectable: boolean statusText: string @@ -113,8 +117,12 @@ export function buildPublishAccountCard( verifiedAt: resolveAccountCheckedAt(account), clientId: account.client_id, clientOnline: account.client_online, + accountSessionOnline: account.account_session_online ?? null, + taskConsumerOnline: account.task_consumer_online ?? null, clientDeviceName: account.client_device_name, clientLastSeenAt: account.client_last_seen_at, + queuedPublishTaskCount: account.queued_publish_task_count ?? 0, + oldestQueuedPublishTaskAt: account.oldest_queued_publish_task_at ?? null, publishState, selectable: publishState !== 'unavailable', statusText: resolvePublishStatusText(account, publishState), @@ -128,7 +136,7 @@ export function resolvePublishState(account: DesktopAccountInfo): PublishState { if (!account.client_id) { return 'unavailable' } - return account.client_online === true ? 'immediate' : 'queued' + return account.task_consumer_online === true ? 'immediate' : 'queued' } export function publishRank(state: PublishState): number { @@ -195,9 +203,10 @@ export function clientStatusLabel(card: PublishAccountCard): string { return '未绑定客户端' } - const state = card.clientOnline ? '在线' : '离线' + const state = card.clientOnline ? '心跳在线' : '心跳离线' + const consumerState = card.taskConsumerOnline ? '任务通道在线' : '任务通道离线' const deviceName = card.clientDeviceName?.trim() - return deviceName ? `${state} · ${deviceName}` : `客户端${state}` + return deviceName ? `${state} · ${consumerState} · ${deviceName}` : `${state} · ${consumerState}` } export function clientStatusTooltip(card: PublishAccountCard): string { @@ -205,16 +214,20 @@ export function clientStatusTooltip(card: PublishAccountCard): string { return '未绑定桌面客户端' } - const state = card.clientOnline ? '在线' : '离线' + const state = card.clientOnline ? '心跳在线' : '心跳离线' + const consumerState = card.taskConsumerOnline ? '任务通道在线' : '任务通道离线' const deviceName = card.clientDeviceName?.trim() - return deviceName ? `${state} · ${deviceName}` : `客户端${state}` + return deviceName ? `${state} · ${consumerState} · ${deviceName}` : `${state} · ${consumerState}` } export function clientStatusColor(card: PublishAccountCard): string { if (!card.clientId) { return 'default' } - return card.clientOnline ? 'green' : 'gold' + if (card.taskConsumerOnline) { + return 'green' + } + return card.clientOnline ? 'orange' : 'gold' } export function accountInitial( @@ -238,10 +251,15 @@ export function resolvePlatformLogoUrl( function resolvePublishStatusText(account: DesktopAccountInfo, publishState: PublishState): string { if (publishState === 'immediate') { - return '客户端在线，RabbitMQ 会立即唤醒桌面端开始发布。' + return '任务消费通道在线，发布任务会立即被桌面端领取执行。' } if (publishState === 'queued') { - return '客户端当前离线，任务会先落库排队，等桌面端上线后自动继续发布。' + const count = account.queued_publish_task_count ?? 0 + const queuePrefix = count > 0 ? `当前已有 ${count} 个发布任务排队。` : '' + if (account.client_online === true) { + return `${queuePrefix}客户端心跳在线，但任务消费通道未就绪；任务会落库排队，通道恢复后自动继续发布。` + } + return `${queuePrefix}客户端当前离线，任务会先落库排队，等桌面端上线后自动继续发布。` } if (resolveAccountHealth(account) !== 'live') { return '该账号授权状态异常，请先在桌面端重新校验或重新授权。' diff --git a/apps/admin-web/src/views/MediaView.vue b/apps/admin-web/src/views/MediaView.vue index fa06aa5..f99f5e8 100644 --- a/apps/admin-web/src/views/MediaView.vue +++ b/apps/admin-web/src/views/MediaView.vue @@ -60,8 +60,12 @@ interface MediaAccountCard { verifiedAt: string | null clientId: string | null clientOnline: boolean | null + accountSessionOnline: boolean | null + taskConsumerOnline: boolean | null clientDeviceName: string | null clientLastSeenAt: string | null + queuedPublishTaskCount: number + oldestQueuedPublishTaskAt: string | null publishState: PublishState publishHint: string deleteRequestedAt: string | null @@ -251,8 +255,12 @@ const mediaAccounts = computed(() => { verifiedAt: resolveAccountCheckedAt(account), clientId: account.client_id, clientOnline: account.client_online, + accountSessionOnline: account.account_session_online ?? null, + taskConsumerOnline: account.task_consumer_online ?? null, clientDeviceName: account.client_device_name, clientLastSeenAt: account.client_last_seen_at, + queuedPublishTaskCount: account.queued_publish_task_count ?? 0, + oldestQueuedPublishTaskAt: account.oldest_queued_publish_task_at ?? null, publishState, publishHint: resolvePublishHint(account, publishState), deleteRequestedAt: account.delete_requested_at, @@ -322,7 +330,7 @@ function resolvePublishState(account: DesktopAccountInfo): PublishState { if (!account.client_id) { return 'unavailable' } - return account.client_online === true ? 'immediate' : 'queued' + return account.task_consumer_online === true ? 'immediate' : 'queued' } function resolvePublishHint(account: DesktopAccountInfo, publishState: PublishState): string { @@ -330,7 +338,12 @@ function resolvePublishHint(account: DesktopAccountInfo, publishState: PublishSt return '' } if (publishState === 'queued') { - return '客户端当前离线，任务会先进入发布队列，等桌面端上线后自动继续。' + const count = account.queued_publish_task_count ?? 0 + const prefix = count > 0 ? `当前已有 ${count} 个发布任务排队。` : '' + if (account.client_online === true) { + return `${prefix}客户端心跳在线，但任务消费通道未就绪；任务会进入发布队列，通道恢复后自动继续。` + } + return `${prefix}客户端当前离线，任务会先进入发布队列，等桌面端上线后自动继续。` } if (resolveAccountHealth(account) !== 'live') { return '授权状态异常，先在桌面端重新校验后再发布。' @@ -402,14 +415,20 @@ function clientStatusLabel(account: MediaAccountCard): string { if (!account.clientId) { return '未绑定' } - return account.clientOnline ? '在线' : '离线' + if (account.taskConsumerOnline) { + return '任务通道在线' + } + return account.clientOnline ? '心跳在线，任务通道离线' : '离线' } function clientStatusColor(account: MediaAccountCard): string { if (!account.clientId) { return 'default' } - return account.clientOnline ? 'green' : 'gold' + if (account.taskConsumerOnline) { + return 'green' + } + return account.clientOnline ? 'orange' : 'gold' } function clientStatusText(account: MediaAccountCard): string { @@ -426,7 +445,7 @@ function publishStateLabel(state: PublishState): string { case 'immediate': return '可立即发布' case 'queued': - return '离线排队' + return '排队等待' default: return '不可发布' } @@ -715,7 +734,7 @@ function releaseSize(value?: number | null): string {
- 离线排队 + 排队等待 {{ overview.queued }}
diff --git a/apps/desktop-client/src/main/runtime-controller.ts b/apps/desktop-client/src/main/runtime-controller.ts index f1a8682..7a8ae26 100644 --- a/apps/desktop-client/src/main/runtime-controller.ts +++ b/apps/desktop-client/src/main/runtime-controller.ts @@ -149,7 +149,8 @@ type MonitorExecutionPhase = const heartbeatIntervalMs = 15_000 const pullIntervalMs = 60_000 -const pumpWatchdogIntervalMs = 15_000 +const publishPullIntervalMs = 30_000 +const pumpWatchdogIntervalMs = 30_000 const leaseExtendIntervalMs = 60_000 const defaultPublishTaskTimeoutMs = 5 * 60_000 const publishTaskStaleProgressMs = 2 * 60_000 @@ -816,11 +817,11 @@ function schedulePullLoop(): void { clearInterval(state.pullTimer) } - setSchedulerNextPull(Date.now() + pullIntervalMs) + setSchedulerNextPull(Date.now() + publishPullIntervalMs) state.pullTimer = setInterval(() => { - setSchedulerNextPull(Date.now() + pullIntervalMs) + setSchedulerNextPull(Date.now() + publishPullIntervalMs) pumpExecutionLoop() - }, pullIntervalMs) + }, publishPullIntervalMs) } function schedulePumpWatchdogLoop(): void { @@ -859,6 +860,7 @@ function connectDispatchWs(): void { setTransportDispatchState('streaming') setSchedulerError(null) recordActivity('success', '任务分发通道已连接', '统一 WebSocket 分发通道已就绪。') + pumpExecutionLoop() void pullNextTask('monitor') if (hasConnectedOnce) { void resumeLeasedMonitoringTasks('reconnect') @@ -872,6 +874,7 @@ function connectDispatchWs(): void { if (isConnectedEvent(payload)) { state.lastServerTime = payload.server_time } + pumpExecutionLoop() }) const dispatchTaskHandler = (payload: unknown) => { @@ -1162,6 +1165,7 @@ async function sendHeartbeat(source: 'startup' | 'timer'): Promise { if (source === 'startup' || previousStatus === 'failed') { recordActivity('success', '心跳连接正常', '客户端心跳已回写到服务端。') } + pumpExecutionLoop() } catch (error) { state.lastHeartbeatAt = Date.now() state.lastHeartbeatStatus = 'failed' @@ -1911,7 +1915,7 @@ async function pullNextTask(kind: 'publish' | 'monitor'): Promise { noteSchedulerPull() if (!leased.task) { - state.publishFallbackBackoffUntil = Date.now() + pullIntervalMs + state.publishFallbackBackoffUntil = Date.now() + publishPullIntervalMs return } @@ -1923,7 +1927,7 @@ async function pullNextTask(kind: 'publish' | 'monitor'): Promise { state.lastPullAt = Date.now() state.lastPullStatus = 'failed' state.lastError = errorMessage(error) - state.publishFallbackBackoffUntil = Date.now() + pullIntervalMs + state.publishFallbackBackoffUntil = Date.now() + publishPullIntervalMs noteTransportPull(false) setSchedulerError(state.lastError) @@ -3505,7 +3509,6 @@ function canStartAnotherPublishExecution(): boolean { function shouldPullPublishFallback(): boolean { return ( - !state.dispatchWsConnected && Date.now() >= state.publishFallbackBackoffUntil && !isLeaseInFlight('publish') && !hasPublishBacklog() && diff --git a/packages/shared-types/src/index.ts b/packages/shared-types/src/index.ts index f4ecc74..77626f5 100644 --- a/packages/shared-types/src/index.ts +++ b/packages/shared-types/src/index.ts @@ -248,8 +248,12 @@ export interface DesktopAccountInfo { deleted_at: string | null delete_requested_at: string | null client_online: boolean | null + account_session_online?: boolean | null + task_consumer_online?: boolean | null client_last_seen_at: string | null client_device_name: string | null + queued_publish_task_count?: number + oldest_queued_publish_task_at?: string | null } export interface UpsertDesktopAccountRequest { diff --git a/server/internal/shared/stream/desktop_dispatch.go b/server/internal/shared/stream/desktop_dispatch.go index 0744871..268fde8 100644 --- a/server/internal/shared/stream/desktop_dispatch.go +++ b/server/internal/shared/stream/desktop_dispatch.go @@ -20,27 +20,28 @@ import ( // the existing SSE handler on the desktop client can treat both channels the // same way and dedupe by task_id. type DesktopDispatchEvent struct { - Type string `json:"type"` - TaskID string `json:"task_id"` - JobID string `json:"job_id,omitempty"` - WorkspaceID int64 `json:"workspace_id"` - TargetAccountID string `json:"target_account_id,omitempty"` - TargetClientID string `json:"target_client_id"` - Platform string `json:"platform,omitempty"` - Title *string `json:"title,omitempty"` - BusinessDate *string `json:"business_date,omitempty"` - SchedulerGroupKey *string `json:"scheduler_group_key,omitempty"` - QuestionText *string `json:"question_text,omitempty"` - Status string `json:"status"` - Kind string `json:"kind"` - Priority int `json:"priority,omitempty"` - Lane string `json:"lane,omitempty"` - InterruptGeneration int `json:"interrupt_generation,omitempty"` - SignalOnly bool `json:"signal_only,omitempty"` - Control string `json:"control,omitempty"` - Reason string `json:"reason,omitempty"` - ReplacementTaskID string `json:"replacement_task_id,omitempty"` - UpdatedAt time.Time `json:"updated_at"` + Type string `json:"type"` + TaskID string `json:"task_id"` + JobID string `json:"job_id,omitempty"` + WorkspaceID int64 `json:"workspace_id"` + TargetAccountID string `json:"target_account_id,omitempty"` + TargetClientID string `json:"target_client_id"` + Platform string `json:"platform,omitempty"` + Title *string `json:"title,omitempty"` + BusinessDate *string `json:"business_date,omitempty"` + SchedulerGroupKey *string `json:"scheduler_group_key,omitempty"` + QuestionText *string `json:"question_text,omitempty"` + Status string `json:"status"` + Kind string `json:"kind"` + Priority int `json:"priority,omitempty"` + Lane string `json:"lane,omitempty"` + InterruptGeneration int `json:"interrupt_generation,omitempty"` + SignalOnly bool `json:"signal_only,omitempty"` + Control string `json:"control,omitempty"` + Reason string `json:"reason,omitempty"` + ReplacementTaskID string `json:"replacement_task_id,omitempty"` + UpdatedAt time.Time `json:"updated_at"` + ServerTime *time.Time `json:"server_time,omitempty"` } // DispatchSubscriber represents a live WebSocket connection bound to a single diff --git a/server/internal/shared/swagger/descriptions.go b/server/internal/shared/swagger/descriptions.go index 6368775..0d33244 100644 --- a/server/internal/shared/swagger/descriptions.go +++ b/server/internal/shared/swagger/descriptions.go @@ -202,6 +202,16 @@ var routeDocs = map[string]routeDoc{ // --- Tenant：媒体 --- "GET /api/tenant/media/platforms": {"媒体平台列表", "返回当前租户可用的媒体平台元数据（图标、能力开关等）。"}, + // --- Tenant：企业站点发布 --- + "GET /api/tenant/enterprise-sites": {"企业站点列表", "查询当前 Workspace 已配置的企业官网/CMS 发布连接。"}, + "POST /api/tenant/enterprise-sites": {"新增企业站点", "保存企业站点 CMS 连接配置和凭证，用于服务端直投文章。"}, + "PATCH /api/tenant/enterprise-sites/:id": {"更新企业站点", "更新指定企业站点连接配置、状态或凭证。"}, + "DELETE /api/tenant/enterprise-sites/:id": {"删除企业站点", "删除指定企业站点连接配置。"}, + "POST /api/tenant/enterprise-sites/:id/ping": {"检测企业站点连接", "调用目标 CMS 探测接口，验证站点凭证和发布能力。"}, + "GET /api/tenant/enterprise-sites/:id/categories": {"企业站点栏目列表", "读取已同步的企业站点栏目。"}, + "POST /api/tenant/enterprise-sites/:id/categories/sync": {"同步企业站点栏目", "从目标 CMS 拉取栏目树并写入本地缓存。"}, + "POST /api/tenant/enterprise-sites/:id/publish": {"发布文章到企业站点", "将指定文章内容直投到企业站点 CMS，并回写发布记录。"}, + // --- Tenant：桌面客户端 --- "GET /api/tenant/desktop-client/releases/downloadable": {"可下载桌面客户端版本", "返回当前租户可下载的已启用桌面客户端版本列表。"}, diff --git a/server/internal/tenant/app/desktop_account_service.go b/server/internal/tenant/app/desktop_account_service.go index af5f037..c15a90d 100644 --- a/server/internal/tenant/app/desktop_account_service.go +++ b/server/internal/tenant/app/desktop_account_service.go @@ -12,6 +12,7 @@ import ( "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" @@ -23,6 +24,7 @@ import ( const desktopAccountHealthReportTTL = 24 * time.Hour type DesktopAccountService struct { + pool *pgxpool.Pool repo repository.DesktopAccountRepository clientRepo repository.DesktopClientRepository redis *goredis.Client @@ -31,12 +33,14 @@ type DesktopAccountService struct { } 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, @@ -46,29 +50,33 @@ func NewDesktopAccountService( } 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"` - ClientLastSeenAt *time.Time `json:"client_last_seen_at"` - ClientDeviceName *string `json:"client_device_name"` + 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 UpsertDesktopAccountRequest struct { @@ -143,12 +151,13 @@ func (s *DesktopAccountService) ListByClient(ctx context.Context, client *reposi return nil, response.ErrInternal(50081, "desktop_accounts_query_failed", "failed to list desktop accounts") } - clientMap, presenceMap := s.loadClientState(ctx, client.WorkspaceID, rows) + 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, presenceMap)) + 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 } @@ -162,12 +171,13 @@ func (s *DesktopAccountService) ListForActor(ctx context.Context, actor auth.Act return nil, response.ErrInternal(50081, "desktop_accounts_query_failed", "failed to list desktop accounts") } - clientMap, presenceMap := s.loadClientState(ctx, actor.PrimaryWorkspaceID, rows) + 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, presenceMap)) + 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 } @@ -211,7 +221,7 @@ func (s *DesktopAccountService) Upsert(ctx context.Context, client *repository.D return nil, appErr } - view := s.buildDesktopAccountView(account, nil, nil) + view := s.buildDesktopAccountView(account, nil, nil, nil, nil) return &view, nil } @@ -355,7 +365,7 @@ func (s *DesktopAccountService) Patch(ctx context.Context, client *repository.De return nil, response.ErrInternal(50084, "desktop_account_patch_failed", "failed to patch desktop account") } - view := s.buildDesktopAccountView(account, nil, nil) + view := s.buildDesktopAccountView(account, nil, nil, nil, nil) return &view, nil } @@ -372,7 +382,7 @@ func (s *DesktopAccountService) Tombstone(ctx context.Context, client *repositor return nil, response.ErrInternal(50085, "desktop_account_delete_failed", "failed to delete desktop account") } - view := s.buildDesktopAccountView(account, nil, nil) + view := s.buildDesktopAccountView(account, nil, nil, nil, nil) return &view, nil } @@ -398,7 +408,7 @@ func (s *DesktopAccountService) RequestDelete(ctx context.Context, actor auth.Ac return nil, response.ErrInternal(50086, "desktop_account_request_delete_failed", "failed to update desktop account delete request") } - view := s.buildDesktopAccountView(account, nil, nil) + view := s.buildDesktopAccountView(account, nil, nil, nil, nil) return &view, nil } @@ -527,6 +537,99 @@ func (s *DesktopAccountService) applyRuntimeHealth(ctx context.Context, workspac } } +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 @@ -551,10 +654,14 @@ func effectiveDesktopAccountRuntimeHealth(report bufferedDesktopAccountHealthRep func (s *DesktopAccountService) buildDesktopAccountView( account *repository.DesktopAccount, clientMap map[uuid.UUID]*repository.DesktopClient, - presenceClientMap map[uuid.UUID]uuid.UUID, + 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 @@ -562,40 +669,55 @@ func (s *DesktopAccountService) buildDesktopAccountView( value := account.ClientID.String() clientID = &value - online := false - clientOnline = &online + 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 - if presenceClientMap != nil { - if activeClientID, ok := presenceClientMap[account.DesktopID]; ok && activeClientID == *account.ClientID { - online = true - } - } + 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, - ClientLastSeenAt: clientLastSeenAt, - ClientDeviceName: clientDeviceName, + 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, } } @@ -603,7 +725,7 @@ 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]*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) @@ -621,8 +743,8 @@ func (s *DesktopAccountService) loadClientState( clientIDs = append(clientIDs, *account.ClientID) } - presenceClientMap := loadDesktopAccountPresence(ctx, s.redis, accountIDs) - for _, clientID := range presenceClientMap { + accountPresenceMap := loadDesktopAccountPresence(ctx, s.redis, accountIDs) + for _, clientID := range accountPresenceMap { if _, ok := seen[clientID]; ok { continue } @@ -644,7 +766,16 @@ func (s *DesktopAccountService) loadClientState( } } - return clientMap, presenceClientMap + 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( diff --git a/server/internal/tenant/app/desktop_account_service_test.go b/server/internal/tenant/app/desktop_account_service_test.go index 2f178c4..d51955f 100644 --- a/server/internal/tenant/app/desktop_account_service_test.go +++ b/server/internal/tenant/app/desktop_account_service_test.go @@ -58,7 +58,9 @@ func TestResolveDesktopClientOnline_StaleHeartbeatIsOffline(t *testing.T) { } func TestBuildDesktopAccountView_StoredClientOwnershipWinsOverPresence(t *testing.T) { - service := &DesktopAccountService{} + service := &DesktopAccountService{now: func() time.Time { + return time.Date(2026, 4, 19, 21, 30, 0, 0, time.UTC) + }} accountID := uuid.New() staleClientID := uuid.New() activeClientID := uuid.New() @@ -84,6 +86,12 @@ func TestBuildDesktopAccountView_StoredClientOwnershipWinsOverPresence(t *testin map[uuid.UUID]uuid.UUID{ accountID: activeClientID, }, + map[uuid.UUID]bool{ + activeClientID: true, + }, + map[uuid.UUID]bool{ + activeClientID: true, + }, ) if assert.NotNil(t, view.ClientID) { @@ -92,16 +100,23 @@ func TestBuildDesktopAccountView_StoredClientOwnershipWinsOverPresence(t *testin if assert.NotNil(t, view.ClientOnline) { assert.False(t, *view.ClientOnline) } + if assert.NotNil(t, view.AccountSessionOnline) { + assert.False(t, *view.AccountSessionOnline) + } + if assert.NotNil(t, view.TaskConsumerOnline) { + assert.False(t, *view.TaskConsumerOnline) + } assert.Equal(t, "4181862206", view.PlatformUID) assert.Nil(t, view.ClientLastSeenAt) assert.Nil(t, view.ClientDeviceName) } func TestBuildDesktopAccountView_FallsBackToStoredClientWhenNoPresence(t *testing.T) { - service := &DesktopAccountService{} + now := time.Date(2026, 4, 19, 21, 30, 0, 0, time.UTC) + service := &DesktopAccountService{now: func() time.Time { return now }} accountID := uuid.New() storedClientID := uuid.New() - lastSeen := time.Date(2026, 4, 19, 21, 30, 0, 0, time.UTC) + lastSeen := now.Add(-desktopClientPresenceTTL / 2) view := service.buildDesktopAccountView( &repository.DesktopAccount{ @@ -119,17 +134,67 @@ func TestBuildDesktopAccountView_FallsBackToStoredClientWhenNoPresence(t *testin }, }, nil, + nil, + nil, ) if assert.NotNil(t, view.ClientID) { assert.Equal(t, storedClientID.String(), *view.ClientID) } if assert.NotNil(t, view.ClientOnline) { - assert.False(t, *view.ClientOnline) + assert.True(t, *view.ClientOnline) } assert.Equal(t, &lastSeen, view.ClientLastSeenAt) } +func TestBuildDesktopAccountView_SplitsHeartbeatAccountSessionAndTaskConsumer(t *testing.T) { + now := time.Date(2026, 4, 19, 21, 30, 0, 0, time.UTC) + service := &DesktopAccountService{now: func() time.Time { return now }} + accountID := uuid.New() + clientID := uuid.New() + lastSeen := now.Add(-time.Minute) + deviceName := "Windows Workstation" + + view := service.buildDesktopAccountView( + &repository.DesktopAccount{ + DesktopID: accountID, + ClientID: &clientID, + Platform: "sohuhao", + PlatformUID: "110424528162", + DisplayName: "全屋定制老炮", + Health: "live", + }, + map[uuid.UUID]*repository.DesktopClient{ + clientID: { + ID: clientID, + LastSeenAt: &lastSeen, + DeviceName: &deviceName, + }, + }, + map[uuid.UUID]uuid.UUID{ + accountID: clientID, + }, + map[uuid.UUID]bool{ + clientID: true, + }, + map[uuid.UUID]bool{ + clientID: false, + }, + ) + + if assert.NotNil(t, view.ClientOnline) { + assert.True(t, *view.ClientOnline) + } + if assert.NotNil(t, view.AccountSessionOnline) { + assert.True(t, *view.AccountSessionOnline) + } + if assert.NotNil(t, view.TaskConsumerOnline) { + assert.False(t, *view.TaskConsumerOnline) + } + assert.Equal(t, &lastSeen, view.ClientLastSeenAt) + assert.Equal(t, &deviceName, view.ClientDeviceName) +} + func TestShouldApplyDesktopAccountRuntimeHealth_IgnoresReportOlderThanVerifiedAt(t *testing.T) { verifiedAt := time.Date(2026, 4, 30, 9, 30, 0, 0, time.UTC) reportCheckedAt := verifiedAt.Add(-time.Minute) diff --git a/server/internal/tenant/app/desktop_presence.go b/server/internal/tenant/app/desktop_presence.go index d71e816..54321a9 100644 --- a/server/internal/tenant/app/desktop_presence.go +++ b/server/internal/tenant/app/desktop_presence.go @@ -14,11 +14,16 @@ import ( // as offline within one TTL even when the explicit offline request is missed. const desktopClientPresenceTTL = 30 * time.Second const desktopAccountPresenceTTL = 30 * time.Second +const desktopTaskConsumerPresenceTTL = 75 * time.Second func desktopClientPresenceKey(clientID uuid.UUID) string { return "desktop:presence:client:" + clientID.String() } +func desktopTaskConsumerPresenceKey(clientID uuid.UUID) string { + return "desktop:presence:task_consumer:" + clientID.String() +} + func desktopAccountPresenceKey(accountID uuid.UUID) string { return "desktop:presence:account:" + accountID.String() } @@ -42,6 +47,18 @@ func markDesktopClientPresent(ctx context.Context, redis *goredis.Client, client _ = redis.Set(ctx, desktopClientPresenceKey(clientID), time.Now().UTC().Format(time.RFC3339), desktopClientPresenceTTL).Err() } +func markDesktopTaskConsumerPresent(ctx context.Context, redis *goredis.Client, clientID uuid.UUID) { + if redis == nil || clientID == uuid.Nil { + return + } + + _ = redis.Set(ctx, desktopTaskConsumerPresenceKey(clientID), time.Now().UTC().Format(time.RFC3339), desktopTaskConsumerPresenceTTL).Err() +} + +func MarkDesktopTaskConsumerPresent(ctx context.Context, redis *goredis.Client, clientID uuid.UUID) { + markDesktopTaskConsumerPresent(ctx, redis, clientID) +} + func markDesktopClientPresentForUser(ctx context.Context, redis *goredis.Client, clientID uuid.UUID, tenantID, workspaceID, userID int64) { markDesktopClientPresent(ctx, redis, clientID) if redis == nil || clientID == uuid.Nil || tenantID == 0 || workspaceID == 0 || userID == 0 { @@ -66,7 +83,7 @@ func clearDesktopClientPresence(ctx context.Context, redis *goredis.Client, clie return } - _ = redis.Del(ctx, desktopClientPresenceKey(clientID)).Err() + _ = redis.Del(ctx, desktopClientPresenceKey(clientID), desktopTaskConsumerPresenceKey(clientID)).Err() } func clearDesktopClientPresenceStateForUser(ctx context.Context, redis *goredis.Client, clientID uuid.UUID, tenantID, workspaceID, userID int64) { @@ -115,6 +132,44 @@ func loadDesktopClientPresence(ctx context.Context, redis *goredis.Client, clien return result } +func loadDesktopTaskConsumerPresence(ctx context.Context, redis *goredis.Client, clientIDs []uuid.UUID) map[uuid.UUID]bool { + if redis == nil || len(clientIDs) == 0 { + return nil + } + + uniqueIDs := make([]uuid.UUID, 0, len(clientIDs)) + seen := make(map[uuid.UUID]struct{}, len(clientIDs)) + for _, clientID := range clientIDs { + if clientID == uuid.Nil { + continue + } + if _, ok := seen[clientID]; ok { + continue + } + seen[clientID] = struct{}{} + uniqueIDs = append(uniqueIDs, clientID) + } + if len(uniqueIDs) == 0 { + return nil + } + + pipe := redis.Pipeline() + cmds := make(map[uuid.UUID]*goredis.IntCmd, len(uniqueIDs)) + for _, clientID := range uniqueIDs { + cmds[clientID] = pipe.Exists(ctx, desktopTaskConsumerPresenceKey(clientID)) + } + + if _, err := pipe.Exec(ctx); err != nil && err != goredis.Nil { + return nil + } + + result := make(map[uuid.UUID]bool, len(cmds)) + for clientID, cmd := range cmds { + result[clientID] = cmd.Val() > 0 + } + return result +} + func loadDesktopUserOnlineClientCount(ctx context.Context, redis *goredis.Client, tenantID, workspaceID, userID int64, now time.Time) (int, bool) { if redis == nil || tenantID == 0 || workspaceID == 0 || userID == 0 { return 0, false diff --git a/server/internal/tenant/app/desktop_presence_test.go b/server/internal/tenant/app/desktop_presence_test.go index f1cee7f..b1a45a5 100644 --- a/server/internal/tenant/app/desktop_presence_test.go +++ b/server/internal/tenant/app/desktop_presence_test.go @@ -84,3 +84,30 @@ func TestDesktopUserOnlineClientCountPrunesStaleRedisPresence(t *testing.T) { require.True(t, ok) assert.Equal(t, 1, count) } + +func TestDesktopTaskConsumerPresenceIsSeparateFromHeartbeatPresence(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) + + clientPresence := loadDesktopClientPresence(ctx, redis, []uuid.UUID{clientID}) + taskConsumerPresence := loadDesktopTaskConsumerPresence(ctx, redis, []uuid.UUID{clientID}) + + require.NotNil(t, clientPresence) + require.NotNil(t, taskConsumerPresence) + assert.True(t, clientPresence[clientID]) + assert.False(t, taskConsumerPresence[clientID]) + + markDesktopTaskConsumerPresent(ctx, redis, clientID) + taskConsumerPresence = loadDesktopTaskConsumerPresence(ctx, redis, []uuid.UUID{clientID}) + + require.NotNil(t, taskConsumerPresence) + assert.True(t, taskConsumerPresence[clientID]) +} diff --git a/server/internal/tenant/app/desktop_task_events.go b/server/internal/tenant/app/desktop_task_events.go index 47f0246..5106326 100644 --- a/server/internal/tenant/app/desktop_task_events.go +++ b/server/internal/tenant/app/desktop_task_events.go @@ -8,6 +8,8 @@ import ( "time" "github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq" + "github.com/geo-platform/tenant-api/internal/shared/stream" + "github.com/geo-platform/tenant-api/internal/tenant/repository" ) type DesktopTaskEvent struct { @@ -43,6 +45,32 @@ func publishDesktopTaskEvent(ctx context.Context, client *rabbitmq.Client, event return client.PublishDesktopTaskEvent(ctx, payload) } +func DesktopDispatchEventFromTask(task *repository.DesktopTask, eventType string) stream.DesktopDispatchEvent { + if task == nil { + return stream.DesktopDispatchEvent{} + } + title, businessDate, schedulerGroupKey, questionText := deriveDesktopTaskEventMetadataFromPayload(task.Kind, task.Payload) + return stream.DesktopDispatchEvent{ + Type: eventType, + TaskID: task.DesktopID.String(), + JobID: task.JobID.String(), + WorkspaceID: task.WorkspaceID, + TargetAccountID: task.TargetAccountID.String(), + TargetClientID: task.TargetClientID.String(), + 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, + } +} + func deriveDesktopTaskEventMetadataFromPayload( kind string, payload []byte, diff --git a/server/internal/tenant/app/desktop_task_service.go b/server/internal/tenant/app/desktop_task_service.go index 1a14d03..79ceb0c 100644 --- a/server/internal/tenant/app/desktop_task_service.go +++ b/server/internal/tenant/app/desktop_task_service.go @@ -157,6 +157,7 @@ func (s *DesktopTaskService) Lease(ctx context.Context, client *repository.Deskt if client == nil { return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required") } + markDesktopTaskConsumerPresent(ctx, s.redis, client.ID) taskID := routeTaskID if taskID == nil && req.TaskID != nil && strings.TrimSpace(*req.TaskID) != "" { diff --git a/server/internal/tenant/app/publish_job_service.go b/server/internal/tenant/app/publish_job_service.go index f2b75a0..f2a738c 100644 --- a/server/internal/tenant/app/publish_job_service.go +++ b/server/internal/tenant/app/publish_job_service.go @@ -337,7 +337,7 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr invalidateArticleCaches(ctx, s.cache, actor.TenantID, &articleID) for _, task := range createdTasks { - title, businessDate, schedulerGroupKey, questionText := deriveDesktopTaskEventMetadataFromPayload(task.Kind, task.Payload) + dispatchEvent := DesktopDispatchEventFromTask(task, "task_available") if publishErr := publishDesktopTaskEvent(context.Background(), s.messaging, DesktopTaskEvent{ Type: "task_available", TaskID: task.DesktopID.String(), @@ -346,10 +346,10 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr TargetAccountID: task.TargetAccountID.String(), TargetClientID: task.TargetClientID.String(), Platform: task.Platform, - Title: title, - BusinessDate: businessDate, - SchedulerGroupKey: schedulerGroupKey, - QuestionText: questionText, + Title: dispatchEvent.Title, + BusinessDate: dispatchEvent.BusinessDate, + SchedulerGroupKey: dispatchEvent.SchedulerGroupKey, + QuestionText: dispatchEvent.QuestionText, Status: task.Status, Kind: task.Kind, UpdatedAt: task.UpdatedAt, @@ -357,22 +357,7 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr s.logWarn("desktop task available event publish failed", publishErr, zap.String("task_id", task.DesktopID.String())) } - s.publishDispatchAvailable(task, stream.DesktopDispatchEvent{ - Type: "task_available", - TaskID: task.DesktopID.String(), - JobID: task.JobID.String(), - WorkspaceID: task.WorkspaceID, - TargetAccountID: task.TargetAccountID.String(), - TargetClientID: task.TargetClientID.String(), - Platform: task.Platform, - Title: title, - BusinessDate: businessDate, - SchedulerGroupKey: schedulerGroupKey, - QuestionText: questionText, - Status: task.Status, - Kind: task.Kind, - UpdatedAt: task.UpdatedAt, - }) + s.publishDispatchAvailable(task, dispatchEvent) } return &CreatePublishJobResponse{ diff --git a/server/internal/tenant/transport/desktop_account_handler.go b/server/internal/tenant/transport/desktop_account_handler.go index c435e90..acbfcb4 100644 --- a/server/internal/tenant/transport/desktop_account_handler.go +++ b/server/internal/tenant/transport/desktop_account_handler.go @@ -20,6 +20,7 @@ type DesktopAccountHandler struct { func NewDesktopAccountHandler(a *bootstrap.App) *DesktopAccountHandler { return &DesktopAccountHandler{ svc: app.NewDesktopAccountService( + a.DB, repository.NewDesktopAccountRepository(a.DB), repository.NewDesktopClientRepository(a.DB), a.Redis, diff --git a/server/internal/tenant/transport/desktop_dispatch_handler.go b/server/internal/tenant/transport/desktop_dispatch_handler.go index fc5e558..be7f2f6 100644 --- a/server/internal/tenant/transport/desktop_dispatch_handler.go +++ b/server/internal/tenant/transport/desktop_dispatch_handler.go @@ -1,6 +1,7 @@ package transport import ( + "context" "encoding/json" "net/http" "time" @@ -11,6 +12,8 @@ import ( "github.com/geo-platform/tenant-api/internal/bootstrap" "github.com/geo-platform/tenant-api/internal/shared/stream" + tenantapp "github.com/geo-platform/tenant-api/internal/tenant/app" + "github.com/geo-platform/tenant-api/internal/tenant/repository" ) const ( @@ -19,6 +22,7 @@ const ( dispatchPingPeriod = 25 * time.Second dispatchReadLimitBytes = 2048 dispatchCloseGracePause = 500 * time.Millisecond + dispatchReplayLimit = 20 ) type DesktopDispatchHandler struct { @@ -66,6 +70,7 @@ func (h *DesktopDispatchHandler) Dispatch(c *gin.Context) { subscriber, cancel := hub.Subscribe(client.ID.String()) defer cancel() + tenantapp.MarkDesktopTaskConsumerPresent(c.Request.Context(), h.app.Redis, client.ID) conn.SetReadLimit(dispatchReadLimitBytes) _ = conn.SetReadDeadline(time.Now().Add(dispatchPongWait)) @@ -86,16 +91,19 @@ func (h *DesktopDispatchHandler) Dispatch(c *gin.Context) { } }() + serverTime := time.Now().UTC() connected := stream.DesktopDispatchEvent{ Type: "connected", TargetClientID: client.ID.String(), WorkspaceID: client.WorkspaceID, - UpdatedAt: time.Now().UTC(), + UpdatedAt: serverTime, + ServerTime: &serverTime, } if err := writeDispatchFrame(conn, connected); err != nil { _ = conn.Close() return } + h.replayQueuedPublishTasks(c.Request.Context(), client, conn) pingTicker := time.NewTicker(dispatchPingPeriod) defer pingTicker.Stop() @@ -139,3 +147,154 @@ func writeDispatchFrame(conn *websocket.Conn, event stream.DesktopDispatchEvent) _ = conn.SetWriteDeadline(time.Now().Add(dispatchWriteWait)) return conn.WriteMessage(websocket.TextMessage, body) } + +func (h *DesktopDispatchHandler) replayQueuedPublishTasks(ctx context.Context, client *repository.DesktopClient, conn *websocket.Conn) { + if h == nil || h.app == nil || h.app.DB == nil || client == nil || conn == nil { + return + } + + rows, err := h.app.DB.Query(ctx, ` + SELECT + dt.desktop_id, + dt.job_id, + dt.tenant_id, + dt.workspace_id, + dt.target_account_id, + dt.target_client_id, + dt.platform_id, + dt.kind, + dt.payload, + dt.status, + dt.priority, + dt.lane, + dt.lane_weight, + dt.source, + dt.scheduler_group_key, + dt.monitor_task_id, + dt.supersedes_task_id, + dt.control_flags, + dt.interrupt_generation, + dt.dedup_key, + dt.active_attempt_id, + dt.lease_expires_at, + dt.attempts, + dt.result, + dt.error, + dt.started_at, + dt.interrupted_at, + dt.interrupt_reason, + dt.enqueued_at, + dt.created_at, + dt.updated_at + FROM desktop_tasks AS dt + WHERE dt.tenant_id = $1 + AND dt.workspace_id = $2 + AND dt.target_client_id = $3 + AND dt.kind = 'publish' + AND dt.status = 'queued' + 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' + ) + ORDER BY dt.priority DESC, + COALESCE(dt.enqueued_at, dt.created_at) ASC, + dt.created_at ASC, + dt.desktop_id ASC + LIMIT $4 + `, client.TenantID, client.WorkspaceID, client.ID, dispatchReplayLimit) + if err != nil { + if h.app.Logger != nil { + h.app.Logger.Warn("desktop dispatch replay query failed", + zap.String("client_id", client.ID.String()), + zap.Error(err), + ) + } + return + } + defer rows.Close() + + replayed := 0 + for rows.Next() { + task, scanErr := scanDispatchReplayDesktopTask(rows) + if scanErr != nil { + if h.app.Logger != nil { + h.app.Logger.Warn("desktop dispatch replay scan failed", + zap.String("client_id", client.ID.String()), + zap.Error(scanErr), + ) + } + return + } + if err := writeDispatchFrame(conn, tenantapp.DesktopDispatchEventFromTask(task, "task_available")); err != nil { + if h.app.Logger != nil { + h.app.Logger.Warn("desktop dispatch replay write failed", + zap.String("client_id", client.ID.String()), + zap.String("task_id", task.DesktopID.String()), + zap.Error(err), + ) + } + return + } + replayed++ + } + if err := rows.Err(); err != nil && h.app.Logger != nil { + h.app.Logger.Warn("desktop dispatch replay rows failed", + zap.String("client_id", client.ID.String()), + zap.Error(err), + ) + } + if replayed > 0 && h.app.Logger != nil { + h.app.Logger.Info("desktop dispatch replayed queued publish tasks", + zap.String("client_id", client.ID.String()), + zap.Int("count", replayed), + ) + } +} + +type dispatchReplayDesktopTaskScanner interface { + Scan(dest ...any) error +} + +func scanDispatchReplayDesktopTask(row dispatchReplayDesktopTaskScanner) (*repository.DesktopTask, error) { + var task repository.DesktopTask + err := row.Scan( + &task.DesktopID, + &task.JobID, + &task.TenantID, + &task.WorkspaceID, + &task.TargetAccountID, + &task.TargetClientID, + &task.Platform, + &task.Kind, + &task.Payload, + &task.Status, + &task.Priority, + &task.Lane, + &task.LaneWeight, + &task.Source, + &task.SchedulerGroupKey, + &task.MonitorTaskID, + &task.SupersedesTaskID, + &task.ControlFlags, + &task.InterruptGeneration, + &task.DedupKey, + &task.ActiveAttemptID, + &task.LeaseExpiresAt, + &task.Attempts, + &task.Result, + &task.Error, + &task.StartedAt, + &task.InterruptedAt, + &task.InterruptReason, + &task.EnqueuedAt, + &task.CreatedAt, + &task.UpdatedAt, + ) + if err != nil { + return nil, err + } + return &task, nil +}