From c69a243f75e0f38fd88ab87b484258d0b8b0595a Mon Sep 17 00:00:00 2001 From: Stephen Adamson Date: Sun, 21 Jun 2026 19:28:18 +0100 Subject: [PATCH] feat(backend): add PoolProxy, Querier, and per-test transaction infrastructure New types and helpers for context-aware DB routing and per-test transactions: - PoolProxy: wraps pgxpool.Pool, routes queries through context transaction when active - Querier: interface accepted by fixture/helper functions for decoupling - ContextWithTx / TxFromContext: store/extract pgx.Tx in context.Context - SetupTestTx (in testutils): begins tx, stores in context, auto-rolls back on cleanup - SetupTestTx (in testtx): package-level variant with semaphore for parallel safety Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus --- backend/db/context.go | 24 ++++++++++ backend/db/proxy.go | 83 ++++++++++++++++++++++++++++++++++ backend/testutils/testtx/tx.go | 32 +++++++++++++ backend/testutils/tx.go | 52 +++++++++++++++++++++ 4 files changed, 191 insertions(+) create mode 100644 backend/db/context.go create mode 100644 backend/db/proxy.go create mode 100644 backend/testutils/testtx/tx.go create mode 100644 backend/testutils/tx.go diff --git a/backend/db/context.go b/backend/db/context.go new file mode 100644 index 0000000..4c54586 --- /dev/null +++ b/backend/db/context.go @@ -0,0 +1,24 @@ +package db + +import ( + "context" + + "github.com/jackc/pgx/v5" +) + +type ctxKey string + +const txCtxKey ctxKey = "poolproxy_tx" + +// ContextWithTx stores a pgx.Tx in the context for PoolProxy routing. +// When PoolProxy.Exec/Query/QueryRow sees this context key, it routes the +// call through the stored transaction instead of the pool. +func ContextWithTx(ctx context.Context, tx pgx.Tx) context.Context { + return context.WithValue(ctx, txCtxKey, tx) +} + +// TxFromContext extracts a pgx.Tx from context (returns nil if none active). +func TxFromContext(ctx context.Context) pgx.Tx { + tx, _ := ctx.Value(txCtxKey).(pgx.Tx) + return tx +} diff --git a/backend/db/proxy.go b/backend/db/proxy.go new file mode 100644 index 0000000..82863c1 --- /dev/null +++ b/backend/db/proxy.go @@ -0,0 +1,83 @@ +package db + +import ( + "context" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" +) + +// Querier is implemented by *pgxpool.Pool, pgx.Tx, and *PoolProxy. +// Fixture functions and internal helpers that need to run queries should +// accept Querier to remain decoupled from transaction state. +type Querier interface { + Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) + Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) + QueryRow(ctx context.Context, sql string, args ...any) pgx.Row +} + +// PoolProxy wraps *pgxpool.Pool and routes DB operations through an active +// transaction stored in context.Context. If no transaction is present, +// it delegates to the underlying pool directly. +// +// This enables per-test transactions: tests store a pgx.Tx in the request +// context, and all handler DB calls (via db.Conn.Exec/Query/QueryRow) route +// through that transaction automatically, rolling back on test cleanup. +type PoolProxy struct { + pool *pgxpool.Pool +} + +// NewPoolProxy creates a PoolProxy wrapping the given pool. +func NewPoolProxy(pool *pgxpool.Pool) *PoolProxy { + return &PoolProxy{pool: pool} +} + +// Pool returns the underlying pool, used for test setup and startup code. +func (p *PoolProxy) Pool() *pgxpool.Pool { return p.pool } + +// Exec runs a query, routing through a context transaction if one is active. +func (p *PoolProxy) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) { + if tx := TxFromContext(ctx); tx != nil { + return tx.Exec(ctx, sql, args...) + } + return p.pool.Exec(ctx, sql, args...) +} + +// Query runs a query, routing through a context transaction if one is active. +func (p *PoolProxy) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) { + if tx := TxFromContext(ctx); tx != nil { + return tx.Query(ctx, sql, args...) + } + return p.pool.Query(ctx, sql, args...) +} + +// QueryRow runs a query, routing through a context transaction if one is active. +func (p *PoolProxy) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row { + if tx := TxFromContext(ctx); tx != nil { + return tx.QueryRow(ctx, sql, args...) + } + return p.pool.QueryRow(ctx, sql, args...) +} + +// Begin starts a transaction. If the context already has an active transaction, +// it creates a savepoint (nested transaction) instead. This allows production +// code that calls db.Conn.Begin() to work inside per-test transactions. +func (p *PoolProxy) Begin(ctx context.Context) (pgx.Tx, error) { + if tx := TxFromContext(ctx); tx != nil { + return tx.Begin(ctx) + } + return p.pool.Begin(ctx) +} + +// Ping always goes to the underlying pool — it's a health check, not a query. +func (p *PoolProxy) Ping(ctx context.Context) error { + return p.pool.Ping(ctx) +} + +// Acquire always goes to the underlying pool — pgx.Tx has no Acquire method. +// The only production caller is the payment advisory lock, which needs a +// dedicated connection orthogonal to any transaction context. +func (p *PoolProxy) Acquire(ctx context.Context) (*pgxpool.Conn, error) { + return p.pool.Acquire(ctx) +} diff --git a/backend/testutils/testtx/tx.go b/backend/testutils/testtx/tx.go new file mode 100644 index 0000000..272d499 --- /dev/null +++ b/backend/testutils/testtx/tx.go @@ -0,0 +1,32 @@ +//go:build test +// +build test + +package testtx + +import ( + "context" + "testing" + + "crussell/db" +) + +// txSemaphore limits concurrent test transactions per package. +var txSemaphore = make(chan struct{}, 8) + +// SetupTestTx begins a transaction, stores it in context, and auto-rolls back. +func SetupTestTx(t *testing.T) (context.Context, db.Querier) { + t.Helper() + txSemaphore <- struct{}{} + pool := db.Conn.Pool() + tx, err := pool.Begin(context.Background()) + if err != nil { + <-txSemaphore + t.Fatalf("SetupTestTx: failed to begin transaction: %v", err) + } + ctx := db.ContextWithTx(context.Background(), tx) + t.Cleanup(func() { + tx.Rollback(context.Background()) + <-txSemaphore + }) + return ctx, tx +} diff --git a/backend/testutils/tx.go b/backend/testutils/tx.go new file mode 100644 index 0000000..6d68531 --- /dev/null +++ b/backend/testutils/tx.go @@ -0,0 +1,52 @@ +//go:build test +// +build test + +package testutils + +import ( + "context" + "testing" + + "crussell/db" +) + +// txSemaphore limits concurrent test transactions per package to prevent +// pgxpool connection exhaustion when many tests use t.Parallel(). +// The cap (8) should match the -parallel N flag used in test invocations +// and should be well below MaxConns per pool (currently 16). +var txSemaphore = make(chan struct{}, 8) + +// SetupTestTx begins a transaction on the db.Conn pool, stores it in the +// returned context (for PoolProxy routing), and returns the raw tx for use +// with fixture functions (e.g. fixtures.CreateTestUser(tx)). +// +// The transaction is automatically rolled back in t.Cleanup. After calling +// SetupTestTx, the test should call t.Parallel() and pass the returned context +// to any request helper functions (MakeRequest, makeAdminRequest, etc.). +// +// Example: +// +// func TestMyHandler(t *testing.T) { +// t.Parallel() +// ctx, tx := testutils.SetupTestTx(t) +// userID, _ := fixtures.CreateTestUser(tx) +// w := testutils.MakeUserRequest(handler, "GET", "/api/foo", nil, userID, ctx) +// assert.Equal(t, 200, w.Code) +// } +func SetupTestTx(t *testing.T) (context.Context, db.Querier) { + t.Helper() + txSemaphore <- struct{}{} + pool := db.Conn.Pool() + tx, err := pool.Begin(context.Background()) + if err != nil { + <-txSemaphore + t.Fatalf("SetupTestTx: failed to begin transaction: %v", err) + } + ctx := db.ContextWithTx(context.Background(), tx) + t.Cleanup(func() { + tx.Rollback(context.Background()) + <-txSemaphore + }) + return ctx, tx +} +