Agent-System/internal/docduty/revoke.go

package docduty

import (
	"context"
	"errors"
	"fmt"

	"github.com/jackc/pgx/v5"
)

// MemberDutyChange 成员职责/在任状态变更，供显式撤销文档槽位。
type MemberDutyChange struct {
	UserID         int64
	PreviousDuty   string
	NextDuty       string
	Exited         bool
	PriorHolderIDs map[string]int64
}

// SlotRevocation 从受控文档上撤销某用户的审核/批准槽位；永不针对 author。
type SlotRevocation struct {
	UserID int64
	Duty   string
}

// PriorDutyHolders 更新前在任成员的职责持有者。
func PriorDutyHolders(members []Member) map[string]int64 {
	out := make(map[string]int64, 3)
	for _, m := range members {
		if !isActiveMember(m) {
			continue
		}
		switch m.DocumentDuty {
		case DutyAuthor, DutyReviewer, DutyApprover:
			out[m.DocumentDuty] = m.UserID
		}
	}
	return out
}

// PlanSlotRevocations 按更新前 duty 显式计算要撤销的审核/批准槽位。
// reviewer/approver → viewer/其他职责、职责转移、退出成员都会产生撤销；author 永不进入结果。
func PlanSlotRevocations(change MemberDutyChange) []SlotRevocation {
	seen := make(map[string]struct{}, 4)
	out := make([]SlotRevocation, 0, 2)
	add := func(userID int64, duty string) {
		if userID <= 0 || (duty != DutyReviewer && duty != DutyApprover) {
			return
		}
		key := fmt.Sprintf("%d:%s", userID, duty)
		if _, ok := seen[key]; ok {
			return
		}
		seen[key] = struct{}{}
		out = append(out, SlotRevocation{UserID: userID, Duty: duty})
	}
	if change.Exited || change.PreviousDuty != change.NextDuty {
		add(change.UserID, change.PreviousDuty)
	}
	if change.NextDuty == DutyReviewer || change.NextDuty == DutyApprover {
		if prior := change.PriorHolderIDs[change.NextDuty]; prior > 0 && prior != change.UserID {
			add(prior, change.NextDuty)
		}
	}
	return out
}

// 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
}

// projectDutyLockSQL 成员写与 HealProjectDocuments 共用同一键：pg_advisory_xact_lock(project_id)。
// 必须先于成员/文档读写；禁止改成两段 int 锁或其它键。
const projectDutyLockSQL = `SELECT pg_advisory_xact_lock($1)`

// projectDutyLockHook 仅测试使用：在申请锁之前同步，便于证明 heal 会阻塞在成员写事务之后。
var projectDutyLockHook func()

// SetProjectDutyLockHookForTest installs a pre-lock hook for PG concurrency tests.
func SetProjectDutyLockHookForTest(fn func()) {
	projectDutyLockHook = fn
}

// AcquireProjectDutyLock 获取项目职责互斥锁；调用方须已 Begin，锁随事务结束释放。
func AcquireProjectDutyLock(ctx context.Context, tx pgx.Tx, projectID int64) error {
	if tx == nil || projectID <= 0 {
		return fmt.Errorf("docduty: invalid project lock")
	}
	if hook := projectDutyLockHook; hook != nil {
		hook()
	}
	if _, err := tx.Exec(ctx, projectDutyLockSQL, projectID); err != nil {
		return fmt.Errorf("docduty: lock project %d: %w", projectID, err)
	}
	return nil
}