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) <noreply@anthropic.com>
2026-04-27 21:33:36 +08:00
parent 94f185c3a1
commit 051976e4a9
10 changed files with 647 additions and 51 deletions
@@ -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,