From 5ff2e2e74cb7c5902c8aacd380e10c86672647a5 Mon Sep 17 00:00:00 2001 From: liangxu Date: Mon, 20 Apr 2026 13:52:35 +0800 Subject: [PATCH] refactor(tenant): drop legacy plugin_installations, migrate monitoring to desktop_clients Hard cutover from the browser-extension plugin flow to desktop clients: remove plugin_installations/plugin_sessions tables and related service, handler, router, and generated model code; migrate monitoring quotas and collector types to desktop_clients (UUID primary_client_id); recreate platform_access_snapshots keyed by client_id; update dev-seed and callback types accordingly; mark legacy design docs as historical. Co-Authored-By: Claude Opus 4.7 (1M context) --- docs/ai-brand-monitoring-tech-design-v4.md | 6 +- docs/ai-brand-monitoring-tech-design-v5.md | 6 +- ...platform-ops-admin-tech-architecture-v1.md | 2 + docs/media-binding-and-publish-v1.md | 4 +- docs/media-publisher-extension-runtime-v1.md | 4 +- .../2026-04-20-legacy-plugin-design-status.md | 36 + server/cmd/dev-seed/main.go | 98 +- server/internal/tenant/app/media_service.go | 960 +----------------- .../tenant/app/monitoring_callback_service.go | 125 +-- .../monitoring_result_ingest_worker_test.go | 6 +- .../internal/tenant/app/monitoring_service.go | 103 +- .../tenant/app/publish_record_support.go | 67 ++ .../tenant/repository/generated/models.go | 36 +- .../tenant/transport/media_handler.go | 134 +-- server/internal/tenant/transport/router.go | 10 +- ...000_create_media_publisher_tables.down.sql | 2 - ...40000_create_media_publisher_tables.up.sql | 44 - ...epair_plugin_installations_schema.down.sql | 5 +- ..._repair_plugin_installations_schema.up.sql | 40 +- ...0410103000_create_monitoring_tables.up.sql | 4 +- ...plugin_installations_and_sessions.down.sql | 53 + ...e_plugin_installations_and_sessions.up.sql | 32 + ...0410103000_create_monitoring_tables.up.sql | 16 +- ...ate_monitoring_to_desktop_clients.down.sql | 63 ++ ...grate_monitoring_to_desktop_clients.up.sql | 63 ++ 25 files changed, 535 insertions(+), 1384 deletions(-) create mode 100644 docs/worklogs/2026-04-20-legacy-plugin-design-status.md create mode 100644 server/migrations/20260421100030_remove_plugin_installations_and_sessions.down.sql create mode 100644 server/migrations/20260421100030_remove_plugin_installations_and_sessions.up.sql create mode 100644 server/migrations_monitoring/20260421100030_migrate_monitoring_to_desktop_clients.down.sql create mode 100644 server/migrations_monitoring/20260421100030_migrate_monitoring_to_desktop_clients.up.sql diff --git a/docs/ai-brand-monitoring-tech-design-v4.md b/docs/ai-brand-monitoring-tech-design-v4.md index 3ed82b4..42d9226 100644 --- a/docs/ai-brand-monitoring-tech-design-v4.md +++ b/docs/ai-brand-monitoring-tech-design-v4.md @@ -1,4 +1,6 @@ -# AI 品牌曝光监测系统技术方案 V4 +# AI 品牌曝光监测系统技术方案 V4（旧版设计） + +> 旧版设计说明：本文描述的是基于浏览器插件采集、`installation_token`、`/api/plugin/monitoring/*` 的旧监控方案。自 2026-04-20 起，当前实现已切换到 `desktop_clients` / desktop client 架构。本文保留用于工作记录与方案追溯，不再作为当前实现依据。 ## 1. 文档信息 @@ -6,7 +8,7 @@ | --- | --- | | 文档名称 | AI 品牌曝光监测系统技术方案 V4（插件采集 + 配额约束） | | 文档版本 | V4.0 | -| 文档状态 | 待评审 | +| 文档状态 | 旧版设计（保留工作记录） | | 创建日期 | 2026-04-08 | | 基线文档 | `docs/ai-brand-monitoring-tech-design-v3.md` | | 适用范围 | 数据追踪模块全部页面 | diff --git a/docs/ai-brand-monitoring-tech-design-v5.md b/docs/ai-brand-monitoring-tech-design-v5.md index 03d9386..f17d1c7 100644 --- a/docs/ai-brand-monitoring-tech-design-v5.md +++ b/docs/ai-brand-monitoring-tech-design-v5.md @@ -1,4 +1,6 @@ -# AI 品牌曝光监测系统技术方案 V5 +# AI 品牌曝光监测系统技术方案 V5（旧版设计） + +> 旧版设计说明：本文描述的是基于浏览器插件采样、`plugin_installations`、`primary_installation_id`、`/api/plugin/monitoring/*` 的旧监控方案。自 2026-04-20 起，当前实现已切换到 `desktop_clients` / desktop client 架构。本文保留用于工作记录与方案追溯，不再作为当前实现依据。 ## 1. 文档信息 @@ -6,7 +8,7 @@ | -------- | --------------------------------------------------------------------- | | 文档名称 | AI 品牌曝光监测系统技术方案 V5（插件采样首期交付 / API 精确采集预留） | | 文档版本 | V5.9 | -| 文档状态 | 待评审 | +| 文档状态 | 旧版设计（保留工作记录） | | 创建日期 | 2026-04-09 | | 基线文档 | `docs/ai-brand-monitoring-tech-design-v4.md` | | 适用范围 | 数据追踪模块全部页面 | diff --git a/docs/geo-platform-ops-admin-tech-architecture-v1.md b/docs/geo-platform-ops-admin-tech-architecture-v1.md index 035a84b..4618ae3 100644 --- a/docs/geo-platform-ops-admin-tech-architecture-v1.md +++ b/docs/geo-platform-ops-admin-tech-architecture-v1.md @@ -1,5 +1,7 @@ # GEO 双后台与后端技术架构方案 V1 +> 工作记录说明：本文仍保留作为历史架构记录。其中涉及浏览器插件执行链路、`/api/callback/plugin/*`、`plugin_sessions`、`installation_token` 的相关章节，自 2026-04-20 起视为旧版设计；当前实现请以 desktop client 方案和最新代码为准。 + ## 1. 文档信息 | 项目 | 内容 | diff --git a/docs/media-binding-and-publish-v1.md b/docs/media-binding-and-publish-v1.md index 69f7894..a7ed006 100644 --- a/docs/media-binding-and-publish-v1.md +++ b/docs/media-binding-and-publish-v1.md @@ -1,4 +1,6 @@ -# V1 技术方案：绑定媒体账号与手动发送文章 +# V1 技术方案：绑定媒体账号与手动发送文章（旧版设计） + +> 旧版设计说明：本文描述的是基于浏览器插件回调、`plugin_sessions`、`/api/callback/plugin/*` 的旧方案。自 2026-04-20 起，当前实现已切换到 desktop client 架构。本文保留用于工作记录与方案追溯，不再作为当前实现依据。 > 口径说明：本文的插件执行流程、接口路径与会话模型已同步到主架构文档 [geo-platform-ops-admin-tech-architecture-v1.md](D:/project/geo/docs/geo-platform-ops-admin-tech-architecture-v1.md) 的 `§13.8 浏览器插件执行面闭环`、`§13.6.3 插件回调与内部接口`、`§13.6.4 统一错误响应与错误码`。若两份文档存在任何冲突，以主架构文档为准。 diff --git a/docs/media-publisher-extension-runtime-v1.md b/docs/media-publisher-extension-runtime-v1.md index 30368bb..17ec566 100644 --- a/docs/media-publisher-extension-runtime-v1.md +++ b/docs/media-publisher-extension-runtime-v1.md @@ -1,4 +1,6 @@ -# Media Publisher Extension Runtime V1 +# Media Publisher Extension Runtime V1（旧版设计） + +> 旧版设计说明：本文描述的是基于浏览器插件、`plugin_installations`、`plugin_sessions` 的旧执行方案。自 2026-04-20 起，当前实现已切换到 `desktop_clients` / desktop client 架构。本文保留用于工作记录与方案追溯，不再作为当前实现依据。 ## Goal diff --git a/docs/worklogs/2026-04-20-legacy-plugin-design-status.md b/docs/worklogs/2026-04-20-legacy-plugin-design-status.md new file mode 100644 index 0000000..0e12153 --- /dev/null +++ b/docs/worklogs/2026-04-20-legacy-plugin-design-status.md @@ -0,0 +1,36 @@ +# 2026-04-20 Legacy Plugin Design Status + +## 背景 + +2026-04-20 这次代码与数据库清理，已经移除了旧浏览器插件链路中的 `plugin_installations`、`plugin_sessions` 及其主要业务引用，并将监控主链路切换到 `desktop_clients` / desktop client 体系。 + +## 决策 + +本次**不删除**历史设计文档。 + +保留原因： + +- 作为架构演进的工作记录，方便回溯旧方案的边界、假设和历史实现背景 +- 便于对照为什么当前实现从 browser extension 迁移到 desktop client +- 避免后续排查历史字段、旧迁移、旧接口时失去上下文 + +## 本次文档处理方式 + +对明显描述旧插件执行模型的文档，统一补充“旧版设计”标识；对仍有其他参考价值的综合架构文档，补充“相关章节为旧版设计”的说明，而不删除正文内容。 + +## 已标记文档 + +- `docs/media-publisher-extension-runtime-v1.md` +- `docs/media-binding-and-publish-v1.md` +- `docs/ai-brand-monitoring-tech-design-v4.md` +- `docs/ai-brand-monitoring-tech-design-v5.md` +- `docs/geo-platform-ops-admin-tech-architecture-v1.md` + +## 当前口径 + +当前实现以 desktop client 架构为准，旧插件设计文档仅用于历史追溯，不再作为新开发或现状判断依据。 + +建议优先参考： + +- `docs/superpowers/specs/2026-04-18-electron-desktop-client-design.md` +- 代码中的 `desktop_clients`、`/api/desktop/*`、监控与发布最新实现 diff --git a/server/cmd/dev-seed/main.go b/server/cmd/dev-seed/main.go index e09ad16..624b519 100644 --- a/server/cmd/dev-seed/main.go +++ b/server/cmd/dev-seed/main.go @@ -11,31 +11,31 @@ import ( "strings" "time" + "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "golang.org/x/crypto/bcrypt" - sharedauth "github.com/geo-platform/tenant-api/internal/shared/auth" "github.com/geo-platform/tenant-api/internal/shared/config" "github.com/geo-platform/tenant-api/internal/tenant/prompts" "github.com/geo-platform/tenant-api/internal/tenant/repository" ) type seedState struct { - TenantID int64 - UserID int64 - WorkspaceID int64 - TestTenantID int64 - TestUserID int64 - PlanID int64 - BrandID int64 - KeywordPrimaryID int64 - KeywordSecondaryID int64 - QuestionPrimaryID int64 - QuestionSecondaryID int64 - QuestionTertiaryID int64 - ArticleID int64 - PluginInstallationID int64 + TenantID int64 + UserID int64 + WorkspaceID int64 + TestTenantID int64 + TestUserID int64 + PlanID int64 + BrandID int64 + KeywordPrimaryID int64 + KeywordSecondaryID int64 + QuestionPrimaryID int64 + QuestionSecondaryID int64 + QuestionTertiaryID int64 + ArticleID int64 + DesktopClientID uuid.UUID } type monitoringRunSeed struct { @@ -137,7 +137,7 @@ func main() { fmt.Printf(" Test User ID: %d (test@geo.local / Test@123)\n", state.TestUserID) fmt.Printf(" Plan ID: %d (free)\n", state.PlanID) fmt.Printf(" Brand ID: %d (轻氧口腔)\n", state.BrandID) - fmt.Printf(" Plugin Installation ID: %d\n", state.PluginInstallationID) + fmt.Printf(" Desktop Client ID: %s\n", state.DesktopClientID.String()) fmt.Println(" Templates: 4 platform presets") fmt.Println(" Monitoring: brand daily snapshots + question runs + citation facts") } @@ -241,7 +241,7 @@ func seedBusinessData(ctx context.Context, pool *pgxpool.Pool, membershipCfg con return nil, err } - state.PluginInstallationID, err = ensurePluginInstallation(ctx, tx, state.TenantID, state.UserID) + state.DesktopClientID, err = ensureDesktopClient(ctx, tx, state.TenantID, state.WorkspaceID, state.UserID) if err != nil { return nil, err } @@ -641,47 +641,55 @@ func ensureArticle(ctx context.Context, tx pgx.Tx, tenantID int64) (int64, error return articleID, nil } -func ensurePluginInstallation(ctx context.Context, tx pgx.Tx, tenantID, userID int64) (int64, error) { - installationKey := "demo-monitoring-installation" - tokenHash := sharedauth.HashToken("demo-monitoring-installation-token") +func ensureDesktopClient(ctx context.Context, tx pgx.Tx, tenantID, workspaceID, userID int64) (uuid.UUID, error) { + clientID := uuid.NewSHA1(uuid.NameSpaceOID, []byte("geo-rankly-demo-monitoring-client")) + tokenHash := sha256.Sum256([]byte("demo-monitoring-client-token")) - var installationID int64 + var ensuredID uuid.UUID err := tx.QueryRow(ctx, ` WITH ensured AS ( - INSERT INTO plugin_installations ( - tenant_id, user_id, installation_key, installation_name, browser_name, - client_version, installation_token_hash, status, last_seen_at + INSERT INTO desktop_clients ( + id, tenant_id, workspace_id, user_id, token_hash, device_name, os, cpu_arch, client_version, channel, last_seen_at ) VALUES ( - $1, $2, $3, 'GEO Monitoring Chrome', 'Chrome', - '0.1.0', $4, 'active', NOW() + $1, $2, $3, $4, $5, 'GEO Monitoring Desktop', 'macOS', 'arm64', '0.1.0', 'stable', NOW() ) - ON CONFLICT DO NOTHING + ON CONFLICT (id) DO UPDATE + SET tenant_id = EXCLUDED.tenant_id, + workspace_id = EXCLUDED.workspace_id, + user_id = EXCLUDED.user_id, + token_hash = EXCLUDED.token_hash, + device_name = EXCLUDED.device_name, + os = EXCLUDED.os, + cpu_arch = EXCLUDED.cpu_arch, + client_version = EXCLUDED.client_version, + channel = EXCLUDED.channel, + last_seen_at = NOW(), + revoked_at = NULL RETURNING id ) SELECT id FROM ensured UNION ALL SELECT id - FROM plugin_installations - WHERE tenant_id = $1 AND installation_key = $3 AND deleted_at IS NULL + FROM desktop_clients + WHERE id = $1 LIMIT 1 - `, tenantID, userID, installationKey, tokenHash).Scan(&installationID) + `, clientID, tenantID, workspaceID, userID, tokenHash[:]).Scan(&ensuredID) if err != nil { - return 0, wrapSeedErr("insert plugin installation", err) + return uuid.Nil, wrapSeedErr("insert desktop client", err) } _, err = tx.Exec(ctx, ` - UPDATE plugin_installations - SET status = 'active', - installation_token_hash = $2, + UPDATE desktop_clients + SET token_hash = $2, last_seen_at = NOW(), - updated_at = NOW() + revoked_at = NULL WHERE id = $1 - `, installationID, tokenHash) + `, ensuredID, tokenHash[:]) if err != nil { - return 0, wrapSeedErr("touch plugin installation", err) + return uuid.Nil, wrapSeedErr("touch desktop client", err) } - return installationID, nil + return ensuredID, nil } func ensureMonitoringQuota(ctx context.Context, tx pgx.Tx, state *seedState) error { @@ -689,20 +697,20 @@ func ensureMonitoringQuota(ctx context.Context, tx pgx.Tx, state *seedState) err _, err := tx.Exec(ctx, ` INSERT INTO tenant_monitoring_quotas ( tenant_id, workspace_id, max_brands, collection_mode, question_selection_mode, collect_frequency, - enabled_platforms, primary_installation_id, task_daily_hard_cap, plan_tier + enabled_platforms, primary_client_id, task_daily_hard_cap, plan_tier ) - VALUES ($1, $2, 1, 'plugin', 'best_effort_all', 'daily', $3::jsonb, $4, 36, 'free') + VALUES ($1, $2, 1, 'desktop', 'best_effort_all', 'daily', $3::jsonb, $4, 36, 'free') ON CONFLICT (tenant_id) DO UPDATE - SET workspace_id = COALESCE(tenant_monitoring_quotas.workspace_id, EXCLUDED.workspace_id), + SET workspace_id = EXCLUDED.workspace_id, collection_mode = EXCLUDED.collection_mode, question_selection_mode = EXCLUDED.question_selection_mode, collect_frequency = EXCLUDED.collect_frequency, enabled_platforms = EXCLUDED.enabled_platforms, - primary_installation_id = EXCLUDED.primary_installation_id, + primary_client_id = EXCLUDED.primary_client_id, task_daily_hard_cap = EXCLUDED.task_daily_hard_cap, plan_tier = EXCLUDED.plan_tier, updated_at = NOW() - `, state.TenantID, state.WorkspaceID, string(enabledPlatforms), state.PluginInstallationID) + `, state.TenantID, state.WorkspaceID, string(enabledPlatforms), state.DesktopClientID) return wrapSeedErr("upsert monitoring quota", err) } @@ -799,11 +807,11 @@ func insertPlatformAccessSnapshots(ctx context.Context, tx pgx.Tx, state *seedSt for _, item := range entries { if _, err := tx.Exec(ctx, ` INSERT INTO monitoring_platform_access_snapshots ( - tenant_id, installation_id, ai_platform_id, business_date, access_status, + tenant_id, client_id, ai_platform_id, business_date, access_status, connected, accessible, detected_at, reason_code, reason_message ) VALUES ($1, $2, $3, $4::date, $5, $6, $7, NOW(), $8, $9) - `, state.TenantID, state.PluginInstallationID, item.PlatformID, businessDate, item.AccessStatus, item.Connected, item.Accessible, item.ReasonCode, item.ReasonMessage); err != nil { + `, state.TenantID, state.DesktopClientID, item.PlatformID, businessDate, item.AccessStatus, item.Connected, item.Accessible, item.ReasonCode, item.ReasonMessage); err != nil { return wrapSeedErr("insert platform access snapshot", err) } } diff --git a/server/internal/tenant/app/media_service.go b/server/internal/tenant/app/media_service.go index c517888..3f4264a 100644 --- a/server/internal/tenant/app/media_service.go +++ b/server/internal/tenant/app/media_service.go @@ -4,35 +4,20 @@ import ( "context" "crypto/rand" "encoding/hex" - "encoding/json" - "errors" - "strings" "time" - "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" - "go.uber.org/zap" "github.com/geo-platform/tenant-api/internal/shared/auth" - sharedcache "github.com/geo-platform/tenant-api/internal/shared/cache" "github.com/geo-platform/tenant-api/internal/shared/response" ) -const minBrowserExtensionVersion = "0.1.0" - type MediaService struct { - pool *pgxpool.Pool - logger *zap.Logger - cache sharedcache.Cache + pool *pgxpool.Pool } -func NewMediaService(pool *pgxpool.Pool, logger *zap.Logger) *MediaService { - return &MediaService{pool: pool, logger: logger} -} - -func (s *MediaService) WithCache(c sharedcache.Cache) *MediaService { - s.cache = c - return s +func NewMediaService(pool *pgxpool.Pool) *MediaService { + return &MediaService{pool: pool} } type MediaPlatformResponse struct { @@ -48,63 +33,6 @@ type MediaPlatformResponse struct { SortOrder int `json:"sort_order"` } -type PlatformAccountResponse struct { - ID int64 `json:"id"` - PlatformID string `json:"platform_id"` - PlatformName string `json:"platform_name"` - PlatformCategory string `json:"platform_category"` - PlatformUID string `json:"platform_uid"` - Nickname string `json:"nickname"` - AvatarURL *string `json:"avatar_url"` - Status string `json:"status"` - LoginURL *string `json:"login_url"` - LastCheckAt *time.Time `json:"last_check_at"` - Metadata map[string]interface{} `json:"metadata,omitempty"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` -} - -type CreatePluginSessionRequest struct { - ActionType string `json:"action_type" binding:"required"` - PlatformID string `json:"platform_id" binding:"required"` - PlatformAccountID *int64 `json:"platform_account_id"` - PluginInstallationID *int64 `json:"plugin_installation_id"` - ResourceType string `json:"resource_type"` - ResourceID *int64 `json:"resource_id"` -} - -type CreatePluginSessionResponse struct { - PluginSessionID int64 `json:"plugin_session_id"` - SessionToken string `json:"session_token"` - ExpireAt time.Time `json:"expire_at"` - MinPluginVersion string `json:"min_plugin_version"` -} - -type CreatePublishBatchRequest struct { - PlatformAccountIDs []int64 `json:"platform_account_ids" binding:"required"` - PluginInstallationID *int64 `json:"plugin_installation_id"` - PublishType string `json:"publish_type"` - CoverAssetURL *string `json:"cover_asset_url"` -} - -type PublishBatchTask struct { - PublishRecordID int64 `json:"publish_record_id"` - PlatformAccountID int64 `json:"platform_account_id"` - PlatformID string `json:"platform_id"` - PlatformName string `json:"platform_name"` - PlatformUID string `json:"platform_uid"` - PlatformNickname string `json:"platform_nickname"` - PluginSessionID int64 `json:"plugin_session_id"` - SessionToken string `json:"session_token"` - ExpireAt time.Time `json:"expire_at"` -} - -type CreatePublishBatchResponse struct { - PublishBatchID int64 `json:"publish_batch_id"` - Status string `json:"status"` - Tasks []PublishBatchTask `json:"tasks"` -} - type PublishRecordResponse struct { ID int64 `json:"id"` PublishBatchID int64 `json:"publish_batch_id"` @@ -123,82 +51,6 @@ type PublishRecordResponse struct { UpdatedAt time.Time `json:"updated_at"` } -type PluginBindCallbackRequest struct { - PluginSessionID int64 `json:"plugin_session_id" binding:"required"` - SessionToken string `json:"session_token" binding:"required"` - PlatformID string `json:"platform_id" binding:"required"` - PlatformUID string `json:"platform_uid" binding:"required"` - Nickname string `json:"nickname" binding:"required"` - AvatarURL *string `json:"avatar_url"` - ClientVersion *string `json:"client_version"` - Metadata map[string]interface{} `json:"metadata"` - Status string `json:"status"` -} - -type PluginBindCallbackResponse struct { - PlatformAccountID int64 `json:"platform_account_id"` - Status string `json:"status"` -} - -type PluginPublishCallbackRequest struct { - PluginSessionID int64 `json:"plugin_session_id" binding:"required"` - SessionToken string `json:"session_token" binding:"required"` - ArticleID int64 `json:"article_id" binding:"required"` - PublishRecordID int64 `json:"publish_record_id" binding:"required"` - PlatformAccountID int64 `json:"platform_account_id" binding:"required"` - PlatformID string `json:"platform_id" binding:"required"` - Status string `json:"status" binding:"required"` - ExternalArticleID *string `json:"external_article_id"` - ExternalArticleURL *string `json:"external_article_url"` - ExternalManageURL *string `json:"external_manage_url"` - PublishedAt *time.Time `json:"published_at"` - ClientVersion *string `json:"client_version"` - RequestPayload map[string]interface{} `json:"request_payload"` - ResponsePayload map[string]interface{} `json:"response_payload"` - ErrorMessage *string `json:"error_message"` -} - -type PluginPublishCallbackResponse struct { - PublishRecordID int64 `json:"publish_record_id"` - BatchStatus string `json:"batch_status"` - ArticleStatus string `json:"article_status"` -} - -type RegisterPluginInstallationRequest struct { - InstallationKey string `json:"installation_key" binding:"required"` - InstallationName string `json:"installation_name" binding:"required"` - BrowserName string `json:"browser_name"` - ClientVersion string `json:"client_version"` -} - -type RegisterPluginInstallationResponse struct { - PluginInstallationID int64 `json:"plugin_installation_id"` - InstallationToken string `json:"installation_token"` - Status string `json:"status"` - LastSeenAt time.Time `json:"last_seen_at"` -} - -type platformAccountSeed struct { - ID int64 - PlatformID string - PlatformName string - PlatformUID string - Nickname string -} - -type pluginSessionRow struct { - ID int64 - TenantID int64 - UserID int64 - PluginInstallationID *int64 - ActionType string - ResourceType *string - ResourceID *int64 - PlatformAccountID *int64 - PlatformID string - ExpireAt time.Time -} - func (s *MediaService) ListPlatforms(ctx context.Context) ([]MediaPlatformResponse, error) { rows, err := s.pool.Query(ctx, ` SELECT id, platform_id, name, category, short_name, accent_color, login_url, logo_url, status, sort_order @@ -233,334 +85,6 @@ func (s *MediaService) ListPlatforms(ctx context.Context) ([]MediaPlatformRespon return items, nil } -func (s *MediaService) RegisterPluginInstallation(ctx context.Context, req RegisterPluginInstallationRequest) (*RegisterPluginInstallationResponse, error) { - actor := auth.MustActor(ctx) - installationKey := strings.TrimSpace(req.InstallationKey) - installationName := strings.TrimSpace(req.InstallationName) - if installationKey == "" || installationName == "" { - return nil, response.ErrBadRequest(40038, "invalid_plugin_installation", "installation_key and installation_name are required") - } - - installationToken, err := newSessionToken() - if err != nil { - return nil, response.ErrInternal(50060, "installation_token_failed", "failed to generate installation token") - } - tokenHash := auth.HashToken(installationToken) - lastSeenAt := time.Now().UTC() - browserName := limitStringPointer(nilIfBlank(req.BrowserName), 64) - clientVersion := limitStringPointer(nilIfBlank(req.ClientVersion), 32) - - tx, err := s.pool.Begin(ctx) - if err != nil { - return nil, response.ErrInternal(50061, "installation_begin_failed", "failed to start installation registration") - } - defer tx.Rollback(ctx) - - var installationID int64 - err = tx.QueryRow(ctx, ` - SELECT id - FROM plugin_installations - WHERE tenant_id = $1 AND installation_key = $2 AND deleted_at IS NULL - LIMIT 1 - `, actor.TenantID, installationKey).Scan(&installationID) - - switch { - case errors.Is(err, pgx.ErrNoRows): - if err := tx.QueryRow(ctx, ` - INSERT INTO plugin_installations ( - tenant_id, user_id, installation_key, installation_name, browser_name, - client_version, installation_token_hash, status, last_seen_at - ) - VALUES ($1, $2, $3, $4, $5, $6, $7, 'active', $8) - RETURNING id - `, actor.TenantID, actor.UserID, installationKey, installationName, browserName, clientVersion, tokenHash, lastSeenAt).Scan(&installationID); err != nil { - return nil, response.ErrInternal(50062, "installation_insert_failed", "failed to register plugin installation") - } - case err != nil: - return nil, response.ErrInternal(50063, "installation_lookup_failed", "failed to lookup plugin installation") - default: - if _, err := tx.Exec(ctx, ` - UPDATE plugin_installations - SET user_id = $1, - installation_name = $2, - browser_name = $3, - client_version = $4, - installation_token_hash = $5, - status = 'active', - last_seen_at = $6, - updated_at = NOW() - WHERE id = $7 - `, actor.UserID, installationName, browserName, clientVersion, tokenHash, lastSeenAt, installationID); err != nil { - return nil, response.ErrInternal(50064, "installation_update_failed", "failed to refresh plugin installation") - } - } - - if err := ensureMonitoringPrimaryInstallation(ctx, tx, actor.TenantID, installationID); err != nil { - return nil, err - } - - if err := tx.Commit(ctx); err != nil { - return nil, response.ErrInternal(50065, "installation_commit_failed", "failed to commit plugin installation") - } - - return &RegisterPluginInstallationResponse{ - PluginInstallationID: installationID, - InstallationToken: installationToken, - Status: "active", - LastSeenAt: lastSeenAt, - }, nil -} - -func ensureMonitoringPrimaryInstallation(ctx context.Context, tx pgx.Tx, tenantID, installationID int64) error { - if _, err := tx.Exec(ctx, ` - INSERT INTO tenant_monitoring_quotas ( - tenant_id, workspace_id, primary_installation_id - ) - VALUES ( - $1, - (SELECT id FROM workspaces WHERE tenant_id = $1 AND is_default = TRUE LIMIT 1), - $2 - ) - ON CONFLICT (tenant_id) DO UPDATE - SET workspace_id = COALESCE(tenant_monitoring_quotas.workspace_id, EXCLUDED.workspace_id), - primary_installation_id = COALESCE(tenant_monitoring_quotas.primary_installation_id, EXCLUDED.primary_installation_id), - updated_at = NOW() - `, tenantID, installationID); err != nil { - return response.ErrInternal(50068, "monitoring_quota_update_failed", "failed to assign monitoring plugin installation") - } - return nil -} - -func (s *MediaService) ListPlatformAccounts(ctx context.Context) ([]PlatformAccountResponse, error) { - actor := auth.MustActor(ctx) - rows, err := s.pool.Query(ctx, ` - SELECT pa.id, pa.platform_id, mp.name, mp.category, pa.platform_uid, pa.nickname, - pa.avatar_url, pa.status, mp.login_url, pa.last_check_at, pa.metadata_json, - pa.created_at, pa.updated_at - FROM platform_accounts pa - JOIN media_platforms mp ON mp.platform_id = pa.platform_id - WHERE pa.tenant_id = $1 AND pa.deleted_at IS NULL - ORDER BY mp.sort_order ASC, pa.created_at DESC - `, actor.TenantID) - if err != nil { - return nil, response.ErrInternal(50031, "platform_account_query_failed", "failed to list platform accounts") - } - defer rows.Close() - - items := make([]PlatformAccountResponse, 0) - for rows.Next() { - var item PlatformAccountResponse - var metadataJSON []byte - if err := rows.Scan( - &item.ID, - &item.PlatformID, - &item.PlatformName, - &item.PlatformCategory, - &item.PlatformUID, - &item.Nickname, - &item.AvatarURL, - &item.Status, - &item.LoginURL, - &item.LastCheckAt, - &metadataJSON, - &item.CreatedAt, - &item.UpdatedAt, - ); err != nil { - return nil, response.ErrInternal(50031, "platform_account_scan_failed", err.Error()) - } - if len(metadataJSON) > 0 { - item.Metadata = map[string]interface{}{} - _ = json.Unmarshal(metadataJSON, &item.Metadata) - } - items = append(items, item) - } - return items, nil -} - -func (s *MediaService) DeletePlatformAccount(ctx context.Context, id int64) error { - actor := auth.MustActor(ctx) - tag, err := s.pool.Exec(ctx, ` - UPDATE platform_accounts - SET deleted_at = NOW(), updated_at = NOW(), status = 'invalid' - WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL - `, id, actor.TenantID) - if err != nil { - return response.ErrInternal(50032, "platform_account_delete_failed", "failed to unbind platform account") - } - if tag.RowsAffected() == 0 { - return response.ErrNotFound(40431, "platform_account_not_found", "platform account not found") - } - return nil -} - -func (s *MediaService) CreatePluginSession(ctx context.Context, req CreatePluginSessionRequest) (*CreatePluginSessionResponse, error) { - actor := auth.MustActor(ctx) - actionType := strings.TrimSpace(req.ActionType) - platformID := strings.TrimSpace(req.PlatformID) - if actionType == "" || platformID == "" { - return nil, response.ErrBadRequest(40031, "invalid_plugin_session_request", "action_type and platform_id are required") - } - if actionType != "bind" && actionType != "check" { - return nil, response.ErrBadRequest(40032, "invalid_plugin_action", "only bind and check are supported on this endpoint") - } - if err := s.validatePluginInstallationOwnership(ctx, actor.TenantID, req.PluginInstallationID); err != nil { - return nil, err - } - - if req.PlatformAccountID != nil { - var exists bool - if err := s.pool.QueryRow(ctx, ` - SELECT EXISTS( - SELECT 1 FROM platform_accounts - WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL - ) - `, *req.PlatformAccountID, actor.TenantID).Scan(&exists); err != nil { - return nil, response.ErrInternal(50033, "platform_account_check_failed", "failed to validate platform account") - } - if !exists { - return nil, response.ErrNotFound(40431, "platform_account_not_found", "platform account not found") - } - } - - token, err := newSessionToken() - if err != nil { - return nil, response.ErrInternal(50034, "session_token_failed", "failed to generate plugin session token") - } - expireAt := time.Now().Add(10 * time.Minute).UTC() - - var id int64 - if err := s.pool.QueryRow(ctx, ` - INSERT INTO plugin_sessions ( - tenant_id, user_id, plugin_installation_id, action_type, resource_type, resource_id, platform_account_id, - platform_id, session_token, expire_at, status - ) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, 'pending') - RETURNING id - `, actor.TenantID, actor.UserID, req.PluginInstallationID, actionType, nilIfBlank(req.ResourceType), req.ResourceID, req.PlatformAccountID, platformID, token, expireAt).Scan(&id); err != nil { - return nil, response.ErrInternal(50035, "plugin_session_create_failed", "failed to create plugin session") - } - - return &CreatePluginSessionResponse{ - PluginSessionID: id, - SessionToken: token, - ExpireAt: expireAt, - MinPluginVersion: minBrowserExtensionVersion, - }, nil -} - -func (s *MediaService) CreatePublishBatch(ctx context.Context, articleID int64, req CreatePublishBatchRequest) (*CreatePublishBatchResponse, error) { - actor := auth.MustActor(ctx) - if len(req.PlatformAccountIDs) == 0 { - return nil, response.ErrBadRequest(40033, "platform_account_ids_required", "select at least one platform account") - } - if err := s.validatePluginInstallationOwnership(ctx, actor.TenantID, req.PluginInstallationID); err != nil { - return nil, err - } - - publishType := strings.TrimSpace(req.PublishType) - if publishType == "" { - publishType = "publish" - } - - tx, err := s.pool.Begin(ctx) - if err != nil { - return nil, response.ErrInternal(50036, "publish_batch_begin_failed", "failed to start publish batch") - } - defer tx.Rollback(ctx) - - articleMeta, err := loadPublishableArticleMeta(ctx, tx, actor.TenantID, articleID) - if err != nil { - return nil, err - } - - accountSeeds, err := loadPlatformAccountsForPublish(ctx, tx, actor.TenantID, req.PlatformAccountIDs) - if err != nil { - return nil, err - } - coverAssetURL := normalizeStringPointer(req.CoverAssetURL) - if publishBatchRequiresCover(accountSeeds) && strings.TrimSpace(pointerStringValue(coverAssetURL)) == "" { - return nil, response.ErrBadRequest(40044, "publish_cover_required", "baijiahao publishing requires a cover image") - } - - if coverAssetURL == nil { - coverAssetURL = articleMeta.CoverAssetURL - } - - publishBatchID, publishRecordIDs, err := createPublishBatchRecords( - ctx, - tx, - actor.TenantID, - actor.UserID, - articleID, - publishType, - coverAssetURL, - accountSeeds, - ) - if err != nil { - return nil, err - } - - tasks := make([]PublishBatchTask, 0, len(accountSeeds)) - for _, account := range accountSeeds { - token, tokenErr := newSessionToken() - if tokenErr != nil { - return nil, response.ErrInternal(50034, "session_token_failed", "failed to generate plugin session token") - } - expireAt := time.Now().Add(15 * time.Minute).UTC() - - var pluginSessionID int64 - if err := tx.QueryRow(ctx, ` - INSERT INTO plugin_sessions ( - tenant_id, user_id, plugin_installation_id, action_type, resource_type, resource_id, platform_account_id, - platform_id, session_token, expire_at, status - ) - VALUES ($1, $2, $3, 'publish', 'publish_record', $4, $5, $6, $7, $8, 'pending') - RETURNING id - `, actor.TenantID, actor.UserID, req.PluginInstallationID, publishRecordIDs[account.ID], account.ID, account.PlatformID, token, expireAt).Scan(&pluginSessionID); err != nil { - return nil, response.ErrInternal(50040, "publish_session_create_failed", "failed to create publish session") - } - - tasks = append(tasks, PublishBatchTask{ - PublishRecordID: publishRecordIDs[account.ID], - PlatformAccountID: account.ID, - PlatformID: account.PlatformID, - PlatformName: account.PlatformName, - PlatformUID: account.PlatformUID, - PlatformNickname: account.Nickname, - PluginSessionID: pluginSessionID, - SessionToken: token, - ExpireAt: expireAt, - }) - } - - if err := tx.Commit(ctx); err != nil { - return nil, response.ErrInternal(50042, "publish_batch_commit_failed", "failed to commit publish batch") - } - invalidateArticleCaches(ctx, s.cache, actor.TenantID, &articleID) - - return &CreatePublishBatchResponse{ - PublishBatchID: publishBatchID, - Status: "publishing", - Tasks: tasks, - }, nil -} - -func publishBatchRequiresCover(accounts []platformAccountSeed) bool { - for _, account := range accounts { - if strings.TrimSpace(account.PlatformID) == "baijiahao" { - return true - } - } - return false -} - -func pointerStringValue(value *string) string { - if value == nil { - return "" - } - return *value -} - func (s *MediaService) ListArticlePublishRecords(ctx context.Context, articleID int64) ([]PublishRecordResponse, error) { actor := auth.MustActor(ctx) return s.listArticlePublishRecords(ctx, actor.TenantID, articleID) @@ -609,489 +133,37 @@ func (s *MediaService) listArticlePublishRecords(ctx context.Context, tenantID, return items, nil } -func (s *MediaService) HandleBindCallback(ctx context.Context, req PluginBindCallbackRequest) (*PluginBindCallbackResponse, error) { - session, err := s.validatePluginSession(ctx, req.PluginSessionID, req.SessionToken, "bind") - if err != nil { - return nil, err - } - if session.PlatformID != req.PlatformID { - return nil, response.ErrConflict(40936, "plugin_session_platform_mismatch", "platform does not match plugin session") - } - - metadataJSON, err := marshalJSON(req.Metadata) - if err != nil { - return nil, response.ErrBadRequest(40034, "invalid_bind_metadata", "metadata must be serializable") - } - - status := strings.TrimSpace(req.Status) - if status == "" { - status = "active" - } - - tx, err := s.pool.Begin(ctx) - if err != nil { - return nil, response.ErrInternal(50044, "bind_callback_begin_failed", "failed to start bind callback transaction") - } - defer tx.Rollback(ctx) - - var platformAccountID int64 - err = tx.QueryRow(ctx, ` - SELECT id - FROM platform_accounts - WHERE tenant_id = $1 - AND workspace_id = (SELECT id FROM workspaces WHERE tenant_id = $1 AND is_default = TRUE LIMIT 1) - AND platform_id = $2 - AND platform_uid = $3 - AND deleted_at IS NULL - LIMIT 1 - `, session.TenantID, req.PlatformID, req.PlatformUID).Scan(&platformAccountID) - - switch { - case errors.Is(err, pgx.ErrNoRows): - if err := tx.QueryRow(ctx, ` - INSERT INTO platform_accounts ( - tenant_id, workspace_id, user_id, platform_id, platform_uid, nickname, avatar_url, status, - metadata_json, last_check_at, desktop_id, client_id, account_fingerprint, - display_name, health, verified_at, tags, sync_version - ) - VALUES ( - $1, - (SELECT id FROM workspaces WHERE tenant_id = $1 AND is_default = TRUE LIMIT 1), - $2, $3, $4, $5, $6, $7, $8, NOW(), - gen_random_uuid(), NULL, NULL, - $5, - CASE - WHEN $7 = 'active' THEN 'live' - WHEN $7 = 'captcha' THEN 'captcha' - WHEN $7 = 'risk' THEN 'risk' - ELSE 'expired' - END, - NOW(), - CASE - WHEN $8 IS NOT NULL AND jsonb_typeof($8 -> 'tags') = 'array' THEN $8 -> 'tags' - ELSE NULL - END, - 1 - ) - RETURNING id - `, session.TenantID, session.UserID, req.PlatformID, req.PlatformUID, req.Nickname, req.AvatarURL, status, metadataJSON).Scan(&platformAccountID); err != nil { - return nil, response.ErrInternal(50045, "platform_account_insert_failed", "failed to create platform account") - } - case err != nil: - return nil, response.ErrInternal(50046, "platform_account_lookup_failed", "failed to lookup platform account") - default: - if _, err := tx.Exec(ctx, ` - UPDATE platform_accounts - SET user_id = $1, - nickname = $2, - display_name = $2, - avatar_url = $3, - status = $4, - health = CASE - WHEN $4 = 'active' THEN 'live' - WHEN $4 = 'captcha' THEN 'captcha' - WHEN $4 = 'risk' THEN 'risk' - ELSE 'expired' - END, - metadata_json = $5, - last_check_at = NOW(), - verified_at = NOW(), - tags = CASE - WHEN $5 IS NOT NULL AND jsonb_typeof($5 -> 'tags') = 'array' THEN $5 -> 'tags' - ELSE tags - END, - sync_version = sync_version + 1, - updated_at = NOW() - WHERE id = $6 - `, session.UserID, req.Nickname, req.AvatarURL, status, metadataJSON, platformAccountID); err != nil { - return nil, response.ErrInternal(50047, "platform_account_update_failed", "failed to update platform account") +func publishBatchRequiresCover(accounts []platformAccountSeed) bool { + for _, account := range accounts { + if account.PlatformID == "baijiahao" { + return true } } - - if _, err := tx.Exec(ctx, ` - UPDATE plugin_sessions - SET client_version = COALESCE($1, client_version), status = 'completed', updated_at = NOW() - WHERE id = $2 - `, req.ClientVersion, session.ID); err != nil { - return nil, response.ErrInternal(50048, "plugin_session_update_failed", "failed to finalize bind session") - } - if err := touchPluginInstallation(ctx, tx, session.PluginInstallationID); err != nil { - return nil, err - } - - if err := tx.Commit(ctx); err != nil { - return nil, response.ErrInternal(50049, "bind_callback_commit_failed", "failed to commit bind callback") - } - - return &PluginBindCallbackResponse{ - PlatformAccountID: platformAccountID, - Status: status, - }, nil + return false } -func (s *MediaService) HandlePublishCallback(ctx context.Context, req PluginPublishCallbackRequest) (*PluginPublishCallbackResponse, error) { - session, err := s.validatePluginSession(ctx, req.PluginSessionID, req.SessionToken, "publish") - if err != nil { - return nil, err +func pointerStringValue(value *string) string { + if value == nil { + return "" } - if session.ResourceType == nil || *session.ResourceType != "publish_record" || session.ResourceID == nil || *session.ResourceID != req.PublishRecordID { - return nil, response.ErrConflict(40931, "plugin_session_resource_mismatch", "publish record does not match plugin session") - } - if session.PlatformAccountID == nil || *session.PlatformAccountID != req.PlatformAccountID { - return nil, response.ErrConflict(40932, "plugin_session_account_mismatch", "platform account does not match plugin session") - } - if session.PlatformID != req.PlatformID { - return nil, response.ErrConflict(40933, "plugin_session_platform_mismatch", "platform does not match plugin session") - } - - reqStatus := normalizePublishStatus(req.Status) - requestPayloadJSON, err := marshalJSON(req.RequestPayload) - if err != nil { - return nil, response.ErrBadRequest(40035, "invalid_publish_request_payload", "request_payload must be serializable") - } - responsePayloadJSON, err := marshalJSON(req.ResponsePayload) - if err != nil { - return nil, response.ErrBadRequest(40036, "invalid_publish_response_payload", "response_payload must be serializable") - } - - tx, err := s.pool.Begin(ctx) - if err != nil { - return nil, response.ErrInternal(50050, "publish_callback_begin_failed", "failed to start publish callback transaction") - } - defer tx.Rollback(ctx) - - var publishBatchID int64 - var articleID int64 - if err := tx.QueryRow(ctx, ` - SELECT publish_batch_id, article_id - FROM publish_records - WHERE id = $1 AND tenant_id = $2 AND platform_account_id = $3 - `, req.PublishRecordID, session.TenantID, req.PlatformAccountID).Scan(&publishBatchID, &articleID); err != nil { - if errors.Is(err, pgx.ErrNoRows) { - return nil, response.ErrNotFound(40432, "publish_record_not_found", "publish record not found") - } - return nil, response.ErrInternal(50051, "publish_record_lookup_failed", "failed to lookup publish record") - } - if req.ArticleID != articleID { - return nil, response.ErrConflict(40934, "publish_article_mismatch", "article does not match publish record") - } - - publishedAt := req.PublishedAt - if reqStatus == "success" && publishedAt == nil { - now := time.Now().UTC() - publishedAt = &now - } - - if reqStatus != "success" && s.logger != nil { - s.logger.Warn("publisher callback reported non-success status", - zap.Int64("plugin_session_id", req.PluginSessionID), - zap.Int64("publish_record_id", req.PublishRecordID), - zap.Int64("platform_account_id", req.PlatformAccountID), - zap.Int64("article_id", req.ArticleID), - zap.String("platform_id", req.PlatformID), - zap.String("status", reqStatus), - zap.Stringp("error_message", req.ErrorMessage), - zap.Stringp("external_article_id", req.ExternalArticleID), - zap.Stringp("external_article_url", req.ExternalArticleURL), - zap.Any("request_payload", req.RequestPayload), - zap.Any("response_payload", req.ResponsePayload), - ) - } - - if _, err := tx.Exec(ctx, ` - UPDATE publish_records - SET status = $1, - external_article_id = $2, - external_article_url = $3, - external_manage_url = $4, - published_at = $5, - request_payload_json = $6, - response_payload_json = $7, - error_message = $8, - updated_at = NOW() - WHERE id = $9 - `, reqStatus, normalizeStringPointer(req.ExternalArticleID), normalizeStringPointer(req.ExternalArticleURL), normalizeStringPointer(req.ExternalManageURL), publishedAt, requestPayloadJSON, responsePayloadJSON, normalizeStringPointer(req.ErrorMessage), req.PublishRecordID); err != nil { - return nil, response.ErrInternal(50052, "publish_record_update_failed", "failed to update publish record") - } - - pluginStatus := "completed" - if reqStatus == "failed" { - pluginStatus = "failed" - } - if _, err := tx.Exec(ctx, ` - UPDATE plugin_sessions - SET client_version = COALESCE($1, client_version), status = $2, updated_at = NOW() - WHERE id = $3 - `, req.ClientVersion, pluginStatus, session.ID); err != nil { - return nil, response.ErrInternal(50053, "plugin_session_finalize_failed", "failed to finalize publish session") - } - if err := touchPluginInstallation(ctx, tx, session.PluginInstallationID); err != nil { - return nil, err - } - - batchStatus, err := recalculatePublishBatchStatus(ctx, tx, publishBatchID) - if err != nil { - return nil, err - } - if _, err := tx.Exec(ctx, ` - UPDATE publish_batches - SET status = $1, updated_at = NOW() - WHERE id = $2 - `, batchStatus, publishBatchID); err != nil { - return nil, response.ErrInternal(50054, "publish_batch_update_failed", "failed to update publish batch") - } - - articleStatus := batchStatus - if batchStatus == "success" { - articleStatus = "success" - } - if batchStatus == "failed" { - articleStatus = "failed" - } - if _, err := tx.Exec(ctx, ` - UPDATE articles - SET publish_status = $1, updated_at = NOW() - WHERE id = $2 AND tenant_id = $3 AND deleted_at IS NULL - `, articleStatus, articleID, session.TenantID); err != nil { - return nil, response.ErrInternal(50055, "article_publish_update_failed", "failed to update article publish status") - } - - if err := tx.Commit(ctx); err != nil { - return nil, response.ErrInternal(50056, "publish_callback_commit_failed", "failed to commit publish callback") - } - invalidateArticleCaches(ctx, s.cache, session.TenantID, &articleID) - - return &PluginPublishCallbackResponse{ - PublishRecordID: req.PublishRecordID, - BatchStatus: batchStatus, - ArticleStatus: articleStatus, - }, nil -} - -func (s *MediaService) validatePluginSession(ctx context.Context, pluginSessionID int64, sessionToken, expectedAction string) (*pluginSessionRow, error) { - sessionToken = strings.TrimSpace(sessionToken) - if sessionToken == "" { - return nil, response.ErrUnauthorized(40131, "plugin_session_invalid", "plugin session token is required") - } - - row := &pluginSessionRow{} - err := s.pool.QueryRow(ctx, ` - SELECT id, tenant_id, user_id, plugin_installation_id, action_type, resource_type, resource_id, platform_account_id, platform_id, expire_at - FROM plugin_sessions - WHERE id = $1 AND session_token = $2 - LIMIT 1 - `, pluginSessionID, sessionToken).Scan( - &row.ID, - &row.TenantID, - &row.UserID, - &row.PluginInstallationID, - &row.ActionType, - &row.ResourceType, - &row.ResourceID, - &row.PlatformAccountID, - &row.PlatformID, - &row.ExpireAt, - ) - if err != nil { - if errors.Is(err, pgx.ErrNoRows) { - return nil, response.ErrUnauthorized(40132, "plugin_session_not_found", "plugin session not found") - } - return nil, response.ErrInternal(50057, "plugin_session_lookup_failed", "failed to lookup plugin session") - } - if row.ActionType != expectedAction { - return nil, response.ErrConflict(40935, "plugin_session_action_mismatch", "plugin session action mismatch") - } - if time.Now().UTC().After(row.ExpireAt) { - return nil, response.ErrUnauthorized(40133, "plugin_session_expired", "plugin session has expired") - } - return row, nil -} - -func (s *MediaService) validatePluginInstallationOwnership(ctx context.Context, tenantID int64, pluginInstallationID *int64) error { - if pluginInstallationID == nil { - return nil - } - - var exists bool - if err := s.pool.QueryRow(ctx, ` - SELECT EXISTS( - SELECT 1 - FROM plugin_installations - WHERE id = $1 AND tenant_id = $2 AND status = 'active' AND deleted_at IS NULL - ) - `, *pluginInstallationID, tenantID).Scan(&exists); err != nil { - return response.ErrInternal(50066, "plugin_installation_lookup_failed", "failed to validate plugin installation") - } - if !exists { - return response.ErrNotFound(40433, "plugin_installation_not_found", "plugin installation not found") - } - return nil -} - -func touchPluginInstallation(ctx context.Context, tx pgx.Tx, pluginInstallationID *int64) error { - if pluginInstallationID == nil { - return nil - } - if _, err := tx.Exec(ctx, ` - UPDATE plugin_installations - SET last_seen_at = NOW(), updated_at = NOW() - WHERE id = $1 - `, *pluginInstallationID); err != nil { - return response.ErrInternal(50067, "plugin_installation_touch_failed", "failed to update plugin installation activity") - } - return nil -} - -func loadPlatformAccountsForPublish(ctx context.Context, tx pgx.Tx, tenantID int64, ids []int64) ([]platformAccountSeed, error) { - rows, err := tx.Query(ctx, ` - SELECT pa.id, pa.platform_id, mp.name, pa.platform_uid, pa.nickname - FROM platform_accounts pa - JOIN media_platforms mp ON mp.platform_id = pa.platform_id - WHERE pa.tenant_id = $1 - AND pa.deleted_at IS NULL - AND pa.status = 'active' - AND pa.id = ANY($2) - ORDER BY mp.sort_order ASC, pa.id ASC - `, tenantID, ids) - if err != nil { - return nil, response.ErrInternal(50058, "platform_account_publish_query_failed", "failed to load platform accounts") - } - defer rows.Close() - - items := make([]platformAccountSeed, 0, len(ids)) - for rows.Next() { - var item platformAccountSeed - if err := rows.Scan(&item.ID, &item.PlatformID, &item.PlatformName, &item.PlatformUID, &item.Nickname); err != nil { - return nil, response.ErrInternal(50058, "platform_account_publish_scan_failed", err.Error()) - } - items = append(items, item) - } - if len(items) != len(uniqueInt64(ids)) { - return nil, response.ErrBadRequest(40037, "invalid_platform_accounts", "one or more selected platform accounts are unavailable") - } - return items, nil -} - -func recalculatePublishBatchStatus(ctx context.Context, tx pgx.Tx, publishBatchID int64) (string, error) { - rows, err := tx.Query(ctx, ` - SELECT status - FROM publish_records - WHERE publish_batch_id = $1 - `, publishBatchID) - if err != nil { - return "", response.ErrInternal(50059, "publish_batch_status_query_failed", "failed to aggregate publish batch status") - } - defer rows.Close() - - statuses := make([]string, 0) - for rows.Next() { - var status string - if err := rows.Scan(&status); err != nil { - return "", response.ErrInternal(50059, "publish_batch_status_scan_failed", err.Error()) - } - statuses = append(statuses, normalizePublishStatus(status)) - } - return aggregatePublishStatuses(statuses), nil -} - -func aggregatePublishStatuses(statuses []string) string { - if len(statuses) == 0 { - return "pending" - } - - counts := map[string]int{} - for _, status := range statuses { - counts[normalizePublishStatus(status)]++ - } - - if counts["queued"] > 0 || counts["publishing"] > 0 { - return "publishing" - } - if counts["failed"] == len(statuses) { - return "failed" - } - if counts["success"] == len(statuses) { - return "success" - } - if counts["failed"] > 0 && counts["success"] > 0 { - return "partial_success" - } - if counts["success"] > 0 { - return "success" - } - return "publishing" + return *value } func normalizePublishStatus(status string) string { - switch strings.TrimSpace(status) { - case "queued": - return "queued" - case "publishing", "running": - return "publishing" + switch status { case "success", "published", "publish_success": return "success" - case "failed", "publish_failed", "pending_review": - return "failed" + case "publishing", "pending", "queued", "running": + return "publishing" default: return "failed" } } -func normalizeStringPointer(value *string) *string { - if value == nil { - return nil - } - trimmed := strings.TrimSpace(*value) - if trimmed == "" { - return nil - } - return &trimmed -} - -func nilIfBlank(value string) *string { - value = strings.TrimSpace(value) - if value == "" { - return nil - } - return &value -} - -func limitStringPointer(value *string, maxLen int) *string { - if value == nil || maxLen <= 0 { - return value - } - runes := []rune(*value) - if len(runes) <= maxLen { - return value - } - trimmed := string(runes[:maxLen]) - return &trimmed -} - -func marshalJSON(value interface{}) ([]byte, error) { - if value == nil { - return nil, nil - } - return json.Marshal(value) -} - func newSessionToken() (string, error) { - buf := make([]byte, 24) + buf := make([]byte, 16) if _, err := rand.Read(buf); err != nil { return "", err } return hex.EncodeToString(buf), nil } - -func uniqueInt64(values []int64) []int64 { - seen := make(map[int64]struct{}, len(values)) - result := make([]int64, 0, len(values)) - for _, value := range values { - if _, ok := seen[value]; ok { - continue - } - seen[value] = struct{}{} - result = append(result, value) - } - return result -} diff --git a/server/internal/tenant/app/monitoring_callback_service.go b/server/internal/tenant/app/monitoring_callback_service.go index c184e09..b152b17 100644 --- a/server/internal/tenant/app/monitoring_callback_service.go +++ b/server/internal/tenant/app/monitoring_callback_service.go @@ -100,7 +100,7 @@ type MonitoringHeartbeatPlatform struct { } type MonitoringHeartbeatResponse struct { - InstallationID int64 `json:"installation_id"` + InstallationID string `json:"installation_id"` BusinessDate string `json:"business_date"` UpdatedPlatformCount int `json:"updated_platform_count"` ReconciledSkippedTaskCount int64 `json:"reconciled_skipped_task_count"` @@ -166,8 +166,9 @@ type MonitoringSkipTaskResponse struct { } type monitoringInstallationIdentity struct { - ID int64 - TenantID int64 + ID string + TenantID int64 + WorkspaceID int64 } type monitoringPlatformAccessSnapshotState struct { @@ -236,7 +237,7 @@ type monitoringTaskResultEnvelope struct { Version string `json:"version"` TaskID int64 `json:"task_id"` TenantID int64 `json:"tenant_id"` - InstallationID int64 `json:"installation_id"` + InstallationID string `json:"installation_id"` ReceivedAt time.Time `json:"received_at"` QueueStatus string `json:"queue_status"` QueueClaimedAt *time.Time `json:"queue_claimed_at"` @@ -256,7 +257,7 @@ type monitoringTaskResultQueueMetaPatch struct { type MonitoringTaskResultQueueMetaPatch = monitoringTaskResultQueueMetaPatch -func (s *MonitoringCallbackService) HandleTaskResult(ctx context.Context, installationID int64, installationToken string, taskID int64, req MonitoringTaskResultRequest) (*MonitoringTaskResultResponse, error) { +func (s *MonitoringCallbackService) HandleTaskResult(ctx context.Context, installationID string, installationToken string, taskID int64, req MonitoringTaskResultRequest) (*MonitoringTaskResultResponse, error) { installation, err := s.authenticateInstallation(ctx, installationID, installationToken) if err != nil { return nil, err @@ -268,10 +269,10 @@ func (s *MonitoringCallbackService) HandleTaskResult(ctx context.Context, instal } if task.CollectorType != monitoringCollectorType { - return nil, response.ErrConflict(40941, "collector_type_mismatch", "task collector type is not plugin") + return nil, response.ErrConflict(40941, "collector_type_mismatch", "task collector type is not desktop") } - if task.LeasedToExecutor.Valid && strings.TrimSpace(task.LeasedToExecutor.String) != "" && task.LeasedToExecutor.String != strconv.FormatInt(installation.ID, 10) { + if task.LeasedToExecutor.Valid && strings.TrimSpace(task.LeasedToExecutor.String) != "" && task.LeasedToExecutor.String != installation.ID { return nil, response.ErrUnauthorized(40141, "lease_owner_mismatch", "task is leased to another installation") } @@ -356,13 +357,13 @@ func (s *MonitoringCallbackService) HandleTaskResult(ctx context.Context, instal }, nil } -func (s *MonitoringCallbackService) Heartbeat(ctx context.Context, installationID int64, installationToken string, req MonitoringHeartbeatRequest) (*MonitoringHeartbeatResponse, error) { +func (s *MonitoringCallbackService) Heartbeat(ctx context.Context, installationID string, installationToken string, req MonitoringHeartbeatRequest) (*MonitoringHeartbeatResponse, error) { installation, err := s.authenticateInstallation(ctx, installationID, installationToken) if err != nil { return nil, err } - isPrimaryInstallation, err := s.resolveHeartbeatPrimaryInstallation(ctx, installation.TenantID, installation.ID) + isPrimaryInstallation, err := s.resolveHeartbeatPrimaryInstallation(ctx, installation.TenantID, installation) if err != nil { return nil, err } @@ -442,7 +443,7 @@ func (s *MonitoringCallbackService) Heartbeat(ctx context.Context, installationI }, nil } -func (s *MonitoringCallbackService) LeaseTasks(ctx context.Context, installationID int64, installationToken string, req MonitoringLeaseTasksRequest) (*MonitoringLeaseTasksResponse, error) { +func (s *MonitoringCallbackService) LeaseTasks(ctx context.Context, installationID string, installationToken string, req MonitoringLeaseTasksRequest) (*MonitoringLeaseTasksResponse, error) { installation, err := s.authenticateInstallation(ctx, installationID, installationToken) if err != nil { return nil, err @@ -522,7 +523,7 @@ func (s *MonitoringCallbackService) LeaseTasks(ctx context.Context, installation }, nil } -func (s *MonitoringCallbackService) ResumeTasks(ctx context.Context, installationID int64, installationToken string, req MonitoringResumeTasksRequest) (*MonitoringResumeTasksResponse, error) { +func (s *MonitoringCallbackService) ResumeTasks(ctx context.Context, installationID string, installationToken string, req MonitoringResumeTasksRequest) (*MonitoringResumeTasksResponse, error) { installation, err := s.authenticateInstallation(ctx, installationID, installationToken) if err != nil { return nil, err @@ -601,7 +602,7 @@ func (s *MonitoringCallbackService) ResumeTasks(ctx context.Context, installatio }, nil } -func (s *MonitoringCallbackService) SkipTask(ctx context.Context, installationID int64, installationToken string, taskID int64, req MonitoringSkipTaskRequest) (*MonitoringSkipTaskResponse, error) { +func (s *MonitoringCallbackService) SkipTask(ctx context.Context, installationID string, installationToken string, taskID int64, req MonitoringSkipTaskRequest) (*MonitoringSkipTaskResponse, error) { installation, err := s.authenticateInstallation(ctx, installationID, installationToken) if err != nil { return nil, err @@ -704,48 +705,47 @@ func (s *MonitoringCallbackService) ProcessQueuedTaskResult(ctx context.Context, return s.processTaskResult(ctx, task, envelope.CallbackResult) } -func (s *MonitoringCallbackService) authenticateInstallation(ctx context.Context, installationID int64, installationToken string) (*monitoringInstallationIdentity, error) { - tokenHash := sharedauth.HashToken(strings.TrimSpace(installationToken)) +func (s *MonitoringCallbackService) authenticateInstallation(ctx context.Context, installationID string, installationToken string) (*monitoringInstallationIdentity, error) { + tokenHash := HashDesktopClientToken(strings.TrimSpace(installationToken)) row := &monitoringInstallationIdentity{} if err := s.businessPool.QueryRow(ctx, ` - SELECT id, tenant_id - FROM plugin_installations - WHERE id = $1 - AND installation_token_hash = $2 - AND status = 'active' - AND deleted_at IS NULL - `, installationID, tokenHash).Scan(&row.ID, &row.TenantID); err != nil { + SELECT id::text, tenant_id, workspace_id + FROM desktop_clients + WHERE id = $1::uuid + AND token_hash = $2 + AND revoked_at IS NULL + `, strings.TrimSpace(installationID), tokenHash).Scan(&row.ID, &row.TenantID, &row.WorkspaceID); err != nil { if err == pgx.ErrNoRows { - return nil, response.ErrUnauthorized(40141, "plugin_installation_unauthorized", "plugin installation is invalid or expired") + return nil, response.ErrUnauthorized(40141, "desktop_client_unauthorized", "desktop client is invalid or expired") } - return nil, response.ErrInternal(50041, "plugin_installation_lookup_failed", "failed to validate plugin installation") + return nil, response.ErrInternal(50041, "desktop_client_lookup_failed", "failed to validate desktop client") } return row, nil } -func (s *MonitoringCallbackService) resolveHeartbeatPrimaryInstallation(ctx context.Context, tenantID, installationID int64) (bool, error) { - var preferred sql.NullInt64 +func (s *MonitoringCallbackService) resolveHeartbeatPrimaryInstallation(ctx context.Context, tenantID int64, installation *monitoringInstallationIdentity) (bool, error) { + var preferred sql.NullString err := s.businessPool.QueryRow(ctx, ` - SELECT primary_installation_id + SELECT primary_client_id::text FROM tenant_monitoring_quotas WHERE tenant_id = $1 `, tenantID).Scan(&preferred) if err != nil && err != pgx.ErrNoRows { - return false, response.ErrInternal(50041, "query_failed", "failed to inspect primary monitoring installation") + return false, response.ErrInternal(50041, "query_failed", "failed to inspect primary monitoring desktop client") } - if !preferred.Valid || preferred.Int64 == installationID { + if !preferred.Valid || preferred.String == installation.ID { if !preferred.Valid { - if err := s.persistHeartbeatPrimaryInstallationID(ctx, tenantID, installationID); err != nil { + if err := s.persistHeartbeatPrimaryInstallationID(ctx, tenantID, installation); err != nil { return false, err } } return true, nil } - online, err := s.isMonitoringInstallationRecentlyOnline(ctx, tenantID, preferred.Int64) + online, err := s.isMonitoringInstallationRecentlyOnline(ctx, tenantID, installation.WorkspaceID, preferred.String) if err != nil { return false, err } @@ -753,27 +753,27 @@ func (s *MonitoringCallbackService) resolveHeartbeatPrimaryInstallation(ctx cont return false, nil } - if err := s.persistHeartbeatPrimaryInstallationID(ctx, tenantID, installationID); err != nil { + if err := s.persistHeartbeatPrimaryInstallationID(ctx, tenantID, installation); err != nil { return false, err } return true, nil } -func (s *MonitoringCallbackService) isMonitoringInstallationRecentlyOnline(ctx context.Context, tenantID, installationID int64) (bool, error) { +func (s *MonitoringCallbackService) isMonitoringInstallationRecentlyOnline(ctx context.Context, tenantID, workspaceID int64, installationID string) (bool, error) { var lastSeenAt sql.NullTime err := s.businessPool.QueryRow(ctx, ` SELECT last_seen_at - FROM plugin_installations - WHERE id = $1 + FROM desktop_clients + WHERE id = $1::uuid AND tenant_id = $2 - AND status = 'active' - AND deleted_at IS NULL - `, installationID, tenantID).Scan(&lastSeenAt) + AND workspace_id = $3 + AND revoked_at IS NULL + `, strings.TrimSpace(installationID), tenantID, workspaceID).Scan(&lastSeenAt) if err != nil { if err == pgx.ErrNoRows { return false, nil } - return false, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring installation heartbeat") + return false, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring desktop client heartbeat") } if !lastSeenAt.Valid { @@ -782,22 +782,22 @@ func (s *MonitoringCallbackService) isMonitoringInstallationRecentlyOnline(ctx c return time.Since(lastSeenAt.Time) <= monitoringOnlineDuration, nil } -func (s *MonitoringCallbackService) persistHeartbeatPrimaryInstallationID(ctx context.Context, tenantID, installationID int64) error { +func (s *MonitoringCallbackService) persistHeartbeatPrimaryInstallationID(ctx context.Context, tenantID int64, installation *monitoringInstallationIdentity) error { if _, err := s.businessPool.Exec(ctx, ` INSERT INTO tenant_monitoring_quotas ( - tenant_id, workspace_id, primary_installation_id + tenant_id, workspace_id, primary_client_id ) VALUES ( $1, - (SELECT id FROM workspaces WHERE tenant_id = $1 AND is_default = TRUE LIMIT 1), - $2 + $2, + $3::uuid ) ON CONFLICT (tenant_id) DO UPDATE - SET workspace_id = COALESCE(tenant_monitoring_quotas.workspace_id, EXCLUDED.workspace_id), - primary_installation_id = EXCLUDED.primary_installation_id, + SET workspace_id = EXCLUDED.workspace_id, + primary_client_id = EXCLUDED.primary_client_id, updated_at = NOW() - `, tenantID, installationID); err != nil { - return response.ErrInternal(50041, "update_failed", "failed to persist primary monitoring installation") + `, tenantID, installation.WorkspaceID, installation.ID); err != nil { + return response.ErrInternal(50041, "update_failed", "failed to persist primary monitoring desktop client") } return nil } @@ -1020,11 +1020,12 @@ func (s *MonitoringCallbackService) selectPendingLeaseTasks( func (s *MonitoringCallbackService) selectLeasedTasksForInstallation( ctx context.Context, tx pgx.Tx, - tenantID, installationID int64, + tenantID int64, + installationID string, businessDate string, eligiblePlatformIDs []string, ) ([]monitoringLeasedTaskRow, error) { - queryArgs := []interface{}{tenantID, monitoringCollectorType, businessDate, strconv.FormatInt(installationID, 10)} + queryArgs := []interface{}{tenantID, monitoringCollectorType, businessDate, installationID} query := ` SELECT t.id, @@ -1108,7 +1109,7 @@ func (s *MonitoringCallbackService) selectLeasedTasksForInstallation( return result, nil } -func (s *MonitoringCallbackService) markTaskLeased(ctx context.Context, tx pgx.Tx, installationID, taskID int64, leaseToken string, leasedAt, leaseExpiresAt time.Time) error { +func (s *MonitoringCallbackService) markTaskLeased(ctx context.Context, tx pgx.Tx, installationID string, taskID int64, leaseToken string, leasedAt, leaseExpiresAt time.Time) error { if _, err := tx.Exec(ctx, ` UPDATE monitoring_collect_tasks SET status = 'leased', @@ -1120,7 +1121,7 @@ func (s *MonitoringCallbackService) markTaskLeased(ctx context.Context, tx pgx.T error_message = NULL, updated_at = NOW() WHERE id = $1 - `, taskID, sharedauth.HashToken(strings.TrimSpace(leaseToken)), strconv.FormatInt(installationID, 10), leasedAt, leaseExpiresAt); err != nil { + `, taskID, sharedauth.HashToken(strings.TrimSpace(leaseToken)), installationID, leasedAt, leaseExpiresAt); err != nil { return response.ErrInternal(50041, "lease_update_failed", "failed to update monitoring lease task") } return nil @@ -1165,11 +1166,11 @@ func (s *MonitoringCallbackService) upsertPlatformAccessSnapshot(ctx context.Con if _, err := tx.Exec(ctx, ` INSERT INTO monitoring_platform_access_snapshots ( - tenant_id, installation_id, ai_platform_id, business_date, access_status, + tenant_id, client_id, ai_platform_id, business_date, access_status, connected, accessible, detected_at, reason_code, reason_message ) - VALUES ($1, $2, $3, $4::date, $5, $6, $7, $8, $9, $10) - ON CONFLICT (tenant_id, installation_id, ai_platform_id, business_date) + VALUES ($1, $2::uuid, $3, $4::date, $5, $6, $7, $8, $9, $10) + ON CONFLICT (tenant_id, client_id, ai_platform_id, business_date) DO UPDATE SET access_status = EXCLUDED.access_status, connected = EXCLUDED.connected, @@ -1195,7 +1196,7 @@ func (s *MonitoringCallbackService) upsertPlatformAccessSnapshot(ctx context.Con return hasPlatformAccessSnapshotChanged(previous, nextState), nil } -func (s *MonitoringCallbackService) loadPlatformAccessSnapshotState(ctx context.Context, tx pgx.Tx, tenantID, installationID int64, businessDate, aiPlatformID string) (*monitoringPlatformAccessSnapshotState, error) { +func (s *MonitoringCallbackService) loadPlatformAccessSnapshotState(ctx context.Context, tx pgx.Tx, tenantID int64, installationID string, businessDate, aiPlatformID string) (*monitoringPlatformAccessSnapshotState, error) { item := &monitoringPlatformAccessSnapshotState{} var reasonCode sql.NullString var reasonMessage sql.NullString @@ -1203,7 +1204,7 @@ func (s *MonitoringCallbackService) loadPlatformAccessSnapshotState(ctx context. SELECT access_status, connected, accessible, reason_code, reason_message FROM monitoring_platform_access_snapshots WHERE tenant_id = $1 - AND installation_id = $2 + AND client_id = $2::uuid AND ai_platform_id = $3 AND business_date = $4::date `, tenantID, installationID, aiPlatformID, businessDate).Scan( @@ -1721,11 +1722,11 @@ func (s *MonitoringCallbackService) loadExistingRunSummary(ctx context.Context, return &value, citationSourceCount, contentCitationCount, nil } -func (s *MonitoringCallbackService) touchInstallation(ctx context.Context, installationID int64) error { +func (s *MonitoringCallbackService) touchInstallation(ctx context.Context, installationID string) error { _, err := s.businessPool.Exec(ctx, ` - UPDATE plugin_installations - SET last_seen_at = NOW(), updated_at = NOW() - WHERE id = $1 + UPDATE desktop_clients + SET last_seen_at = NOW() + WHERE id = $1::uuid `, installationID) return err } @@ -1821,7 +1822,7 @@ func buildMonitoringRawPayload(req MonitoringTaskResultRequest) ([]byte, []Monit return data, sourceInputs } -func newMonitoringTaskResultEnvelope(task *monitoringCollectTask, installationID int64, receivedAt time.Time, req MonitoringTaskResultRequest) monitoringTaskResultEnvelope { +func newMonitoringTaskResultEnvelope(task *monitoringCollectTask, installationID string, receivedAt time.Time, req MonitoringTaskResultRequest) monitoringTaskResultEnvelope { return monitoringTaskResultEnvelope{ Version: "v1", TaskID: task.ID, @@ -2090,8 +2091,8 @@ func normalizeMonitoringAccessStatus(accessStatus *string, connected *bool, acce return "", false, false, response.ErrBadRequest(40041, "invalid_access_status", "access status requires access_status, connected, or accessible") } -func validateTaskLease(task *monitoringCollectTask, installationID int64, leaseToken string) error { - if task.LeasedToExecutor.Valid && strings.TrimSpace(task.LeasedToExecutor.String) != "" && task.LeasedToExecutor.String != strconv.FormatInt(installationID, 10) { +func validateTaskLease(task *monitoringCollectTask, installationID string, leaseToken string) error { + if task.LeasedToExecutor.Valid && strings.TrimSpace(task.LeasedToExecutor.String) != "" && task.LeasedToExecutor.String != installationID { return response.ErrUnauthorized(40141, "lease_owner_mismatch", "task is leased to another installation") } diff --git a/server/internal/tenant/app/monitoring_result_ingest_worker_test.go b/server/internal/tenant/app/monitoring_result_ingest_worker_test.go index 0057441..d2232c4 100644 --- a/server/internal/tenant/app/monitoring_result_ingest_worker_test.go +++ b/server/internal/tenant/app/monitoring_result_ingest_worker_test.go @@ -48,12 +48,12 @@ func TestNewMonitoringTaskResultEnvelope(t *testing.T) { TenantID: 34, } - envelope := newMonitoringTaskResultEnvelope(task, 56, receivedAt, req) + envelope := newMonitoringTaskResultEnvelope(task, "00000000-0000-0000-0000-000000000056", receivedAt, req) assert.Equal(t, "v1", envelope.Version) assert.Equal(t, int64(12), envelope.TaskID) assert.Equal(t, int64(34), envelope.TenantID) - assert.Equal(t, int64(56), envelope.InstallationID) + assert.Equal(t, "00000000-0000-0000-0000-000000000056", envelope.InstallationID) assert.Equal(t, receivedAt, envelope.ReceivedAt) assert.Equal(t, "pending", envelope.QueueStatus) assert.Nil(t, envelope.QueueEnqueuedAt) @@ -69,7 +69,7 @@ func TestDecodeMonitoringTaskResultEnvelope(t *testing.T) { Version: "v1", TaskID: 12, TenantID: 34, - InstallationID: 56, + InstallationID: "00000000-0000-0000-0000-000000000056", ReceivedAt: time.Date(2026, 4, 10, 12, 0, 0, 0, time.UTC), QueueStatus: "queued", CallbackResult: MonitoringTaskResultRequest{LeaseToken: "lease-token", Status: "succeeded"}, diff --git a/server/internal/tenant/app/monitoring_service.go b/server/internal/tenant/app/monitoring_service.go index c91f07c..25cad54 100644 --- a/server/internal/tenant/app/monitoring_service.go +++ b/server/internal/tenant/app/monitoring_service.go @@ -10,6 +10,7 @@ import ( "strings" "time" + "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" @@ -20,7 +21,7 @@ import ( const ( defaultTrackingDays = 7 maxTrackingDays = 30 - monitoringCollectorType = "plugin" + monitoringCollectorType = "desktop" monitoringOnlineDuration = 20 * time.Minute monitoringCollectNowQuestionLimit = 5 ) @@ -189,9 +190,9 @@ type MonitoringCollectNowResponse struct { } type monitoringQuotaConfig struct { - CollectionMode string - PrimaryInstallationID *int64 - EnabledPlatforms []string + CollectionMode string + PrimaryClientID *uuid.UUID + EnabledPlatforms []string } type monitoringBrandIdentity struct { @@ -301,7 +302,7 @@ func (s *MonitoringService) DashboardComposite( return nil, err } - accessStates, err := s.loadAccessStates(ctx, actor.TenantID, quota.PrimaryInstallationID, endDate) + accessStates, err := s.loadAccessStates(ctx, actor.TenantID, quota.PrimaryClientID, endDate) if err != nil { return nil, err } @@ -378,7 +379,7 @@ func (s *MonitoringService) QuestionDetail(ctx context.Context, brandID, questio } } - accessStates, err := s.loadAccessStates(ctx, actor.TenantID, quota.PrimaryInstallationID, toDate) + accessStates, err := s.loadAccessStates(ctx, actor.TenantID, quota.PrimaryClientID, toDate) if err != nil { return nil, err } @@ -497,15 +498,15 @@ func (s *MonitoringService) CollectNow(ctx context.Context, brandID int64, keywo platforms := resolveMonitoringPlatforms(quota.EnabledPlatforms) questionIDs := configuredQuestionIDs(configuredQuestions) - installationID, err := s.resolveCollectNowInstallationID(ctx, actor.TenantID, quota.PrimaryInstallationID) + clientID, err := s.resolveCollectNowClientID(ctx, actor.TenantID, actor.PrimaryWorkspaceID, quota.PrimaryClientID) if err != nil { return nil, err } - if installationID == nil { - return nil, response.ErrConflict(40941, "installation_offline", "primary monitoring installation is offline") + if clientID == nil { + return nil, response.ErrConflict(40941, "desktop_client_offline", "primary monitoring desktop client is offline") } - if err := s.ensureInstallationOnline(ctx, actor.TenantID, *installationID); err != nil { + if err := s.ensureClientOnline(ctx, actor.TenantID, actor.PrimaryWorkspaceID, *clientID); err != nil { return nil, err } @@ -584,9 +585,9 @@ func (s *MonitoringService) CollectNow(ctx context.Context, brandID int64, keywo }, nil } -func (s *MonitoringService) resolveCollectNowInstallationID(ctx context.Context, tenantID int64, preferred *int64) (*int64, error) { +func (s *MonitoringService) resolveCollectNowClientID(ctx context.Context, tenantID, workspaceID int64, preferred *uuid.UUID) (*uuid.UUID, error) { if preferred != nil { - online, err := s.isInstallationOnline(ctx, tenantID, *preferred) + online, err := s.isClientOnline(ctx, tenantID, workspaceID, *preferred) if err != nil { return nil, err } @@ -595,7 +596,7 @@ func (s *MonitoringService) resolveCollectNowInstallationID(ctx context.Context, } } - fallbackID, err := s.findLatestOnlineInstallation(ctx, tenantID) + fallbackID, err := s.findLatestOnlineClient(ctx, tenantID, workspaceID) if err != nil { return nil, err } @@ -604,7 +605,7 @@ func (s *MonitoringService) resolveCollectNowInstallationID(ctx context.Context, } if preferred == nil || *preferred != *fallbackID { - if err := s.persistPrimaryInstallationID(ctx, tenantID, *fallbackID); err != nil { + if err := s.persistPrimaryClientID(ctx, tenantID, workspaceID, *fallbackID); err != nil { return nil, err } } @@ -612,44 +613,44 @@ func (s *MonitoringService) resolveCollectNowInstallationID(ctx context.Context, return fallbackID, nil } -func (s *MonitoringService) findLatestOnlineInstallation(ctx context.Context, tenantID int64) (*int64, error) { - var installationID int64 +func (s *MonitoringService) findLatestOnlineClient(ctx context.Context, tenantID, workspaceID int64) (*uuid.UUID, error) { + var clientID uuid.UUID err := s.businessPool.QueryRow(ctx, ` SELECT id - FROM plugin_installations + FROM desktop_clients WHERE tenant_id = $1 - AND status = 'active' - AND deleted_at IS NULL + AND workspace_id = $2 + AND revoked_at IS NULL AND last_seen_at IS NOT NULL - AND last_seen_at >= NOW() - $2::interval - ORDER BY last_seen_at DESC, updated_at DESC, id DESC + AND last_seen_at >= NOW() - $3::interval + ORDER BY last_seen_at DESC, created_at DESC, id DESC LIMIT 1 - `, tenantID, formatPgInterval(monitoringOnlineDuration)).Scan(&installationID) + `, tenantID, workspaceID, formatPgInterval(monitoringOnlineDuration)).Scan(&clientID) if err != nil { if err == pgx.ErrNoRows { return nil, nil } - return nil, response.ErrInternal(50041, "query_failed", "failed to inspect online monitoring installation") + return nil, response.ErrInternal(50041, "query_failed", "failed to inspect online monitoring desktop client") } - return &installationID, nil + return &clientID, nil } -func (s *MonitoringService) persistPrimaryInstallationID(ctx context.Context, tenantID, installationID int64) error { +func (s *MonitoringService) persistPrimaryClientID(ctx context.Context, tenantID, workspaceID int64, clientID uuid.UUID) error { if _, err := s.businessPool.Exec(ctx, ` INSERT INTO tenant_monitoring_quotas ( - tenant_id, workspace_id, primary_installation_id + tenant_id, workspace_id, primary_client_id ) VALUES ( $1, - (SELECT id FROM workspaces WHERE tenant_id = $1 AND is_default = TRUE LIMIT 1), - $2 + $2, + $3 ) ON CONFLICT (tenant_id) DO UPDATE - SET workspace_id = COALESCE(tenant_monitoring_quotas.workspace_id, EXCLUDED.workspace_id), - primary_installation_id = EXCLUDED.primary_installation_id, + SET workspace_id = EXCLUDED.workspace_id, + primary_client_id = EXCLUDED.primary_client_id, updated_at = NOW() - `, tenantID, installationID); err != nil { - return response.ErrInternal(50041, "update_failed", "failed to persist primary monitoring installation") + `, tenantID, workspaceID, clientID); err != nil { + return response.ErrInternal(50041, "update_failed", "failed to persist primary monitoring desktop client") } return nil } @@ -894,12 +895,12 @@ func (s *MonitoringService) loadQuota(ctx context.Context, tenantID int64) (moni } var enabledJSON []byte - var installationID sql.NullInt64 + var clientID uuid.NullUUID err := s.businessPool.QueryRow(ctx, ` - SELECT collection_mode, enabled_platforms, primary_installation_id + SELECT collection_mode, enabled_platforms, primary_client_id FROM tenant_monitoring_quotas WHERE tenant_id = $1 - `, tenantID).Scan(&cfg.CollectionMode, &enabledJSON, &installationID) + `, tenantID).Scan(&cfg.CollectionMode, &enabledJSON, &clientID) if err != nil { if err == pgx.ErrNoRows { return cfg, nil @@ -913,9 +914,9 @@ func (s *MonitoringService) loadQuota(ctx context.Context, tenantID int64) (moni cfg.EnabledPlatforms = enabled } } - if installationID.Valid { - id := installationID.Int64 - cfg.PrimaryInstallationID = &id + if clientID.Valid { + id := clientID.UUID + cfg.PrimaryClientID = &id } if strings.TrimSpace(cfg.CollectionMode) == "" { cfg.CollectionMode = monitoringCollectorType @@ -1700,9 +1701,9 @@ func (s *MonitoringService) loadTimeBucketsForPlatform( return buckets, nil } -func (s *MonitoringService) loadAccessStates(ctx context.Context, tenantID int64, installationID *int64, businessDate time.Time) (map[string]monitoringAccessState, error) { +func (s *MonitoringService) loadAccessStates(ctx context.Context, tenantID int64, clientID *uuid.UUID, businessDate time.Time) (map[string]monitoringAccessState, error) { states := map[string]monitoringAccessState{} - if installationID == nil { + if clientID == nil { return states, nil } @@ -1710,9 +1711,9 @@ func (s *MonitoringService) loadAccessStates(ctx context.Context, tenantID int64 SELECT ai_platform_id, access_status FROM monitoring_platform_access_snapshots WHERE tenant_id = $1 - AND installation_id = $2 + AND client_id = $2 AND business_date = $3::date - `, tenantID, *installationID, businessDate.Format("2006-01-02")) + `, tenantID, *clientID, businessDate.Format("2006-01-02")) if err != nil { return nil, response.ErrInternal(50041, "query_failed", "failed to load platform access states") } @@ -2452,32 +2453,32 @@ func (s *MonitoringService) loadQuestionContentCitations(ctx context.Context, te return items, nil } -func (s *MonitoringService) ensureInstallationOnline(ctx context.Context, tenantID, installationID int64) error { - online, err := s.isInstallationOnline(ctx, tenantID, installationID) +func (s *MonitoringService) ensureClientOnline(ctx context.Context, tenantID, workspaceID int64, clientID uuid.UUID) error { + online, err := s.isClientOnline(ctx, tenantID, workspaceID, clientID) if err != nil { return err } if !online { - return response.ErrConflict(40941, "installation_offline", "primary monitoring installation is offline") + return response.ErrConflict(40941, "desktop_client_offline", "primary monitoring desktop client is offline") } return nil } -func (s *MonitoringService) isInstallationOnline(ctx context.Context, tenantID, installationID int64) (bool, error) { +func (s *MonitoringService) isClientOnline(ctx context.Context, tenantID, workspaceID int64, clientID uuid.UUID) (bool, error) { var lastSeenAt sql.NullTime err := s.businessPool.QueryRow(ctx, ` SELECT last_seen_at - FROM plugin_installations + FROM desktop_clients WHERE id = $1 AND tenant_id = $2 - AND status = 'active' - AND deleted_at IS NULL - `, installationID, tenantID).Scan(&lastSeenAt) + AND workspace_id = $3 + AND revoked_at IS NULL + `, clientID, tenantID, workspaceID).Scan(&lastSeenAt) if err != nil { if err == pgx.ErrNoRows { return false, nil } - return false, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring installation") + return false, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring desktop client") } if !lastSeenAt.Valid || time.Since(lastSeenAt.Time) > monitoringOnlineDuration { diff --git a/server/internal/tenant/app/publish_record_support.go b/server/internal/tenant/app/publish_record_support.go index 126cbc5..420b9e7 100644 --- a/server/internal/tenant/app/publish_record_support.go +++ b/server/internal/tenant/app/publish_record_support.go @@ -19,6 +19,14 @@ type publishableArticleMeta struct { CoverAssetURL *string } +type platformAccountSeed struct { + ID int64 + PlatformID string + PlatformName string + PlatformUID string + Nickname string +} + type desktopPublishSyncOutcome struct { TenantID int64 ArticleID int64 @@ -242,6 +250,65 @@ func publishBatchStatusToArticleStatus(batchStatus string) string { } } +func normalizeStringPointer(value *string) *string { + if value == nil { + return nil + } + trimmed := strings.TrimSpace(*value) + if trimmed == "" { + return nil + } + return &trimmed +} + +func recalculatePublishBatchStatus(ctx context.Context, tx pgx.Tx, publishBatchID int64) (string, error) { + rows, err := tx.Query(ctx, ` + SELECT status + FROM publish_records + WHERE publish_batch_id = $1 + `, publishBatchID) + if err != nil { + return "", response.ErrInternal(50054, "publish_batch_status_query_failed", "failed to inspect publish batch status") + } + defer rows.Close() + + statuses := make([]string, 0) + for rows.Next() { + var status string + if scanErr := rows.Scan(&status); scanErr != nil { + return "", response.ErrInternal(50054, "publish_batch_status_scan_failed", "failed to parse publish batch status") + } + statuses = append(statuses, normalizePublishStatus(status)) + } + if err := rows.Err(); err != nil { + return "", response.ErrInternal(50054, "publish_batch_status_scan_failed", "failed to iterate publish batch status") + } + + if len(statuses) == 0 { + return "publishing", nil + } + + counts := map[string]int{ + "success": 0, + "failed": 0, + "publishing": 0, + } + for _, status := range statuses { + counts[normalizePublishStatus(status)]++ + } + + switch { + case counts["publishing"] > 0: + return "publishing", nil + case counts["success"] > 0 && counts["failed"] > 0: + return "partial_success", nil + case counts["success"] > 0: + return "success", nil + default: + return "failed", nil + } +} + func resolvePublishArticleID(contentRef map[string]any) (int64, error) { if articleID, ok := extractInt64FromMap(contentRef, "article_id", "id"); ok && articleID > 0 { return articleID, nil diff --git a/server/internal/tenant/repository/generated/models.go b/server/internal/tenant/repository/generated/models.go index eadfafe..6b48ceb 100644 --- a/server/internal/tenant/repository/generated/models.go +++ b/server/internal/tenant/repository/generated/models.go @@ -484,40 +484,6 @@ type PlatformUserRole struct { DeletedAt pgtype.Timestamptz `json:"deleted_at"` } -type PluginInstallation struct { - ID int64 `json:"id"` - TenantID int64 `json:"tenant_id"` - UserID int64 `json:"user_id"` - InstallationKey string `json:"installation_key"` - InstallationName string `json:"installation_name"` - BrowserName pgtype.Text `json:"browser_name"` - ClientVersion pgtype.Text `json:"client_version"` - InstallationTokenHash string `json:"installation_token_hash"` - Status string `json:"status"` - LastSeenAt pgtype.Timestamptz `json:"last_seen_at"` - CreatedAt pgtype.Timestamptz `json:"created_at"` - UpdatedAt pgtype.Timestamptz `json:"updated_at"` - DeletedAt pgtype.Timestamptz `json:"deleted_at"` -} - -type PluginSession struct { - ID int64 `json:"id"` - TenantID int64 `json:"tenant_id"` - UserID int64 `json:"user_id"` - PluginInstallationID pgtype.Int8 `json:"plugin_installation_id"` - ActionType string `json:"action_type"` - ResourceType pgtype.Text `json:"resource_type"` - ResourceID pgtype.Int8 `json:"resource_id"` - PlatformAccountID pgtype.Int8 `json:"platform_account_id"` - PlatformID string `json:"platform_id"` - SessionToken string `json:"session_token"` - ClientVersion pgtype.Text `json:"client_version"` - Status string `json:"status"` - ExpireAt pgtype.Timestamptz `json:"expire_at"` - CreatedAt pgtype.Timestamptz `json:"created_at"` - UpdatedAt pgtype.Timestamptz `json:"updated_at"` -} - type PromptRule struct { ID int64 `json:"id"` TenantID int64 `json:"tenant_id"` @@ -681,7 +647,7 @@ type TenantMonitoringQuota struct { QuestionSelectionMode string `json:"question_selection_mode"` CollectFrequency string `json:"collect_frequency"` EnabledPlatforms []byte `json:"enabled_platforms"` - PrimaryInstallationID pgtype.Int8 `json:"primary_installation_id"` + PrimaryClientID pgtype.UUID `json:"primary_client_id"` TaskDailyHardCap int32 `json:"task_daily_hard_cap"` PlanTier string `json:"plan_tier"` CreatedAt pgtype.Timestamptz `json:"created_at"` diff --git a/server/internal/tenant/transport/media_handler.go b/server/internal/tenant/transport/media_handler.go index fb222c2..20d98c8 100644 --- a/server/internal/tenant/transport/media_handler.go +++ b/server/internal/tenant/transport/media_handler.go @@ -4,7 +4,6 @@ import ( "strconv" "github.com/gin-gonic/gin" - "go.uber.org/zap" "github.com/geo-platform/tenant-api/internal/bootstrap" "github.com/geo-platform/tenant-api/internal/shared/response" @@ -12,14 +11,12 @@ import ( ) type MediaHandler struct { - svc *app.MediaService - logger *zap.Logger + svc *app.MediaService } func NewMediaHandler(a *bootstrap.App) *MediaHandler { return &MediaHandler{ - svc: app.NewMediaService(a.DB, a.Logger).WithCache(a.Cache), - logger: a.Logger, + svc: app.NewMediaService(a.DB), } } @@ -32,75 +29,6 @@ func (h *MediaHandler) ListPlatforms(c *gin.Context) { response.Success(c, data) } -func (h *MediaHandler) ListPlatformAccounts(c *gin.Context) { - data, err := h.svc.ListPlatformAccounts(c.Request.Context()) - if err != nil { - response.Error(c, err) - return - } - response.Success(c, data) -} - -func (h *MediaHandler) DeletePlatformAccount(c *gin.Context) { - id, err := strconv.ParseInt(c.Param("id"), 10, 64) - if err != nil { - response.Error(c, response.ErrBadRequest(40001, "invalid_id", "platform account id must be a number")) - return - } - if err := h.svc.DeletePlatformAccount(c.Request.Context(), id); err != nil { - response.Error(c, err) - return - } - response.Success(c, nil) -} - -func (h *MediaHandler) CreatePluginSession(c *gin.Context) { - var req app.CreatePluginSessionRequest - if err := c.ShouldBindJSON(&req); err != nil { - response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error())) - return - } - data, err := h.svc.CreatePluginSession(c.Request.Context(), req) - if err != nil { - response.Error(c, err) - return - } - response.Success(c, data) -} - -func (h *MediaHandler) RegisterPluginInstallation(c *gin.Context) { - var req app.RegisterPluginInstallationRequest - if err := c.ShouldBindJSON(&req); err != nil { - response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error())) - return - } - data, err := h.svc.RegisterPluginInstallation(c.Request.Context(), req) - if err != nil { - response.Error(c, err) - return - } - response.Success(c, data) -} - -func (h *MediaHandler) CreatePublishBatch(c *gin.Context) { - articleID, err := strconv.ParseInt(c.Param("id"), 10, 64) - if err != nil { - response.Error(c, response.ErrBadRequest(40001, "invalid_id", "article id must be a number")) - return - } - var req app.CreatePublishBatchRequest - if err := c.ShouldBindJSON(&req); err != nil { - response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error())) - return - } - data, err := h.svc.CreatePublishBatch(c.Request.Context(), articleID, req) - if err != nil { - response.Error(c, err) - return - } - response.SuccessWithStatus(c, 201, data) -} - func (h *MediaHandler) ListPublishRecords(c *gin.Context) { articleID, err := strconv.ParseInt(c.Param("id"), 10, 64) if err != nil { @@ -114,61 +42,3 @@ func (h *MediaHandler) ListPublishRecords(c *gin.Context) { } response.Success(c, data) } - -func (h *MediaHandler) PluginBindCallback(c *gin.Context) { - var req app.PluginBindCallbackRequest - if err := c.ShouldBindJSON(&req); err != nil { - response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error())) - return - } - data, err := h.svc.HandleBindCallback(c.Request.Context(), req) - if err != nil { - if h.logger != nil { - appErr := response.Normalize(err) - h.logger.Warn("publisher bind callback failed", - zap.String("request_id", c.GetString("request_id")), - zap.Int64("plugin_session_id", req.PluginSessionID), - zap.String("platform_id", req.PlatformID), - zap.String("platform_uid", req.PlatformUID), - zap.Int("code", appErr.Code), - zap.String("message", appErr.Message), - zap.String("detail", appErr.Detail), - ) - } - response.Error(c, err) - return - } - response.Success(c, data) -} - -func (h *MediaHandler) PluginPublishCallback(c *gin.Context) { - var req app.PluginPublishCallbackRequest - if err := c.ShouldBindJSON(&req); err != nil { - response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error())) - return - } - data, err := h.svc.HandlePublishCallback(c.Request.Context(), req) - if err != nil { - if h.logger != nil { - appErr := response.Normalize(err) - h.logger.Warn("publisher publish callback failed", - zap.String("request_id", c.GetString("request_id")), - zap.Int64("plugin_session_id", req.PluginSessionID), - zap.Int64("publish_record_id", req.PublishRecordID), - zap.Int64("platform_account_id", req.PlatformAccountID), - zap.Int64("article_id", req.ArticleID), - zap.String("platform_id", req.PlatformID), - zap.String("status", req.Status), - zap.Stringp("error_message", req.ErrorMessage), - zap.Int("code", appErr.Code), - zap.String("message", appErr.Message), - zap.String("detail", appErr.Detail), - zap.Any("request_payload", req.RequestPayload), - zap.Any("response_payload", req.ResponsePayload), - ) - } - response.Error(c, err) - return - } - response.Success(c, data) -} diff --git a/server/internal/tenant/transport/router.go b/server/internal/tenant/transport/router.go index 4ca9ae8..c022f7f 100644 --- a/server/internal/tenant/transport/router.go +++ b/server/internal/tenant/transport/router.go @@ -16,10 +16,7 @@ func RegisterRoutes(app *bootstrap.App) { publicAssets := NewAssetHandler(app) app.Engine.GET("/api/public/assets/:token", publicAssets.Serve) - pluginHandler := NewMediaHandler(app) - callbacks := app.Engine.Group("/api/callback/plugin") - callbacks.POST("/bind", pluginHandler.PluginBindCallback) - callbacks.POST("/publish", pluginHandler.PluginPublishCallback) + mediaHandler := NewMediaHandler(app) protected := app.Engine.Group("/api") protected.Use(auth.Middleware(app.JWT, app.Sessions)) @@ -65,7 +62,7 @@ func RegisterRoutes(app *bootstrap.App) { tenantProtected.POST("/publish-jobs", publishJobHandler.Create) media := tenantProtected.Group("/media") - media.GET("/platforms", pluginHandler.ListPlatforms) + media.GET("/platforms", mediaHandler.ListPlatforms) workspace := tenantProtected.Group("/workspace") wsHandler := NewWorkspaceHandler(app) @@ -149,8 +146,7 @@ func RegisterRoutes(app *bootstrap.App) { articles.PUT("/:id", artHandler.Update) articles.GET("/:id/stream", artHandler.Stream) articles.GET("/:id/versions", artHandler.Versions) - articles.GET("/:id/publish-records", pluginHandler.ListPublishRecords) - articles.POST("/:id/publish-batches", pluginHandler.CreatePublishBatch) + articles.GET("/:id/publish-records", mediaHandler.ListPublishRecords) articles.DELETE("/:id", artHandler.Delete) brands := tenantProtected.Group("/brands") diff --git a/server/migrations/20260402140000_create_media_publisher_tables.down.sql b/server/migrations/20260402140000_create_media_publisher_tables.down.sql index 587bcaa..f7351d0 100644 --- a/server/migrations/20260402140000_create_media_publisher_tables.down.sql +++ b/server/migrations/20260402140000_create_media_publisher_tables.down.sql @@ -1,6 +1,4 @@ DROP TABLE IF EXISTS publish_records; DROP TABLE IF EXISTS publish_batches; -DROP TABLE IF EXISTS plugin_sessions; -DROP TABLE IF EXISTS plugin_installations; DROP TABLE IF EXISTS platform_accounts; DROP TABLE IF EXISTS media_platforms; diff --git a/server/migrations/20260402140000_create_media_publisher_tables.up.sql b/server/migrations/20260402140000_create_media_publisher_tables.up.sql index 82cf3b3..78c8a9a 100644 --- a/server/migrations/20260402140000_create_media_publisher_tables.up.sql +++ b/server/migrations/20260402140000_create_media_publisher_tables.up.sql @@ -53,50 +53,6 @@ CREATE UNIQUE INDEX uk_platform_accounts_identity_active WHERE deleted_at IS NULL; CREATE INDEX idx_platform_accounts_tenant_platform ON platform_accounts(tenant_id, platform_id, status, created_at DESC); -CREATE TABLE plugin_installations ( - id BIGSERIAL PRIMARY KEY, - tenant_id BIGINT NOT NULL REFERENCES tenants(id), - user_id BIGINT NOT NULL REFERENCES users(id), - installation_key VARCHAR(128) NOT NULL, - installation_name VARCHAR(255) NOT NULL, - browser_name VARCHAR(64), - client_version VARCHAR(32), - installation_token_hash VARCHAR(128) NOT NULL, - status VARCHAR(20) NOT NULL DEFAULT 'active', - last_seen_at TIMESTAMPTZ, - created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - deleted_at TIMESTAMPTZ -); - -CREATE UNIQUE INDEX uk_plugin_installations_key_active - ON plugin_installations(tenant_id, installation_key) - WHERE deleted_at IS NULL; -CREATE INDEX idx_plugin_installations_tenant_status - ON plugin_installations(tenant_id, status, created_at DESC); - -CREATE TABLE plugin_sessions ( - id BIGSERIAL PRIMARY KEY, - tenant_id BIGINT NOT NULL REFERENCES tenants(id), - user_id BIGINT NOT NULL REFERENCES users(id), - plugin_installation_id BIGINT REFERENCES plugin_installations(id), - action_type VARCHAR(20) NOT NULL, - resource_type VARCHAR(30), - resource_id BIGINT, - platform_account_id BIGINT REFERENCES platform_accounts(id), - platform_id VARCHAR(64) NOT NULL REFERENCES media_platforms(platform_id), - session_token VARCHAR(128) NOT NULL, - client_version VARCHAR(32), - status VARCHAR(20) NOT NULL DEFAULT 'pending', - expire_at TIMESTAMPTZ NOT NULL, - created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() -); - -CREATE UNIQUE INDEX uk_plugin_sessions_token ON plugin_sessions(session_token); -CREATE INDEX idx_plugin_sessions_status_expire ON plugin_sessions(status, expire_at); -CREATE INDEX idx_plugin_sessions_tenant_action ON plugin_sessions(tenant_id, action_type, created_at DESC); - CREATE TABLE publish_batches ( id BIGSERIAL PRIMARY KEY, tenant_id BIGINT NOT NULL REFERENCES tenants(id), diff --git a/server/migrations/20260402153000_repair_plugin_installations_schema.down.sql b/server/migrations/20260402153000_repair_plugin_installations_schema.down.sql index 72e5efb..bb4d4a3 100644 --- a/server/migrations/20260402153000_repair_plugin_installations_schema.down.sql +++ b/server/migrations/20260402153000_repair_plugin_installations_schema.down.sql @@ -1,4 +1 @@ --- No-op rollback. --- This migration repairs databases that already applied an older variant of --- 20260402140000 without plugin_installations. Dropping these objects on --- rollback would break databases created from the current baseline migration. +-- Legacy no-op rollback. diff --git a/server/migrations/20260402153000_repair_plugin_installations_schema.up.sql b/server/migrations/20260402153000_repair_plugin_installations_schema.up.sql index 7fc3954..0539306 100644 --- a/server/migrations/20260402153000_repair_plugin_installations_schema.up.sql +++ b/server/migrations/20260402153000_repair_plugin_installations_schema.up.sql @@ -1,39 +1 @@ -CREATE TABLE IF NOT EXISTS plugin_installations ( - id BIGSERIAL PRIMARY KEY, - tenant_id BIGINT NOT NULL REFERENCES tenants(id), - user_id BIGINT NOT NULL REFERENCES users(id), - installation_key VARCHAR(128) NOT NULL, - installation_name VARCHAR(255) NOT NULL, - browser_name VARCHAR(64), - client_version VARCHAR(32), - installation_token_hash VARCHAR(128) NOT NULL, - status VARCHAR(20) NOT NULL DEFAULT 'active', - last_seen_at TIMESTAMPTZ, - created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - deleted_at TIMESTAMPTZ -); - -CREATE UNIQUE INDEX IF NOT EXISTS uk_plugin_installations_key_active - ON plugin_installations(tenant_id, installation_key) - WHERE deleted_at IS NULL; - -CREATE INDEX IF NOT EXISTS idx_plugin_installations_tenant_status - ON plugin_installations(tenant_id, status, created_at DESC); - -ALTER TABLE plugin_sessions - ADD COLUMN IF NOT EXISTS plugin_installation_id BIGINT; - -DO $$ -BEGIN - IF NOT EXISTS ( - SELECT 1 - FROM pg_constraint - WHERE conname = 'plugin_sessions_plugin_installation_id_fkey' - ) THEN - ALTER TABLE plugin_sessions - ADD CONSTRAINT plugin_sessions_plugin_installation_id_fkey - FOREIGN KEY (plugin_installation_id) - REFERENCES plugin_installations(id); - END IF; -END $$; +-- Legacy no-op. diff --git a/server/migrations/20260410103000_create_monitoring_tables.up.sql b/server/migrations/20260410103000_create_monitoring_tables.up.sql index 1a9e0fe..a5dfcb9 100644 --- a/server/migrations/20260410103000_create_monitoring_tables.up.sql +++ b/server/migrations/20260410103000_create_monitoring_tables.up.sql @@ -1,11 +1,11 @@ CREATE TABLE tenant_monitoring_quotas ( tenant_id BIGINT PRIMARY KEY REFERENCES tenants(id), max_brands INT NOT NULL DEFAULT 1, - collection_mode VARCHAR(20) NOT NULL DEFAULT 'plugin', + collection_mode VARCHAR(20) NOT NULL DEFAULT 'desktop', question_selection_mode VARCHAR(20) NOT NULL DEFAULT 'best_effort_all', collect_frequency VARCHAR(20) NOT NULL DEFAULT 'daily', enabled_platforms JSONB NOT NULL DEFAULT '["deepseek","qwen","doubao"]'::jsonb, - primary_installation_id BIGINT REFERENCES plugin_installations(id), + primary_client_id UUID, task_daily_hard_cap INT NOT NULL DEFAULT 36, plan_tier VARCHAR(20) NOT NULL DEFAULT 'free', created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), diff --git a/server/migrations/20260421100030_remove_plugin_installations_and_sessions.down.sql b/server/migrations/20260421100030_remove_plugin_installations_and_sessions.down.sql new file mode 100644 index 0000000..06dced5 --- /dev/null +++ b/server/migrations/20260421100030_remove_plugin_installations_and_sessions.down.sql @@ -0,0 +1,53 @@ +CREATE TABLE IF NOT EXISTS plugin_installations ( + id BIGSERIAL PRIMARY KEY, + tenant_id BIGINT NOT NULL REFERENCES tenants(id), + user_id BIGINT NOT NULL REFERENCES users(id), + installation_key VARCHAR(128) NOT NULL, + installation_name VARCHAR(255) NOT NULL, + browser_name VARCHAR(64), + client_version VARCHAR(32), + installation_token_hash VARCHAR(128) NOT NULL, + status VARCHAR(20) NOT NULL DEFAULT 'active', + last_seen_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + deleted_at TIMESTAMPTZ +); + +CREATE UNIQUE INDEX IF NOT EXISTS uk_plugin_installations_key_active + ON plugin_installations(tenant_id, installation_key) + WHERE deleted_at IS NULL; + +CREATE TABLE IF NOT EXISTS plugin_sessions ( + id BIGSERIAL PRIMARY KEY, + tenant_id BIGINT NOT NULL REFERENCES tenants(id), + user_id BIGINT NOT NULL REFERENCES users(id), + plugin_installation_id BIGINT REFERENCES plugin_installations(id), + action_type VARCHAR(20) NOT NULL, + resource_type VARCHAR(30), + resource_id BIGINT, + platform_account_id BIGINT REFERENCES platform_accounts(id), + platform_id VARCHAR(64) NOT NULL REFERENCES media_platforms(platform_id), + session_token VARCHAR(128) NOT NULL, + client_version VARCHAR(32), + status VARCHAR(20) NOT NULL DEFAULT 'pending', + expire_at TIMESTAMPTZ NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +ALTER TABLE tenant_monitoring_quotas + DROP CONSTRAINT IF EXISTS tenant_monitoring_quotas_primary_client_id_fkey; + +ALTER TABLE tenant_monitoring_quotas + ADD COLUMN IF NOT EXISTS primary_installation_id BIGINT REFERENCES plugin_installations(id); + +UPDATE tenant_monitoring_quotas +SET collection_mode = 'plugin' +WHERE collection_mode = 'desktop'; + +ALTER TABLE tenant_monitoring_quotas + ALTER COLUMN collection_mode SET DEFAULT 'plugin'; + +ALTER TABLE tenant_monitoring_quotas + DROP COLUMN IF EXISTS primary_client_id; diff --git a/server/migrations/20260421100030_remove_plugin_installations_and_sessions.up.sql b/server/migrations/20260421100030_remove_plugin_installations_and_sessions.up.sql new file mode 100644 index 0000000..7851c42 --- /dev/null +++ b/server/migrations/20260421100030_remove_plugin_installations_and_sessions.up.sql @@ -0,0 +1,32 @@ +ALTER TABLE tenant_monitoring_quotas + DROP CONSTRAINT IF EXISTS tenant_monitoring_quotas_primary_installation_id_fkey; + +ALTER TABLE tenant_monitoring_quotas + ADD COLUMN IF NOT EXISTS primary_client_id UUID; + +UPDATE tenant_monitoring_quotas +SET collection_mode = 'desktop' +WHERE collection_mode = 'plugin'; + +ALTER TABLE tenant_monitoring_quotas + ALTER COLUMN collection_mode SET DEFAULT 'desktop'; + +ALTER TABLE tenant_monitoring_quotas + DROP COLUMN IF EXISTS primary_installation_id; + +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'tenant_monitoring_quotas_primary_client_id_fkey' + ) THEN + ALTER TABLE tenant_monitoring_quotas + ADD CONSTRAINT tenant_monitoring_quotas_primary_client_id_fkey + FOREIGN KEY (primary_client_id) + REFERENCES desktop_clients(id); + END IF; +END $$; + +DROP TABLE IF EXISTS plugin_sessions; +DROP TABLE IF EXISTS plugin_installations; diff --git a/server/migrations_monitoring/20260410103000_create_monitoring_tables.up.sql b/server/migrations_monitoring/20260410103000_create_monitoring_tables.up.sql index c6dc327..c7487e4 100644 --- a/server/migrations_monitoring/20260410103000_create_monitoring_tables.up.sql +++ b/server/migrations_monitoring/20260410103000_create_monitoring_tables.up.sql @@ -21,7 +21,7 @@ CREATE INDEX idx_monitoring_question_snapshot_active CREATE TABLE monitoring_platform_access_snapshots ( id BIGSERIAL PRIMARY KEY, tenant_id BIGINT NOT NULL, - installation_id BIGINT NOT NULL, + client_id UUID NOT NULL, ai_platform_id VARCHAR(30) NOT NULL, business_date DATE NOT NULL, access_status VARCHAR(20) NOT NULL, @@ -32,7 +32,7 @@ CREATE TABLE monitoring_platform_access_snapshots ( reason_message TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - CONSTRAINT uk_platform_access_snapshot UNIQUE (tenant_id, installation_id, ai_platform_id, business_date) + CONSTRAINT uk_platform_access_snapshot UNIQUE (tenant_id, client_id, ai_platform_id, business_date) ); CREATE INDEX idx_platform_access_snapshot_lookup @@ -45,9 +45,9 @@ CREATE TABLE monitoring_collect_tasks ( question_id BIGINT NOT NULL, question_hash BYTEA NOT NULL, ai_platform_id VARCHAR(30) NOT NULL, - collector_type VARCHAR(20) NOT NULL DEFAULT 'plugin', + collector_type VARCHAR(20) NOT NULL DEFAULT 'desktop', trigger_source VARCHAR(20) NOT NULL DEFAULT 'automatic', - run_mode VARCHAR(30) NOT NULL DEFAULT 'plugin_standard', + run_mode VARCHAR(30) NOT NULL DEFAULT 'desktop_standard', business_date DATE NOT NULL, planned_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), status VARCHAR(20) NOT NULL DEFAULT 'pending', @@ -77,9 +77,9 @@ CREATE TABLE question_monitor_runs ( question_text_snapshot TEXT NOT NULL, question_hash BYTEA NOT NULL, ai_platform_id VARCHAR(30) NOT NULL, - collector_type VARCHAR(20) NOT NULL DEFAULT 'plugin', + collector_type VARCHAR(20) NOT NULL DEFAULT 'desktop', trigger_source VARCHAR(20) NOT NULL DEFAULT 'automatic', - run_mode VARCHAR(30) NOT NULL DEFAULT 'plugin_standard', + run_mode VARCHAR(30) NOT NULL DEFAULT 'desktop_standard', business_date DATE NOT NULL, provider_model VARCHAR(100), provider_request_id VARCHAR(100), @@ -187,7 +187,7 @@ CREATE TABLE monitoring_brand_daily ( id BIGSERIAL PRIMARY KEY, tenant_id BIGINT NOT NULL, brand_id BIGINT NOT NULL, - collector_type VARCHAR(20) NOT NULL DEFAULT 'plugin', + collector_type VARCHAR(20) NOT NULL DEFAULT 'desktop', business_date DATE NOT NULL, sample_status VARCHAR(20) NOT NULL, desired_sample_count INT NOT NULL DEFAULT 0, @@ -221,7 +221,7 @@ CREATE TABLE monitoring_brand_platform_daily ( tenant_id BIGINT NOT NULL, brand_id BIGINT NOT NULL, ai_platform_id VARCHAR(30) NOT NULL, - collector_type VARCHAR(20) NOT NULL DEFAULT 'plugin', + collector_type VARCHAR(20) NOT NULL DEFAULT 'desktop', business_date DATE NOT NULL, platform_sample_status VARCHAR(20) NOT NULL, planned_sample_count INT NOT NULL DEFAULT 0, diff --git a/server/migrations_monitoring/20260421100030_migrate_monitoring_to_desktop_clients.down.sql b/server/migrations_monitoring/20260421100030_migrate_monitoring_to_desktop_clients.down.sql new file mode 100644 index 0000000..a96a173 --- /dev/null +++ b/server/migrations_monitoring/20260421100030_migrate_monitoring_to_desktop_clients.down.sql @@ -0,0 +1,63 @@ +DROP TABLE IF EXISTS monitoring_platform_access_snapshots; + +CREATE TABLE monitoring_platform_access_snapshots ( + id BIGSERIAL PRIMARY KEY, + tenant_id BIGINT NOT NULL, + installation_id BIGINT NOT NULL, + ai_platform_id VARCHAR(30) NOT NULL, + business_date DATE NOT NULL, + access_status VARCHAR(20) NOT NULL, + connected BOOLEAN NOT NULL DEFAULT false, + accessible BOOLEAN NOT NULL DEFAULT false, + detected_at TIMESTAMPTZ NOT NULL, + reason_code VARCHAR(50), + reason_message TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT uk_platform_access_snapshot UNIQUE (tenant_id, installation_id, ai_platform_id, business_date) +); + +CREATE INDEX idx_platform_access_snapshot_lookup + ON monitoring_platform_access_snapshots(tenant_id, business_date, ai_platform_id, access_status); + +ALTER TABLE monitoring_collect_tasks + ALTER COLUMN collector_type SET DEFAULT 'plugin'; + +ALTER TABLE monitoring_collect_tasks + ALTER COLUMN run_mode SET DEFAULT 'plugin_standard'; + +UPDATE monitoring_collect_tasks +SET collector_type = 'plugin' +WHERE collector_type = 'desktop'; + +UPDATE monitoring_collect_tasks +SET run_mode = 'plugin_standard' +WHERE run_mode = 'desktop_standard'; + +ALTER TABLE question_monitor_runs + ALTER COLUMN collector_type SET DEFAULT 'plugin'; + +ALTER TABLE question_monitor_runs + ALTER COLUMN run_mode SET DEFAULT 'plugin_standard'; + +UPDATE question_monitor_runs +SET collector_type = 'plugin' +WHERE collector_type = 'desktop'; + +UPDATE question_monitor_runs +SET run_mode = 'plugin_standard' +WHERE run_mode = 'desktop_standard'; + +ALTER TABLE monitoring_brand_daily + ALTER COLUMN collector_type SET DEFAULT 'plugin'; + +UPDATE monitoring_brand_daily +SET collector_type = 'plugin' +WHERE collector_type = 'desktop'; + +ALTER TABLE monitoring_brand_platform_daily + ALTER COLUMN collector_type SET DEFAULT 'plugin'; + +UPDATE monitoring_brand_platform_daily +SET collector_type = 'plugin' +WHERE collector_type = 'desktop'; diff --git a/server/migrations_monitoring/20260421100030_migrate_monitoring_to_desktop_clients.up.sql b/server/migrations_monitoring/20260421100030_migrate_monitoring_to_desktop_clients.up.sql new file mode 100644 index 0000000..b5a4d62 --- /dev/null +++ b/server/migrations_monitoring/20260421100030_migrate_monitoring_to_desktop_clients.up.sql @@ -0,0 +1,63 @@ +DROP TABLE IF EXISTS monitoring_platform_access_snapshots; + +CREATE TABLE monitoring_platform_access_snapshots ( + id BIGSERIAL PRIMARY KEY, + tenant_id BIGINT NOT NULL, + client_id UUID NOT NULL, + ai_platform_id VARCHAR(30) NOT NULL, + business_date DATE NOT NULL, + access_status VARCHAR(20) NOT NULL, + connected BOOLEAN NOT NULL DEFAULT false, + accessible BOOLEAN NOT NULL DEFAULT false, + detected_at TIMESTAMPTZ NOT NULL, + reason_code VARCHAR(50), + reason_message TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT uk_platform_access_snapshot UNIQUE (tenant_id, client_id, ai_platform_id, business_date) +); + +CREATE INDEX idx_platform_access_snapshot_lookup + ON monitoring_platform_access_snapshots(tenant_id, business_date, ai_platform_id, access_status); + +ALTER TABLE monitoring_collect_tasks + ALTER COLUMN collector_type SET DEFAULT 'desktop'; + +ALTER TABLE monitoring_collect_tasks + ALTER COLUMN run_mode SET DEFAULT 'desktop_standard'; + +UPDATE monitoring_collect_tasks +SET collector_type = 'desktop' +WHERE collector_type = 'plugin'; + +UPDATE monitoring_collect_tasks +SET run_mode = 'desktop_standard' +WHERE run_mode = 'plugin_standard'; + +ALTER TABLE question_monitor_runs + ALTER COLUMN collector_type SET DEFAULT 'desktop'; + +ALTER TABLE question_monitor_runs + ALTER COLUMN run_mode SET DEFAULT 'desktop_standard'; + +UPDATE question_monitor_runs +SET collector_type = 'desktop' +WHERE collector_type = 'plugin'; + +UPDATE question_monitor_runs +SET run_mode = 'desktop_standard' +WHERE run_mode = 'plugin_standard'; + +ALTER TABLE monitoring_brand_daily + ALTER COLUMN collector_type SET DEFAULT 'desktop'; + +UPDATE monitoring_brand_daily +SET collector_type = 'desktop' +WHERE collector_type = 'plugin'; + +ALTER TABLE monitoring_brand_platform_daily + ALTER COLUMN collector_type SET DEFAULT 'desktop'; + +UPDATE monitoring_brand_platform_daily +SET collector_type = 'desktop' +WHERE collector_type = 'plugin';