test(desktop): cover deepseek account name extraction for wechat + phone logins
Desktop Client Build / Resolve Build Metadata (push) Successful in 28s
Frontend CI / Frontend (push) Successful in 4m5s
Backend CI / Backend (push) Successful in 17m7s
Desktop Client Build / Publish Client Artifacts to NAS (push) Has been cancelled
Desktop Client Build / Build Desktop Client (push) Has been cancelled
Desktop Client Build / Resolve Build Metadata (push) Successful in 28s
Frontend CI / Frontend (push) Successful in 4m5s
Backend CI / Backend (push) Successful in 17m7s
Desktop Client Build / Publish Client Artifacts to NAS (push) Has been cancelled
Desktop Client Build / Build Desktop Client (push) Has been cancelled
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) <noreply@anthropic.com>
This commit is contained in:
@@ -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")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user