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 <clio-agent@sisyphuslabs.ai>
This commit is contained in:
2026-06-21 19:28:18 +01:00
co-authored by Sisyphus
parent f2e8e3eb77
commit c69a243f75
4 changed files with 191 additions and 0 deletions
+24
View File
@@ -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
}
+83
View File
@@ -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)
}
+32
View File
@@ -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
}
+52
View File
@@ -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
}