diff --git a/packages/shared-types/src/index.ts b/packages/shared-types/src/index.ts index dbb9626..ea730d5 100644 --- a/packages/shared-types/src/index.ts +++ b/packages/shared-types/src/index.ts @@ -405,8 +405,12 @@ export interface CreatePublishJobRequest { } export interface CreatePublishJobResponse { + /** Empty when all requested targets were already covered by existing publish tasks. */ job_id: string + /** Target-ordered union of created_task_ids and existing_task_ids. */ task_ids: string[] + created_task_ids: string[] + existing_task_ids: string[] } export interface WorkspaceOverview { @@ -991,6 +995,7 @@ export interface PublishRecord { publish_batch_id: number article_id: number platform_account_id: number + desktop_account_id: string | null platform_id: string platform_name: string platform_nickname: string diff --git a/server/internal/tenant/app/media_service.go b/server/internal/tenant/app/media_service.go index 0a82252..0eb636f 100644 --- a/server/internal/tenant/app/media_service.go +++ b/server/internal/tenant/app/media_service.go @@ -39,6 +39,7 @@ type PublishRecordResponse struct { PublishBatchID int64 `json:"publish_batch_id"` ArticleID int64 `json:"article_id"` PlatformAccountID int64 `json:"platform_account_id"` + DesktopAccountID *string `json:"desktop_account_id"` PlatformID string `json:"platform_id"` PlatformName string `json:"platform_name"` PlatformNickname string `json:"platform_nickname"` @@ -97,7 +98,7 @@ func (s *MediaService) ListArticlePublishRecords(ctx context.Context, articleID func (s *MediaService) listArticlePublishRecords(ctx context.Context, tenantID, brandID, 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, + SELECT pr.id, pr.publish_batch_id, pr.article_id, pr.platform_account_id, pa.desktop_id::text, 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 @@ -115,11 +116,13 @@ func (s *MediaService) listArticlePublishRecords(ctx context.Context, tenantID, items := make([]PublishRecordResponse, 0) for rows.Next() { var item PublishRecordResponse + var desktopAccountID string if err := rows.Scan( &item.ID, &item.PublishBatchID, &item.ArticleID, &item.PlatformAccountID, + &desktopAccountID, &item.PlatformID, &item.PlatformName, &item.PlatformNickname, @@ -134,6 +137,9 @@ func (s *MediaService) listArticlePublishRecords(ctx context.Context, tenantID, ); err != nil { return nil, response.ErrInternal(50043, "publish_record_scan_failed", err.Error()) } + if strings.TrimSpace(desktopAccountID) != "" { + item.DesktopAccountID = &desktopAccountID + } normalizePublishRecordForResponse(&item) items = append(items, item) } diff --git a/server/internal/tenant/app/publish_job_service.go b/server/internal/tenant/app/publish_job_service.go index f8683dd..8cfaa2f 100644 --- a/server/internal/tenant/app/publish_job_service.go +++ b/server/internal/tenant/app/publish_job_service.go @@ -4,11 +4,14 @@ import ( "context" "encoding/json" "errors" + "sort" + "strconv" "strings" "time" "github.com/google/uuid" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgxpool" goredis "github.com/redis/go-redis/v9" "go.uber.org/zap" @@ -79,10 +82,29 @@ type CreatePublishJobRequest struct { } type CreatePublishJobResponse struct { - JobID string `json:"job_id"` - TaskIDs []string `json:"task_ids"` + // JobID is set only when this request created a new publish job. It is empty + // when every requested target was already covered by an existing publish task. + JobID string `json:"job_id"` + // TaskIDs is the target-ordered union of created and already-existing tasks. + TaskIDs []string `json:"task_ids"` + CreatedTaskIDs []string `json:"created_task_ids"` + ExistingTaskIDs []string `json:"existing_task_ids"` } +type publishTarget struct { + account *repository.DesktopAccount + accountSeed *platformAccountSeed + targetClientID uuid.UUID + dedupKey string +} + +type existingPublishRecordTarget struct { + TaskID uuid.UUID + PlatformAccountID int64 +} + +const desktopPublishDedupActiveConstraint = "uq_desktop_tasks_publish_dedup_active" + func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req CreatePublishJobRequest) (*CreatePublishJobResponse, error) { if actor.TenantID == 0 || actor.PrimaryWorkspaceID == 0 || actor.UserID == 0 { return nil, response.ErrUnauthorized(40132, "workspace_scope_missing", "workspace context is required") @@ -105,13 +127,9 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr return nil, err } - requestedAccountIDs := make([]uuid.UUID, 0, len(req.Accounts)) - for _, accountReq := range req.Accounts { - accountID, parseErr := uuid.Parse(strings.TrimSpace(accountReq.AccountID)) - if parseErr != nil { - return nil, response.ErrBadRequest(40092, "invalid_desktop_account_id", "account_id must be a uuid") - } - requestedAccountIDs = append(requestedAccountIDs, accountID) + requestedAccountIDs, err := parseUniquePublishAccountIDs(req.Accounts) + if err != nil { + return nil, err } runtimeHealth := loadDesktopAccountRuntimeHealth(ctx, s.redis, actor.PrimaryWorkspaceID, requestedAccountIDs) @@ -128,13 +146,7 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr return nil, err } - type publishTarget struct { - account *repository.DesktopAccount - accountSeed *platformAccountSeed - targetClientID uuid.UUID - } targets := make([]publishTarget, 0, len(req.Accounts)) - targetPlatformSet := make(map[string]struct{}) for _, accountID := range requestedAccountIDs { account, lookupErr := accountRepo.GetByDesktopID(ctx, accountID, actor.PrimaryWorkspaceID) @@ -165,20 +177,15 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr account: account, accountSeed: accountSeed, targetClientID: *targetClientID, + dedupKey: desktopPublishDedupKey(actor.TenantID, actor.PrimaryWorkspaceID, articleID, accountSeed.PlatformID, account.DesktopID), }) - if platformID := strings.TrimSpace(accountSeed.PlatformID); platformID != "" { - targetPlatformSet[platformID] = struct{}{} - } } effectiveVersionID, err := s.compliance.ResolvePublishableVersion(ctx, actor.TenantID, articleID, req.ArticleVersionID) if err != nil { return nil, err } - targetPlatforms := make([]string, 0, len(targetPlatformSet)) - for platformID := range targetPlatformSet { - targetPlatforms = append(targetPlatforms, platformID) - } + targetPlatforms := publishTargetPlatforms(targets) gate, err := s.compliance.GateForPublish(ctx, tenantcompliance.GateForPublishRequest{ TenantID: actor.TenantID, ArticleID: articleID, @@ -200,6 +207,50 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr } } + dedupKeys := desktopPublishDedupKeysFromTargets(targets) + if len(targets) > 0 && len(dedupKeys) == 0 { + return nil, response.ErrInternal(50130, "desktop_publish_dedup_key_missing", "failed to build desktop publish deduplication key") + } + if err := lockDesktopPublishDedupKeys(ctx, tx, dedupKeys); err != nil { + return nil, err + } + existingTasks, err := loadExistingPublishRecordTargets(ctx, tx, actor.TenantID, actor.PrimaryWorkspaceID, articleID, targets) + if err != nil { + return nil, err + } + newTargets := make([]publishTarget, 0, len(targets)) + taskIDsByAccount := make(map[uuid.UUID]string, len(targets)) + existingTaskIDs := make([]string, 0, len(targets)) + responseJobID := "" + for _, target := range targets { + if existing, ok := existingTasks[target.accountSeed.ID]; ok { + taskIDsByAccount[target.account.DesktopID] = existing.TaskID.String() + existingTaskIDs = append(existingTaskIDs, existing.TaskID.String()) + continue + } + newTargets = append(newTargets, target) + } + if len(newTargets) == 0 { + // No new task or job was created, so leave compliance acknowledgements + // untouched and let the deferred rollback release xact locks. + taskIDs := publishTaskIDsInTargetOrder(targets, taskIDsByAccount) + return &CreatePublishJobResponse{ + JobID: responseJobID, + TaskIDs: taskIDs, + CreatedTaskIDs: []string{}, + ExistingTaskIDs: existingTaskIDs, + }, nil + } + + accountSeeds := make([]platformAccountSeed, 0, len(newTargets)) + for _, target := range newTargets { + accountSeeds = append(accountSeeds, *target.accountSeed) + } + + if publishBatchRequiresCover(accountSeeds) && strings.TrimSpace(pointerStringValue(articleMeta.CoverAssetURL)) == "" { + return nil, response.ErrBadRequest(40044, "publish_cover_required", "selected publishing platform requires a cover image") + } + jobID := uuid.New() job, err := taskRepo.CreatePublishJob(ctx, repository.CreateDesktopPublishJobParams{ DesktopID: jobID, @@ -215,14 +266,7 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr if err != nil { return nil, response.ErrInternal(50097, "desktop_publish_job_create_failed", "failed to create desktop publish job") } - - accountSeeds := make([]platformAccountSeed, 0, len(targets)) - for _, target := range targets { - accountSeeds = append(accountSeeds, *target.accountSeed) - } - if publishBatchRequiresCover(accountSeeds) && strings.TrimSpace(pointerStringValue(articleMeta.CoverAssetURL)) == "" { - return nil, response.ErrBadRequest(40044, "publish_cover_required", "selected publishing platform requires a cover image") - } + responseJobID = job.DesktopID.String() publishBatchID, publishRecordIDs, err := createPublishBatchRecords( ctx, @@ -238,9 +282,9 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr return nil, err } - taskIDs := make([]string, 0, len(targets)) - createdTasks := make([]*repository.DesktopTask, 0, len(targets)) - for _, target := range targets { + createdTasks := make([]*repository.DesktopTask, 0, len(newTargets)) + createdTaskIDs := make([]string, 0, len(newTargets)) + for _, target := range newTargets { publishRecordID := publishRecordIDs[target.accountSeed.ID] payloadJSON, payloadErr := marshalOptionalJSON(map[string]any{ "title": req.Title, @@ -268,34 +312,22 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr Kind: "publish", Payload: payloadJSON, Status: "queued", + DedupKey: &target.dedupKey, }) if createErr != nil { + if isPostgresUniqueViolation(createErr, desktopPublishDedupActiveConstraint) { + return nil, response.ErrConflict(40995, "desktop_publish_duplicate", "article has already been submitted for this platform account") + } return nil, response.ErrInternal(50099, "desktop_task_create_failed", "failed to create desktop publish task") } createdTasks = append(createdTasks, task) - taskIDs = append(taskIDs, task.DesktopID.String()) + taskIDsByAccount[target.account.DesktopID] = task.DesktopID.String() + createdTaskIDs = append(createdTaskIDs, task.DesktopID.String()) } - if req.AckRecordID != nil { - if gate == nil || gate.Result == nil { - return nil, response.ErrBadRequest(41003, "compliance_ack_mismatch", "ack record cannot be consumed without a matching compliance result") - } - consumed, consumeErr := s.compliance.ConsumeAckTx(ctx, tx, *req.AckRecordID, tenantcompliance.ConsumeAckQuery{ - TenantID: actor.TenantID, - ArticleID: articleID, - ArticleVersionID: effectiveVersionID, - CheckRecordID: gate.Result.RecordID, - ContentHash: gate.Result.ContentHash, - PolicyFingerprint: gate.Result.PolicyFingerprint, - ConsumedByJobID: jobID, - }) - if consumeErr != nil { - return nil, response.ErrInternal(51004, "compliance_ack_lookup_failed", "failed to consume compliance ack") - } - if !consumed { - return nil, response.ErrBadRequest(41003, "compliance_ack_mismatch", "ack record is expired, consumed, or no longer matches this publish request") - } + if err := s.consumePublishAckTx(ctx, tx, req.AckRecordID, gate, actor.TenantID, articleID, effectiveVersionID, &jobID); err != nil { + return nil, err } if err := tx.Commit(ctx); err != nil { @@ -343,8 +375,10 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr } return &CreatePublishJobResponse{ - JobID: job.DesktopID.String(), - TaskIDs: taskIDs, + JobID: responseJobID, + TaskIDs: publishTaskIDsInTargetOrder(targets, taskIDsByAccount), + CreatedTaskIDs: createdTaskIDs, + ExistingTaskIDs: existingTaskIDs, }, nil } @@ -362,6 +396,249 @@ func compliancePublishError(err error, gate *tenantcompliance.GateOutcome) error return ©Err } +func (s *PublishJobService) consumePublishAckTx( + ctx context.Context, + tx pgx.Tx, + ackRecordID *int64, + gate *tenantcompliance.GateOutcome, + tenantID int64, + articleID int64, + articleVersionID int64, + consumedByJobID *uuid.UUID, +) error { + if ackRecordID == nil { + return nil + } + if gate == nil || gate.Result == nil { + return response.ErrBadRequest(41003, "compliance_ack_mismatch", "ack record cannot be consumed without a matching compliance result") + } + consumed, consumeErr := s.compliance.ConsumeAckTx(ctx, tx, *ackRecordID, tenantcompliance.ConsumeAckQuery{ + TenantID: tenantID, + ArticleID: articleID, + ArticleVersionID: articleVersionID, + CheckRecordID: gate.Result.RecordID, + ContentHash: gate.Result.ContentHash, + PolicyFingerprint: gate.Result.PolicyFingerprint, + ConsumedByJobID: consumedByJobID, + }) + if consumeErr != nil { + return response.ErrInternal(51004, "compliance_ack_lookup_failed", "failed to consume compliance ack") + } + if !consumed { + return response.ErrBadRequest(41003, "compliance_ack_mismatch", "ack record is expired, consumed, or no longer matches this publish request") + } + return nil +} + +func parseUniquePublishAccountIDs(accounts []CreatePublishJobAccountRequest) ([]uuid.UUID, error) { + if len(accounts) == 0 { + return nil, response.ErrBadRequest(40092, "invalid_desktop_account_id", "at least one account is required") + } + seen := make(map[uuid.UUID]struct{}, len(accounts)) + ids := make([]uuid.UUID, 0, len(accounts)) + for _, accountReq := range accounts { + accountID, parseErr := uuid.Parse(strings.TrimSpace(accountReq.AccountID)) + if parseErr != nil { + return nil, response.ErrBadRequest(40092, "invalid_desktop_account_id", "account_id must be a uuid") + } + if _, ok := seen[accountID]; ok { + continue + } + seen[accountID] = struct{}{} + ids = append(ids, accountID) + } + if len(ids) == 0 { + return nil, response.ErrBadRequest(40092, "invalid_desktop_account_id", "at least one account is required") + } + return ids, nil +} + +func desktopPublishDedupKey(tenantID, workspaceID, articleID int64, platformID string, accountID uuid.UUID) string { + parts := []string{ + "publish", + strconv.FormatInt(tenantID, 10), + strconv.FormatInt(workspaceID, 10), + strconv.FormatInt(articleID, 10), + strings.TrimSpace(platformID), + accountID.String(), + } + encoded, err := json.Marshal(parts) + if err != nil { + return "" + } + return string(encoded) +} + +func desktopPublishDedupKeysFromTargets(targets []publishTarget) []string { + seen := make(map[string]struct{}, len(targets)) + keys := make([]string, 0, len(targets)) + for _, target := range targets { + key := strings.TrimSpace(target.dedupKey) + if key == "" { + continue + } + if _, ok := seen[key]; ok { + continue + } + seen[key] = struct{}{} + keys = append(keys, key) + } + sort.Strings(keys) + return keys +} + +func publishTargetPlatforms(targets []publishTarget) []string { + seen := make(map[string]struct{}, len(targets)) + platforms := make([]string, 0, len(targets)) + for _, target := range targets { + if target.accountSeed == nil { + continue + } + platformID := strings.TrimSpace(target.accountSeed.PlatformID) + if platformID == "" { + continue + } + if _, ok := seen[platformID]; ok { + continue + } + seen[platformID] = struct{}{} + platforms = append(platforms, platformID) + } + sort.Strings(platforms) + return platforms +} + +func lockDesktopPublishDedupKeys(ctx context.Context, tx pgx.Tx, keys []string) error { + if len(keys) == 0 { + return nil + } + if _, err := tx.Exec(ctx, ` + SELECT pg_advisory_xact_lock(hashtextextended(dedup_key, 0)) + FROM ( + SELECT dedup_key + FROM unnest($1::text[]) AS dedup_keys(dedup_key) + ORDER BY dedup_key + ) AS ordered_dedup_keys + `, keys); err != nil { + return response.ErrInternal(50128, "desktop_publish_dedup_lock_failed", "failed to lock desktop publish deduplication key") + } + return nil +} + +func loadExistingPublishRecordTargets( + ctx context.Context, + tx pgx.Tx, + tenantID int64, + workspaceID int64, + articleID int64, + targets []publishTarget, +) (map[int64]existingPublishRecordTarget, error) { + if len(targets) == 0 { + return map[int64]existingPublishRecordTarget{}, nil + } + + platformAccountIDs := make([]int64, 0, len(targets)) + platformAccountIDSet := make(map[int64]struct{}, len(targets)) + for _, target := range targets { + if target.accountSeed == nil || target.accountSeed.ID <= 0 { + continue + } + if _, ok := platformAccountIDSet[target.accountSeed.ID]; ok { + continue + } + platformAccountIDSet[target.accountSeed.ID] = struct{}{} + platformAccountIDs = append(platformAccountIDs, target.accountSeed.ID) + } + if len(platformAccountIDs) == 0 { + return map[int64]existingPublishRecordTarget{}, nil + } + + rows, err := tx.Query(ctx, ` + SELECT DISTINCT ON (pr.platform_account_id) + dt.desktop_id, + pr.platform_account_id + FROM publish_records pr + JOIN platform_accounts pa + ON pa.id = pr.platform_account_id + AND pa.tenant_id = pr.tenant_id + JOIN desktop_tasks dt + ON dt.tenant_id = pr.tenant_id + AND dt.tenant_id = $1 + AND dt.workspace_id = $2 + AND dt.kind = 'publish' + AND dt.payload ? 'publish_record_id' + -- Keep this expression aligned with idx_desktop_tasks_publish_record_payload. + AND CASE + WHEN (dt.payload->>'publish_record_id') ~ '^[0-9]+$' + THEN (dt.payload->>'publish_record_id')::bigint + ELSE NULL + END = pr.id + WHERE pr.tenant_id = $1 + AND pa.workspace_id = $2 + AND pr.article_id = $3 + AND pr.platform_account_id = ANY($4::bigint[]) + AND pr.status IN ('queued', 'publishing', 'success') + AND dt.status IN ('queued', 'in_progress', 'succeeded', 'unknown') + ORDER BY pr.platform_account_id, + CASE + WHEN pr.status = 'success' OR dt.status = 'succeeded' THEN 0 + WHEN pr.status IN ('queued', 'publishing') OR dt.status IN ('queued', 'in_progress') THEN 1 + ELSE 2 + END, + GREATEST( + COALESCE(dt.updated_at, '-infinity'::timestamptz), + COALESCE(pr.updated_at, '-infinity'::timestamptz) + ) DESC, + GREATEST( + COALESCE(dt.created_at, '-infinity'::timestamptz), + COALESCE(pr.created_at, '-infinity'::timestamptz) + ) DESC, + dt.created_at DESC + `, tenantID, workspaceID, articleID, platformAccountIDs) + if err != nil { + return nil, response.ErrInternal(50129, "desktop_publish_dedup_lookup_failed", "failed to inspect existing publish records") + } + defer rows.Close() + + existing := make(map[int64]existingPublishRecordTarget, len(platformAccountIDs)) + for rows.Next() { + var item existingPublishRecordTarget + if scanErr := rows.Scan( + &item.TaskID, + &item.PlatformAccountID, + ); scanErr != nil { + return nil, response.ErrInternal(50129, "desktop_publish_dedup_lookup_failed", "failed to parse existing publish record") + } + existing[item.PlatformAccountID] = item + } + if err := rows.Err(); err != nil { + return nil, response.ErrInternal(50129, "desktop_publish_dedup_lookup_failed", "failed to iterate existing publish records") + } + return existing, nil +} + +func publishTaskIDsInTargetOrder(targets []publishTarget, taskIDsByAccount map[uuid.UUID]string) []string { + taskIDs := make([]string, 0, len(targets)) + seen := make(map[string]struct{}, len(targets)) + for _, target := range targets { + taskID := strings.TrimSpace(taskIDsByAccount[target.account.DesktopID]) + if taskID == "" { + continue + } + if _, ok := seen[taskID]; ok { + continue + } + seen[taskID] = struct{}{} + taskIDs = append(taskIDs, taskID) + } + return taskIDs +} + +func isPostgresUniqueViolation(err error, constraintName string) bool { + var pgErr *pgconn.PgError + return errors.As(err, &pgErr) && pgErr.Code == "23505" && pgErr.ConstraintName == constraintName +} + func (s *PublishJobService) RetryByDesktopTask( ctx context.Context, client *repository.DesktopClient, @@ -397,13 +674,18 @@ func (s *PublishJobService) RetryByDesktopTask( if err != nil { return nil, err } + brandID, err := s.resolveRetryArticleBrandID(ctx, client.TenantID, articleID) + if err != nil { + return nil, err + } title := strings.TrimSpace(stringPointerValue(extractStringPointer(payload, "title"))) if title == "" { title = "文章发布" } - return s.Create(ctx, auth.Actor{ + publishCtx := auth.WithCurrentBrandID(ctx, brandID) + return s.Create(publishCtx, auth.Actor{ TenantID: client.TenantID, UserID: client.UserID, PrimaryWorkspaceID: client.WorkspaceID, @@ -534,6 +816,27 @@ func (s *PublishJobService) authorizePublishTaskForActor( return nil } +func (s *PublishJobService) resolveRetryArticleBrandID(ctx context.Context, tenantID int64, articleID int64) (int64, error) { + var brandID int64 + err := s.pool.QueryRow(ctx, ` + SELECT brand_id + FROM articles + WHERE id = $1 + AND tenant_id = $2 + AND deleted_at IS NULL + `, articleID, tenantID).Scan(&brandID) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return 0, response.ErrNotFound(40411, "article_not_found", "article not found") + } + return 0, response.ErrInternal(50037, "article_check_failed", "failed to validate article") + } + if brandID <= 0 { + return 0, response.ErrInternal(50037, "article_check_failed", "failed to validate article") + } + return brandID, nil +} + func (s *PublishJobService) resolveRetryArticleID(ctx context.Context, tenantID int64, payload map[string]any) (int64, error) { if articleID, ok := resolveRetryArticleIDFromPayload(payload); ok { return articleID, nil diff --git a/server/internal/tenant/app/publish_job_service_test.go b/server/internal/tenant/app/publish_job_service_test.go index 4c43fc7..e1cb47b 100644 --- a/server/internal/tenant/app/publish_job_service_test.go +++ b/server/internal/tenant/app/publish_job_service_test.go @@ -1,6 +1,17 @@ package app -import "testing" +import ( + "context" + "fmt" + "strings" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + + "github.com/geo-platform/tenant-api/internal/tenant/repository" +) func TestResolveRetryArticleIDFromPayload(t *testing.T) { t.Parallel() @@ -61,3 +72,215 @@ func TestResolveRetryArticleIDFromPayload(t *testing.T) { }) } } + +func TestParseUniquePublishAccountIDsDedupesInRequestOrder(t *testing.T) { + t.Parallel() + + first := uuid.MustParse("11111111-1111-1111-1111-111111111111") + second := uuid.MustParse("22222222-2222-2222-2222-222222222222") + + got, err := parseUniquePublishAccountIDs([]CreatePublishJobAccountRequest{ + {AccountID: " " + first.String() + " "}, + {AccountID: second.String()}, + {AccountID: first.String()}, + }) + if err != nil { + t.Fatalf("parseUniquePublishAccountIDs() error = %v", err) + } + if len(got) != 2 || got[0] != first || got[1] != second { + t.Fatalf("parseUniquePublishAccountIDs() = %v, want [%s %s]", got, first, second) + } +} + +func TestDesktopPublishDedupKeyUsesArticlePlatformAndAccount(t *testing.T) { + t.Parallel() + + accountID := uuid.MustParse("33333333-3333-3333-3333-333333333333") + got := desktopPublishDedupKey(7, 11, 897, " sohuhao ", accountID) + want := `["publish","7","11","897","sohuhao","33333333-3333-3333-3333-333333333333"]` + if got != want { + t.Fatalf("desktopPublishDedupKey() = %q, want %q", got, want) + } +} + +func TestDesktopPublishDedupKeyEncodesAmbiguousPlatformIDs(t *testing.T) { + t.Parallel() + + accountID := uuid.MustParse("33333333-3333-3333-3333-333333333333") + got := desktopPublishDedupKey(7, 11, 897, "sohu:hao", accountID) + want := `["publish","7","11","897","sohu:hao","33333333-3333-3333-3333-333333333333"]` + if got != want { + t.Fatalf("desktopPublishDedupKey() = %q, want %q", got, want) + } +} + +func TestDesktopPublishDedupKeysFromTargetsSortsAndDedupes(t *testing.T) { + t.Parallel() + + got := desktopPublishDedupKeysFromTargets([]publishTarget{ + {dedupKey: "publish:1:1:2:b:account"}, + {dedupKey: "publish:1:1:2:a:account"}, + {dedupKey: "publish:1:1:2:b:account"}, + {dedupKey: " "}, + }) + want := []string{"publish:1:1:2:a:account", "publish:1:1:2:b:account"} + if len(got) != len(want) { + t.Fatalf("desktopPublishDedupKeysFromTargets() length = %d, want %d; got %v", len(got), len(want), got) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("desktopPublishDedupKeysFromTargets()[%d] = %q, want %q; got %v", i, got[i], want[i], got) + } + } +} + +func TestPublishTargetPlatformsSortsAndDedupes(t *testing.T) { + t.Parallel() + + got := publishTargetPlatforms([]publishTarget{ + {accountSeed: &platformAccountSeed{PlatformID: "xiaohongshu"}}, + {accountSeed: &platformAccountSeed{PlatformID: " weixin_channels "}}, + {accountSeed: &platformAccountSeed{PlatformID: "xiaohongshu"}}, + {accountSeed: &platformAccountSeed{PlatformID: " "}}, + {}, + }) + want := []string{"weixin_channels", "xiaohongshu"} + if len(got) != len(want) { + t.Fatalf("publishTargetPlatforms() length = %d, want %d; got %v", len(got), len(want), got) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("publishTargetPlatforms()[%d] = %q, want %q; got %v", i, got[i], want[i], got) + } + } +} + +func TestLoadExistingPublishRecordTargetsDedupeUnknownTasks(t *testing.T) { + t.Parallel() + + tx := &capturePublishDedupLookupTx{rows: emptyPublishDedupRows{}} + accountID := uuid.MustParse("44444444-4444-4444-4444-444444444444") + targets := []publishTarget{ + {accountSeed: &platformAccountSeed{ID: 77}, account: &repository.DesktopAccount{DesktopID: accountID}}, + } + + _, err := loadExistingPublishRecordTargets(context.Background(), tx, 1, 2, 3, targets) + if err != nil { + t.Fatalf("loadExistingPublishRecordTargets() error = %v", err) + } + normalizedSQL := normalizeSQLWhitespace(tx.sql) + if !strings.Contains(normalizedSQL, "dt.status IN ('queued', 'in_progress', 'succeeded', 'unknown')") { + t.Fatalf("publish dedup lookup should include unknown task status; query:\n%s", tx.sql) + } + if !strings.Contains(normalizedSQL, "AND dt.tenant_id = $1") { + t.Fatalf("publish dedup lookup should constrain desktop_tasks by tenant for the payload index; query:\n%s", tx.sql) + } + if strings.Contains(normalizedSQL, "desktop_publish_jobs") { + t.Fatalf("publish dedup lookup should not load unused publish job metadata; query:\n%s", tx.sql) + } +} + +func TestLockDesktopPublishDedupKeysUsesSingleArrayQuery(t *testing.T) { + t.Parallel() + + tx := &capturePublishDedupLookupTx{} + err := lockDesktopPublishDedupKeys(context.Background(), tx, []string{"a", "b"}) + if err != nil { + t.Fatalf("lockDesktopPublishDedupKeys() error = %v", err) + } + if tx.execCount != 1 { + t.Fatalf("lockDesktopPublishDedupKeys() exec count = %d, want 1", tx.execCount) + } + if !strings.Contains(tx.execSQL, "unnest($1::text[])") { + t.Fatalf("lockDesktopPublishDedupKeys() should use array unnest query; query:\n%s", tx.execSQL) + } + gotKeys, ok := tx.execArgs[0].([]string) + if !ok || len(gotKeys) != 2 || gotKeys[0] != "a" || gotKeys[1] != "b" { + t.Fatalf("lockDesktopPublishDedupKeys() args = %#v, want [a b]", tx.execArgs) + } +} + +func TestLockDesktopPublishDedupKeysSkipsEmptyList(t *testing.T) { + t.Parallel() + + tx := &capturePublishDedupLookupTx{} + err := lockDesktopPublishDedupKeys(context.Background(), tx, nil) + if err != nil { + t.Fatalf("lockDesktopPublishDedupKeys() error = %v", err) + } + if tx.execCount != 0 { + t.Fatalf("lockDesktopPublishDedupKeys() exec count = %d, want 0", tx.execCount) + } +} + +func TestIsPostgresUniqueViolationMatchesConstraint(t *testing.T) { + t.Parallel() + + err := fmt.Errorf("wrapped: %w", &pgconn.PgError{ + Code: "23505", + ConstraintName: desktopPublishDedupActiveConstraint, + }) + if !isPostgresUniqueViolation(err, desktopPublishDedupActiveConstraint) { + t.Fatal("isPostgresUniqueViolation() should match the dedup constraint") + } + if isPostgresUniqueViolation(&pgconn.PgError{Code: "23505", ConstraintName: "desktop_tasks_desktop_id_key"}, desktopPublishDedupActiveConstraint) { + t.Fatal("isPostgresUniqueViolation() should ignore other unique constraints") + } + if isPostgresUniqueViolation(&pgconn.PgError{Code: "23503", ConstraintName: desktopPublishDedupActiveConstraint}, desktopPublishDedupActiveConstraint) { + t.Fatal("isPostgresUniqueViolation() should require SQLSTATE 23505") + } +} + +type capturePublishDedupLookupTx struct { + sql string + args []any + execSQL string + execArgs []any + execCount int + rows pgx.Rows +} + +func (tx *capturePublishDedupLookupTx) Begin(context.Context) (pgx.Tx, error) { return nil, nil } +func (tx *capturePublishDedupLookupTx) Commit(context.Context) error { return nil } +func (tx *capturePublishDedupLookupTx) Rollback(context.Context) error { return nil } +func (tx *capturePublishDedupLookupTx) CopyFrom(context.Context, pgx.Identifier, []string, pgx.CopyFromSource) (int64, error) { + return 0, nil +} +func (tx *capturePublishDedupLookupTx) SendBatch(context.Context, *pgx.Batch) pgx.BatchResults { + return nil +} +func (tx *capturePublishDedupLookupTx) LargeObjects() pgx.LargeObjects { return pgx.LargeObjects{} } +func (tx *capturePublishDedupLookupTx) Prepare(context.Context, string, string) (*pgconn.StatementDescription, error) { + return nil, nil +} +func (tx *capturePublishDedupLookupTx) Exec(_ context.Context, sql string, arguments ...any) (pgconn.CommandTag, error) { + tx.execCount++ + tx.execSQL = sql + tx.execArgs = arguments + return pgconn.CommandTag{}, nil +} +func (tx *capturePublishDedupLookupTx) Query(_ context.Context, sql string, args ...any) (pgx.Rows, error) { + tx.sql = sql + tx.args = args + return tx.rows, nil +} +func (tx *capturePublishDedupLookupTx) QueryRow(context.Context, string, ...any) pgx.Row { + return nil +} +func (tx *capturePublishDedupLookupTx) Conn() *pgx.Conn { return nil } + +type emptyPublishDedupRows struct{} + +func (r emptyPublishDedupRows) Close() {} +func (r emptyPublishDedupRows) Err() error { return nil } +func (r emptyPublishDedupRows) CommandTag() pgconn.CommandTag { return pgconn.CommandTag{} } +func (r emptyPublishDedupRows) FieldDescriptions() []pgconn.FieldDescription { return nil } +func (r emptyPublishDedupRows) Next() bool { return false } +func (r emptyPublishDedupRows) Scan(...any) error { return nil } +func (r emptyPublishDedupRows) Values() ([]any, error) { return nil, nil } +func (r emptyPublishDedupRows) RawValues() [][]byte { return nil } +func (r emptyPublishDedupRows) Conn() *pgx.Conn { return nil } + +func normalizeSQLWhitespace(sql string) string { + return strings.Join(strings.Fields(sql), " ") +} diff --git a/server/internal/tenant/compliance/service.go b/server/internal/tenant/compliance/service.go index 33f76e6..463a888 100644 --- a/server/internal/tenant/compliance/service.go +++ b/server/internal/tenant/compliance/service.go @@ -120,7 +120,7 @@ type ConsumeAckQuery struct { CheckRecordID int64 ContentHash string PolicyFingerprint string - ConsumedByJobID uuid.UUID + ConsumedByJobID *uuid.UUID } type CreateManualReviewRequest struct {