test(docduty): add isolated PG heal, backfill, and concurrency fixtures

Require TEST_DATABASE_URL and refuse agent_system_dev. Script creates a throwaway database, runs packages serially, then drops it. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-13 22:16:44 +00:00
parent 7ef34a685d
commit c2d4e9a55c
4 changed files with 386 additions and 14 deletions
@@ -0,0 +1,129 @@
+package authz
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"chat-agent-system/internal/db"
+	"chat-agent-system/internal/docduty"
+)
+
+func TestUpsertProjectMemberDuty_ConcurrentLastWriter_PG(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping PostgreSQL duty concurrency in -short")
+	}
+	dsn := db.SkipIfNoTestDatabaseURL(t)
+	ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
+	defer cancel()
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatalf("postgres unavailable: %v", err)
+	}
+	defer pool.Close()
+	if err := pool.Ping(ctx); err != nil {
+		t.Fatalf("postgres ping failed: %v", err)
+	}
+	if err := db.RunMigrationsOnceForTest(ctx, pool, dsn); err != nil {
+		t.Fatalf("migration failed: %v", err)
+	}
+
+	suffix := time.Now().UnixNano()
+	ownerID := insertAuthzUser(t, ctx, pool, suffix, "owner")
+	reviewerA := insertAuthzUser(t, ctx, pool, suffix, "ra")
+	reviewerB := insertAuthzUser(t, ctx, pool, suffix, "rb")
+
+	var projectID int64
+	if err := pool.QueryRow(ctx, `
+		INSERT INTO projects (code, name, status, creator_user_id)
+		VALUES ($1, $2, 'active', $3)
+		RETURNING id
+	`, fmt.Sprintf("duty-race-%d", suffix), "concurrent reviewer", ownerID).Scan(&projectID); err != nil {
+		t.Fatalf("insert project: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), `DELETE FROM project_members WHERE project_id = $1`, projectID)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id = $1`, projectID)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id IN ($1,$2,$3)`, ownerID, reviewerA, reviewerB)
+	})
+
+	a := NewAuthorizer(pool)
+	dutyAuthor := docduty.DutyAuthor
+	if err := a.UpsertProjectMemberSpec(ctx, projectID, ownerID, MemberUpsert{
+		ProjectRole: RoleOwner, DocumentDuty: &dutyAuthor,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if err := a.UpsertProjectMember(ctx, projectID, reviewerA, RoleMember); err != nil {
+		t.Fatal(err)
+	}
+	if err := a.UpsertProjectMember(ctx, projectID, reviewerB, RoleMember); err != nil {
+		t.Fatal(err)
+	}
+
+	duty := docduty.DutyReviewer
+	var wg sync.WaitGroup
+	errCh := make(chan error, 16)
+	targets := []int64{reviewerA, reviewerB}
+	for i := 0; i < 12; i++ {
+		target := targets[i%2]
+		wg.Add(1)
+		go func(uid int64) {
+			defer wg.Done()
+			if err := a.UpsertProjectMemberSpec(ctx, projectID, uid, MemberUpsert{
+				ProjectRole: RoleMember, DocumentDuty: &duty,
+			}); err != nil {
+				errCh <- err
+			}
+		}(target)
+	}
+	wg.Wait()
+	close(errCh)
+	for upsertErr := range errCh {
+		t.Fatalf("concurrent upsert failed after retry: %v", upsertErr)
+	}
+
+	rows, err := pool.Query(ctx, `
+		SELECT user_id FROM project_members
+		WHERE project_id = $1 AND status = 'active' AND document_duty = 'reviewer'
+		ORDER BY user_id
+	`, projectID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rows.Close()
+	var holders []int64
+	for rows.Next() {
+		var id int64
+		if err := rows.Scan(&id); err != nil {
+			t.Fatal(err)
+		}
+		holders = append(holders, id)
+	}
+	if err := rows.Err(); err != nil {
+		t.Fatal(err)
+	}
+	if len(holders) != 1 {
+		t.Fatalf("want exactly one active reviewer, got %v", holders)
+	}
+	if holders[0] != reviewerA && holders[0] != reviewerB {
+		t.Fatalf("unexpected reviewer %d", holders[0])
+	}
+}
+
+func insertAuthzUser(t *testing.T, ctx context.Context, pool *pgxpool.Pool, suffix int64, kind string) int64 {
+	t.Helper()
+	var id int64
+	if err := pool.QueryRow(ctx, `
+		INSERT INTO users (account_id, name, role)
+		VALUES ($1, $2, 'engineer')
+		RETURNING id
+	`, fmt.Sprintf("authz_%s_%d", kind, suffix), "用户"+kind).Scan(&id); err != nil {
+		t.Fatalf("insert user %s: %v", kind, err)
+	}
+	return id
+}
@@ -1,11 +1,33 @@
 package db

 import (
+	"context"
 	"os"
 	"strings"
+	"sync"
 	"testing"
+
+	"github.com/jackc/pgx/v5/pgxpool"
 )

+var (
+	migrateOnceMu sync.Mutex
+	migratedDSNs  = map[string]error{}
+)
+
+// RunMigrationsOnceForTest 同一 DSN 只迁一次，避免并行包抢同一隔离库。
+func RunMigrationsOnceForTest(ctx context.Context, pool *pgxpool.Pool, dsn string) error {
+	key := strings.TrimSpace(dsn)
+	migrateOnceMu.Lock()
+	defer migrateOnceMu.Unlock()
+	if err, ok := migratedDSNs[key]; ok {
+		return err
+	}
+	err := RunMigrations(ctx, pool, ResolveMigrationsDir())
+	migratedDSNs[key] = err
+	return err
+}
+
 const testDatabaseEnv = "TEST_DATABASE_URL"

 // RequireTestDatabaseURL 集成测试必须显式提供隔离测试库，禁止默认连开发库。
@@ -11,30 +11,45 @@ import (
 	"chat-agent-system/internal/db"
 )

-func TestProjectReviewerDutySync_PG(t *testing.T) {
+func openDutyTestPool(t *testing.T) (context.Context, context.CancelFunc, *pgxpool.Pool) {
+	t.Helper()
 	if testing.Short() {
 		t.Skip("skipping PostgreSQL duty sync integration in -short")
 	}
 	dsn := db.SkipIfNoTestDatabaseURL(t)
 	ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
-	defer cancel()
 	pool, err := pgxpool.New(ctx, dsn)
 	if err != nil {
-		t.Skipf("postgres unavailable: %v", err)
+		cancel()
+		t.Fatalf("postgres unavailable: %v", err)
 	}
-	defer pool.Close()
 	if err := pool.Ping(ctx); err != nil {
-		t.Skipf("postgres ping failed: %v", err)
+		pool.Close()
+		cancel()
+		t.Fatalf("postgres ping failed: %v", err)
 	}
-	if err := db.RunMigrations(ctx, pool, db.ResolveMigrationsDir()); err != nil {
-		t.Skipf("migration failed: %v", err)
+	if err := db.RunMigrationsOnceForTest(ctx, pool, dsn); err != nil {
+		pool.Close()
+		cancel()
+		t.Fatalf("migration failed: %v", err)
 	}
+	t.Cleanup(func() {
+		pool.Close()
+		cancel()
+	})
+	return ctx, cancel, pool
+}
+
+func TestProjectReviewerDutySync_PG(t *testing.T) {
+	ctx, _, pool := openDutyTestPool(t)

 	suffix := time.Now().UnixNano()
 	authorID := insertDutyUser(t, ctx, pool, suffix, "author")
 	reviewerID := insertDutyUser(t, ctx, pool, suffix, "reviewer")
 	outsiderID := insertDutyUser(t, ctx, pool, suffix, "none")
 	placeholderID := insertDutyUser(t, ctx, pool, suffix, "placeholder")
+	wrongApproverID := insertDutyUser(t, ctx, pool, suffix, "wrong_appr")
+	assignedApproverID := insertDutyUser(t, ctx, pool, suffix, "assigned_appr")

 	var projectID int64
 	if err := pool.QueryRow(ctx, `
@@ -48,14 +63,16 @@ func TestProjectReviewerDutySync_PG(t *testing.T) {
 		_, _ = pool.Exec(context.Background(), `DELETE FROM document_records WHERE project_id = $1`, projectID)
 		_, _ = pool.Exec(context.Background(), `DELETE FROM project_members WHERE project_id = $1`, projectID)
 		_, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id = $1`, projectID)
-		_, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id IN ($1,$2,$3,$4)`, authorID, reviewerID, outsiderID, placeholderID)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id IN ($1,$2,$3,$4,$5,$6)`, authorID, reviewerID, outsiderID, placeholderID, wrongApproverID, assignedApproverID)
 	})

 	if _, err := pool.Exec(ctx, `
 		INSERT INTO project_members (project_id, user_id, project_role, document_duty, status, joined_at, updated_at)
 		VALUES ($1, $2, 'owner', 'author', 'active', now(), now()),
-		       ($1, $3, 'member', 'reviewer', 'active', now(), now())
-	`, projectID, authorID, reviewerID); err != nil {
+		       ($1, $3, 'member', 'reviewer', 'active', now(), now()),
+		       ($1, $4, 'member', '', 'active', now(), now()),
+		       ($1, $5, 'member', 'approver', 'active', now(), now())
+	`, projectID, authorID, reviewerID, wrongApproverID, assignedApproverID); err != nil {
 		t.Fatalf("insert members: %v", err)
 	}

@@ -72,8 +89,8 @@ func TestProjectReviewerDutySync_PG(t *testing.T) {

 	intakeID := fmt.Sprintf("doc_proj%d_intake", projectID)
 	secondID := fmt.Sprintf("doc_proj%d_prelim_design", projectID)
-	insertDutyDoc(t, ctx, pool, projectID, intakeID, "intake", authorID, placeholderID, 0)
-	insertDutyDoc(t, ctx, pool, projectID, secondID, "prelim_design", authorID, placeholderID, 0)
+	insertDutyDoc(t, ctx, pool, projectID, intakeID, "intake", authorID, placeholderID, wrongApproverID)
+	insertDutyDoc(t, ctx, pool, projectID, secondID, "prelim_design", authorID, placeholderID, wrongApproverID)

 	assigned, n, err := HealProjectDocuments(ctx, pool, projectID)
 	if err != nil {
@@ -87,6 +104,9 @@ func TestProjectReviewerDutySync_PG(t *testing.T) {
 	}
 	assertDocReviewer(t, ctx, pool, intakeID, reviewerID)
 	assertDocReviewer(t, ctx, pool, secondID, reviewerID)
+	assertDocApprover(t, ctx, pool, intakeID, assignedApproverID)
+	assertDocApprover(t, ctx, pool, secondID, assignedApproverID)
+	assertDocAuthorName(t, ctx, pool, intakeID, "用户author")

 	_, n2, err := HealProjectDocuments(ctx, pool, projectID)
 	if err != nil {
@@ -96,6 +116,94 @@ func TestProjectReviewerDutySync_PG(t *testing.T) {
 		t.Fatalf("heal must be idempotent, second pass updated %d rows", n2)
 	}
 	assertDocReviewer(t, ctx, pool, secondID, reviewerID)
+	assertDocApprover(t, ctx, pool, secondID, assignedApproverID)
+}
+
+func TestHealBinding_EmptyAssignedKeepsMember_PG(t *testing.T) {
+	ctx, _, pool := openDutyTestPool(t)
+	suffix := time.Now().UnixNano()
+	authorID := insertDutyUser(t, ctx, pool, suffix, "keep_author")
+	memberID := insertDutyUser(t, ctx, pool, suffix, "keep_member")
+
+	var projectID int64
+	if err := pool.QueryRow(ctx, `
+		INSERT INTO projects (code, name, status, creator_user_id)
+		VALUES ($1, $2, 'active', $3)
+		RETURNING id
+	`, fmt.Sprintf("duty-keep-%d", suffix), "empty assigned keep", authorID).Scan(&projectID); err != nil {
+		t.Fatalf("insert project: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), `DELETE FROM document_records WHERE project_id = $1`, projectID)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM project_members WHERE project_id = $1`, projectID)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id = $1`, projectID)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id IN ($1,$2)`, authorID, memberID)
+	})
+	if _, err := pool.Exec(ctx, `
+		INSERT INTO project_members (project_id, user_id, project_role, document_duty, status, joined_at, updated_at)
+		VALUES ($1, $2, 'owner', 'author', 'active', now(), now()),
+		       ($1, $3, 'member', '', 'active', now(), now())
+	`, projectID, authorID, memberID); err != nil {
+		t.Fatalf("insert members: %v", err)
+	}
+	docID := fmt.Sprintf("doc_proj%d_intake", projectID)
+	insertDutyDoc(t, ctx, pool, projectID, docID, "intake", authorID, memberID, memberID)
+	if _, err := pool.Exec(ctx, `UPDATE document_records SET author_name = $2 WHERE doc_id = $1`, docID, "用户keep_author"); err != nil {
+		t.Fatal(err)
+	}
+
+	_, n, err := HealProjectDocuments(ctx, pool, projectID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n != 0 {
+		t.Fatalf("empty assigned reviewer/approver must keep existing member, updated %d", n)
+	}
+	assertDocReviewer(t, ctx, pool, docID, memberID)
+	assertDocApprover(t, ctx, pool, docID, memberID)
+}
+
+func TestOwnerAuthorBackfill_MultiOwnerDeterministic_PG(t *testing.T) {
+	ctx, _, pool := openDutyTestPool(t)
+	suffix := time.Now().UnixNano()
+	lowOwner := insertDutyUser(t, ctx, pool, suffix, "low_owner")
+	creatorOwner := insertDutyUser(t, ctx, pool, suffix, "creator_owner")
+	if lowOwner >= creatorOwner {
+		t.Fatalf("expected inserted low owner id < creator, got %d >= %d", lowOwner, creatorOwner)
+	}
+
+	var projectID int64
+	if err := pool.QueryRow(ctx, `
+		INSERT INTO projects (code, name, status, creator_user_id)
+		VALUES ($1, $2, 'active', $3)
+		RETURNING id
+	`, fmt.Sprintf("duty-owners-%d", suffix), "multi owner backfill", creatorOwner).Scan(&projectID); err != nil {
+		t.Fatalf("insert project: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), `DELETE FROM project_members WHERE project_id = $1`, projectID)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id = $1`, projectID)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id IN ($1,$2)`, lowOwner, creatorOwner)
+	})
+	if _, err := pool.Exec(ctx, `
+		INSERT INTO project_members (project_id, user_id, project_role, document_duty, status, joined_at, updated_at)
+		VALUES ($1, $2, 'owner', '', 'active', now(), now()),
+		       ($1, $3, 'owner', '', 'active', now(), now())
+	`, projectID, lowOwner, creatorOwner); err != nil {
+		t.Fatalf("insert owners: %v", err)
+	}
+
+	if _, err := pool.Exec(ctx, OwnerAuthorBackfillOneProjectSQL, projectID); err != nil {
+		t.Fatalf("owner author backfill: %v", err)
+	}
+	assertMemberDuty(t, ctx, pool, projectID, creatorOwner, DutyAuthor)
+	assertMemberDuty(t, ctx, pool, projectID, lowOwner, DutyNone)
+
+	if _, err := pool.Exec(ctx, OwnerAuthorBackfillOneProjectSQL, projectID); err != nil {
+		t.Fatalf("second backfill: %v", err)
+	}
+	assertMemberDuty(t, ctx, pool, projectID, creatorOwner, DutyAuthor)
+	assertMemberDuty(t, ctx, pool, projectID, lowOwner, DutyNone)
 }

 func insertDutyUser(t *testing.T, ctx context.Context, pool *pgxpool.Pool, suffix int64, kind string) int64 {
@@ -134,12 +242,47 @@ func insertDutyDoc(t *testing.T, ctx context.Context, pool *pgxpool.Pool, projec
 }

 func assertDocReviewer(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID string, want int64) {
+	t.Helper()
+	assertDocSlot(t, ctx, pool, docID, "reviewer_id", want)
+}
+
+func assertDocApprover(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID string, want int64) {
+	t.Helper()
+	assertDocSlot(t, ctx, pool, docID, "approver_id", want)
+}
+
+func assertDocSlot(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID, column string, want int64) {
 	t.Helper()
 	var got int64
-	if err := pool.QueryRow(ctx, `SELECT COALESCE(reviewer_id, 0) FROM document_records WHERE doc_id = $1`, docID).Scan(&got); err != nil {
+	q := fmt.Sprintf(`SELECT COALESCE(%s, 0) FROM document_records WHERE doc_id = $1`, column)
+	if err := pool.QueryRow(ctx, q, docID).Scan(&got); err != nil {
 		t.Fatal(err)
 	}
 	if got != want {
-		t.Fatalf("%s reviewer_id=%d want %d", docID, got, want)
+		t.Fatalf("%s %s=%d want %d", docID, column, got, want)
+	}
+}
+
+func assertDocAuthorName(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID, want string) {
+	t.Helper()
+	var got string
+	if err := pool.QueryRow(ctx, `SELECT COALESCE(author_name, '') FROM document_records WHERE doc_id = $1`, docID).Scan(&got); err != nil {
+		t.Fatal(err)
+	}
+	if got != want {
+		t.Fatalf("%s author_name=%q want %q", docID, got, want)
+	}
+}
+
+func assertMemberDuty(t *testing.T, ctx context.Context, pool *pgxpool.Pool, projectID, userID int64, want string) {
+	t.Helper()
+	var got string
+	if err := pool.QueryRow(ctx, `
+		SELECT COALESCE(document_duty, '') FROM project_members WHERE project_id = $1 AND user_id = $2
+	`, projectID, userID).Scan(&got); err != nil {
+		t.Fatal(err)
+	}
+	if got != want {
+		t.Fatalf("user %d document_duty=%q want %q", userID, got, want)
 	}
 }
@@ -0,0 +1,78 @@
+#!/usr/bin/env bash
+# 隔离库跑 document_duty 相关 PostgreSQL 集成测试。
+# 不启停/重建 compose，不连 agent_system_dev / 共享对话库。
+#
+# 用法 A（已有隔离库）：
+#   TEST_DATABASE_URL='postgres://app:pass@127.0.0.1:55432/agent_duty_sync_itest?sslmode=disable' \
+#     ./scripts/run-isolated-duty-sync-pg-tests.sh
+#
+# 用法 B（Codex/本机用管理连接建临时库，测完删除）：
+#   PG_ADMIN_URL='postgres://postgres:pass@127.0.0.1:55432/postgres?sslmode=disable' \
+#     ./scripts/run-isolated-duty-sync-pg-tests.sh
+#
+# 禁止：把 TEST_DATABASE_URL 指到 agent_system_dev、5433 默认开发库。
+set -euo pipefail
+
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "${ROOT_DIR}"
+
+refuse_shared_dsn() {
+  local dsn="$1"
+  if [[ "${dsn}" == *"agent_system_dev"* || "${dsn}" == *"/agent_system_test"* || "${dsn}" == *"/agent_system_main"* ]]; then
+    echo "[错误] 拒绝共享环境库名（agent_system_dev/test/main）。请 CREATE DATABASE 临时库。" >&2
+    exit 2
+  fi
+}
+
+rewrite_db_name() {
+  local src="$1"
+  local dbname="$2"
+  local prefix query
+  if [[ "${src}" == *"?"* ]]; then
+    prefix="${src%%\?*}"
+    query="${src#*\?}"
+    echo "${prefix%/*}/${dbname}?${query}"
+  else
+    echo "${src%/*}/${dbname}"
+  fi
+}
+
+CREATED_DB=""
+ADMIN_URL="${PG_ADMIN_URL:-}"
+TEST_URL="${TEST_DATABASE_URL:-}"
+
+if [[ -n "${ADMIN_URL}" ]]; then
+  refuse_shared_dsn "${ADMIN_URL}"
+  if ! command -v psql >/dev/null 2>&1; then
+    echo "[错误] 需要 psql 才能用 PG_ADMIN_URL 建临时库" >&2
+    exit 1
+  fi
+  CREATED_DB="agent_duty_sync_itest_$$"
+  echo "[info] creating isolated database ${CREATED_DB}"
+  psql "${ADMIN_URL}" -v ON_ERROR_STOP=1 -c "CREATE DATABASE ${CREATED_DB};"
+  cleanup() {
+    if [[ -n "${CREATED_DB}" ]]; then
+      echo "[info] dropping isolated database ${CREATED_DB}"
+      psql "${ADMIN_URL}" -v ON_ERROR_STOP=1 -c "DROP DATABASE IF EXISTS ${CREATED_DB};" || true
+    fi
+  }
+  trap cleanup EXIT
+  TEST_URL="$(rewrite_db_name "${ADMIN_URL}" "${CREATED_DB}")"
+elif [[ -n "${TEST_URL}" ]]; then
+  refuse_shared_dsn "${TEST_URL}"
+else
+  echo "[错误] 请设置 TEST_DATABASE_URL（隔离库）或 PG_ADMIN_URL（自动建临时库）" >&2
+  echo "示例：" >&2
+  echo "  createdb agent_duty_sync_itest" >&2
+  echo "  TEST_DATABASE_URL='postgres://app:pass@127.0.0.1:55432/agent_duty_sync_itest?sslmode=disable' \\" >&2
+  echo "    go test ./internal/docduty ./internal/authz ./internal/server -count=1 -run 'Duty|ReviewerDuty|DocumentDuty|ConcurrentLastWriter|OwnerAuthor'" >&2
+  exit 1
+fi
+
+export TEST_DATABASE_URL="${TEST_URL}"
+echo "[info] running isolated duty-sync PostgreSQL tests"
+
+# 包串行，避免并行 goose 抢同一临时库。
+go test ./internal/docduty -count=1 -run 'Duty|ReviewerDuty|DocumentDuty|OwnerAuthor|HealBinding'
+go test ./internal/authz -count=1 -run 'ConcurrentLastWriter|DocumentDuty'
+go test ./internal/server -count=1 -run 'AuthorizeDocRevision|ResolveUpdateProjectRole'