package app 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" "github.com/geo-platform/tenant-api/internal/shared/response" ) const minBrowserExtensionVersion = "0.1.0" type MediaService struct { pool *pgxpool.Pool logger *zap.Logger } func NewMediaService(pool *pgxpool.Pool, logger *zap.Logger) *MediaService { return &MediaService{pool: pool, logger: logger} } type MediaPlatformResponse struct { ID int64 `json:"id"` PlatformID string `json:"platform_id"` Name string `json:"name"` Category string `json:"category"` ShortName string `json:"short_name"` AccentColor string `json:"accent_color"` LoginURL *string `json:"login_url"` LogoURL *string `json:"logo_url"` Status string `json:"status"` 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"` ArticleID int64 `json:"article_id"` PlatformAccountID int64 `json:"platform_account_id"` PlatformID string `json:"platform_id"` PlatformName string `json:"platform_name"` PlatformNickname string `json:"platform_nickname"` Status string `json:"status"` ExternalArticleID *string `json:"external_article_id"` ExternalArticleURL *string `json:"external_article_url"` ExternalManageURL *string `json:"external_manage_url"` PublishedAt *time.Time `json:"published_at"` ErrorMessage *string `json:"error_message"` CreatedAt time.Time `json:"created_at"` 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 FROM media_platforms WHERE status = 'active' ORDER BY sort_order ASC, id ASC `) if err != nil { return nil, response.ErrInternal(50030, "media_platform_query_failed", "failed to list media platforms") } defer rows.Close() items := make([]MediaPlatformResponse, 0) for rows.Next() { var item MediaPlatformResponse if err := rows.Scan( &item.ID, &item.PlatformID, &item.Name, &item.Category, &item.ShortName, &item.AccentColor, &item.LoginURL, &item.LogoURL, &item.Status, &item.SortOrder, ); err != nil { return nil, response.ErrInternal(50030, "media_platform_scan_failed", err.Error()) } items = append(items, item) } 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 := 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 (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) var articleExists bool var generateStatus string if err := tx.QueryRow(ctx, ` SELECT EXISTS(SELECT 1 FROM articles WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL), COALESCE((SELECT generate_status FROM articles WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL), '') `, articleID, actor.TenantID).Scan(&articleExists, &generateStatus); err != nil { return nil, response.ErrInternal(50037, "article_check_failed", "failed to validate article") } if !articleExists { return nil, response.ErrNotFound(40411, "article_not_found", "article not found") } if generateStatus != "completed" { return nil, response.ErrConflict(40913, "article_not_publishable", "only completed articles can be published") } accountSeeds, err := loadPlatformAccountsForPublish(ctx, tx, actor.TenantID, req.PlatformAccountIDs) if err != nil { return nil, err } if publishBatchRequiresCover(accountSeeds) && strings.TrimSpace(pointerStringValue(req.CoverAssetURL)) == "" { return nil, response.ErrBadRequest(40044, "publish_cover_required", "baijiahao publishing requires a cover image") } var publishBatchID int64 if err := tx.QueryRow(ctx, ` INSERT INTO publish_batches (tenant_id, article_id, initiator_user_id, status, publish_type, cover_asset_url) VALUES ($1, $2, $3, 'publishing', $4, $5) RETURNING id `, actor.TenantID, articleID, actor.UserID, publishType, normalizeStringPointer(req.CoverAssetURL)).Scan(&publishBatchID); err != nil { return nil, response.ErrInternal(50038, "publish_batch_create_failed", "failed to create publish batch") } tasks := make([]PublishBatchTask, 0, len(accountSeeds)) for _, account := range accountSeeds { var publishRecordID int64 if err := tx.QueryRow(ctx, ` INSERT INTO publish_records ( tenant_id, publish_batch_id, article_id, platform_account_id, platform_id, status ) VALUES ($1, $2, $3, $4, $5, 'queued') RETURNING id `, actor.TenantID, publishBatchID, articleID, account.ID, account.PlatformID).Scan(&publishRecordID); err != nil { return nil, response.ErrInternal(50039, "publish_record_create_failed", "failed to create publish record") } 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, publishRecordID, 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: publishRecordID, PlatformAccountID: account.ID, PlatformID: account.PlatformID, PlatformName: account.PlatformName, PlatformUID: account.PlatformUID, PlatformNickname: account.Nickname, PluginSessionID: pluginSessionID, SessionToken: token, ExpireAt: expireAt, }) } if _, err := tx.Exec(ctx, ` UPDATE articles SET publish_status = 'publishing', updated_at = NOW() WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL `, articleID, actor.TenantID); err != nil { return nil, response.ErrInternal(50041, "article_publish_status_failed", "failed to update article publish status") } if err := tx.Commit(ctx); err != nil { return nil, response.ErrInternal(50042, "publish_batch_commit_failed", "failed to commit publish batch") } 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) } func (s *MediaService) listArticlePublishRecords(ctx context.Context, tenantID, articleID int64) ([]PublishRecordResponse, error) { rows, err := s.pool.Query(ctx, ` SELECT pr.id, pr.publish_batch_id, pr.article_id, pr.platform_account_id, pr.platform_id, mp.name, pa.nickname, pr.status, pr.external_article_id, pr.external_article_url, pr.external_manage_url, pr.published_at, pr.error_message, pr.created_at, pr.updated_at FROM publish_records pr JOIN media_platforms mp ON mp.platform_id = pr.platform_id JOIN platform_accounts pa ON pa.id = pr.platform_account_id WHERE pr.tenant_id = $1 AND pr.article_id = $2 ORDER BY pr.created_at DESC, pr.id DESC `, tenantID, articleID) if err != nil { return nil, response.ErrInternal(50043, "publish_record_query_failed", "failed to list publish records") } defer rows.Close() items := make([]PublishRecordResponse, 0) for rows.Next() { var item PublishRecordResponse if err := rows.Scan( &item.ID, &item.PublishBatchID, &item.ArticleID, &item.PlatformAccountID, &item.PlatformID, &item.PlatformName, &item.PlatformNickname, &item.Status, &item.ExternalArticleID, &item.ExternalArticleURL, &item.ExternalManageURL, &item.PublishedAt, &item.ErrorMessage, &item.CreatedAt, &item.UpdatedAt, ); err != nil { return nil, response.ErrInternal(50043, "publish_record_scan_failed", err.Error()) } items = append(items, item) } 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 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, user_id, platform_id, platform_uid, nickname, avatar_url, status, metadata_json, last_check_at ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NOW()) 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, avatar_url = $3, status = $4, metadata_json = $5, last_check_at = NOW(), 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") } } 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 } 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 } 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") } 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["pending_review"] == len(statuses) { return "pending_review" } if counts["failed"] > 0 && (counts["success"] > 0 || counts["pending_review"] > 0) { return "partial_success" } if counts["pending_review"] > 0 { return "pending_review" } if counts["success"] > 0 { return "success" } return "publishing" } func normalizePublishStatus(status string) string { switch strings.TrimSpace(status) { case "queued": return "queued" case "publishing", "running": return "publishing" case "success", "published", "publish_success": return "success" case "failed", "publish_failed": return "failed" case "pending_review": return "pending_review" 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) 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 }