From 051976e4a983d883d6bc243d8c43b198eadf9b85 Mon Sep 17 00:00:00 2001 From: liangxu Date: Mon, 27 Apr 2026 21:33:36 +0800 Subject: [PATCH] feat(server/desktop): ingest desktop account health reports Add a POST /desktop/accounts/health-reports endpoint that buffers reports in Redis, dedupes via signature, and publishes change events to a new desktop.account.health RabbitMQ queue. A sink worker drains the queue, and the account service overlays runtime health onto database health when serving desktop accounts. Publish job creation now consults runtime health so a stale DB row no longer blocks publishing once the desktop client reports it healthy. Co-Authored-By: Claude Opus 4.7 (1M context) --- server/cmd/tenant-api/main.go | 1 + server/configs/config.local.yaml.example | 6 + server/configs/config.yaml | 6 + server/internal/shared/config/config.go | 112 ++++--- .../shared/messaging/rabbitmq/client.go | 20 ++ .../app/desktop_account_health_sink_worker.go | 211 ++++++++++++ .../tenant/app/desktop_account_service.go | 302 ++++++++++++++++++ .../tenant/app/publish_job_service.go | 23 +- .../transport/desktop_account_handler.go | 16 + server/internal/tenant/transport/router.go | 1 + 10 files changed, 647 insertions(+), 51 deletions(-) create mode 100644 server/internal/tenant/app/desktop_account_health_sink_worker.go diff --git a/server/cmd/tenant-api/main.go b/server/cmd/tenant-api/main.go index dbdc0a3..454ee0c 100644 --- a/server/cmd/tenant-api/main.go +++ b/server/cmd/tenant-api/main.go @@ -37,6 +37,7 @@ func main() { tenantapp.NewMonitoringCollectOutboxWorker(monitoringService, app.Logger).Start(workerCtx) tenantapp.NewMonitoringResultIngestWorker(monitoringCallbackService, app.RabbitMQ, app.Logger, app.Config.MonitoringWorkers).Start(workerCtx) tenantapp.NewMonitoringProjectionRebuildWorker(monitoringCallbackService, app.RabbitMQ, app.Logger, app.Config.MonitoringWorkers).Start(workerCtx) + tenantapp.NewDesktopAccountHealthSinkWorker(app.DB, app.RabbitMQ, app.Logger).Start(workerCtx) imageService := tenantapp.NewImageService(app.DB, app.ObjectStorage, app.RabbitMQ, app.Logger) tenantapp.NewImageAssetWorker(imageService, app.RabbitMQ, app.Logger).Start(workerCtx) diff --git a/server/configs/config.local.yaml.example b/server/configs/config.local.yaml.example index 9177c4e..d5dda19 100644 --- a/server/configs/config.local.yaml.example +++ b/server/configs/config.local.yaml.example @@ -37,6 +37,12 @@ rabbitmq: generation_dlq: generation.task.run.dlq generation_dlq_routing_key: generation.task.run.dlq generation_event_exchange: generation.event + desktop_account_health_exchange: desktop.account.health + desktop_account_health_routing_key: desktop.account.health.report + desktop_account_health_queue: desktop.account.health.report + desktop_account_health_dlx: desktop.account.health.dlx + desktop_account_health_dlq: desktop.account.health.report.dlq + desktop_account_health_dlq_routing_key: desktop.account.health.report.dlq template_assist_exchange: template.assist template_assist_routing_key: template.assist.run template_assist_queue: template.assist.run diff --git a/server/configs/config.yaml b/server/configs/config.yaml index 91bfd59..2017a29 100644 --- a/server/configs/config.yaml +++ b/server/configs/config.yaml @@ -47,6 +47,12 @@ rabbitmq: desktop_dispatch_exchange: desktop.task.dispatch desktop_dispatch_publish_prefix: publish desktop_dispatch_monitor_prefix: monitor + desktop_account_health_exchange: desktop.account.health + desktop_account_health_routing_key: desktop.account.health.report + desktop_account_health_queue: desktop.account.health.report + desktop_account_health_dlx: desktop.account.health.dlx + desktop_account_health_dlq: desktop.account.health.report.dlq + desktop_account_health_dlq_routing_key: desktop.account.health.report.dlq template_assist_exchange: template.assist template_assist_routing_key: template.assist.run template_assist_queue: template.assist.run diff --git a/server/internal/shared/config/config.go b/server/internal/shared/config/config.go index 7fd9ae5..7642c45 100644 --- a/server/internal/shared/config/config.go +++ b/server/internal/shared/config/config.go @@ -58,50 +58,56 @@ type RedisConfig struct { } type RabbitMQConfig struct { - URL string `mapstructure:"url"` - MonitorResultExchange string `mapstructure:"monitor_result_exchange"` - MonitorResultRoutingKey string `mapstructure:"monitor_result_routing_key"` - MonitorResultQueue string `mapstructure:"monitor_result_queue"` - MonitorResultDLX string `mapstructure:"monitor_result_dlx"` - MonitorResultDLQ string `mapstructure:"monitor_result_dlq"` - MonitorResultDLQRouteKey string `mapstructure:"monitor_result_dlq_routing_key"` - MonitorProjectionExchange string `mapstructure:"monitor_projection_exchange"` - MonitorProjectionRoutingKey string `mapstructure:"monitor_projection_routing_key"` - MonitorProjectionQueue string `mapstructure:"monitor_projection_queue"` - MonitorProjectionDLX string `mapstructure:"monitor_projection_dlx"` - MonitorProjectionDLQ string `mapstructure:"monitor_projection_dlq"` - MonitorProjectionDLQRouteKey string `mapstructure:"monitor_projection_dlq_routing_key"` - GenerationExchange string `mapstructure:"generation_exchange"` - GenerationRoutingKey string `mapstructure:"generation_routing_key"` - GenerationQueue string `mapstructure:"generation_queue"` - GenerationDLX string `mapstructure:"generation_dlx"` - GenerationDLQ string `mapstructure:"generation_dlq"` - GenerationDLQRouteKey string `mapstructure:"generation_dlq_routing_key"` - GenerationEventExchange string `mapstructure:"generation_event_exchange"` - DesktopTaskEventExchange string `mapstructure:"desktop_task_event_exchange"` - DesktopDispatchExchange string `mapstructure:"desktop_dispatch_exchange"` - DesktopDispatchPublishPrefix string `mapstructure:"desktop_dispatch_publish_prefix"` - DesktopDispatchMonitorPrefix string `mapstructure:"desktop_dispatch_monitor_prefix"` - TemplateAssistExchange string `mapstructure:"template_assist_exchange"` - TemplateAssistRoutingKey string `mapstructure:"template_assist_routing_key"` - TemplateAssistQueue string `mapstructure:"template_assist_queue"` - TemplateAssistDLX string `mapstructure:"template_assist_dlx"` - TemplateAssistDLQ string `mapstructure:"template_assist_dlq"` - TemplateAssistDLQRouteKey string `mapstructure:"template_assist_dlq_routing_key"` - KolAssistExchange string `mapstructure:"kol_assist_exchange"` - KolAssistRoutingKey string `mapstructure:"kol_assist_routing_key"` - KolAssistQueue string `mapstructure:"kol_assist_queue"` - KolAssistDLX string `mapstructure:"kol_assist_dlx"` - KolAssistDLQ string `mapstructure:"kol_assist_dlq"` - KolAssistDLQRouteKey string `mapstructure:"kol_assist_dlq_routing_key"` - ImageAssetExchange string `mapstructure:"image_asset_exchange"` - ImageAssetRoutingKey string `mapstructure:"image_asset_routing_key"` - ImageAssetQueue string `mapstructure:"image_asset_queue"` - ImageAssetDLX string `mapstructure:"image_asset_dlx"` - ImageAssetDLQ string `mapstructure:"image_asset_dlq"` - ImageAssetDLQRouteKey string `mapstructure:"image_asset_dlq_routing_key"` - ConsumerPrefetch int `mapstructure:"consumer_prefetch"` - PublishChannelPoolSize int `mapstructure:"publish_channel_pool_size"` + URL string `mapstructure:"url"` + MonitorResultExchange string `mapstructure:"monitor_result_exchange"` + MonitorResultRoutingKey string `mapstructure:"monitor_result_routing_key"` + MonitorResultQueue string `mapstructure:"monitor_result_queue"` + MonitorResultDLX string `mapstructure:"monitor_result_dlx"` + MonitorResultDLQ string `mapstructure:"monitor_result_dlq"` + MonitorResultDLQRouteKey string `mapstructure:"monitor_result_dlq_routing_key"` + MonitorProjectionExchange string `mapstructure:"monitor_projection_exchange"` + MonitorProjectionRoutingKey string `mapstructure:"monitor_projection_routing_key"` + MonitorProjectionQueue string `mapstructure:"monitor_projection_queue"` + MonitorProjectionDLX string `mapstructure:"monitor_projection_dlx"` + MonitorProjectionDLQ string `mapstructure:"monitor_projection_dlq"` + MonitorProjectionDLQRouteKey string `mapstructure:"monitor_projection_dlq_routing_key"` + GenerationExchange string `mapstructure:"generation_exchange"` + GenerationRoutingKey string `mapstructure:"generation_routing_key"` + GenerationQueue string `mapstructure:"generation_queue"` + GenerationDLX string `mapstructure:"generation_dlx"` + GenerationDLQ string `mapstructure:"generation_dlq"` + GenerationDLQRouteKey string `mapstructure:"generation_dlq_routing_key"` + GenerationEventExchange string `mapstructure:"generation_event_exchange"` + DesktopTaskEventExchange string `mapstructure:"desktop_task_event_exchange"` + DesktopDispatchExchange string `mapstructure:"desktop_dispatch_exchange"` + DesktopDispatchPublishPrefix string `mapstructure:"desktop_dispatch_publish_prefix"` + DesktopDispatchMonitorPrefix string `mapstructure:"desktop_dispatch_monitor_prefix"` + DesktopAccountHealthExchange string `mapstructure:"desktop_account_health_exchange"` + DesktopAccountHealthRoutingKey string `mapstructure:"desktop_account_health_routing_key"` + DesktopAccountHealthQueue string `mapstructure:"desktop_account_health_queue"` + DesktopAccountHealthDLX string `mapstructure:"desktop_account_health_dlx"` + DesktopAccountHealthDLQ string `mapstructure:"desktop_account_health_dlq"` + DesktopAccountHealthDLQRouteKey string `mapstructure:"desktop_account_health_dlq_routing_key"` + TemplateAssistExchange string `mapstructure:"template_assist_exchange"` + TemplateAssistRoutingKey string `mapstructure:"template_assist_routing_key"` + TemplateAssistQueue string `mapstructure:"template_assist_queue"` + TemplateAssistDLX string `mapstructure:"template_assist_dlx"` + TemplateAssistDLQ string `mapstructure:"template_assist_dlq"` + TemplateAssistDLQRouteKey string `mapstructure:"template_assist_dlq_routing_key"` + KolAssistExchange string `mapstructure:"kol_assist_exchange"` + KolAssistRoutingKey string `mapstructure:"kol_assist_routing_key"` + KolAssistQueue string `mapstructure:"kol_assist_queue"` + KolAssistDLX string `mapstructure:"kol_assist_dlx"` + KolAssistDLQ string `mapstructure:"kol_assist_dlq"` + KolAssistDLQRouteKey string `mapstructure:"kol_assist_dlq_routing_key"` + ImageAssetExchange string `mapstructure:"image_asset_exchange"` + ImageAssetRoutingKey string `mapstructure:"image_asset_routing_key"` + ImageAssetQueue string `mapstructure:"image_asset_queue"` + ImageAssetDLX string `mapstructure:"image_asset_dlx"` + ImageAssetDLQ string `mapstructure:"image_asset_dlq"` + ImageAssetDLQRouteKey string `mapstructure:"image_asset_dlq_routing_key"` + ConsumerPrefetch int `mapstructure:"consumer_prefetch"` + PublishChannelPoolSize int `mapstructure:"publish_channel_pool_size"` } type SchedulerConfig struct { @@ -523,6 +529,24 @@ func normalizeRabbitMQConfig(cfg *RabbitMQConfig) { if strings.TrimSpace(cfg.DesktopDispatchMonitorPrefix) == "" { cfg.DesktopDispatchMonitorPrefix = "monitor" } + if strings.TrimSpace(cfg.DesktopAccountHealthExchange) == "" { + cfg.DesktopAccountHealthExchange = "desktop.account.health" + } + if strings.TrimSpace(cfg.DesktopAccountHealthRoutingKey) == "" { + cfg.DesktopAccountHealthRoutingKey = "desktop.account.health.report" + } + if strings.TrimSpace(cfg.DesktopAccountHealthQueue) == "" { + cfg.DesktopAccountHealthQueue = "desktop.account.health.report" + } + if strings.TrimSpace(cfg.DesktopAccountHealthDLX) == "" { + cfg.DesktopAccountHealthDLX = "desktop.account.health.dlx" + } + if strings.TrimSpace(cfg.DesktopAccountHealthDLQ) == "" { + cfg.DesktopAccountHealthDLQ = "desktop.account.health.report.dlq" + } + if strings.TrimSpace(cfg.DesktopAccountHealthDLQRouteKey) == "" { + cfg.DesktopAccountHealthDLQRouteKey = "desktop.account.health.report.dlq" + } if strings.TrimSpace(cfg.TemplateAssistExchange) == "" { cfg.TemplateAssistExchange = "template.assist" } diff --git a/server/internal/shared/messaging/rabbitmq/client.go b/server/internal/shared/messaging/rabbitmq/client.go index 36484fb..fea21c2 100644 --- a/server/internal/shared/messaging/rabbitmq/client.go +++ b/server/internal/shared/messaging/rabbitmq/client.go @@ -116,6 +116,10 @@ func (c *Client) PublishDesktopDispatch(ctx context.Context, routingKey string, return c.publish(ctx, c.cfg.DesktopDispatchExchange, routingKey, body) } +func (c *Client) PublishDesktopAccountHealth(ctx context.Context, body []byte) error { + return c.publish(ctx, c.cfg.DesktopAccountHealthExchange, c.cfg.DesktopAccountHealthRoutingKey, body) +} + // ConsumeDesktopDispatch declares a transient per-process queue bound to the // desktop dispatch topic exchange using bindingKeys, and begins consuming from // it. Every tenant-api instance gets its own queue so every instance sees every @@ -283,6 +287,10 @@ func (c *Client) ConsumeDesktopTaskEvents(consumerName string) (<-chan amqp.Deli return c.consumeFanout(consumerName, c.cfg.DesktopTaskEventExchange) } +func (c *Client) ConsumeDesktopAccountHealth(consumerName string) (<-chan amqp.Delivery, *amqp.Channel, error) { + return c.consume(consumerName, c.cfg.DesktopAccountHealthQueue) +} + func (c *Client) ConsumeTemplateAssistTask(consumerName string) (<-chan amqp.Delivery, *amqp.Channel, error) { return c.consume(consumerName, c.cfg.TemplateAssistQueue) } @@ -685,6 +693,18 @@ func (c *Client) ensureTopology(conn *amqp.Connection) error { return fmt.Errorf("declare rabbitmq exchange %s: %w", c.cfg.DesktopDispatchExchange, err) } + if err := c.ensureWorkQueue( + ch, + c.cfg.DesktopAccountHealthExchange, + c.cfg.DesktopAccountHealthRoutingKey, + c.cfg.DesktopAccountHealthQueue, + c.cfg.DesktopAccountHealthDLX, + c.cfg.DesktopAccountHealthDLQ, + c.cfg.DesktopAccountHealthDLQRouteKey, + ); err != nil { + return err + } + if err := c.ensureWorkQueue( ch, c.cfg.TemplateAssistExchange, diff --git a/server/internal/tenant/app/desktop_account_health_sink_worker.go b/server/internal/tenant/app/desktop_account_health_sink_worker.go new file mode 100644 index 0000000..7fca65e --- /dev/null +++ b/server/internal/tenant/app/desktop_account_health_sink_worker.go @@ -0,0 +1,211 @@ +package app + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "strings" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" + amqp "github.com/rabbitmq/amqp091-go" + "go.uber.org/zap" + + "github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq" +) + +const ( + defaultDesktopAccountHealthSinkRetryInterval = 5 * time.Second + defaultDesktopAccountHealthSinkProcessTimeout = 30 * time.Second +) + +var errDesktopAccountHealthConsumerClosed = errors.New("desktop account health consumer channel closed") + +type DesktopAccountHealthSinkWorker struct { + db *pgxpool.Pool + rabbitMQ *rabbitmq.Client + logger *zap.Logger + retryInterval time.Duration + processTimeout time.Duration + consumerPrefix string + concurrency int +} + +func NewDesktopAccountHealthSinkWorker(db *pgxpool.Pool, rabbitMQClient *rabbitmq.Client, logger *zap.Logger) *DesktopAccountHealthSinkWorker { + return &DesktopAccountHealthSinkWorker{ + db: db, + rabbitMQ: rabbitMQClient, + logger: logger, + retryInterval: defaultDesktopAccountHealthSinkRetryInterval, + processTimeout: defaultDesktopAccountHealthSinkProcessTimeout, + consumerPrefix: fmt.Sprintf("tenant-api-account-health-%d", os.Getpid()), + concurrency: 1, + } +} + +func (w *DesktopAccountHealthSinkWorker) Start(ctx context.Context) { + if w == nil { + return + } + for i := 0; i < w.concurrency; i++ { + consumerName := fmt.Sprintf("%s-%d", w.consumerPrefix, i+1) + go w.run(ctx, consumerName) + } +} + +func (w *DesktopAccountHealthSinkWorker) run(ctx context.Context, consumerName string) { + if w.db == nil || w.rabbitMQ == nil { + return + } + + for { + if err := w.consumeOnce(ctx, consumerName); err != nil && ctx.Err() == nil { + if w.logger != nil { + if errors.Is(err, errDesktopAccountHealthConsumerClosed) { + w.logger.Info("desktop account health consumer channel closed, retrying", + zap.String("consumer", consumerName), + ) + } else { + w.logger.Warn("desktop account health sink worker stopped unexpectedly", + zap.Error(err), + zap.String("consumer", consumerName), + ) + } + } + } + + select { + case <-ctx.Done(): + return + case <-time.After(w.retryInterval): + } + } +} + +func (w *DesktopAccountHealthSinkWorker) consumeOnce(ctx context.Context, consumerName string) error { + deliveries, ch, err := w.rabbitMQ.ConsumeDesktopAccountHealth(consumerName) + if err != nil { + return err + } + defer ch.Close() + + for { + select { + case <-ctx.Done(): + return nil + case delivery, ok := <-deliveries: + if !ok { + return errDesktopAccountHealthConsumerClosed + } + w.handleDelivery(ctx, delivery) + } + } +} + +func (w *DesktopAccountHealthSinkWorker) handleDelivery(parent context.Context, delivery amqp.Delivery) { + var report bufferedDesktopAccountHealthReport + if err := json.Unmarshal(delivery.Body, &report); err != nil { + w.rejectDelivery(delivery, false, err, "") + return + } + + if err := validateBufferedDesktopAccountHealthReport(report); err != nil { + w.rejectDelivery(delivery, false, err, report.AccountID) + return + } + + ctx, cancel := context.WithTimeout(parent, w.processTimeout) + defer cancel() + + if err := sinkDesktopAccountHealthSnapshot(ctx, w.db, report); err != nil { + w.rejectDelivery(delivery, !delivery.Redelivered, err, report.AccountID) + return + } + + if err := delivery.Ack(false); err != nil && w.logger != nil { + w.logger.Warn("desktop account health ack failed", + zap.Error(err), + zap.String("account_id", report.AccountID), + zap.Int64("workspace_id", report.WorkspaceID), + ) + } +} + +func (w *DesktopAccountHealthSinkWorker) rejectDelivery(delivery amqp.Delivery, requeue bool, err error, accountID string) { + if nackErr := delivery.Nack(false, requeue); nackErr != nil && w.logger != nil { + w.logger.Warn("desktop account health nack failed", + zap.Error(nackErr), + zap.String("account_id", accountID), + ) + } + if w.logger != nil { + w.logger.Warn("desktop account health processing failed", + zap.Error(err), + zap.Bool("requeue", requeue), + zap.Bool("redelivered", delivery.Redelivered), + zap.String("account_id", accountID), + ) + } +} + +func validateBufferedDesktopAccountHealthReport(report bufferedDesktopAccountHealthReport) error { + if _, err := uuid.Parse(strings.TrimSpace(report.AccountID)); err != nil { + return fmt.Errorf("invalid account_id: %w", err) + } + if report.WorkspaceID == 0 { + return fmt.Errorf("workspace_id is required") + } + if strings.TrimSpace(report.Health) == "" || strings.TrimSpace(report.AuthState) == "" || strings.TrimSpace(report.ProbeState) == "" { + return fmt.Errorf("health, auth_state and probe_state are required") + } + if report.CheckedAt.IsZero() { + return fmt.Errorf("checked_at is required") + } + return nil +} + +func sinkDesktopAccountHealthSnapshot(ctx context.Context, db *pgxpool.Pool, report bufferedDesktopAccountHealthReport) error { + if db == nil { + return fmt.Errorf("database pool is required") + } + + checkedAt := report.CheckedAt.UTC() + var verifiedAt any + if report.VerifiedAt != nil { + value := report.VerifiedAt.UTC() + verifiedAt = value + } + + _, err := db.Exec(ctx, ` +UPDATE platform_accounts +SET health = $1, + status = $2, + verified_at = COALESCE($3::timestamptz, verified_at), + last_check_at = $4::timestamptz, + updated_at = now() +WHERE desktop_id = $5::uuid + AND workspace_id = $6 + AND deleted_at IS NULL + AND (last_check_at IS NULL OR last_check_at <= $4::timestamptz) +`, report.Health, desktopAccountLegacyStatusFromHealth(report.Health), verifiedAt, checkedAt, report.AccountID, report.WorkspaceID) + if err != nil { + return fmt.Errorf("sink desktop account health snapshot: %w", err) + } + return nil +} + +func desktopAccountLegacyStatusFromHealth(health string) string { + switch health { + case "live": + return "active" + case "captcha": + return "captcha" + case "risk": + return "risk" + default: + return "expired" + } +} diff --git a/server/internal/tenant/app/desktop_account_service.go b/server/internal/tenant/app/desktop_account_service.go index 24320bf..eede445 100644 --- a/server/internal/tenant/app/desktop_account_service.go +++ b/server/internal/tenant/app/desktop_account_service.go @@ -2,7 +2,11 @@ package app import ( "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" "errors" + "fmt" "strings" "time" @@ -11,14 +15,18 @@ import ( goredis "github.com/redis/go-redis/v9" "github.com/geo-platform/tenant-api/internal/shared/auth" + "github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq" "github.com/geo-platform/tenant-api/internal/shared/response" "github.com/geo-platform/tenant-api/internal/tenant/repository" ) +const desktopAccountHealthReportTTL = 24 * time.Hour + type DesktopAccountService struct { repo repository.DesktopAccountRepository clientRepo repository.DesktopClientRepository redis *goredis.Client + rabbitMQ *rabbitmq.Client now func() time.Time } @@ -26,11 +34,13 @@ func NewDesktopAccountService( repo repository.DesktopAccountRepository, clientRepo repository.DesktopClientRepository, redis *goredis.Client, + rabbitMQClient *rabbitmq.Client, ) *DesktopAccountService { return &DesktopAccountService{ repo: repo, clientRepo: clientRepo, redis: redis, + rabbitMQ: rabbitMQClient, now: time.Now, } } @@ -42,6 +52,13 @@ type DesktopAccountView struct { DisplayName string `json:"display_name"` AvatarURL *string `json:"avatar_url"` Health string `json:"health"` + RuntimeHealth *string `json:"runtime_health"` + RuntimeVerifiedAt *time.Time `json:"runtime_verified_at"` + RuntimeCheckedAt *time.Time `json:"runtime_checked_at"` + RuntimeAuthState *string `json:"runtime_auth_state"` + RuntimeProbeState *string `json:"runtime_probe_state"` + RuntimeAuthReason *string `json:"runtime_auth_reason"` + HealthSource string `json:"health_source"` ClientID *string `json:"client_id"` AccountFingerprint *string `json:"account_fingerprint"` VerifiedAt *time.Time `json:"verified_at"` @@ -74,6 +91,48 @@ type PatchDesktopAccountRequest struct { IfSyncVersion int64 `json:"if_sync_version" binding:"required"` } +type DesktopAccountHealthReportRequest struct { + AccountID string `json:"account_id" binding:"required"` + Platform string `json:"platform" binding:"required"` + PlatformUID string `json:"platform_uid" binding:"required"` + Health string `json:"health" binding:"required,oneof=live expired captcha risk"` + AuthState string `json:"auth_state" binding:"required"` + ProbeState string `json:"probe_state" binding:"required"` + AuthReason *string `json:"auth_reason"` + DisplayName *string `json:"display_name"` + AvatarURL *string `json:"avatar_url"` + VerifiedAt *time.Time `json:"verified_at"` + CheckedAt time.Time `json:"checked_at" binding:"required"` +} + +type ReportDesktopAccountHealthRequest struct { + Reports []DesktopAccountHealthReportRequest `json:"reports" binding:"required,min=1,max=100"` +} + +type ReportDesktopAccountHealthResponse struct { + AcceptedCount int `json:"accepted_count"` + BufferedCount int `json:"buffered_count"` +} + +type bufferedDesktopAccountHealthReport struct { + TenantID int64 `json:"tenant_id"` + WorkspaceID int64 `json:"workspace_id"` + UserID int64 `json:"user_id"` + ClientID string `json:"client_id"` + AccountID string `json:"account_id"` + Platform string `json:"platform"` + PlatformUID string `json:"platform_uid"` + Health string `json:"health"` + AuthState string `json:"auth_state"` + ProbeState string `json:"probe_state"` + AuthReason *string `json:"auth_reason"` + DisplayName *string `json:"display_name"` + AvatarURL *string `json:"avatar_url"` + VerifiedAt *time.Time `json:"verified_at"` + CheckedAt time.Time `json:"checked_at"` + ReceivedAt time.Time `json:"received_at"` +} + func (s *DesktopAccountService) ListByUser(ctx context.Context, client *repository.DesktopClient) ([]DesktopAccountView, error) { if client == nil { return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required") @@ -89,6 +148,7 @@ func (s *DesktopAccountService) ListByUser(ctx context.Context, client *reposito for _, item := range rows { items = append(items, s.buildDesktopAccountView(&item, clientMap, presenceMap)) } + s.applyRuntimeHealth(ctx, client.WorkspaceID, items) return items, nil } @@ -107,6 +167,7 @@ func (s *DesktopAccountService) ListForActor(ctx context.Context, actor auth.Act for _, item := range rows { items = append(items, s.buildDesktopAccountView(&item, clientMap, presenceMap)) } + s.applyRuntimeHealth(ctx, actor.PrimaryWorkspaceID, items) return items, nil } @@ -154,6 +215,124 @@ func (s *DesktopAccountService) Upsert(ctx context.Context, client *repository.D return &view, nil } +func (s *DesktopAccountService) ReportHealth(ctx context.Context, client *repository.DesktopClient, req ReportDesktopAccountHealthRequest) (*ReportDesktopAccountHealthResponse, error) { + if client == nil { + return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required") + } + if len(req.Reports) == 0 { + return nil, response.ErrBadRequest(40087, "empty_account_health_reports", "at least one account health report is required") + } + if len(req.Reports) > 100 { + return nil, response.ErrBadRequest(40087, "too_many_account_health_reports", "at most 100 account health reports are allowed") + } + + accepted := 0 + buffered := 0 + now := s.now().UTC() + type preparedHealthReport struct { + accountID uuid.UUID + latestKey string + signature string + payload []byte + report bufferedDesktopAccountHealthReport + } + prepared := make([]preparedHealthReport, 0, len(req.Reports)) + for _, item := range req.Reports { + accountID, err := uuid.Parse(strings.TrimSpace(item.AccountID)) + if err != nil { + return nil, response.ErrBadRequest(40087, "invalid_account_health_report_account_id", "account_id must be a uuid") + } + platform := strings.TrimSpace(item.Platform) + platformUID := strings.TrimSpace(item.PlatformUID) + if platform == "" || platformUID == "" { + return nil, response.ErrBadRequest(40087, "invalid_account_health_report_identity", "platform and platform_uid are required") + } + + report := bufferedDesktopAccountHealthReport{ + TenantID: client.TenantID, + WorkspaceID: client.WorkspaceID, + UserID: client.UserID, + ClientID: client.ID.String(), + AccountID: accountID.String(), + Platform: platform, + PlatformUID: platformUID, + Health: item.Health, + AuthState: strings.TrimSpace(item.AuthState), + ProbeState: strings.TrimSpace(item.ProbeState), + AuthReason: trimOptionalString(item.AuthReason), + DisplayName: trimOptionalString(item.DisplayName), + AvatarURL: trimOptionalString(item.AvatarURL), + VerifiedAt: item.VerifiedAt, + CheckedAt: item.CheckedAt.UTC(), + ReceivedAt: now, + } + if report.AuthState == "" || report.ProbeState == "" || report.CheckedAt.IsZero() { + return nil, response.ErrBadRequest(40087, "invalid_account_health_report_state", "auth_state, probe_state and checked_at are required") + } + + payload, err := json.Marshal(report) + if err != nil { + return nil, response.ErrInternal(50087, "account_health_report_encode_failed", "failed to encode account health report") + } + + latestKey := desktopAccountHealthLatestKey(client.WorkspaceID, accountID) + prepared = append(prepared, preparedHealthReport{ + accountID: accountID, + latestKey: latestKey, + signature: desktopAccountHealthSignature(report), + payload: payload, + report: report, + }) + accepted += 1 + } + changed := make([]preparedHealthReport, 0, len(prepared)) + if s.redis != nil { + signaturePipe := s.redis.Pipeline() + signatureCommands := make(map[uuid.UUID]*goredis.StringCmd, len(prepared)) + for _, item := range prepared { + signatureCommands[item.accountID] = signaturePipe.Get(ctx, item.latestKey+":signature") + } + if _, err := signaturePipe.Exec(ctx); err != nil && !errors.Is(err, goredis.Nil) { + return nil, response.ErrInternal(50087, "account_health_report_cache_failed", "failed to inspect account health report cache") + } + + pipe := s.redis.Pipeline() + for _, item := range prepared { + previousSignature, signatureErr := signatureCommands[item.accountID].Result() + if signatureErr != nil && !errors.Is(signatureErr, goredis.Nil) { + return nil, response.ErrInternal(50087, "account_health_report_cache_failed", "failed to inspect account health report cache") + } + + pipe.Set(ctx, item.latestKey, item.payload, desktopAccountHealthReportTTL) + if previousSignature != item.signature { + changed = append(changed, item) + } + } + if _, err := pipe.Exec(ctx); err != nil { + return nil, response.ErrInternal(50087, "account_health_report_buffer_failed", "failed to buffer account health reports") + } + } else { + changed = append(changed, prepared...) + } + + if s.rabbitMQ == nil { + return &ReportDesktopAccountHealthResponse{AcceptedCount: accepted, BufferedCount: 0}, nil + } + + for _, item := range changed { + if err := s.rabbitMQ.PublishDesktopAccountHealth(ctx, item.payload); err != nil { + return nil, response.ErrInternal(50087, "account_health_report_queue_failed", "failed to enqueue account health report") + } + buffered += 1 + if s.redis != nil { + if err := s.redis.Set(ctx, item.latestKey+":signature", item.signature, desktopAccountHealthReportTTL).Err(); err != nil { + return nil, response.ErrInternal(50087, "account_health_report_signature_failed", "failed to record account health report signature") + } + } + } + return &ReportDesktopAccountHealthResponse{AcceptedCount: accepted, BufferedCount: buffered}, nil +} + func (s *DesktopAccountService) Patch(ctx context.Context, client *repository.DesktopClient, desktopID uuid.UUID, req PatchDesktopAccountRequest) (*DesktopAccountView, error) { if client == nil { return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required") @@ -222,6 +401,128 @@ func (s *DesktopAccountService) RequestDelete(ctx context.Context, actor auth.Ac return &view, nil } +func desktopAccountHealthLatestKey(workspaceID int64, accountID uuid.UUID) string { + return fmt.Sprintf("desktop:account-health:latest:%d:%s", workspaceID, accountID.String()) +} + +func desktopAccountHealthSignature(report bufferedDesktopAccountHealthReport) string { + payload, _ := json.Marshal(struct { + AccountID string `json:"account_id"` + Platform string `json:"platform"` + PlatformUID string `json:"platform_uid"` + Health string `json:"health"` + AuthState string `json:"auth_state"` + ProbeState string `json:"probe_state"` + AuthReason *string `json:"auth_reason"` + DisplayName *string `json:"display_name"` + AvatarURL *string `json:"avatar_url"` + }{ + AccountID: report.AccountID, + Platform: report.Platform, + PlatformUID: report.PlatformUID, + Health: report.Health, + AuthState: report.AuthState, + ProbeState: report.ProbeState, + AuthReason: report.AuthReason, + DisplayName: report.DisplayName, + AvatarURL: report.AvatarURL, + }) + sum := sha256.Sum256(payload) + return hex.EncodeToString(sum[:]) +} + +func trimOptionalString(value *string) *string { + if value == nil { + return nil + } + trimmed := strings.TrimSpace(*value) + if trimmed == "" { + return nil + } + return &trimmed +} + +func loadDesktopAccountRuntimeHealth( + ctx context.Context, + redis *goredis.Client, + workspaceID int64, + accountIDs []uuid.UUID, +) map[uuid.UUID]bufferedDesktopAccountHealthReport { + if redis == nil || len(accountIDs) == 0 { + return nil + } + + uniqueIDs := uniqueUUIDs(accountIDs) + pipe := redis.Pipeline() + commands := make(map[uuid.UUID]*goredis.StringCmd, len(uniqueIDs)) + for _, accountID := range uniqueIDs { + commands[accountID] = pipe.Get(ctx, desktopAccountHealthLatestKey(workspaceID, accountID)) + } + + if _, err := pipe.Exec(ctx); err != nil && !errors.Is(err, goredis.Nil) { + return nil + } + + result := make(map[uuid.UUID]bufferedDesktopAccountHealthReport, len(commands)) + for accountID, cmd := range commands { + payload, err := cmd.Bytes() + if err != nil { + continue + } + var report bufferedDesktopAccountHealthReport + if err := json.Unmarshal(payload, &report); err != nil { + continue + } + if report.WorkspaceID != workspaceID || report.AccountID != accountID.String() || report.CheckedAt.IsZero() { + continue + } + result[accountID] = report + } + + return result +} + +func (s *DesktopAccountService) applyRuntimeHealth(ctx context.Context, workspaceID int64, items []DesktopAccountView) { + if s.redis == nil || len(items) == 0 { + return + } + + accountIDs := make([]uuid.UUID, 0, len(items)) + accountIndex := make(map[uuid.UUID]int, len(items)) + for index, item := range items { + accountID, err := uuid.Parse(item.ID) + if err != nil { + continue + } + accountIDs = append(accountIDs, accountID) + accountIndex[accountID] = index + } + + reports := loadDesktopAccountRuntimeHealth(ctx, s.redis, workspaceID, accountIDs) + for accountID, report := range reports { + index, ok := accountIndex[accountID] + if !ok { + continue + } + + health := report.Health + authState := report.AuthState + probeState := report.ProbeState + checkedAt := report.CheckedAt + items[index].Health = health + items[index].RuntimeHealth = &health + items[index].RuntimeVerifiedAt = report.VerifiedAt + items[index].RuntimeCheckedAt = &checkedAt + items[index].RuntimeAuthState = &authState + items[index].RuntimeProbeState = &probeState + items[index].RuntimeAuthReason = report.AuthReason + items[index].HealthSource = "runtime" + if report.VerifiedAt != nil { + items[index].VerifiedAt = report.VerifiedAt + } + } +} + func (s *DesktopAccountService) buildDesktopAccountView( account *repository.DesktopAccount, clientMap map[uuid.UUID]*repository.DesktopClient, @@ -266,6 +567,7 @@ func (s *DesktopAccountService) buildDesktopAccountView( DisplayName: account.DisplayName, AvatarURL: account.AvatarURL, Health: account.Health, + HealthSource: "database", ClientID: clientID, AccountFingerprint: account.AccountFingerprint, VerifiedAt: account.VerifiedAt, diff --git a/server/internal/tenant/app/publish_job_service.go b/server/internal/tenant/app/publish_job_service.go index b2d1a47..0bf006e 100644 --- a/server/internal/tenant/app/publish_job_service.go +++ b/server/internal/tenant/app/publish_job_service.go @@ -81,6 +81,16 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr return nil, err } + requestedAccountIDs := make([]uuid.UUID, 0, len(req.Accounts)) + for _, accountReq := range req.Accounts { + accountID, parseErr := uuid.Parse(strings.TrimSpace(accountReq.AccountID)) + if parseErr != nil { + return nil, response.ErrBadRequest(40092, "invalid_desktop_account_id", "account_id must be a uuid") + } + requestedAccountIDs = append(requestedAccountIDs, accountID) + } + runtimeHealth := loadDesktopAccountRuntimeHealth(ctx, s.redis, actor.PrimaryWorkspaceID, requestedAccountIDs) + tx, err := s.pool.Begin(ctx) if err != nil { return nil, response.ErrInternal(50096, "desktop_publish_job_begin_failed", "failed to start desktop publish job transaction") @@ -101,12 +111,7 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr } targets := make([]publishTarget, 0, len(req.Accounts)) - for _, accountReq := range req.Accounts { - accountID, parseErr := uuid.Parse(strings.TrimSpace(accountReq.AccountID)) - if parseErr != nil { - return nil, response.ErrBadRequest(40092, "invalid_desktop_account_id", "account_id must be a uuid") - } - + for _, accountID := range requestedAccountIDs { account, lookupErr := accountRepo.GetByDesktopID(ctx, accountID, actor.PrimaryWorkspaceID) if lookupErr != nil { if errors.Is(lookupErr, pgx.ErrNoRows) { @@ -114,7 +119,11 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr } return nil, response.ErrInternal(50098, "desktop_account_lookup_failed", "failed to load selected desktop account") } - if account.Health != "live" { + accountHealth := account.Health + if report, ok := runtimeHealth[account.DesktopID]; ok && strings.TrimSpace(report.Health) != "" { + accountHealth = report.Health + } + if accountHealth != "live" { return nil, response.ErrConflict(40993, "desktop_account_not_publishable", "one or more selected desktop accounts require re-authorization before publishing") } targetClientID := s.resolveTargetClientID(ctx, account) diff --git a/server/internal/tenant/transport/desktop_account_handler.go b/server/internal/tenant/transport/desktop_account_handler.go index f52bebf..7a3bc66 100644 --- a/server/internal/tenant/transport/desktop_account_handler.go +++ b/server/internal/tenant/transport/desktop_account_handler.go @@ -23,6 +23,7 @@ func NewDesktopAccountHandler(a *bootstrap.App) *DesktopAccountHandler { repository.NewDesktopAccountRepository(a.DB), repository.NewDesktopClientRepository(a.DB), a.Redis, + a.RabbitMQ, ), } } @@ -65,6 +66,21 @@ func (h *DesktopAccountHandler) Upsert(c *gin.Context) { response.SuccessWithStatus(c, 201, data) } +func (h *DesktopAccountHandler) ReportHealth(c *gin.Context) { + var req app.ReportDesktopAccountHealthRequest + if err := c.ShouldBindJSON(&req); err != nil { + response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error())) + return + } + + data, err := h.svc.ReportHealth(c.Request.Context(), MustDesktopClient(c.Request.Context()), req) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, data) +} + func (h *DesktopAccountHandler) Patch(c *gin.Context) { desktopID, err := uuid.Parse(c.Param("id")) if err != nil { diff --git a/server/internal/tenant/transport/router.go b/server/internal/tenant/transport/router.go index d0b2ee2..ecdd5a8 100644 --- a/server/internal/tenant/transport/router.go +++ b/server/internal/tenant/transport/router.go @@ -48,6 +48,7 @@ func RegisterRoutes(app *bootstrap.App) { desktopAuth.GET("/content/articles/:id", desktopContentHandler.Article) desktopAuth.GET("/content/assets/:token", publicAssets.ServeDesktop) desktopAuth.GET("/publish-tasks", desktopTaskHandler.ListPublish) + desktopAuth.POST("/accounts/health-reports", desktopAccountHandler.ReportHealth) desktopAuth.POST("/accounts", desktopAccountHandler.Upsert) desktopAuth.PATCH("/accounts/:id", desktopAccountHandler.Patch) desktopAuth.DELETE("/accounts/:id", desktopAccountHandler.Delete)