09295d11a1
Add ai_platforms table + shared catalog (yuanbao/kimi/wenxin/deepseek/doubao/qwen), drop FKs from platform_accounts and desktop_tasks to media_platforms, and wire target_account_id + platform into DesktopTaskEvent so the desktop runtime no longer has to infer them from local state. Desktop client gains generic AI page detection for binding, drops stale-business-date monitor tasks at the edge, and refactors AccountsView/AiPlatformsView around the shared catalog. Admin tracking view surfaces per-platform sampling status cards. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
355 lines
12 KiB
Go
355 lines
12 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5"
|
|
goredis "github.com/redis/go-redis/v9"
|
|
|
|
"github.com/geo-platform/tenant-api/internal/shared/auth"
|
|
"github.com/geo-platform/tenant-api/internal/shared/response"
|
|
"github.com/geo-platform/tenant-api/internal/tenant/repository"
|
|
)
|
|
|
|
type DesktopAccountService struct {
|
|
repo repository.DesktopAccountRepository
|
|
clientRepo repository.DesktopClientRepository
|
|
redis *goredis.Client
|
|
now func() time.Time
|
|
}
|
|
|
|
func NewDesktopAccountService(
|
|
repo repository.DesktopAccountRepository,
|
|
clientRepo repository.DesktopClientRepository,
|
|
redis *goredis.Client,
|
|
) *DesktopAccountService {
|
|
return &DesktopAccountService{
|
|
repo: repo,
|
|
clientRepo: clientRepo,
|
|
redis: redis,
|
|
now: time.Now,
|
|
}
|
|
}
|
|
|
|
type DesktopAccountView struct {
|
|
ID string `json:"id"`
|
|
Platform string `json:"platform"`
|
|
PlatformUID string `json:"platform_uid"`
|
|
DisplayName string `json:"display_name"`
|
|
AvatarURL *string `json:"avatar_url"`
|
|
Health string `json:"health"`
|
|
ClientID *string `json:"client_id"`
|
|
AccountFingerprint *string `json:"account_fingerprint"`
|
|
VerifiedAt *time.Time `json:"verified_at"`
|
|
Tags []string `json:"tags"`
|
|
SyncVersion int64 `json:"sync_version"`
|
|
DeletedAt *time.Time `json:"deleted_at"`
|
|
DeleteRequestedAt *time.Time `json:"delete_requested_at"`
|
|
ClientOnline *bool `json:"client_online"`
|
|
ClientLastSeenAt *time.Time `json:"client_last_seen_at"`
|
|
ClientDeviceName *string `json:"client_device_name"`
|
|
}
|
|
|
|
type UpsertDesktopAccountRequest struct {
|
|
Platform string `json:"platform" binding:"required"`
|
|
PlatformUID string `json:"platform_uid" binding:"required"`
|
|
AccountFingerprint *string `json:"account_fingerprint"`
|
|
DisplayName string `json:"display_name" binding:"required"`
|
|
AvatarURL *string `json:"avatar_url"`
|
|
Health string `json:"health" binding:"required,oneof=live expired captcha risk"`
|
|
VerifiedAt *time.Time `json:"verified_at"`
|
|
Tags []string `json:"tags"`
|
|
IfSyncVersion *int64 `json:"if_sync_version"`
|
|
}
|
|
|
|
type PatchDesktopAccountRequest struct {
|
|
DisplayName *string `json:"display_name"`
|
|
Health *string `json:"health" binding:"omitempty,oneof=live expired captcha risk"`
|
|
VerifiedAt *time.Time `json:"verified_at"`
|
|
Tags *[]string `json:"tags"`
|
|
IfSyncVersion int64 `json:"if_sync_version" binding:"required"`
|
|
}
|
|
|
|
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")
|
|
}
|
|
|
|
rows, err := s.repo.ListByUser(ctx, client.WorkspaceID, client.UserID)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50081, "desktop_accounts_query_failed", "failed to list desktop accounts")
|
|
}
|
|
|
|
clientMap, presenceMap := s.loadClientState(ctx, client.WorkspaceID, rows)
|
|
items := make([]DesktopAccountView, 0, len(rows))
|
|
for _, item := range rows {
|
|
items = append(items, s.buildDesktopAccountView(&item, clientMap, presenceMap))
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
func (s *DesktopAccountService) ListForActor(ctx context.Context, actor auth.Actor) ([]DesktopAccountView, error) {
|
|
if actor.PrimaryWorkspaceID == 0 || actor.UserID == 0 {
|
|
return nil, response.ErrUnauthorized(40132, "workspace_scope_missing", "workspace context is required")
|
|
}
|
|
|
|
rows, err := s.repo.ListByUser(ctx, actor.PrimaryWorkspaceID, actor.UserID)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50081, "desktop_accounts_query_failed", "failed to list desktop accounts")
|
|
}
|
|
|
|
clientMap, presenceMap := s.loadClientState(ctx, actor.PrimaryWorkspaceID, rows)
|
|
items := make([]DesktopAccountView, 0, len(rows))
|
|
for _, item := range rows {
|
|
items = append(items, s.buildDesktopAccountView(&item, clientMap, presenceMap))
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
func (s *DesktopAccountService) Upsert(ctx context.Context, client *repository.DesktopClient, req UpsertDesktopAccountRequest) (*DesktopAccountView, error) {
|
|
if client == nil {
|
|
return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required")
|
|
}
|
|
|
|
if req.IfSyncVersion != nil {
|
|
existing, err := s.repo.GetByIdentity(ctx, client.WorkspaceID, req.Platform, req.PlatformUID)
|
|
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, response.ErrInternal(50082, "desktop_account_lookup_failed", "failed to inspect desktop account before upsert")
|
|
}
|
|
if existing == nil || errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, response.ErrConflict(40981, "desktop_account_sync_version_required_existing", "if_sync_version is only valid when account already exists")
|
|
}
|
|
}
|
|
|
|
account, err := s.repo.Upsert(ctx, repository.UpsertDesktopAccountParams{
|
|
DesktopID: uuid.New(),
|
|
TenantID: client.TenantID,
|
|
WorkspaceID: client.WorkspaceID,
|
|
UserID: client.UserID,
|
|
ClientID: client.ID,
|
|
Platform: req.Platform,
|
|
PlatformUID: req.PlatformUID,
|
|
AccountFingerprint: req.AccountFingerprint,
|
|
DisplayName: req.DisplayName,
|
|
AvatarURL: req.AvatarURL,
|
|
Health: req.Health,
|
|
VerifiedAt: req.VerifiedAt,
|
|
Tags: req.Tags,
|
|
IfSyncVersion: req.IfSyncVersion,
|
|
})
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, response.ErrConflict(40982, "desktop_account_sync_conflict", "desktop account sync version conflict")
|
|
}
|
|
appErr := response.ErrInternal(50083, "desktop_account_upsert_failed", "failed to upsert desktop account")
|
|
appErr.Cause = err
|
|
return nil, appErr
|
|
}
|
|
|
|
view := s.buildDesktopAccountView(account, nil, nil)
|
|
return &view, 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")
|
|
}
|
|
|
|
account, err := s.repo.Patch(ctx, repository.PatchDesktopAccountParams{
|
|
DesktopID: desktopID,
|
|
WorkspaceID: client.WorkspaceID,
|
|
DisplayName: req.DisplayName,
|
|
Health: req.Health,
|
|
VerifiedAt: req.VerifiedAt,
|
|
Tags: req.Tags,
|
|
IfSyncVersion: req.IfSyncVersion,
|
|
})
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, response.ErrConflict(40982, "desktop_account_sync_conflict", "desktop account sync version conflict")
|
|
}
|
|
return nil, response.ErrInternal(50084, "desktop_account_patch_failed", "failed to patch desktop account")
|
|
}
|
|
|
|
view := s.buildDesktopAccountView(account, nil, nil)
|
|
return &view, nil
|
|
}
|
|
|
|
func (s *DesktopAccountService) Tombstone(ctx context.Context, client *repository.DesktopClient, desktopID uuid.UUID, ifSyncVersion int64) (*DesktopAccountView, error) {
|
|
if client == nil {
|
|
return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required")
|
|
}
|
|
|
|
account, err := s.repo.Tombstone(ctx, desktopID, client.WorkspaceID, ifSyncVersion)
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, response.ErrConflict(40982, "desktop_account_sync_conflict", "desktop account sync version conflict")
|
|
}
|
|
return nil, response.ErrInternal(50085, "desktop_account_delete_failed", "failed to delete desktop account")
|
|
}
|
|
|
|
view := s.buildDesktopAccountView(account, nil, nil)
|
|
return &view, nil
|
|
}
|
|
|
|
func (s *DesktopAccountService) RequestDelete(ctx context.Context, actor auth.Actor, desktopID uuid.UUID, undo bool) (*DesktopAccountView, error) {
|
|
if actor.PrimaryWorkspaceID == 0 {
|
|
return nil, response.ErrUnauthorized(40132, "workspace_scope_missing", "workspace context is required")
|
|
}
|
|
|
|
var (
|
|
account *repository.DesktopAccount
|
|
err error
|
|
)
|
|
|
|
if undo {
|
|
account, err = s.repo.ClearDeleteRequested(ctx, desktopID, actor.PrimaryWorkspaceID)
|
|
} else {
|
|
account, err = s.repo.MarkDeleteRequested(ctx, desktopID, actor.PrimaryWorkspaceID)
|
|
}
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, response.ErrNotFound(40481, "desktop_account_not_found", "desktop account not found")
|
|
}
|
|
return nil, response.ErrInternal(50086, "desktop_account_request_delete_failed", "failed to update desktop account delete request")
|
|
}
|
|
|
|
view := s.buildDesktopAccountView(account, nil, nil)
|
|
return &view, nil
|
|
}
|
|
|
|
func (s *DesktopAccountService) buildDesktopAccountView(
|
|
account *repository.DesktopAccount,
|
|
clientMap map[uuid.UUID]*repository.DesktopClient,
|
|
presenceClientMap map[uuid.UUID]uuid.UUID,
|
|
) DesktopAccountView {
|
|
var clientID *string
|
|
var clientOnline *bool
|
|
var clientLastSeenAt *time.Time
|
|
var clientDeviceName *string
|
|
|
|
resolvedClientID := account.ClientID
|
|
if presenceClientMap != nil {
|
|
if activeClientID, ok := presenceClientMap[account.DesktopID]; ok {
|
|
resolvedClientID = &activeClientID
|
|
}
|
|
}
|
|
|
|
if resolvedClientID != nil {
|
|
value := resolvedClientID.String()
|
|
clientID = &value
|
|
|
|
online := false
|
|
clientOnline = &online
|
|
|
|
if clientMap != nil {
|
|
if client, ok := clientMap[*resolvedClientID]; ok && client != nil {
|
|
clientLastSeenAt = client.LastSeenAt
|
|
clientDeviceName = client.DeviceName
|
|
if presenceClientMap != nil {
|
|
if activeClientID, ok := presenceClientMap[account.DesktopID]; ok && activeClientID == *resolvedClientID {
|
|
online = true
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
return DesktopAccountView{
|
|
ID: account.DesktopID.String(),
|
|
Platform: account.Platform,
|
|
PlatformUID: normalizePlatformUID(account.PlatformUID),
|
|
DisplayName: account.DisplayName,
|
|
AvatarURL: account.AvatarURL,
|
|
Health: account.Health,
|
|
ClientID: clientID,
|
|
AccountFingerprint: account.AccountFingerprint,
|
|
VerifiedAt: account.VerifiedAt,
|
|
Tags: account.Tags,
|
|
SyncVersion: account.SyncVersion,
|
|
DeletedAt: account.DeletedAt,
|
|
DeleteRequestedAt: account.DeleteRequestedAt,
|
|
ClientOnline: clientOnline,
|
|
ClientLastSeenAt: clientLastSeenAt,
|
|
ClientDeviceName: clientDeviceName,
|
|
}
|
|
}
|
|
|
|
func (s *DesktopAccountService) loadClientState(
|
|
ctx context.Context,
|
|
workspaceID int64,
|
|
accounts []repository.DesktopAccount,
|
|
) (map[uuid.UUID]*repository.DesktopClient, map[uuid.UUID]uuid.UUID) {
|
|
clientIDs := make([]uuid.UUID, 0, len(accounts)*2)
|
|
accountIDs := make([]uuid.UUID, 0, len(accounts))
|
|
seen := make(map[uuid.UUID]struct{}, len(accounts)*2)
|
|
|
|
for _, account := range accounts {
|
|
accountIDs = append(accountIDs, account.DesktopID)
|
|
|
|
if account.ClientID == nil {
|
|
continue
|
|
}
|
|
if _, ok := seen[*account.ClientID]; ok {
|
|
continue
|
|
}
|
|
seen[*account.ClientID] = struct{}{}
|
|
clientIDs = append(clientIDs, *account.ClientID)
|
|
}
|
|
|
|
presenceClientMap := loadDesktopAccountPresence(ctx, s.redis, accountIDs)
|
|
for _, clientID := range presenceClientMap {
|
|
if _, ok := seen[clientID]; ok {
|
|
continue
|
|
}
|
|
seen[clientID] = struct{}{}
|
|
clientIDs = append(clientIDs, clientID)
|
|
}
|
|
|
|
clientMap := make(map[uuid.UUID]*repository.DesktopClient, len(clientIDs))
|
|
if s.clientRepo != nil {
|
|
for _, clientID := range clientIDs {
|
|
client, err := s.clientRepo.GetByID(ctx, clientID, workspaceID)
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
continue
|
|
}
|
|
continue
|
|
}
|
|
clientMap[clientID] = client
|
|
}
|
|
}
|
|
|
|
return clientMap, presenceClientMap
|
|
}
|
|
|
|
func resolveDesktopClientOnline(
|
|
client *repository.DesktopClient,
|
|
presence bool,
|
|
presenceKnown bool,
|
|
now time.Time,
|
|
) bool {
|
|
if client == nil || client.RevokedAt != nil {
|
|
return false
|
|
}
|
|
if presenceKnown && presence {
|
|
return true
|
|
}
|
|
if client.LastSeenAt == nil {
|
|
return false
|
|
}
|
|
return now.UTC().Sub(client.LastSeenAt.UTC()) <= desktopClientPresenceTTL
|
|
}
|
|
|
|
func normalizePlatformUID(value string) string {
|
|
normalized := strings.TrimSpace(value)
|
|
for strings.HasPrefix(strings.ToLower(normalized), "platform:") {
|
|
normalized = strings.TrimSpace(normalized[len("platform:"):])
|
|
}
|
|
return normalized
|
|
}
|