From 0fa4a5784e8c10ccc5b6ab5aa349d23dc450af8f Mon Sep 17 00:00:00 2001 From: weichen Date: Sun, 13 Sep 2026 22:37:24 +0000 Subject: [PATCH] fix(authz): clear reviewer/approver slots in the member-write transaction Duty revoke must share the project advisory lock with reading old holders and writing members, so a post-commit service snapshot cannot miss a transfer. Heal still fills the new holder after commit, and sync failure stays ERR_DUTY_SYNC. Co-authored-by: Cursor --- internal/authz/authorizer.go | 124 +++++++++++---- internal/authz/authorizer_pg_test.go | 156 +++++++++++++++++++ internal/docduty/revoke.go | 69 +++++--- internal/docduty/store_pg_test.go | 111 +++++++------ internal/server/project_members_handlers.go | 74 ++------- internal/server/project_members_role_test.go | 7 - scripts/run-isolated-duty-sync-pg-tests.sh | 6 +- 7 files changed, 364 insertions(+), 183 deletions(-) diff --git a/internal/authz/authorizer.go b/internal/authz/authorizer.go index e78a0d15..c930ce40 100644 --- a/internal/authz/authorizer.go +++ b/internal/authz/authorizer.go @@ -282,6 +282,8 @@ func isRetryableDutyConflict(err error) bool { } // UpsertProjectMemberSpec 写入访问角色，并可选独占 document_duty。 +// 写入 document_duty 时，同一事务内：项目 advisory lock、读取旧/冲突 holder、 +// 成员更新、精确清空 document_records 的 reviewer/approver 槽；不改 author。 // 并发抢同一职责时重试，收敛为 last-writer，并依赖部分唯一索引保证无双 reviewer。 func (a *Authorizer) UpsertProjectMemberSpec(ctx context.Context, projectID, userID int64, spec MemberUpsert) error { if projectID <= 0 || userID <= 0 { @@ -320,41 +322,20 @@ func (a *Authorizer) upsertProjectMemberSpecOnce(ctx context.Context, projectID, defer func() { _ = tx.Rollback(ctx) }() if spec.DocumentDuty != nil { - if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, projectID); err != nil { + if err := applyDutyWriteInTx(ctx, tx, projectID, userID, spec.ProjectRole, duty); err != nil { return err } + return tx.Commit(ctx) } - if spec.DocumentDuty != nil && duty != docduty.DutyNone { - if _, err := tx.Exec(ctx, ` - UPDATE project_members - SET document_duty = '', updated_at = now() - WHERE project_id = $1 AND document_duty = $2 AND user_id <> $3 AND status = $4 - `, projectID, duty, userID, StatusActive); err != nil { - return err - } - } - - if spec.DocumentDuty != nil { - _, err = tx.Exec(ctx, ` - INSERT INTO project_members (project_id, user_id, project_role, document_duty, status, joined_at, updated_at) - VALUES ($1, $2, $3, $4, $5, now(), now()) - ON CONFLICT (project_id, user_id) DO UPDATE SET - project_role = EXCLUDED.project_role, - document_duty = EXCLUDED.document_duty, - status = EXCLUDED.status, - updated_at = now() - `, projectID, userID, spec.ProjectRole, duty, StatusActive) - } else { - _, err = tx.Exec(ctx, ` - INSERT INTO project_members (project_id, user_id, project_role, status, joined_at, updated_at) - VALUES ($1, $2, $3, $4, now(), now()) - ON CONFLICT (project_id, user_id) DO UPDATE SET - project_role = EXCLUDED.project_role, - status = EXCLUDED.status, - updated_at = now() - `, projectID, userID, spec.ProjectRole, StatusActive) - } + _, err = tx.Exec(ctx, ` + INSERT INTO project_members (project_id, user_id, project_role, status, joined_at, updated_at) + VALUES ($1, $2, $3, $4, now(), now()) + ON CONFLICT (project_id, user_id) DO UPDATE SET + project_role = EXCLUDED.project_role, + status = EXCLUDED.status, + updated_at = now() + `, projectID, userID, spec.ProjectRole, StatusActive) if err != nil { return err } @@ -362,6 +343,7 @@ func (a *Authorizer) upsertProjectMemberSpecOnce(ctx context.Context, projectID, } // SetProjectMemberStatus 更新成员状态（如 exited）。 +// 退出时与 upsert 相同：advisory lock、读取旧 duty、状态更新、精确清槽在同一事务。 func (a *Authorizer) SetProjectMemberStatus(ctx context.Context, projectID, userID int64, status string) error { if projectID <= 0 || userID <= 0 { return fmt.Errorf("authz: invalid project or user id") @@ -372,7 +354,32 @@ func (a *Authorizer) SetProjectMemberStatus(ctx context.Context, projectID, user if !a.enabled() { return nil } - tag, err := a.pool.Exec(ctx, ` + const maxAttempts = 4 + var last error + for attempt := 0; attempt < maxAttempts; attempt++ { + last = a.setProjectMemberStatusOnce(ctx, projectID, userID, status) + if last == nil || !isRetryableDutyConflict(last) { + return last + } + } + return last +} + +func (a *Authorizer) setProjectMemberStatusOnce(ctx context.Context, projectID, userID int64, status string) error { + tx, err := a.pool.Begin(ctx) + if err != nil { + return err + } + defer func() { _ = tx.Rollback(ctx) }() + + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, projectID); err != nil { + return err + } + prevDuty, err := docduty.LoadMemberDutyTx(ctx, tx, projectID, userID) + if err != nil { + return err + } + tag, err := tx.Exec(ctx, ` UPDATE project_members SET status = $3, updated_at = now() WHERE project_id = $1 AND user_id = $2 @@ -383,7 +390,58 @@ func (a *Authorizer) SetProjectMemberStatus(ctx context.Context, projectID, user if tag.RowsAffected() == 0 { return fmt.Errorf("authz: project member not found") } - return nil + if status == StatusExited { + if err := docduty.ApplySlotRevocations(ctx, tx, projectID, docduty.PlanSlotRevocations(docduty.MemberDutyChange{ + UserID: userID, PreviousDuty: prevDuty, NextDuty: prevDuty, Exited: true, + })); err != nil { + return err + } + } + return tx.Commit(ctx) +} + +// applyDutyWriteInTx 假定调用方已 Begin：lock → 读旧/冲突 holder → 独占写成员 → 清槽。 +func applyDutyWriteInTx(ctx context.Context, tx pgx.Tx, projectID, userID int64, projectRole, duty string) error { + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, projectID); err != nil { + return err + } + prevDuty, err := docduty.LoadMemberDutyTx(ctx, tx, projectID, userID) + if err != nil { + return err + } + holders, err := docduty.LoadActiveSlotHoldersTx(ctx, tx, projectID) + if err != nil { + return err + } + + if duty != docduty.DutyNone { + if _, err := tx.Exec(ctx, ` + UPDATE project_members + SET document_duty = '', updated_at = now() + WHERE project_id = $1 AND document_duty = $2 AND user_id <> $3 AND status = $4 + `, projectID, duty, userID, StatusActive); err != nil { + return err + } + } + + if _, err := tx.Exec(ctx, ` + INSERT INTO project_members (project_id, user_id, project_role, document_duty, status, joined_at, updated_at) + VALUES ($1, $2, $3, $4, $5, now(), now()) + ON CONFLICT (project_id, user_id) DO UPDATE SET + project_role = EXCLUDED.project_role, + document_duty = EXCLUDED.document_duty, + status = EXCLUDED.status, + updated_at = now() + `, projectID, userID, projectRole, duty, StatusActive); err != nil { + return err + } + + return docduty.ApplySlotRevocations(ctx, tx, projectID, docduty.PlanSlotRevocations(docduty.MemberDutyChange{ + UserID: userID, + PreviousDuty: prevDuty, + NextDuty: duty, + PriorHolderIDs: holders, + })) } // TableExists 检查 project_members 表是否存在。 diff --git a/internal/authz/authorizer_pg_test.go b/internal/authz/authorizer_pg_test.go index 0a6add27..1901ec7f 100644 --- a/internal/authz/authorizer_pg_test.go +++ b/internal/authz/authorizer_pg_test.go @@ -115,6 +115,162 @@ func TestUpsertProjectMemberDuty_ConcurrentLastWriter_PG(t *testing.T) { } } +func TestUpsertProjectMember_AtomicDutyRevoke_PG(t *testing.T) { + if testing.Short() { + t.Skip("skipping PostgreSQL duty revoke atomicity 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() + authorID := insertAuthzUser(t, ctx, pool, suffix, "author") + 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-atomic-%d", suffix), "atomic upsert revoke", 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,$3)`, authorID, reviewerA, reviewerB) + }) + + a := NewAuthorizer(pool) + dutyAuthor := docduty.DutyAuthor + dutyReviewer := docduty.DutyReviewer + dutyNone := docduty.DutyNone + if err := a.UpsertProjectMemberSpec(ctx, projectID, authorID, MemberUpsert{ + ProjectRole: RoleOwner, DocumentDuty: &dutyAuthor, + }); err != nil { + t.Fatal(err) + } + if err := a.UpsertProjectMemberSpec(ctx, projectID, reviewerA, MemberUpsert{ + ProjectRole: RoleMember, DocumentDuty: &dutyReviewer, + }); err != nil { + t.Fatal(err) + } + if err := a.UpsertProjectMemberSpec(ctx, projectID, reviewerB, MemberUpsert{ + ProjectRole: RoleMember, DocumentDuty: &dutyNone, + }); err != nil { + t.Fatal(err) + } + + docID := fmt.Sprintf("doc_proj%d_intake", projectID) + if _, err := pool.Exec(ctx, ` + INSERT INTO document_records ( + doc_id, project_id, task_key, title, stage, version, + current_state, is_locked, is_template_only, + author_id, author_name, reviewer_id, reviewer_name, approver_id, approver_name, + created_at, updated_at + ) VALUES ( + $1, $2, 'intake', 'intake doc', 'V0.5', 1, + 'drafting', false, false, + $3, '编制工程师', $4, '审核负责人', NULL, '', + now(), now() + ) + `, docID, projectID, authorID, reviewerA); err != nil { + t.Fatalf("insert doc: %v", err) + } + + if err := a.UpsertProjectMemberSpec(ctx, projectID, reviewerA, MemberUpsert{ + ProjectRole: RoleMember, DocumentDuty: &dutyNone, + }); err != nil { + t.Fatal(err) + } + assertAuthzDocSlot(t, ctx, pool, docID, "reviewer_id", 0) + assertAuthzDocSlot(t, ctx, pool, docID, "author_id", authorID) + assertAuthzMemberDuty(t, ctx, pool, projectID, reviewerA, docduty.DutyNone) + + if err := a.UpsertProjectMemberSpec(ctx, projectID, reviewerA, MemberUpsert{ + ProjectRole: RoleMember, DocumentDuty: &dutyReviewer, + }); err != nil { + t.Fatal(err) + } + if _, _, err := docduty.HealProjectDocuments(ctx, pool, projectID); err != nil { + t.Fatal(err) + } + assertAuthzDocSlot(t, ctx, pool, docID, "reviewer_id", reviewerA) + + if err := a.UpsertProjectMemberSpec(ctx, projectID, reviewerB, MemberUpsert{ + ProjectRole: RoleMember, DocumentDuty: &dutyReviewer, + }); err != nil { + t.Fatal(err) + } + assertAuthzDocSlot(t, ctx, pool, docID, "reviewer_id", 0) + assertAuthzDocSlot(t, ctx, pool, docID, "author_id", authorID) + assertAuthzMemberDuty(t, ctx, pool, projectID, reviewerA, docduty.DutyNone) + assertAuthzMemberDuty(t, ctx, pool, projectID, reviewerB, docduty.DutyReviewer) + + if _, _, err := docduty.HealProjectDocuments(ctx, pool, projectID); err != nil { + t.Fatal(err) + } + assertAuthzDocSlot(t, ctx, pool, docID, "reviewer_id", reviewerB) + assertAuthzDocSlot(t, ctx, pool, docID, "author_id", authorID) + + if err := a.UpsertProjectMemberSpec(ctx, projectID, reviewerB, MemberUpsert{ + ProjectRole: RoleMember, DocumentDuty: &dutyReviewer, + }); err != nil { + t.Fatal(err) + } + assertAuthzDocSlot(t, ctx, pool, docID, "reviewer_id", reviewerB) + + if err := a.SetProjectMemberStatus(ctx, projectID, reviewerB, StatusExited); err != nil { + t.Fatal(err) + } + assertAuthzDocSlot(t, ctx, pool, docID, "reviewer_id", 0) + assertAuthzDocSlot(t, ctx, pool, docID, "author_id", authorID) + + if err := a.SetProjectMemberStatus(ctx, projectID, reviewerB, StatusExited); err != nil { + t.Fatal(err) + } + assertAuthzDocSlot(t, ctx, pool, docID, "reviewer_id", 0) + assertAuthzDocSlot(t, ctx, pool, docID, "author_id", authorID) +} + +func assertAuthzDocSlot(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID, column string, want int64) { + t.Helper() + var got int64 + 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 %s=%d want %d", docID, column, got, want) + } +} + +func assertAuthzMemberDuty(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) + } +} + func insertAuthzUser(t *testing.T, ctx context.Context, pool *pgxpool.Pool, suffix int64, kind string) int64 { t.Helper() var id int64 diff --git a/internal/docduty/revoke.go b/internal/docduty/revoke.go index fe8527d2..3b77ee85 100644 --- a/internal/docduty/revoke.go +++ b/internal/docduty/revoke.go @@ -2,9 +2,10 @@ package docduty import ( "context" + "errors" "fmt" - "github.com/jackc/pgx/v5/pgxpool" + "github.com/jackc/pgx/v5" ) // MemberDutyChange 成员职责/在任状态变更，供显式撤销文档槽位。 @@ -64,20 +65,56 @@ func PlanSlotRevocations(change MemberDutyChange) []SlotRevocation { return out } -// ReconcileMemberDuty 在项目级 advisory lock 下按更新前 duty 清空对应槽位，再 heal 新职责。 -func ReconcileMemberDuty(ctx context.Context, pool *pgxpool.Pool, projectID int64, change MemberDutyChange) error { - if pool == nil || projectID <= 0 { +// LoadMemberDutyTx 在调用方事务内读取该成员当前 document_duty；无行则空串。 +func LoadMemberDutyTx(ctx context.Context, tx pgx.Tx, projectID, userID int64) (string, error) { + var duty string + err := tx.QueryRow(ctx, ` + SELECT COALESCE(document_duty, '') + FROM project_members + WHERE project_id = $1 AND user_id = $2 + `, projectID, userID).Scan(&duty) + if errors.Is(err, pgx.ErrNoRows) { + return "", nil + } + if err != nil { + return "", fmt.Errorf("docduty: load member duty: %w", err) + } + return duty, nil +} + +// LoadActiveSlotHoldersTx 在调用方事务内读取在任 reviewer/approver（不含 author）。 +func LoadActiveSlotHoldersTx(ctx context.Context, tx pgx.Tx, projectID int64) (map[string]int64, error) { + rows, err := tx.Query(ctx, ` + SELECT user_id, document_duty + FROM project_members + WHERE project_id = $1 AND status = $2 AND document_duty IN ($3, $4) + `, projectID, StatusActive, DutyReviewer, DutyApprover) + if err != nil { + return nil, fmt.Errorf("docduty: load slot holders: %w", err) + } + defer rows.Close() + out := make(map[string]int64, 2) + for rows.Next() { + var userID int64 + var duty string + if err := rows.Scan(&userID, &duty); err != nil { + return nil, fmt.Errorf("docduty: scan slot holder: %w", err) + } + out[duty] = userID + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("docduty: slot holders: %w", err) + } + return out, nil +} + +// ApplySlotRevocations 在调用方已持有的事务内精确清空 reviewer/approver 槽。 +// 不开启新事务、不加锁、不 heal；author 列永不写入。 +func ApplySlotRevocations(ctx context.Context, tx pgx.Tx, projectID int64, revs []SlotRevocation) error { + if tx == nil || projectID <= 0 { return nil } - tx, err := pool.Begin(ctx) - if err != nil { - return fmt.Errorf("docduty: begin reconcile: %w", err) - } - defer func() { _ = tx.Rollback(ctx) }() - if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, projectID); err != nil { - return fmt.Errorf("docduty: lock project %d: %w", projectID, err) - } - for _, rev := range PlanSlotRevocations(change) { + for _, rev := range revs { q, ok := clearDutySlotSQL(rev.Duty) if !ok { continue @@ -86,11 +123,7 @@ func ReconcileMemberDuty(ctx context.Context, pool *pgxpool.Pool, projectID int6 return fmt.Errorf("docduty: clear %s for user %d: %w", rev.Duty, rev.UserID, err) } } - if err := tx.Commit(ctx); err != nil { - return fmt.Errorf("docduty: commit reconcile: %w", err) - } - _, _, err = HealProjectDocuments(ctx, pool, projectID) - return err + return nil } func clearDutySlotSQL(duty string) (string, bool) { diff --git a/internal/docduty/store_pg_test.go b/internal/docduty/store_pg_test.go index 9cf451c2..44778250 100644 --- a/internal/docduty/store_pg_test.go +++ b/internal/docduty/store_pg_test.go @@ -207,104 +207,94 @@ func TestHealBinding_EmptyAssignedClearsExited_PG(t *testing.T) { assertDocAuthorName(t, ctx, pool, docID, "用户ex_author") } -func TestReconcileMemberDuty_ReviewerToViewerAndTransfer_PG(t *testing.T) { +func TestApplySlotRevocations_SameTxnCommitAndRollback_PG(t *testing.T) { ctx, _, pool := openDutyTestPool(t) suffix := time.Now().UnixNano() - authorID := insertDutyUser(t, ctx, pool, suffix, "rec_author") - reviewerA := insertDutyUser(t, ctx, pool, suffix, "rec_ra") - reviewerB := insertDutyUser(t, ctx, pool, suffix, "rec_rb") + authorID := insertDutyUser(t, ctx, pool, suffix, "txn_author") + reviewerA := insertDutyUser(t, ctx, pool, suffix, "txn_ra") 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-rec-%d", suffix), "explicit reconcile", authorID).Scan(&projectID); err != nil { + `, fmt.Sprintf("duty-txn-%d", suffix), "atomic revoke txn", 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,$3)`, authorID, reviewerA, reviewerB) + _, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id IN ($1,$2)`, authorID, reviewerA) }) 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()), - ($1, $4, 'member', '', 'active', now(), now()) - `, projectID, authorID, reviewerA, reviewerB); err != nil { + ($1, $3, 'member', 'reviewer', 'active', now(), now()) + `, projectID, authorID, reviewerA); err != nil { t.Fatalf("insert members: %v", err) } docID := fmt.Sprintf("doc_proj%d_intake", projectID) insertDutyDoc(t, ctx, pool, projectID, docID, "intake", authorID, reviewerA, 0) - if _, err := pool.Exec(ctx, ` - UPDATE project_members SET document_duty = '' WHERE project_id = $1 AND user_id = $2 - `, projectID, reviewerA); err != nil { + tx, err := pool.Begin(ctx) + if err != nil { t.Fatal(err) } - if err := ReconcileMemberDuty(ctx, pool, projectID, MemberDutyChange{ + if _, err := tx.Exec(ctx, ` + UPDATE project_members SET document_duty = '' WHERE project_id = $1 AND user_id = $2 + `, projectID, reviewerA); err != nil { + _ = tx.Rollback(ctx) + t.Fatal(err) + } + if err := ApplySlotRevocations(ctx, tx, projectID, PlanSlotRevocations(MemberDutyChange{ UserID: reviewerA, PreviousDuty: DutyReviewer, NextDuty: DutyNone, - }); err != nil { + })); err != nil { + _ = tx.Rollback(ctx) t.Fatal(err) } - assertDocReviewer(t, ctx, pool, docID, 0) - var authorGot int64 - if err := pool.QueryRow(ctx, `SELECT COALESCE(author_id, 0) FROM document_records WHERE doc_id = $1`, docID).Scan(&authorGot); err != nil { + if err := tx.Rollback(ctx); err != nil { t.Fatal(err) } - if authorGot != authorID { - t.Fatalf("author must stay %d, got %d", authorID, authorGot) - } + assertDocReviewer(t, ctx, pool, docID, reviewerA) + assertMemberDuty(t, ctx, pool, projectID, reviewerA, DutyReviewer) + assertDocAuthorID(t, ctx, pool, docID, authorID) - if _, err := pool.Exec(ctx, ` + tx, err = pool.Begin(ctx) + if err != nil { + t.Fatal(err) + } + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, projectID); err != nil { + _ = tx.Rollback(ctx) + t.Fatal(err) + } + prev, err := LoadMemberDutyTx(ctx, tx, projectID, reviewerA) + if err != nil { + _ = tx.Rollback(ctx) + t.Fatal(err) + } + if prev != DutyReviewer { + _ = tx.Rollback(ctx) + t.Fatalf("locked previous duty=%q", prev) + } + if _, err := tx.Exec(ctx, ` UPDATE project_members SET document_duty = '' WHERE project_id = $1 AND user_id = $2 `, projectID, reviewerA); err != nil { + _ = tx.Rollback(ctx) t.Fatal(err) } - if _, err := pool.Exec(ctx, ` - UPDATE project_members SET document_duty = 'reviewer' WHERE project_id = $1 AND user_id = $2 - `, projectID, reviewerB); err != nil { + if err := ApplySlotRevocations(ctx, tx, projectID, PlanSlotRevocations(MemberDutyChange{ + UserID: reviewerA, PreviousDuty: prev, NextDuty: DutyNone, + })); err != nil { + _ = tx.Rollback(ctx) t.Fatal(err) } - if _, err := pool.Exec(ctx, ` - UPDATE document_records SET reviewer_id = $2, reviewer_name = 'A' WHERE doc_id = $1 - `, docID, reviewerA); err != nil { - t.Fatal(err) - } - if err := ReconcileMemberDuty(ctx, pool, projectID, MemberDutyChange{ - UserID: reviewerB, PreviousDuty: DutyNone, NextDuty: DutyReviewer, - PriorHolderIDs: map[string]int64{DutyReviewer: reviewerA}, - }); err != nil { - t.Fatal(err) - } - assertDocReviewer(t, ctx, pool, docID, reviewerB) - - if err := ReconcileMemberDuty(ctx, pool, projectID, MemberDutyChange{ - UserID: reviewerB, PreviousDuty: DutyReviewer, NextDuty: DutyReviewer, - }); err != nil { - t.Fatal(err) - } - assertDocReviewer(t, ctx, pool, docID, reviewerB) - - if _, err := pool.Exec(ctx, ` - UPDATE project_members SET status = 'exited', document_duty = 'reviewer' WHERE project_id = $1 AND user_id = $2 - `, projectID, reviewerB); err != nil { - t.Fatal(err) - } - if err := ReconcileMemberDuty(ctx, pool, projectID, MemberDutyChange{ - UserID: reviewerB, PreviousDuty: DutyReviewer, NextDuty: DutyReviewer, Exited: true, - }); err != nil { - t.Fatal(err) - } - assertDocReviewer(t, ctx, pool, docID, 0) - if err := ReconcileMemberDuty(ctx, pool, projectID, MemberDutyChange{ - UserID: reviewerB, PreviousDuty: DutyReviewer, NextDuty: DutyReviewer, Exited: true, - }); err != nil { + if err := tx.Commit(ctx); err != nil { t.Fatal(err) } assertDocReviewer(t, ctx, pool, docID, 0) + assertMemberDuty(t, ctx, pool, projectID, reviewerA, DutyNone) + assertDocAuthorID(t, ctx, pool, docID, authorID) } func TestOwnerAuthorBackfill_MultiOwnerDeterministic_PG(t *testing.T) { @@ -407,6 +397,11 @@ func assertDocSlot(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID, } } +func assertDocAuthorID(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID string, want int64) { + t.Helper() + assertDocSlot(t, ctx, pool, docID, "author_id", want) +} + func assertDocAuthorName(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID, want string) { t.Helper() var got string diff --git a/internal/server/project_members_handlers.go b/internal/server/project_members_handlers.go index e321cc07..d2f5e9ea 100644 --- a/internal/server/project_members_handlers.go +++ b/internal/server/project_members_handlers.go @@ -29,27 +29,7 @@ func writeDutySyncError(w http.ResponseWriter) { http.Error(w, `{"error":"ERR_DUTY_SYNC","message":"成员已保存，但受控文档职责同步失败"}`, http.StatusInternalServerError) } -func dutyChangeForRemove(existing *authz.ProjectMemberRecord, userID int64) docduty.MemberDutyChange { - change := docduty.MemberDutyChange{UserID: userID, Exited: true} - if existing != nil { - change.PreviousDuty = existing.DocumentDuty - change.NextDuty = existing.DocumentDuty - } - return change -} - -func (s *Server) snapshotDutyHolders(ctx context.Context, projectID int64) map[string]int64 { - if s == nil || s.svc == nil || s.svc.Pool() == nil || projectID <= 0 { - return nil - } - members, _, err := docduty.LoadProjectMembers(ctx, s.svc.Pool(), projectID) - if err != nil { - return nil - } - return docduty.PriorDutyHolders(members) -} - -func (s *Server) reconcileAndHealProjectDuties(ctx context.Context, projectID int64, change docduty.MemberDutyChange) error { +func (s *Server) healProjectDuties(ctx context.Context, projectID int64) error { if s == nil || s.svc == nil || projectID <= 0 { return nil } @@ -57,9 +37,9 @@ func (s *Server) reconcileAndHealProjectDuties(ctx context.Context, projectID in if pool == nil { return nil } - err := docduty.ReconcileMemberDuty(ctx, pool, projectID, change) + _, _, err := docduty.HealProjectDocuments(ctx, pool, projectID) if err != nil { - log.Printf("[docduty] reconcile project %d after member change failed: %v", projectID, err) + log.Printf("[docduty] heal project %d after member change failed: %v", projectID, err) } return err } @@ -137,7 +117,6 @@ func (s *Server) handleAddProjectMember(w http.ResponseWriter, r *http.Request, if req.ProjectRole == "" { req.ProjectRole = authz.RoleMember } - priors := s.snapshotDutyHolders(r.Context(), projectID) if err := s.authorizer.UpsertProjectMemberSpec(r.Context(), projectID, req.UserID, authz.MemberUpsert{ ProjectRole: req.ProjectRole, DocumentDuty: req.DocumentDuty, @@ -145,13 +124,7 @@ func (s *Server) handleAddProjectMember(w http.ResponseWriter, r *http.Request, http.Error(w, err.Error(), http.StatusBadRequest) return } - nextDuty := "" - if req.DocumentDuty != nil { - nextDuty = *req.DocumentDuty - } - if err := s.reconcileAndHealProjectDuties(r.Context(), projectID, docduty.MemberDutyChange{ - UserID: req.UserID, NextDuty: nextDuty, PriorHolderIDs: priors, - }); err != nil { + if err := s.healProjectDuties(r.Context(), projectID); err != nil { writeDutySyncError(w) return } @@ -179,7 +152,7 @@ func (s *Server) handleUpdateProjectMember(w http.ResponseWriter, r *http.Reques http.Error(w, `{"error":"ERR_INVALID_PARAM","message":"project_role required with document_duty"}`, http.StatusBadRequest) return } - priors := s.snapshotDutyHolders(r.Context(), projectID) + wrote := false if req.ProjectRole != "" || req.DocumentDuty != nil { role, err := resolveUpdateProjectRole(req.ProjectRole, existing) if err != nil { @@ -193,28 +166,17 @@ func (s *Server) handleUpdateProjectMember(w http.ResponseWriter, r *http.Reques http.Error(w, err.Error(), http.StatusBadRequest) return } + wrote = true } if req.Status != "" { if err := s.authorizer.SetProjectMemberStatus(r.Context(), projectID, req.UserID, req.Status); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } + wrote = true } - if req.ProjectRole != "" || req.DocumentDuty != nil || req.Status != "" { - nextDuty := "" - if existing != nil { - nextDuty = existing.DocumentDuty - } - if req.DocumentDuty != nil { - nextDuty = *req.DocumentDuty - } - if err := s.reconcileAndHealProjectDuties(r.Context(), projectID, docduty.MemberDutyChange{ - UserID: req.UserID, - PreviousDuty: existingDuty(existing), - NextDuty: nextDuty, - Exited: req.Status == authz.StatusExited, - PriorHolderIDs: priors, - }); err != nil { + if wrote { + if err := s.healProjectDuties(r.Context(), projectID); err != nil { writeDutySyncError(w) return } @@ -232,29 +194,13 @@ func (s *Server) handleRemoveProjectMember(w http.ResponseWriter, r *http.Reques http.Error(w, `{"error":"ERR_INVALID_PARAM","message":"user_id query required"}`, http.StatusBadRequest) return } - var existing *authz.ProjectMemberRecord - if s.authorizer != nil { - rec, err := s.authorizer.GetProjectMember(r.Context(), projectID, targetID) - if err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) - return - } - existing = rec - } if err := s.authorizer.SetProjectMemberStatus(r.Context(), projectID, targetID, authz.StatusExited); err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } - if err := s.reconcileAndHealProjectDuties(r.Context(), projectID, dutyChangeForRemove(existing, targetID)); err != nil { + if err := s.healProjectDuties(r.Context(), projectID); err != nil { writeDutySyncError(w) return } writeJSONResponse(w, http.StatusOK, map[string]bool{"success": true}) } - -func existingDuty(existing *authz.ProjectMemberRecord) string { - if existing == nil { - return "" - } - return existing.DocumentDuty -} diff --git a/internal/server/project_members_role_test.go b/internal/server/project_members_role_test.go index ab1037b0..a7dbb8d2 100644 --- a/internal/server/project_members_role_test.go +++ b/internal/server/project_members_role_test.go @@ -29,13 +29,6 @@ func TestResolveUpdateProjectRole_RejectsMissingMember(t *testing.T) { } } -func TestDutyChangeForRemove_UsesExistingReviewerDuty(t *testing.T) { - got := dutyChangeForRemove(&authz.ProjectMemberRecord{DocumentDuty: "reviewer"}, 171) - if !got.Exited || got.UserID != 171 || got.PreviousDuty != "reviewer" || got.NextDuty != "reviewer" { - t.Fatalf("remove change: %+v", got) - } -} - func TestWriteDutySyncError_UsesExplicitCode(t *testing.T) { rec := httptest.NewRecorder() writeDutySyncError(rec) diff --git a/scripts/run-isolated-duty-sync-pg-tests.sh b/scripts/run-isolated-duty-sync-pg-tests.sh index 19bb97db..52d81590 100755 --- a/scripts/run-isolated-duty-sync-pg-tests.sh +++ b/scripts/run-isolated-duty-sync-pg-tests.sh @@ -73,6 +73,6 @@ 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|ReconcileMemberDuty' -go test ./internal/authz -count=1 -run 'ConcurrentLastWriter|DocumentDuty' -go test ./internal/server -count=1 -run 'AuthorizeDocRevision|ResolveUpdateProjectRole|DutyChangeForRemove|WriteDutySyncError' +go test ./internal/docduty -count=1 -run 'Duty|ReviewerDuty|DocumentDuty|OwnerAuthor|HealBinding|ApplySlotRevocations' +go test ./internal/authz -count=1 -run 'ConcurrentLastWriter|DocumentDuty|AtomicDutyRevoke|UpsertProjectMember' +go test ./internal/server -count=1 -run 'AuthorizeDocRevision|ResolveUpdateProjectRole|WriteDutySyncError'