feat(publish-dedup): reject duplicate publish jobs and expose target task ids
Build a per-target dedup_key (tenant/workspace/article/platform/account) on publish job creation, take an xact advisory lock on the keys, look up already-active publish_records via the payload index, and skip creating duplicate desktop_tasks. Surface created vs existing task ids on CreatePublishJobResponse (and shared-types) so callers can tell the difference, return `desktop_publish_duplicate` on unique-constraint races, and expose desktop_account_id on PublishRecordResponse so the modal can disable already-published accounts. Make the compliance ack consumer accept a nullable job id since the dedup short-circuit no longer creates one. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -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), " ")
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user