From af35301d9b5adcf08170af0313a3b87b5d66400f Mon Sep 17 00:00:00 2001 From: liangxu Date: Fri, 8 May 2026 20:02:57 +0800 Subject: [PATCH] test(desktop): cover deepseek account name extraction for wechat + phone logins MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refactor readDeepseekAccountIdentity so the JSON payload parser is a pure ts function (extractDeepseekIdentityFromPayload). The webContents IIFE now just polls for userToken and returns the raw API JSON; payload picking lives in ts and is unit-tested. Tests lock the contract for both login types we observed in the wild: - WeChat-bound account: id_profile.name = '阿白', picture present. - Phone-only login: id_profile = null, mobile_number = '177******08', email = '' (empty string, not null) — the case that originally regressed. Plus regressions for the priority order (profile.name > mobile_number > email), the all-empty fallback, and non-record payloads. Co-Authored-By: Claude Opus 4.7 (1M context) --- apps/admin-web/src/views/TrackingView.vue | 35 +- .../deepseek/account-identity.test.ts | 115 +++ .../adapters/deepseek/account-identity.ts | 79 ++- packages/shared-types/src/index.ts | 1 + .../app/monitoring_collect_outbox_worker.go | 158 +++-- .../app/monitoring_phase2_desktop_tasks.go | 63 +- .../monitoring_phase2_desktop_tasks_test.go | 28 + .../internal/tenant/app/monitoring_service.go | 671 +++++++++++++++--- .../app/monitoring_service_date_test.go | 21 + 9 files changed, 963 insertions(+), 208 deletions(-) create mode 100644 apps/desktop-client/src/main/adapters/deepseek/account-identity.test.ts diff --git a/apps/admin-web/src/views/TrackingView.vue b/apps/admin-web/src/views/TrackingView.vue index f8d3c1a..a089172 100644 --- a/apps/admin-web/src/views/TrackingView.vue +++ b/apps/admin-web/src/views/TrackingView.vue @@ -56,6 +56,7 @@ type AIPlatformStatusCard = { accent: string description: string account: DesktopAccountInfo | null + isClientOnline: boolean sampleStatus: string lastSampledAt: string | null actualSampleCount: number @@ -327,6 +328,17 @@ const selectedPlatformAuthorized = computed(() => { ) }) +const selectedPlatformOnlineAuthorized = computed(() => { + const onlinePlatformIDs = + dashboardQuery.data.value?.runtime.online_authorized_monitoring_platform_ids ?? [] + if (selectedPlatformId.value === 'all') { + return onlinePlatformIDs.length > 0 + } + return onlinePlatformIDs.some( + (item) => normalizeMonitoringPlatformId(item) === selectedPlatformId.value, + ) +}) + const collectNowDisabledReason = computed(() => { if (!selectedBrandId.value) { return t('tracking.collectBrandRequired') @@ -356,6 +368,9 @@ const collectNowDisabledReason = computed(() => { if (!currentUserClientOnline.value) { return t('tracking.collectClientOnlineRequired') } + if (dashboardQuery.data.value && !selectedPlatformOnlineAuthorized.value) { + return t('tracking.collectClientOnlineRequired') + } return null }) @@ -385,9 +400,19 @@ const aiPlatformStatusCards = computed(() => { item, ]), ) + const onlineAuthorizedPlatformIDs = new Set( + (dashboardQuery.data.value?.runtime.online_authorized_monitoring_platform_ids ?? []) + .map((item) => normalizeMonitoringPlatformId(item)) + .filter((item): item is string => Boolean(item)), + ) return aiPlatformCatalog.map((platform) => { const matchedAccounts = [...(accountGroups.get(platform.id) ?? [])].sort((left, right) => { + const leftOnline = left.client_online === true ? 1 : 0 + const rightOnline = right.client_online === true ? 1 : 0 + if (leftOnline !== rightOnline) { + return rightOnline - leftOnline + } const leftAt = Date.parse(resolveAccountCheckedAt(left) ?? '') || 0 const rightAt = Date.parse(resolveAccountCheckedAt(right) ?? '') || 0 return rightAt - leftAt @@ -401,6 +426,10 @@ const aiPlatformStatusCards = computed(() => { accent: platform.accent, description: platform.description, account: matchedAccounts[0] ?? null, + isClientOnline: + onlineAuthorizedPlatformIDs.has(platform.id) || + (!dashboardQuery.data.value && + matchedAccounts.some((account) => account.client_online === true)), sampleStatus: breakdown?.platform_sample_status ?? 'pending', lastSampledAt: breakdown?.last_sampled_at ?? null, actualSampleCount: breakdown?.actual_sample_count ?? 0, @@ -671,12 +700,12 @@ function accountAuthColor(account: DesktopAccountInfo | null): string { } } -function desktopStatusLabel(account: DesktopAccountInfo | null): string { +function desktopStatusLabel(account: DesktopAccountInfo | null, isClientOnline: boolean): string { if (!account?.client_id) { return '未绑定桌面端' } - return account.client_online ? '客户端在线' : '客户端离线' + return isClientOnline ? '客户端在线' : '客户端离线' } function formatCount(value: number | null | undefined): string { @@ -999,7 +1028,7 @@ function disableTrackingDate(current: dayjs.Dayjs): boolean {
客户端 - {{ desktopStatusLabel(platform.account) }} + {{ desktopStatusLabel(platform.account, platform.isClientOnline) }}
采样状态 diff --git a/apps/desktop-client/src/main/adapters/deepseek/account-identity.test.ts b/apps/desktop-client/src/main/adapters/deepseek/account-identity.test.ts new file mode 100644 index 0000000..88d0bcf --- /dev/null +++ b/apps/desktop-client/src/main/adapters/deepseek/account-identity.test.ts @@ -0,0 +1,115 @@ +import { describe, expect, it } from 'vitest' + +import { extractDeepseekIdentityFromPayload } from './account-identity' + +describe('extractDeepseekIdentityFromPayload', () => { + it('returns the WeChat profile name + avatar for WECHAT-bound accounts', () => { + const wechatPayload = { + code: 0, + msg: '', + data: { + biz_code: 0, + biz_msg: '', + biz_data: { + id: '7a15e288-1560-49be-a519-3e3da071071c', + email: '', + mobile_number: '130******73', + area_code: '+86', + status: 0, + id_profile: { + provider: 'WECHAT', + id: 'e5a1861a-6c1b-48ae-9d2c-00e643e6b4bd', + name: '阿白', + picture: 'https://static.deepseek.com/user-avatar/LwJ3Jq-HLkgwCGbl0HS1lECW', + locale: 'zh_CN', + email: null, + }, + }, + }, + } + + expect(extractDeepseekIdentityFromPayload(wechatPayload)).toEqual({ + displayName: '阿白', + avatarUrl: 'https://static.deepseek.com/user-avatar/LwJ3Jq-HLkgwCGbl0HS1lECW', + }) + }) + + it('falls back to mobile_number for phone-only logins (id_profile is null, email is empty string)', () => { + const phonePayload = { + code: 0, + msg: '', + data: { + biz_code: 0, + biz_msg: '', + biz_data: { + id: 'd782f68b-7358-4290-b891-38cb7a1e11fa', + email: '', + mobile_number: '177******08', + area_code: '+86', + status: 0, + id_profile: null, + id_profiles: [], + }, + }, + } + + expect(extractDeepseekIdentityFromPayload(phonePayload)).toEqual({ + displayName: '177******08', + avatarUrl: null, + }) + }) + + it('prefers id_profile.name over mobile_number when both are present', () => { + const payload = { + data: { + biz_data: { + mobile_number: '130******73', + id_profile: { + provider: 'WECHAT', + name: '阿白', + picture: null, + }, + }, + }, + } + + expect(extractDeepseekIdentityFromPayload(payload).displayName).toBe('阿白') + }) + + it('uses email as last-resort fallback when id_profile and mobile_number are empty', () => { + const payload = { + data: { + biz_data: { + id_profile: null, + mobile_number: '', + email: 'user@example.com', + }, + }, + } + + expect(extractDeepseekIdentityFromPayload(payload).displayName).toBe('user@example.com') + }) + + it('returns null displayName when nothing usable is present', () => { + expect( + extractDeepseekIdentityFromPayload({ + data: { biz_data: { id_profile: null, email: '', mobile_number: '' } }, + }), + ).toEqual({ displayName: null, avatarUrl: null }) + }) + + it('rejects non-record payloads without throwing', () => { + expect(extractDeepseekIdentityFromPayload(null)).toEqual({ + displayName: null, + avatarUrl: null, + }) + expect(extractDeepseekIdentityFromPayload('hello')).toEqual({ + displayName: null, + avatarUrl: null, + }) + expect(extractDeepseekIdentityFromPayload([])).toEqual({ + displayName: null, + avatarUrl: null, + }) + }) +}) diff --git a/apps/desktop-client/src/main/adapters/deepseek/account-identity.ts b/apps/desktop-client/src/main/adapters/deepseek/account-identity.ts index 9f6f4e6..5fd97ec 100644 --- a/apps/desktop-client/src/main/adapters/deepseek/account-identity.ts +++ b/apps/desktop-client/src/main/adapters/deepseek/account-identity.ts @@ -14,18 +14,52 @@ function normalizeText(value: unknown): string | null { return trimmed || null } +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +function pickString(value: unknown): string | null { + return typeof value === 'string' && value.trim() ? value : null +} + +// Pure parser, exported for tests. Both 微信登录 and 手机号登录 must work: +// 微信: { biz_data: { id_profile: { provider: 'WECHAT', name: '阿白', picture: '...' }, ... } } +// 手机: { biz_data: { id_profile: null, mobile_number: '177******08', email: '', ... } } +// Empty-string fields (email = "") are common, so we use truthy checks rather than ??. +export function extractDeepseekIdentityFromPayload( + payload: unknown, +): DeepseekAccountIdentityHints { + if (!isRecord(payload)) { + return { displayName: null, avatarUrl: null } + } + + const data = isRecord(payload.data) ? payload.data : null + const bizData = data && isRecord(data.biz_data) ? data.biz_data : null + if (!bizData) { + return { displayName: null, avatarUrl: null } + } + + const profile = isRecord(bizData.id_profile) ? bizData.id_profile : null + + const displayName = + pickString(profile?.name) ?? + pickString(bizData.mobile_number) ?? + pickString(bizData.email) + const avatarUrl = pickString(profile?.picture) + + return { + displayName: normalizeText(displayName), + avatarUrl: normalizeText(avatarUrl), + } +} + // DeepSeek (chat.deepseek.com) 没有 Redux/Pinia/window state，登录态只有 localStorage -// 里的 `userToken`（JSON 包了 value 字段）。账号信息要走 /api/v0/users/current， -// 用 Authorization: Bearer 调。Cookie 不带 token，因此只能在页面上下文里 -// 执行 fetch（自带 token）。 -// -// 不同登录方式响应差异较大： -// 微信登录: id_profile = { provider: 'WECHAT', name: '阿白', picture: '...' } -// 手机登录: id_profile = null, mobile_number = '177******08'（其它字段是空串而非 null） -// 所以这里 fallback 必须用 truthy 判断而不是 ??，否则空串会击穿后续兜底。 +// 里的 `userToken`（JSON 包了 value 字段）。账号信息要走 GET /api/v0/users/current， +// 用 Authorization: Bearer 调（cookie 不带 token）。 // // probe 通常在隐藏窗口、刚加载完就跑，React app 可能还没把 userToken 写入 localStorage， -// 所以要 poll 一下直到 token 出现，再调 API。 +// 所以要 poll 一下直到 token 出现，再调 API。webContents 里的 IIFE 仅拉原始 JSON， +// 解析与字段挑选放到 extractDeepseekIdentityFromPayload 里以便单测。 export async function readDeepseekAccountIdentity( webContents: WebContents | null | undefined, ): Promise { @@ -33,7 +67,7 @@ export async function readDeepseekAccountIdentity( return { displayName: null, avatarUrl: null } } - const result = await webContents + const rawJson = await webContents .executeJavaScript( `(async () => { try { @@ -64,17 +98,7 @@ export async function readDeepseekAccountIdentity( headers: { Authorization: 'Bearer ' + token }, }); if (!response.ok) return null; - const payload = await response.json().catch(() => null); - const bizData = payload?.data?.biz_data ?? null; - const profile = bizData?.id_profile ?? null; - const pickString = (value) => - typeof value === 'string' && value.trim() ? value : null; - const name = - pickString(profile?.name) ?? - pickString(bizData?.mobile_number) ?? - pickString(bizData?.email); - const picture = pickString(profile?.picture); - return { displayName: name, avatarUrl: picture }; + return await response.text(); } catch { return null; } @@ -83,13 +107,16 @@ export async function readDeepseekAccountIdentity( ) .catch(() => null) - if (!result || typeof result !== 'object') { + if (typeof rawJson !== 'string' || !rawJson) { return { displayName: null, avatarUrl: null } } - const typed = result as { displayName?: unknown; avatarUrl?: unknown } - return { - displayName: normalizeText(typed.displayName), - avatarUrl: normalizeText(typed.avatarUrl), + let payload: unknown + try { + payload = JSON.parse(rawJson) + } catch { + return { displayName: null, avatarUrl: null } } + + return extractDeepseekIdentityFromPayload(payload) } diff --git a/packages/shared-types/src/index.ts b/packages/shared-types/src/index.ts index 6462c76..3269c80 100644 --- a/packages/shared-types/src/index.ts +++ b/packages/shared-types/src/index.ts @@ -1352,6 +1352,7 @@ export interface MonitoringDashboardRuntime { platform_authorization_status: MonitoringPlatformAuthorizationStatus authorized_monitoring_platform_ids: string[] authorized_monitoring_platform_count: number + online_authorized_monitoring_platform_ids?: string[] } export interface MonitoringDashboardCompositeResponse { diff --git a/server/internal/tenant/app/monitoring_collect_outbox_worker.go b/server/internal/tenant/app/monitoring_collect_outbox_worker.go index 5320442..452fd11 100644 --- a/server/internal/tenant/app/monitoring_collect_outbox_worker.go +++ b/server/internal/tenant/app/monitoring_collect_outbox_worker.go @@ -46,11 +46,12 @@ type monitoringCollectRequestState struct { } type monitoringCollectRequestScope struct { - BrandID int64 `json:"brand_id"` - KeywordID *int64 `json:"keyword_id"` - QuestionIDs []int64 `json:"question_ids"` - PlatformIDs []string `json:"platform_ids"` - Preempt bool `json:"preempt"` + BrandID int64 `json:"brand_id"` + KeywordID *int64 `json:"keyword_id"` + QuestionIDs []int64 `json:"question_ids"` + PlatformIDs []string `json:"platform_ids"` + TargetClientIDs []string `json:"target_client_ids"` + Preempt bool `json:"preempt"` } type monitoringCollectRequestReconcileRow struct { @@ -553,16 +554,18 @@ func (s *MonitoringService) loadCollectNowRequestTaskStats( return stats, nil } - targetClientID := "" - if request.TargetClientID.Valid { - targetClientID = strings.TrimSpace(request.TargetClientID.String) - } - if targetClientID == "" { - return stats, nil - } - + targetClientIDs := normalizeCollectNowScopeTargetClientIDs(scope.TargetClientIDs, request.TargetClientID) _, businessDate := monitoringBusinessDayAndDateAt(request.CreatedAt) - if err := s.monitoringPool.QueryRow(ctx, ` + queryArgs := []any{ + request.TenantID, + request.BrandID, + monitoringCollectorType, + businessDate, + request.InterruptGeneration, + scope.QuestionIDs, + scope.PlatformIDs, + } + query := ` SELECT COUNT(*)::bigint AS total_count, COUNT(*) FILTER (WHERE t.status IN ('completed', 'failed', 'skipped', 'expired'))::bigint AS terminal_count, @@ -579,11 +582,15 @@ func (s *MonitoringService) loadCollectNowRequestTaskStats( AND t.collector_type = $3 AND t.trigger_source = 'manual' AND t.business_date = $4::date - AND t.target_client_id = $5::uuid - AND t.interrupt_generation = $6 - AND t.question_id = ANY($7::bigint[]) - AND t.ai_platform_id = ANY($8::text[]) - `, request.TenantID, request.BrandID, monitoringCollectorType, businessDate, targetClientID, request.InterruptGeneration, scope.QuestionIDs, scope.PlatformIDs).Scan( + AND t.interrupt_generation = $5 + AND t.question_id = ANY($6::bigint[]) + AND t.ai_platform_id = ANY($7::text[]) + ` + if len(targetClientIDs) > 0 { + query += ` AND t.target_client_id = ANY($8::uuid[])` + queryArgs = append(queryArgs, targetClientIDs) + } + if err := s.monitoringPool.QueryRow(ctx, query, queryArgs...).Scan( &stats.Total, &stats.Terminal, &stats.Completed, @@ -614,9 +621,8 @@ func (s *MonitoringService) loadCollectNowRequestTaskStats( FROM desktop_tasks d WHERE d.kind = 'monitor' AND d.status = 'in_progress' - AND d.target_client_id = $1::uuid - AND d.monitor_task_id = ANY($2::bigint[]) - `, targetClientID, monitorTaskIDs).Scan( + AND d.monitor_task_id = ANY($1::bigint[]) + `, monitorTaskIDs).Scan( &stats.ActiveDesktopTaskCount, ); err != nil { return stats, response.ErrInternal(50041, "query_failed", "failed to inspect collect-now desktop task progress") @@ -637,16 +643,18 @@ func (s *MonitoringService) loadCollectNowMonitorTaskIDs( return nil, nil } - targetClientID := "" - if request.TargetClientID.Valid { - targetClientID = strings.TrimSpace(request.TargetClientID.String) - } - if targetClientID == "" { - return nil, nil - } - + targetClientIDs := normalizeCollectNowScopeTargetClientIDs(scope.TargetClientIDs, request.TargetClientID) _, businessDate := monitoringBusinessDayAndDateAt(request.CreatedAt) - rows, err := s.monitoringPool.Query(ctx, ` + queryArgs := []any{ + request.TenantID, + request.BrandID, + monitoringCollectorType, + businessDate, + request.InterruptGeneration, + scope.QuestionIDs, + scope.PlatformIDs, + } + query := ` SELECT t.id FROM monitoring_collect_tasks t WHERE t.tenant_id = $1 @@ -654,11 +662,15 @@ func (s *MonitoringService) loadCollectNowMonitorTaskIDs( AND t.collector_type = $3 AND t.trigger_source = 'manual' AND t.business_date = $4::date - AND t.target_client_id = $5::uuid - AND t.interrupt_generation = $6 - AND t.question_id = ANY($7::bigint[]) - AND t.ai_platform_id = ANY($8::text[]) - `, request.TenantID, request.BrandID, monitoringCollectorType, businessDate, targetClientID, request.InterruptGeneration, scope.QuestionIDs, scope.PlatformIDs) + AND t.interrupt_generation = $5 + AND t.question_id = ANY($6::bigint[]) + AND t.ai_platform_id = ANY($7::text[]) + ` + if len(targetClientIDs) > 0 { + query += ` AND t.target_client_id = ANY($8::uuid[])` + queryArgs = append(queryArgs, targetClientIDs) + } + rows, err := s.monitoringPool.Query(ctx, query, queryArgs...) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load collect-now monitor task ids") } @@ -689,6 +701,37 @@ func decodeCollectNowRequestScope(payload []byte) monitoringCollectRequestScope return scope } +func normalizeCollectNowScopeTargetClientIDs(values []string, fallback sql.NullString) []uuid.UUID { + result := make([]uuid.UUID, 0, len(values)+1) + seen := make(map[string]struct{}, len(values)+1) + for _, value := range values { + value = strings.TrimSpace(value) + if value == "" { + continue + } + parsed, err := uuid.Parse(value) + if err != nil { + continue + } + if _, exists := seen[value]; exists { + continue + } + seen[value] = struct{}{} + result = append(result, parsed) + } + if fallback.Valid { + value := strings.TrimSpace(fallback.String) + if value != "" { + if parsed, err := uuid.Parse(value); err == nil { + if _, exists := seen[value]; !exists { + result = append(result, parsed) + } + } + } + } + return result +} + func normalizeCollectNowStringList(items []string) []string { if len(items) == 0 { return nil @@ -825,7 +868,7 @@ func (s *MonitoringService) enqueueCollectNowOutbox( tx pgx.Tx, requestID uuid.UUID, workspaceID int64, - targetClientID uuid.UUID, + targetClientIDs []uuid.UUID, affectedTaskCount int64, interruptGeneration int, interruptTaskIDs []int64, @@ -834,15 +877,22 @@ func (s *MonitoringService) enqueueCollectNowOutbox( return nil } + targetClientIDs = uniqueUUIDs(targetClientIDs) + if len(targetClientIDs) == 0 { + return nil + } + if affectedTaskCount > 0 { - if _, err := tx.Exec(ctx, ` - INSERT INTO monitoring_collect_dispatch_outbox ( - request_id, workspace_id, target_client_id, event_kind, - lane, priority, interrupt_generation, request_dispatched_count - ) - VALUES ($1, $2, $3, 'dispatch_high', 'high', 5000, $4, $5) - `, requestID, workspaceID, targetClientID, interruptGeneration, affectedTaskCount); err != nil { - return response.ErrInternal(50041, "outbox_insert_failed", "failed to enqueue collect-now dispatch signal") + for _, targetClientID := range targetClientIDs { + if _, err := tx.Exec(ctx, ` + INSERT INTO monitoring_collect_dispatch_outbox ( + request_id, workspace_id, target_client_id, event_kind, + lane, priority, interrupt_generation, request_dispatched_count + ) + VALUES ($1, $2, $3, 'dispatch_high', 'high', 5000, $4, $5) + `, requestID, workspaceID, targetClientID, interruptGeneration, affectedTaskCount); err != nil { + return response.ErrInternal(50041, "outbox_insert_failed", "failed to enqueue collect-now dispatch signal") + } } } @@ -855,14 +905,16 @@ func (s *MonitoringService) enqueueCollectNowOutbox( continue } seenTaskIDs[taskID] = struct{}{} - if _, err := tx.Exec(ctx, ` - INSERT INTO monitoring_collect_dispatch_outbox ( - request_id, workspace_id, target_client_id, event_kind, task_id, - lane, priority, interrupt_generation, reason - ) - VALUES ($1, $2, $3, 'interrupt_requested', $4, 'high', 5000, $5, 'collect_now_preempt') - `, requestID, workspaceID, targetClientID, taskID, interruptGeneration); err != nil { - return response.ErrInternal(50041, "outbox_insert_failed", "failed to enqueue collect-now interrupt signal") + for _, targetClientID := range targetClientIDs { + if _, err := tx.Exec(ctx, ` + INSERT INTO monitoring_collect_dispatch_outbox ( + request_id, workspace_id, target_client_id, event_kind, task_id, + lane, priority, interrupt_generation, reason + ) + VALUES ($1, $2, $3, 'interrupt_requested', $4, 'high', 5000, $5, 'collect_now_preempt') + `, requestID, workspaceID, targetClientID, taskID, interruptGeneration); err != nil { + return response.ErrInternal(50041, "outbox_insert_failed", "failed to enqueue collect-now interrupt signal") + } } } diff --git a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go index 6098a31..82bf28f 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go @@ -39,6 +39,7 @@ type monitorDesktopTaskSpec struct { Priority int Lane string InterruptGeneration int + RequestedByUserID int64 } type monitorDesktopAccountCandidate struct { @@ -46,6 +47,7 @@ type monitorDesktopAccountCandidate struct { AccountID uuid.UUID DBClientID *uuid.UUID HealthRank int + OnlineRank int RecordOrder int } @@ -107,6 +109,7 @@ func (s *MonitoringService) loadMonitorDesktopTaskSpecs( questionIDs []int64, platformIDs []string, targetClientID uuid.UUID, + requestedByUserID int64, ) ([]monitorDesktopTaskSpec, error) { if q == nil || len(questionIDs) == 0 || len(platformIDs) == 0 { return nil, nil @@ -142,11 +145,10 @@ func (s *MonitoringService) loadMonitorDesktopTaskSpecs( AND t.business_date = $3::date AND t.status = 'pending' AND COALESCE(t.execution_owner, 'legacy') = 'desktop_tasks' - AND t.target_client_id = $4 - AND t.question_id = ANY($5) - AND t.ai_platform_id = ANY($6) + AND t.question_id = ANY($4) + AND t.ai_platform_id = ANY($5) ORDER BY t.dispatch_priority DESC, t.id ASC - `, tenantID, monitoringCollectorType, businessDate.Format("2006-01-02"), targetClientID, questionIDs, platformIDs) + `, tenantID, monitoringCollectorType, businessDate.Format("2006-01-02"), questionIDs, platformIDs) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load phase2 monitor desktop task specs") } @@ -177,6 +179,7 @@ func (s *MonitoringService) loadMonitorDesktopTaskSpecs( } spec.WorkspaceID = workspaceID spec.TargetClientID = targetClientID + spec.RequestedByUserID = requestedByUserID spec.BusinessDate = businessDay.Format("2006-01-02") spec.QuestionHash = encodeQuestionHash(questionHash) spec.SchedulerGroupKey = monitoringSchedulerGroupKey(spec.QuestionHash, spec.QuestionID, spec.QuestionText) @@ -196,7 +199,7 @@ func (s *MonitoringService) dispatchMonitorDesktopTasks( return nil, nil, nil } - targets, err := s.loadMonitorDesktopTaskTargets(ctx, specs[0].TenantID, specs[0].WorkspaceID, specs) + targets, err := s.loadMonitorDesktopTaskTargets(ctx, specs[0].TenantID, specs[0].WorkspaceID, specs[0].RequestedByUserID, specs) if err != nil { return nil, nil, err } @@ -240,7 +243,7 @@ func (s *MonitoringService) dispatchMonitorDesktopTasks( func (s *MonitoringService) loadMonitorDesktopTaskTargets( ctx context.Context, - tenantID, workspaceID int64, + tenantID, workspaceID, userID int64, specs []monitorDesktopTaskSpec, ) (map[string]monitorDesktopTaskTarget, error) { platformIDs := uniqueMonitorDesktopTaskPlatformIDs(specs) @@ -253,6 +256,20 @@ func (s *MonitoringService) loadMonitorDesktopTaskTargets( platform_id, desktop_id, COALESCE(client_id::text, '') AS client_id, + CASE + WHEN client_id IS NOT NULL + AND EXISTS ( + SELECT 1 + FROM desktop_clients dc + WHERE dc.id = platform_accounts.client_id + AND dc.workspace_id = platform_accounts.workspace_id + AND dc.revoked_at IS NULL + AND dc.last_seen_at IS NOT NULL + AND dc.last_seen_at >= NOW() - $5::interval + ) + THEN 0 + ELSE 1 + END AS online_rank, CASE health WHEN 'live' THEN 0 WHEN 'risk' THEN 1 @@ -264,7 +281,22 @@ func (s *MonitoringService) loadMonitorDesktopTaskTargets( AND workspace_id = $2 AND deleted_at IS NULL AND platform_id = ANY($3::text[]) + AND ($4::bigint = 0 OR user_id = $4) ORDER BY platform_id ASC, + CASE + WHEN client_id IS NOT NULL + AND EXISTS ( + SELECT 1 + FROM desktop_clients dc + WHERE dc.id = platform_accounts.client_id + AND dc.workspace_id = platform_accounts.workspace_id + AND dc.revoked_at IS NULL + AND dc.last_seen_at IS NOT NULL + AND dc.last_seen_at >= NOW() - $5::interval + ) + THEN 0 + ELSE 1 + END ASC, CASE health WHEN 'live' THEN 0 WHEN 'risk' THEN 1 @@ -274,7 +306,7 @@ func (s *MonitoringService) loadMonitorDesktopTaskTargets( verified_at DESC NULLS LAST, updated_at DESC, created_at DESC - `, tenantID, workspaceID, platformIDs) + `, tenantID, workspaceID, platformIDs, userID, formatPgInterval(desktopClientPresenceTTL)) if err != nil { return nil, response.ErrInternal(50117, "desktop_account_lookup_failed", "failed to resolve monitor desktop accounts") } @@ -286,7 +318,7 @@ func (s *MonitoringService) loadMonitorDesktopTaskTargets( item monitorDesktopAccountCandidate clientIDText string ) - if scanErr := rows.Scan(&item.PlatformID, &item.AccountID, &clientIDText, &item.HealthRank); scanErr != nil { + if scanErr := rows.Scan(&item.PlatformID, &item.AccountID, &clientIDText, &item.OnlineRank, &item.HealthRank); scanErr != nil { return nil, response.ErrInternal(50117, "desktop_account_lookup_failed", "failed to parse monitor desktop accounts") } if parsedClientID, parseErr := uuid.Parse(strings.TrimSpace(clientIDText)); parseErr == nil { @@ -401,23 +433,28 @@ func selectMonitorDesktopTaskTargets( onlineClients map[uuid.UUID]bool, ) map[string]monitorDesktopTaskTarget { result := make(map[string]monitorDesktopTaskTarget) + targetsByPlatform := make(map[string][]monitorDesktopTaskTarget) for _, candidate := range candidates { platformID := normalizeMonitoringPlatformID(candidate.PlatformID) if platformID == "" || candidate.AccountID == uuid.Nil { continue } - if _, exists := result[platformID]; exists { - continue - } if clientID, ok := accountPresence[candidate.AccountID]; ok && clientID != uuid.Nil && onlineClients[clientID] { - result[platformID] = monitorDesktopTaskTarget{AccountID: candidate.AccountID, ClientID: clientID} + targetsByPlatform[platformID] = append(targetsByPlatform[platformID], monitorDesktopTaskTarget{AccountID: candidate.AccountID, ClientID: clientID}) continue } if candidate.DBClientID != nil && *candidate.DBClientID != uuid.Nil && onlineClients[*candidate.DBClientID] { - result[platformID] = monitorDesktopTaskTarget{AccountID: candidate.AccountID, ClientID: *candidate.DBClientID} + targetsByPlatform[platformID] = append(targetsByPlatform[platformID], monitorDesktopTaskTarget{AccountID: candidate.AccountID, ClientID: *candidate.DBClientID}) } } + for platformID, targets := range targetsByPlatform { + if len(targets) == 0 { + continue + } + index := randomIndex(len(targets)) + result[platformID] = targets[index] + } return result } 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 8f518c7..51a55b9 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go @@ -76,3 +76,31 @@ 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"), + } + + 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]}, + }, + nil, + map[uuid.UUID]bool{ + clientIDs[0]: true, + clientIDs[1]: true, + }, + ) + target := targets["qwen"] + assert.Contains(t, clientIDs, target.ClientID) + assert.Contains(t, accountIDs, target.AccountID) + } +} diff --git a/server/internal/tenant/app/monitoring_service.go b/server/internal/tenant/app/monitoring_service.go index 82a696f..7125c67 100644 --- a/server/internal/tenant/app/monitoring_service.go +++ b/server/internal/tenant/app/monitoring_service.go @@ -3,10 +3,12 @@ package app import ( "context" "crypto/md5" + crand "crypto/rand" "database/sql" "encoding/hex" "encoding/json" "fmt" + "math/big" "sort" "strings" "sync" @@ -115,10 +117,11 @@ type MonitoringDashboardWindow struct { } type MonitoringDashboardRuntime struct { - CurrentUserClientOnline bool `json:"current_user_client_online"` - PlatformAuthorizationStatus string `json:"platform_authorization_status"` - AuthorizedMonitoringPlatformIDs []string `json:"authorized_monitoring_platform_ids"` - AuthorizedMonitoringPlatformCount int `json:"authorized_monitoring_platform_count"` + CurrentUserClientOnline bool `json:"current_user_client_online"` + PlatformAuthorizationStatus string `json:"platform_authorization_status"` + AuthorizedMonitoringPlatformIDs []string `json:"authorized_monitoring_platform_ids"` + AuthorizedMonitoringPlatformCount int `json:"authorized_monitoring_platform_count"` + OnlineAuthorizedMonitoringPlatformIDs []string `json:"online_authorized_monitoring_platform_ids"` } type MonitoringOverview struct { @@ -500,20 +503,21 @@ func (s *MonitoringService) CitationSummary( } func (s *MonitoringService) loadDashboardRuntime(ctx context.Context, actor auth.Actor, workspaceID int64, quota monitoringQuotaConfig) (MonitoringDashboardRuntime, error) { - online := false - if quota.PrimaryClientID != nil { - var err error - online, err = s.isClientOnline(ctx, actor.TenantID, workspaceID, *quota.PrimaryClientID) - if err != nil { - return MonitoringDashboardRuntime{}, err - } + online, err := s.hasOnlineClientForUser(ctx, actor.TenantID, workspaceID, actor.UserID) + if err != nil { + return MonitoringDashboardRuntime{}, err + } + onlinePlatformIDs, err := s.loadOnlineMonitoringPlatformIDsForUser(ctx, actor.TenantID, workspaceID, actor.UserID, quota.EnabledPlatforms) + if err != nil { + return MonitoringDashboardRuntime{}, err } return MonitoringDashboardRuntime{ - CurrentUserClientOnline: online, - PlatformAuthorizationStatus: monitoringPlatformAuthorizationStatus(quota), - AuthorizedMonitoringPlatformIDs: append([]string(nil), quota.EnabledPlatforms...), - AuthorizedMonitoringPlatformCount: len(quota.EnabledPlatforms), + CurrentUserClientOnline: online, + PlatformAuthorizationStatus: monitoringPlatformAuthorizationStatus(quota), + AuthorizedMonitoringPlatformIDs: append([]string(nil), quota.EnabledPlatforms...), + AuthorizedMonitoringPlatformCount: len(quota.EnabledPlatforms), + OnlineAuthorizedMonitoringPlatformIDs: onlinePlatformIDs, }, nil } @@ -680,23 +684,7 @@ func (s *MonitoringService) CollectNow( }, nil } - clientID, err := s.resolveCollectNowClientID(ctx, actor.TenantID, workspaceID, actor.UserID, quota.PrimaryClientID, options.TargetClientID) - if err != nil { - return nil, err - } - if clientID == nil { - return nil, response.ErrConflict(40941, "desktop_client_offline", "current user's monitoring desktop client is offline") - } - - if err := s.ensureClientOnline(ctx, actor.TenantID, workspaceID, *clientID); err != nil { - return nil, err - } - - targetPlatformIDs, err := s.loadMonitoringClientPlatformIDs(ctx, actor.TenantID, workspaceID, *clientID) - if err != nil { - return nil, err - } - platforms := resolveMonitoringPlatforms(targetPlatformIDs) + platforms := resolveMonitoringPlatforms(quota.EnabledPlatforms) platforms, err = intersectMonitoringPlatforms(platforms, options.PlatformIDs) if err != nil { return nil, err @@ -714,11 +702,38 @@ func (s *MonitoringService) CollectNow( } questionIDs := configuredQuestionIDs(configuredQuestions) platformIDs := monitoringPlatformIDs(platforms) + targetClientIDsByPlatform, err := s.loadRandomOnlineMonitoringClientIDsByPlatformForUser(ctx, actor.TenantID, workspaceID, actor.UserID, platformIDs) + if err != nil { + return nil, err + } + platforms = filterMonitoringPlatformsByIDSet(platforms, platformIDsFromClientTargetMap(targetClientIDsByPlatform)) + if len(platforms) == 0 { + return nil, response.ErrConflict(40941, "desktop_client_offline", "current user's monitoring desktop client is offline for selected platforms") + } + platformIDs = monitoringPlatformIDs(platforms) + dispatchTargetClientIDs := collectNowDispatchTargetClientIDs(targetClientIDsByPlatform, nil) executionOwner := "legacy" if s.shouldUseDesktopTasksExecution(actor.TenantID) { executionOwner = "desktop_tasks" } + clientID, err := s.resolveCollectNowClientID( + ctx, + actor.TenantID, + workspaceID, + actor.UserID, + quota.PrimaryClientID, + options.TargetClientID, + platformIDs, + ) + if err != nil { + return nil, err + } + if clientID == nil { + return nil, response.ErrConflict(40941, "desktop_client_offline", "current user's monitoring desktop client is offline") + } + dispatchTargetClientIDs = collectNowDispatchTargetClientIDs(targetClientIDsByPlatform, clientID) + now := time.Now().UTC() todayTime, _ := monitoringBusinessDayAndDateAt(now) requestID := uuid.New() @@ -749,32 +764,30 @@ func (s *MonitoringService) CollectNow( return nil, err } if existingRequest != nil { - if strings.TrimSpace(existingRequest.TargetClientID) == clientID.String() { - base := buildCollectNowResponseFromState(quota.CollectionMode, existingRequest) - executing, execErr := s.hasExecutingCollectNowTask(ctx, existingRequest.RequestID) - if execErr != nil && s.logger != nil { - s.logger.Warn("collect-now executing-state inspection failed", - zap.String("request_id", existingRequest.RequestID.String()), - zap.Error(execErr), - ) - } - if executing { - if base == nil { - base = &MonitoringCollectNowResponse{CollectionMode: quota.CollectionMode} - } - base.Message = "任务正在执行中，请等待" - _ = tx.Rollback(ctx) - return base, nil + base := buildCollectNowResponseFromState(quota.CollectionMode, existingRequest) + executing, execErr := s.hasExecutingCollectNowTask(ctx, existingRequest.RequestID) + if execErr != nil && s.logger != nil { + s.logger.Warn("collect-now executing-state inspection failed", + zap.String("request_id", existingRequest.RequestID.String()), + zap.Error(execErr), + ) + } + if executing { + if base == nil { + base = &MonitoringCollectNowResponse{CollectionMode: quota.CollectionMode} } + base.Message = "任务正在执行中，请等待" + _ = tx.Rollback(ctx) + return base, nil } if _, err := tx.Exec(ctx, ` - UPDATE monitoring_collect_requests - SET superseded_by_request_id = $2, - status = CASE - WHEN status IN ('accepted', 'dispatching') THEN 'superseded' - ELSE status - END, + UPDATE monitoring_collect_requests + SET superseded_by_request_id = $2, + status = CASE + WHEN status IN ('accepted', 'dispatching') THEN 'superseded' + ELSE status + END, updated_at = NOW() WHERE request_id = $1 AND superseded_by_request_id IS NULL @@ -791,7 +804,7 @@ func (s *MonitoringService) CollectNow( return nil, err } - interruptGeneration, err := s.nextCollectNowInterruptGeneration(ctx, tx, *clientID, executionOwner) + interruptGeneration, err := s.nextCollectNowInterruptGeneration(ctx, tx, dispatchTargetClientIDs, executionOwner) if err != nil { return nil, err } @@ -804,6 +817,7 @@ func (s *MonitoringService) CollectNow( configuredQuestions, platforms, todayTime, + targetClientIDsByPlatform, *clientID, interruptGeneration, executionOwner, @@ -823,6 +837,7 @@ func (s *MonitoringService) CollectNow( questionIDs, platformIDs, *clientID, + actor.UserID, ) if err != nil { return nil, err @@ -830,29 +845,33 @@ func (s *MonitoringService) CollectNow( } abortedQueuedCount := int64(0) if options.Preempt { - abortedQueuedCount, err = s.deferQueuedNormalTasks( - ctx, - tx, - actor.TenantID, - todayTime, - brand.ID, - configuredQuestionIDs(configuredQuestions), - monitoringPlatformIDs(platforms), - *clientID, - requestID, - ttlExpiresAt, - ) - if err != nil { - return nil, err + for _, targetClientID := range dispatchTargetClientIDs { + deferredCount, deferErr := s.deferQueuedNormalTasks( + ctx, + tx, + actor.TenantID, + todayTime, + brand.ID, + configuredQuestionIDs(configuredQuestions), + monitoringPlatformIDs(platforms), + targetClientID, + requestID, + ttlExpiresAt, + ) + if deferErr != nil { + return nil, deferErr + } + abortedQueuedCount += deferredCount } } requestScopeJSON, err := json.Marshal(map[string]any{ - "brand_id": brand.ID, - "keyword_id": keywordID, - "question_ids": questionIDs, - "platform_ids": platformIDs, - "preempt": options.Preempt, + "brand_id": brand.ID, + "keyword_id": keywordID, + "question_ids": questionIDs, + "platform_ids": platformIDs, + "target_client_ids": uuidStrings(dispatchTargetClientIDs), + "preempt": options.Preempt, }) if err != nil { return nil, response.ErrInternal(50041, "payload_encode_failed", "failed to encode collect-now request scope") @@ -886,14 +905,14 @@ func (s *MonitoringService) CollectNow( tx, requestID, workspaceID, - *clientID, + dispatchTargetClientIDs, affectedTaskCount, interruptGeneration, func() []int64 { if !options.Preempt { return nil } - return interruptTaskIDs + return collectNowInterruptTaskIDs(interruptTaskIDs) }(), ); err != nil { return nil, err @@ -937,12 +956,38 @@ func (s *MonitoringService) CollectNow( for _, spec := range phase2TaskSpecs { excludedMonitorTaskIDs = append(excludedMonitorTaskIDs, spec.MonitorTaskID) } - phase2DeferredTasks, err = s.deferQueuedPhase2MonitorTasks(ctx, *clientID, excludedMonitorTaskIDs) - if err != nil { - return nil, err + phase2TargetClientIDs := desktopTaskClientIDs(phase2PublishedTasks) + if len(phase2TargetClientIDs) == 0 { + phase2TargetClientIDs = []uuid.UUID{*clientID} + } + for _, phase2TargetClientID := range phase2TargetClientIDs { + deferredTasks, deferErr := s.deferQueuedPhase2MonitorTasks(ctx, phase2TargetClientID, excludedMonitorTaskIDs) + if deferErr != nil { + return nil, deferErr + } + phase2DeferredTasks = append(phase2DeferredTasks, deferredTasks...) + interruptTargets, interruptErr := s.requestPhase2MonitorInterrupts(ctx, phase2TargetClientID, excludedMonitorTaskIDs, interruptGeneration) + if interruptErr != nil { + return nil, interruptErr + } + phase2InterruptTargets = append(phase2InterruptTargets, interruptTargets...) } if len(phase2DeferredTasks) > 0 { abortedQueuedCount += int64(len(phase2DeferredTasks)) + } + if len(phase2InterruptTargets) > 0 { + // Keep collect-now request counters in sync after per-platform dispatch + // fan-out chooses more than one desktop client. + if _, updateErr := s.monitoringPool.Exec(ctx, ` + UPDATE monitoring_collect_requests + SET aborted_queued_count = aborted_queued_count + $2, + interrupt_requested_count = interrupt_requested_count + $3, + updated_at = NOW() + WHERE request_id = $1 + `, requestID, len(phase2DeferredTasks), len(phase2InterruptTargets)); updateErr != nil { + return nil, response.ErrInternal(50041, "request_update_failed", "failed to persist phase2 preempt counts") + } + } else if len(phase2DeferredTasks) > 0 { if _, updateErr := s.monitoringPool.Exec(ctx, ` UPDATE monitoring_collect_requests SET aborted_queued_count = aborted_queued_count + $2, @@ -952,20 +997,6 @@ func (s *MonitoringService) CollectNow( return nil, response.ErrInternal(50041, "request_update_failed", "failed to persist phase2 deferred queue count") } } - phase2InterruptTargets, err = s.requestPhase2MonitorInterrupts(ctx, *clientID, excludedMonitorTaskIDs, interruptGeneration) - if err != nil { - return nil, err - } - if len(phase2InterruptTargets) > 0 { - if _, updateErr := s.monitoringPool.Exec(ctx, ` - UPDATE monitoring_collect_requests - SET interrupt_requested_count = interrupt_requested_count + $2, - updated_at = NOW() - WHERE request_id = $1 - `, requestID, len(phase2InterruptTargets)); updateErr != nil { - return nil, response.ErrInternal(50041, "request_update_failed", "failed to persist phase2 interrupt request count") - } - } } for _, task := range phase2PublishedTasks { @@ -1032,13 +1063,27 @@ func (s *MonitoringService) resolveCollectNowClientID( tenantID, workspaceID, userID int64, defaultClientID *uuid.UUID, preferredClientID *uuid.UUID, + platformIDs []string, ) (*uuid.UUID, error) { if preferredClientID != nil { - if err := s.ensureClientBelongsToUser(ctx, tenantID, workspaceID, userID, *preferredClientID); err != nil { + if err := s.ensureClientBelongsToUser(ctx, tenantID, workspaceID, userID, *preferredClientID, platformIDs); err != nil { return nil, err } return preferredClientID, nil } + clientID, err := s.findRandomOnlineClientForUserWithMonitoringPlatforms(ctx, tenantID, workspaceID, userID, platformIDs) + if err != nil { + return nil, err + } + if clientID != nil { + return clientID, nil + } + if defaultClientID == nil { + return nil, nil + } + if err := s.ensureClientBelongsToUser(ctx, tenantID, workspaceID, userID, *defaultClientID, platformIDs); err != nil { + return nil, err + } return defaultClientID, nil } @@ -1050,25 +1095,106 @@ func (s *MonitoringService) hasOnlineClientForUser(ctx context.Context, tenantID return clientID != nil, nil } +func (s *MonitoringService) findRandomOnlineClientForUser( + ctx context.Context, + tenantID, workspaceID, userID int64, +) (*uuid.UUID, error) { + candidates, err := s.loadOnlineClientIDsForUser(ctx, tenantID, workspaceID, userID) + if err != nil { + return nil, err + } + if len(candidates) == 0 { + return nil, nil + } + return randomUUIDFromSlice(candidates), nil +} + +func (s *MonitoringService) findRandomOnlineClientForUserWithMonitoringPlatforms( + ctx context.Context, + tenantID, workspaceID, userID int64, + platformIDs []string, +) (*uuid.UUID, error) { + targets, err := s.loadRandomOnlineMonitoringClientIDsByPlatformForUser(ctx, tenantID, workspaceID, userID, platformIDs) + if err != nil { + return nil, err + } + return randomUUIDFromSlice(collectNowDispatchTargetClientIDs(targets, nil)), nil +} + +func (s *MonitoringService) loadOnlineMonitoringPlatformIDsForUser( + ctx context.Context, + tenantID, workspaceID, userID int64, + platformIDs []string, +) ([]string, error) { + onlineClientIDs, err := s.loadOnlineClientIDsForUser(ctx, tenantID, workspaceID, userID) + if err != nil { + return nil, err + } + if len(onlineClientIDs) == 0 { + return nil, nil + } + return s.loadMonitoringPlatformIDsForClients(ctx, tenantID, workspaceID, userID, onlineClientIDs, platformIDs) +} + +func (s *MonitoringService) loadRandomOnlineMonitoringClientIDsByPlatformForUser( + ctx context.Context, + tenantID, workspaceID, userID int64, + platformIDs []string, +) (map[string]uuid.UUID, error) { + platformIDs = reconcileEnabledMonitoringPlatforms(platformIDs) + if len(platformIDs) == 0 { + platformIDs = monitoringPlatformIDs(defaultMonitoringPlatforms) + } + onlineClientIDs, err := s.loadOnlineClientIDsForUser(ctx, tenantID, workspaceID, userID) + if err != nil { + return nil, err + } + if len(onlineClientIDs) == 0 { + return nil, nil + } + candidatesByPlatform, err := s.loadOnlineMonitoringClientIDCandidatesByPlatformForUser(ctx, tenantID, workspaceID, userID, onlineClientIDs, platformIDs) + if err != nil { + return nil, err + } + result := make(map[string]uuid.UUID, len(candidatesByPlatform)) + for platformID, candidates := range candidatesByPlatform { + if selected := randomUUIDFromSlice(candidates); selected != nil { + result[platformID] = *selected + } + } + return result, nil +} + func (s *MonitoringService) ensureClientBelongsToUser( ctx context.Context, tenantID, workspaceID, userID int64, clientID uuid.UUID, + platformIDs []string, ) error { var exists bool if err := s.businessPool.QueryRow(ctx, ` SELECT EXISTS ( SELECT 1 - FROM desktop_clients - WHERE id = $1 - AND tenant_id = $2 - AND workspace_id = $3 - AND user_id = $4 - AND revoked_at IS NULL - AND last_seen_at IS NOT NULL - AND last_seen_at >= NOW() - $5::interval + FROM desktop_clients dc + WHERE dc.id = $1 + AND dc.tenant_id = $2 + AND dc.workspace_id = $3 + AND dc.user_id = $4 + AND 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 + WHERE pa.tenant_id = dc.tenant_id + AND pa.workspace_id = dc.workspace_id + AND pa.user_id = dc.user_id + AND pa.client_id = dc.id + AND pa.deleted_at IS NULL + AND pa.platform_id = ANY($6::text[]) + ) ) - `, clientID, tenantID, workspaceID, userID, formatPgInterval(desktopClientPresenceTTL)).Scan(&exists); err != nil { + `, clientID, tenantID, workspaceID, userID, formatPgInterval(desktopClientPresenceTTL), reconcileEnabledMonitoringPlatforms(platformIDs)).Scan(&exists); err != nil { return response.ErrInternal(50041, "query_failed", "failed to validate target monitoring desktop client") } if !exists { @@ -1080,25 +1206,29 @@ func (s *MonitoringService) ensureClientBelongsToUser( func (s *MonitoringService) nextCollectNowInterruptGeneration( ctx context.Context, tx pgx.Tx, - clientID uuid.UUID, + clientIDs []uuid.UUID, executionOwner string, ) (int, error) { var next int + clientIDs = uniqueUUIDs(clientIDs) + if len(clientIDs) == 0 { + return 1, nil + } if strings.TrimSpace(executionOwner) == "desktop_tasks" && s.businessPool != nil { if err := s.businessPool.QueryRow(ctx, ` SELECT COALESCE(MAX(interrupt_generation), 0) + 1 FROM desktop_tasks WHERE kind = 'monitor' - AND target_client_id = $1 - `, clientID).Scan(&next); err != nil { + AND target_client_id = ANY($1::uuid[]) + `, clientIDs).Scan(&next); err != nil { return 0, response.ErrInternal(50041, "query_failed", "failed to allocate collect-now interrupt generation") } } else { if err := tx.QueryRow(ctx, ` SELECT COALESCE(MAX(interrupt_generation), 0) + 1 FROM monitoring_collect_tasks - WHERE target_client_id = $1 - `, clientID).Scan(&next); err != nil { + WHERE target_client_id = ANY($1::uuid[]) + `, clientIDs).Scan(&next); err != nil { return 0, response.ErrInternal(50041, "query_failed", "failed to allocate collect-now interrupt generation") } } @@ -1122,6 +1252,33 @@ func monitoringPlatformIDs(items []monitoringPlatformMetadata) []string { return result } +func filterMonitoringPlatformsByIDSet(items []monitoringPlatformMetadata, platformIDs []string) []monitoringPlatformMetadata { + if len(items) == 0 || len(platformIDs) == 0 { + return nil + } + enabled := make(map[string]struct{}, len(platformIDs)) + for _, platformID := range platformIDs { + platformID = normalizeMonitoringPlatformID(platformID) + if platformID == "" { + continue + } + enabled[platformID] = struct{}{} + } + if len(enabled) == 0 { + return nil + } + result := make([]monitoringPlatformMetadata, 0, len(items)) + for _, item := range items { + platformID := normalizeMonitoringPlatformID(item.ID) + if _, ok := enabled[platformID]; !ok { + continue + } + item.ID = platformID + result = append(result, item) + } + return result +} + func defaultMonitoringPlatformMetadata() []monitoringPlatformMetadata { return append([]monitoringPlatformMetadata(nil), defaultMonitoringPlatforms...) } @@ -1402,6 +1559,7 @@ func (s *MonitoringService) ensureCollectNowTasks( questions []monitoringConfiguredQuestion, platforms []monitoringPlatformMetadata, businessDate time.Time, + targetClientIDsByPlatform map[string]uuid.UUID, targetClientID uuid.UUID, interruptGeneration int, executionOwner string, @@ -1418,6 +1576,14 @@ func (s *MonitoringService) ensureCollectNowTasks( for _, question := range questions { for _, platform := range platforms { + platformID := normalizeMonitoringPlatformID(platform.ID) + if platformID == "" { + continue + } + platformTargetClientID, ok := targetClientIDsByPlatform[platformID] + if !ok || platformTargetClientID == uuid.Nil { + platformTargetClientID = targetClientID + } tag, err := tx.Exec(ctx, ` UPDATE monitoring_collect_tasks SET question_hash = $7, @@ -1454,7 +1620,7 @@ func (s *MonitoringService) ensureCollectNowTasks( AND run_mode = $6 AND business_date = $8::date AND status IN ('pending', 'expired', 'failed', 'completed', 'skipped') - `, tenantID, brandID, question.ID, platform.ID, monitoringCollectorType, runMode, question.QuestionHash, dateText, targetClientID, interruptGeneration, executionOwner) + `, tenantID, brandID, question.ID, platformID, monitoringCollectorType, runMode, question.QuestionHash, dateText, platformTargetClientID, interruptGeneration, executionOwner) if err != nil { return 0, 0, 0, 0, nil, response.ErrInternal(50041, "update_failed", "failed to refresh monitoring tasks") } @@ -1487,7 +1653,7 @@ func (s *MonitoringService) ensureCollectNowTasks( AND status = 'leased' AND COALESCE(execution_owner, 'legacy') = 'legacy' RETURNING id - `, tenantID, brandID, question.ID, platform.ID, monitoringCollectorType, runMode, question.QuestionHash, dateText, targetClientID, interruptGeneration) + `, tenantID, brandID, question.ID, platformID, monitoringCollectorType, runMode, question.QuestionHash, dateText, platformTargetClientID, interruptGeneration) if err != nil { return 0, 0, 0, 0, nil, response.ErrInternal(50041, "update_failed", "failed to promote leased monitoring tasks") } @@ -1522,7 +1688,7 @@ func (s *MonitoringService) ensureCollectNowTasks( tenant_id, brand_id, question_id, ai_platform_id, collector_type, run_mode, business_date ) DO NOTHING - `, tenantID, brandID, question.ID, question.QuestionHash, platform.ID, monitoringCollectorType, runMode, dateText, targetClientID, interruptGeneration, fmt.Sprintf("%d:%d:%s", tenantID, brandID, dateText), executionOwner) + `, tenantID, brandID, question.ID, question.QuestionHash, platformID, monitoringCollectorType, runMode, dateText, platformTargetClientID, interruptGeneration, fmt.Sprintf("%d:%d:%s", tenantID, brandID, dateText), executionOwner) if err != nil { return 0, 0, 0, 0, nil, response.ErrInternal(50041, "insert_failed", "failed to create monitoring tasks") } @@ -1594,7 +1760,13 @@ func (s *MonitoringService) loadQuota(ctx context.Context, actor auth.Actor, wor CollectionMode: monitoringCollectorType, } - clientID, err := s.findLatestRegisteredClientWithMonitoringAccountForUser(ctx, actor.TenantID, workspaceID, actor.UserID) + clientID, err := s.findLatestOnlineClientWithMonitoringAccountForUser(ctx, actor.TenantID, workspaceID, actor.UserID) + if err != nil { + return cfg, err + } + if clientID == nil { + clientID, err = s.findLatestRegisteredClientWithMonitoringAccountForUser(ctx, actor.TenantID, workspaceID, actor.UserID) + } if err != nil { return cfg, err } @@ -1608,7 +1780,7 @@ func (s *MonitoringService) loadQuota(ctx context.Context, actor auth.Actor, wor if clientID == nil { return cfg, nil } - platforms, err := s.loadMonitoringClientPlatformIDs(ctx, actor.TenantID, workspaceID, *clientID) + platforms, err := s.loadMonitoringPlatformIDsForUser(ctx, actor.TenantID, workspaceID, actor.UserID) if err != nil { return cfg, err } @@ -1616,6 +1788,20 @@ func (s *MonitoringService) loadQuota(ctx context.Context, actor auth.Actor, wor return cfg, nil } +// findLatestOnlineClientWithMonitoringAccountForUser returns an online client for +// dashboard runtime defaults. Collect-now may still randomize among all online +// clients that can execute the requested platforms. +func (s *MonitoringService) findLatestOnlineClientWithMonitoringAccountForUser(ctx context.Context, tenantID, workspaceID, userID int64) (*uuid.UUID, error) { + clientIDs, err := s.loadOnlineClientIDsForUser(ctx, tenantID, workspaceID, userID) + if err != nil { + return nil, err + } + if len(clientIDs) == 0 { + return nil, nil + } + return &clientIDs[0], nil +} + // findLatestRegisteredClientWithMonitoringAccountForUser returns the newest // non-revoked desktop client that has at least one bound monitoring platform // account. Platform account health is not authorization. @@ -1673,6 +1859,41 @@ func (s *MonitoringService) findLatestRegisteredClientForUser(ctx context.Contex return &clientID, nil } +func (s *MonitoringService) loadMonitoringPlatformIDsForUser(ctx context.Context, tenantID, workspaceID, userID int64) ([]string, error) { + rows, err := s.businessPool.Query(ctx, ` + SELECT DISTINCT pa.platform_id + FROM platform_accounts pa + JOIN desktop_clients dc + ON dc.id = pa.client_id + AND dc.tenant_id = pa.tenant_id + AND dc.workspace_id = pa.workspace_id + WHERE pa.tenant_id = $1 + AND pa.workspace_id = $2 + AND pa.user_id = $3 + AND pa.deleted_at IS NULL + AND pa.platform_id = ANY($4::text[]) + AND dc.user_id = $3 + AND dc.revoked_at IS NULL + `, tenantID, workspaceID, userID, monitoringPlatformIDs(defaultMonitoringPlatforms)) + if err != nil { + return nil, response.ErrInternal(50041, "query_failed", "failed to load monitoring desktop platform accounts") + } + defer rows.Close() + + platformIDs := make([]string, 0) + for rows.Next() { + var platformID string + if scanErr := rows.Scan(&platformID); scanErr != nil { + return nil, response.ErrInternal(50041, "scan_failed", "failed to parse monitoring desktop platform accounts") + } + platformIDs = append(platformIDs, platformID) + } + if err := rows.Err(); err != nil { + return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate monitoring desktop platform accounts") + } + return reconcileEnabledMonitoringPlatforms(platformIDs), nil +} + func (s *MonitoringService) loadMonitoringClientPlatformIDs(ctx context.Context, tenantID, workspaceID int64, clientID uuid.UUID) ([]string, error) { rows, err := s.businessPool.Query(ctx, ` SELECT DISTINCT platform_id @@ -1702,6 +1923,230 @@ func (s *MonitoringService) loadMonitoringClientPlatformIDs(ctx context.Context, return reconcileEnabledMonitoringPlatforms(platformIDs), nil } +func (s *MonitoringService) loadMonitoringPlatformIDsForClients( + ctx context.Context, + tenantID, workspaceID, userID int64, + clientIDs []uuid.UUID, + platformIDs []string, +) ([]string, error) { + clientIDs = uniqueUUIDs(clientIDs) + if len(clientIDs) == 0 { + return nil, nil + } + platformIDs = reconcileEnabledMonitoringPlatforms(platformIDs) + if len(platformIDs) == 0 { + platformIDs = monitoringPlatformIDs(defaultMonitoringPlatforms) + } + + rows, err := s.businessPool.Query(ctx, ` + SELECT DISTINCT platform_id + FROM platform_accounts + WHERE tenant_id = $1 + AND workspace_id = $2 + AND user_id = $3 + AND client_id = ANY($4::uuid[]) + AND deleted_at IS NULL + AND platform_id = ANY($5::text[]) + `, tenantID, workspaceID, userID, clientIDs, platformIDs) + if err != nil { + return nil, response.ErrInternal(50041, "query_failed", "failed to load online monitoring desktop platform accounts") + } + defer rows.Close() + + result := make([]string, 0) + for rows.Next() { + var platformID string + if scanErr := rows.Scan(&platformID); scanErr != nil { + return nil, response.ErrInternal(50041, "scan_failed", "failed to parse online monitoring desktop platform accounts") + } + result = append(result, platformID) + } + if err := rows.Err(); err != nil { + return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate online monitoring desktop platform accounts") + } + return reconcileEnabledMonitoringPlatforms(result), nil +} + +func (s *MonitoringService) loadOnlineMonitoringClientIDCandidatesByPlatformForUser( + ctx context.Context, + tenantID, workspaceID, userID int64, + clientIDs []uuid.UUID, + platformIDs []string, +) (map[string][]uuid.UUID, error) { + clientIDs = uniqueUUIDs(clientIDs) + if len(clientIDs) == 0 { + return nil, nil + } + platformIDs = reconcileEnabledMonitoringPlatforms(platformIDs) + if len(platformIDs) == 0 { + platformIDs = monitoringPlatformIDs(defaultMonitoringPlatforms) + } + + rows, err := s.businessPool.Query(ctx, ` + SELECT DISTINCT platform_id, client_id + FROM platform_accounts + WHERE tenant_id = $1 + AND workspace_id = $2 + AND user_id = $3 + AND client_id = ANY($4::uuid[]) + AND deleted_at IS NULL + AND platform_id = ANY($5::text[]) + `, tenantID, workspaceID, userID, clientIDs, platformIDs) + if err != nil { + return nil, response.ErrInternal(50041, "query_failed", "failed to load online monitoring desktop platform targets") + } + defer rows.Close() + + result := make(map[string][]uuid.UUID) + for rows.Next() { + var ( + platformID string + clientID uuid.UUID + ) + if scanErr := rows.Scan(&platformID, &clientID); scanErr != nil { + return nil, response.ErrInternal(50041, "scan_failed", "failed to parse online monitoring desktop platform targets") + } + platformID = normalizeMonitoringPlatformID(platformID) + if platformID == "" || clientID == uuid.Nil { + continue + } + result[platformID] = append(result[platformID], clientID) + } + if err := rows.Err(); err != nil { + return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate online monitoring desktop platform targets") + } + for platformID, candidates := range result { + result[platformID] = uniqueUUIDs(candidates) + } + return result, nil +} + +func (s *MonitoringService) loadOnlineClientIDsForUser( + ctx context.Context, + tenantID, workspaceID, userID int64, +) ([]uuid.UUID, error) { + rows, err := s.businessPool.Query(ctx, ` + SELECT dc.id + FROM desktop_clients dc + WHERE dc.tenant_id = $1 + AND dc.workspace_id = $2 + AND dc.user_id = $3 + AND dc.revoked_at IS NULL + AND dc.last_seen_at IS NOT NULL + AND dc.last_seen_at >= NOW() - $4::interval + ORDER BY dc.last_seen_at DESC, dc.created_at DESC, dc.id DESC + `, tenantID, workspaceID, userID, formatPgInterval(desktopClientPresenceTTL)) + if err != nil { + return nil, response.ErrInternal(50041, "query_failed", "failed to inspect online monitoring desktop clients") + } + defer rows.Close() + + clientIDs := make([]uuid.UUID, 0) + for rows.Next() { + var clientID uuid.UUID + if scanErr := rows.Scan(&clientID); scanErr != nil { + return nil, response.ErrInternal(50041, "scan_failed", "failed to parse online monitoring desktop clients") + } + clientIDs = append(clientIDs, clientID) + } + if err := rows.Err(); err != nil { + return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate online monitoring desktop clients") + } + return clientIDs, nil +} + +func randomUUIDFromSlice(values []uuid.UUID) *uuid.UUID { + if len(values) == 0 { + return nil + } + selected := values[randomIndex(len(values))] + return &selected +} + +func randomIndex(length int) int { + if length <= 1 { + return 0 + } + if value, err := crand.Int(crand.Reader, big.NewInt(int64(length))); err == nil { + return int(value.Int64()) + } + return int(time.Now().UnixNano() % int64(length)) +} + +func platformIDsFromClientTargetMap(targets map[string]uuid.UUID) []string { + if len(targets) == 0 { + return nil + } + result := make([]string, 0, len(targets)) + for _, platform := range defaultMonitoringPlatforms { + if clientID, ok := targets[platform.ID]; ok && clientID != uuid.Nil { + result = append(result, platform.ID) + } + } + return result +} + +func collectNowDispatchTargetClientIDs(targets map[string]uuid.UUID, fallback *uuid.UUID) []uuid.UUID { + result := make([]uuid.UUID, 0, len(targets)+1) + for _, platform := range defaultMonitoringPlatforms { + if clientID, ok := targets[platform.ID]; ok && clientID != uuid.Nil { + result = append(result, clientID) + } + } + if fallback != nil && *fallback != uuid.Nil { + result = append(result, *fallback) + } + return uniqueUUIDs(result) +} + +func uuidStrings(values []uuid.UUID) []string { + values = uniqueUUIDs(values) + if len(values) == 0 { + return nil + } + result := make([]string, 0, len(values)) + for _, value := range values { + if value == uuid.Nil { + continue + } + result = append(result, value.String()) + } + return result +} + +func collectNowInterruptTaskIDs(taskIDs []int64) []int64 { + if len(taskIDs) == 0 { + return nil + } + seen := make(map[int64]struct{}, len(taskIDs)) + result := make([]int64, 0, len(taskIDs)) + for _, taskID := range taskIDs { + if taskID <= 0 { + continue + } + if _, ok := seen[taskID]; ok { + continue + } + seen[taskID] = struct{}{} + result = append(result, taskID) + } + return result +} + +func desktopTaskClientIDs(tasks []*repository.DesktopTask) []uuid.UUID { + if len(tasks) == 0 { + return nil + } + result := make([]uuid.UUID, 0, len(tasks)) + for _, task := range tasks { + if task == nil || task.TargetClientID == uuid.Nil { + continue + } + result = append(result, task.TargetClientID) + } + return uniqueUUIDs(result) +} + func newMonitoringDerivedMetrics() monitoringDerivedMetrics { return monitoringDerivedMetrics{ ByDate: make(map[string]monitoringDerivedRateStats), diff --git a/server/internal/tenant/app/monitoring_service_date_test.go b/server/internal/tenant/app/monitoring_service_date_test.go index 475a9f3..ecb9349 100644 --- a/server/internal/tenant/app/monitoring_service_date_test.go +++ b/server/internal/tenant/app/monitoring_service_date_test.go @@ -132,6 +132,27 @@ func TestReconcileEnabledMonitoringPlatformsKeepsAuthorizedSubset(t *testing.T) assert.Empty(t, resolveMonitoringPlatforms(nil)) } +func TestFilterMonitoringPlatformsByIDSetKeepsCatalogOrder(t *testing.T) { + platforms := defaultMonitoringPlatformMetadata() + + filtered := filterMonitoringPlatformsByIDSet(platforms, []string{"qwen", "yuanbao"}) + + assert.Equal(t, []string{"yuanbao", "qwen"}, monitoringPlatformIDs(filtered)) +} + +func TestCollectNowDispatchTargetClientIDsDeduplicatesByPlatformOrder(t *testing.T) { + sharedClientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000501") + qwenClientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000502") + + targets := collectNowDispatchTargetClientIDs(map[string]uuid.UUID{ + "yuanbao": sharedClientID, + "qwen": qwenClientID, + "doubao": sharedClientID, + }, &sharedClientID) + + assert.Equal(t, []uuid.UUID{sharedClientID, qwenClientID}, targets) +} + func TestMonitoringPlatformAuthorizationStatus(t *testing.T) { clientID := uuid.New()