diff --git a/internal/acp/agentkind_service_test.go b/internal/acp/agentkind_service_test.go index 1bfac4b..f12d6c3 100644 --- a/internal/acp/agentkind_service_test.go +++ b/internal/acp/agentkind_service_test.go @@ -6,6 +6,7 @@ import ( "time" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -124,6 +125,14 @@ func (f *fakeAgentKindRepo) PurgeEventsBefore(context.Context, time.Time) (int64 panic("n/a") } +func (f *fakeAgentKindRepo) InTx(context.Context, func(context.Context, pgx.Tx) error) error { + panic("n/a") +} + +func (f *fakeAgentKindRepo) WithTx(pgx.Tx) acp.Repository { + return f +} + // agentKindRecordingRecorder:accumulate audit entries type agentKindRecordingRecorder struct{ entries []audit.Entry } diff --git a/internal/acp/relay_test.go b/internal/acp/relay_test.go index 8bd1f04..9c573f0 100644 --- a/internal/acp/relay_test.go +++ b/internal/acp/relay_test.go @@ -18,6 +18,7 @@ import ( "time" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -112,6 +113,14 @@ func (f *fakeEventRepo) PurgeEventsBefore(context.Context, time.Time) (int64, er panic("n/a") } +func (f *fakeEventRepo) InTx(context.Context, func(context.Context, pgx.Tx) error) error { + panic("n/a") +} + +func (f *fakeEventRepo) WithTx(pgx.Tx) acp.Repository { + return f +} + // relayTestRig wires up a Relay with two io.Pipe pairs that simulate the agent // subprocess. Returns: // - r: the relay under test diff --git a/internal/acp/repository.go b/internal/acp/repository.go index ecb069b..8f117b8 100644 --- a/internal/acp/repository.go +++ b/internal/acp/repository.go @@ -47,6 +47,9 @@ type Repository interface { InsertEvent(ctx context.Context, e *Event) (*Event, error) ListEventsSince(ctx context.Context, sessionID uuid.UUID, sinceID int64, limit int32) ([]*Event, error) PurgeEventsBefore(ctx context.Context, before time.Time) (int64, error) + + InTx(ctx context.Context, fn func(ctx context.Context, tx pgx.Tx) error) error + WithTx(tx pgx.Tx) Repository } // IsUniqueViolation reports whether err is a PG unique constraint violation @@ -74,6 +77,23 @@ func NewPostgresRepository(pool *pgxpool.Pool) Repository { return &pgRepo{pool: pool, q: acpsqlc.New(pool)} } +// InTx runs fn in a single transaction. Commits on nil error; rolls back +// otherwise. Used by SessionService.Create to atomically insert acp_sessions +// + mcp_tokens (spec §6.1). +func (r *pgRepo) InTx(ctx context.Context, fn func(ctx context.Context, tx pgx.Tx) error) error { + if r.pool == nil { + return errs.New(errs.CodeInternal, "acp.Repository.InTx: nil pool (tx-bound instance)") + } + return pgx.BeginFunc(ctx, r.pool, func(tx pgx.Tx) error { + return fn(ctx, tx) + }) +} + +// WithTx 返回绑定到指定事务的 Repository 副本。 +func (r *pgRepo) WithTx(tx pgx.Tx) Repository { + return &pgRepo{pool: nil, q: r.q.WithTx(tx)} +} + // ===== 类型转换 helper ===== func toPgUUID(u uuid.UUID) pgtype.UUID { return pgtype.UUID{Bytes: u, Valid: true} } diff --git a/internal/mcp/repository.go b/internal/mcp/repository.go index dd52634..0a4b6a4 100644 --- a/internal/mcp/repository.go +++ b/internal/mcp/repository.go @@ -32,6 +32,9 @@ type Repository interface { // 测试辅助:直接物理删(生产不暴露) DeleteByID(ctx context.Context, id uuid.UUID) error + + // WithTx 返回绑定到指定事务的 Repository。详见 pgRepo.WithTx 文档。 + WithTx(tx pgx.Tx) Repository } // NewPostgresRepository 用 pgxpool 构造 Repository。 @@ -44,6 +47,13 @@ type pgRepo struct { q *mcpsqlc.Queries } +// WithTx 返回一个绑定到指定事务的 Repository 副本。Caller 在外部用 pgx.BeginFunc +// 或 acp.Repository.InTx 取得 tx;这里只负责生成一个把所有 sqlc 调用切到 tx 的轻 +// 实例。tx-bound 实例不持有 pool(防止误用同一实例混 pool/tx 两路)。 +func (r *pgRepo) WithTx(tx pgx.Tx) Repository { + return &pgRepo{pool: nil, q: r.q.WithTx(tx)} +} + // ===== 类型转换 helper ===== func toPgUUID(u uuid.UUID) pgtype.UUID { diff --git a/internal/mcp/token_service_test.go b/internal/mcp/token_service_test.go index 0105b2a..334e12b 100644 --- a/internal/mcp/token_service_test.go +++ b/internal/mcp/token_service_test.go @@ -8,6 +8,7 @@ import ( "time" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -122,6 +123,10 @@ func (r *fakeRepo) DeleteByID(ctx context.Context, id uuid.UUID) error { return nil } +func (r *fakeRepo) WithTx(tx pgx.Tx) mcp.Repository { + return r +} + // ===== fake audit ===== type fakeAudit struct {