Storage

github.com/mbvlabs/andurel/pkg/storage is the data-infrastructure boundary: one pgx pool, PostgreSQL configuration, transactions, migrations, pagination, River queue clients, and test databases. SQL itself lives with the domain that owns it and is compiled by narsilc; see SQL Queries.

Connection lifecycle

Construct one *storage.Postgres per process and pass it around as storage.Connection. The constructor validates configuration, builds the pool, and pings PostgreSQL; a failed ping closes the pool and stops startup.

1db, err := storage.NewPostgres(ctx, cfg.Database, storage.WithTelemetry(tel))
2if err != nil {
3    return nil, err
4}
5app.DB = db
6app.closers.Add(func(context.Context) error { return db.Close() })

Connection is the application database handle:

1type Connection interface {
2    Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)
3    Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
4    QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
5    Health(ctx context.Context) error
6    BeginTransaction(ctx context.Context) (Transaction, error)
7}

The query methods match pgx's DBTX, so narsilc's generated client accepts the connection directly: queries.New(db). Use Health for a readiness check. Close belongs to the concrete *storage.Postgres; call it once, at shutdown.

Do not open a second pool for a library or a worker. Limits, telemetry, shutdown and transactions would then diverge. The queue process builds its own connection because it is a separate process, not because it needs a different pool inside one process.

Configuration and options

Start from storage.DefaultConfig() and replace the application identity and credentials. Defaults are a 5-second connect timeout, statement and description caches of 512, 25 open and idle connections, a 1-hour maximum lifetime, and a 30-minute maximum idle time.

Config.Validate requires a PostgreSQL kind, host, port, name, user, and a supported SSL mode (disable, allow, prefer, require, verify-ca, verify-full). Counts and durations cannot be negative. DatabaseURL() validates first and then builds the URL.

1cfg := storage.DefaultConfig()
2cfg.Name = "orbit_production"
3cfg.User = "orbit"
4cfg.Password = secret
5cfg.SSLMode = "verify-full"
6cfg.ApplicationName = "orbit-web"
7
8db, err := storage.NewPostgres(ctx, cfg)

Programmatic options override config fields: WithDatabaseURL, WithTLSConfig, WithRuntimeParameters, WithApplicationName, WithConnectTimeout, WithMaxOpenConnections, WithMaxIdleConnections (kept for compatibility; pgxpool has no max-idle analog), WithConnectionMaxLifetime, WithConnectionMaxIdleTime, and the two cache capacities. A custom URL keeps its pgx query parameters unless an explicit option overrides the same setting.

Telemetry is opt-in through config or an option:

• Config.OpenTelemetry enables default instrumentation.
• WithTelemetry(provider) accepts any TelemetryProvider; *telemetry.Telemetry satisfies it.
• WithOpenTelemetry(config) accepts explicit tracer and meter providers, attributes, SQL span naming, and query-parameter capture.
• WithoutOpenTelemetry() disables instrumentation.

Query parameters can contain secrets or personal data, so enable SQL parameter capture deliberately.

Queries and entities

Only domains/ imports the generated domains/internal/queries package. A domain receiver is constructed from the connection and binds the generated client once:

1type Users struct {
2    transactor storage.Transactor
3    queries    *queries.Queries
4}
5
6func NewUsers(db storage.Connection) Users {
7    return Users{transactor: storage.NewTransactor(db), queries: queries.New(db)}
8}

SQL lives beside the domain (domains/user.sql) and is compiled with andurel sync queries. Generated methods return a typed row or entity, so the domain maps expected failures once:

1func (u Users) Find(ctx context.Context, id uuid.UUID) (User, error) {
2    return storage.Row(u.queries.GetUser[User](ctx, id))
3}

Writes follow the same shape: the domain builds and validates the entity, then calls the generated method and maps a unique violation:

1created, err := u.queries.CreateUser[User](ctx, params)
2return storage.Row(created, storage.Conflict(err, "users_email_key"))

storage.Row maps pgx.ErrNoRows to storage.ErrNotFound (re-exported as domains.ErrNotFound) and passes other errors through. storage.Conflict wraps storage.ErrConflict only for a unique violation of the named constraint. Both sentinels are translated into HTTP responses by handlers.MapError; see Controllers.

Use storage.Page for list endpoints. It clamps the page to 1, the size to 10 by default and 100 at most, and keeps the offset inside int32 so a huge page cannot wrap into a negative OFFSET. storage.NewPaginated adds the total count and page metadata.

Transactions

storage.AtomicOps is the only way to commit or roll back. storage.Tx deliberately has no query, commit or rollback methods: holding one lets you bind work to the transaction, nothing else.

A service binds its dependencies at the top of the callback and uses only the bound locals:

 1err := storage.AtomicOps(ctx, i.transactor, func(ctx context.Context, tx storage.Tx) error {
 2    users, queue := i.users.InTx(tx), i.queue.InTx(tx)
 3
 4    created, err := users.Create(ctx, i.pepper, domains.CreateUserData{
 5        Email: data.Email,
 6        PasswordPair: domains.PasswordPair{
 7            Password:        data.Password,
 8            ConfirmPassword: data.ConfirmPassword,
 9        },
10    })
11    if err != nil {
12        return fmt.Errorf("create user: %w", err)
13    }
14
15    // The job is visible only if the transaction commits.
16    if _, err := queue.Insert(ctx, jobs.SendVerificationEmailArgs{UserID: created.ID}, nil); err != nil {
17        return fmt.Errorf("queue verification email: %w", err)
18    }
19
20    return nil
21})

The callback's result decides the outcome:

• return nil: commit; a commit failure is returned wrapped;
• return an error: roll back and return that error unchanged, so errors.Is still works;
• panic: roll back, then re-panic.

A domain receiver that must be atomic on its own holds a storage.Transactor. Its InTx rebinds the transactor, so the same method joins a caller's transaction instead of opening a second one:

1func (t Tokens) InTx(tx storage.Tx) Tokens {
2    return Tokens{
3        transactor: t.transactor.InTx(tx),
4        queries:    queries.New(storage.Unwrap(tx)),
5    }
6}

Only domains/ may call storage.Unwrap. andurel verify enforces that, along with the rules that services never name storage.Connection and that a callback uses only dependencies bound inside it.

Side effects outside PostgreSQL — HTTP calls, cache writes, direct email sends — do not roll back. Perform them after AtomicOps returns nil, or enqueue them on the bound queue so they only run after commit. Email is the important case; see Email Package.

Migrations and test databases

RunMigrations applies Goose migrations from an fs.FS. The scaffold embeds SQL with migrations.Migrations, so tests, the CLI and production binaries read the same history. Migrations are a one-shot process, never part of cmd/app startup:

1if err := storage.RunMigrations(ctx, db, migrations.Migrations, "."); err != nil {
2    return err
3}

andurel db migrate up runs the same path during development. See Database Commands.

Tests use pkg/storage/storagetest, which lives outside storage so production binaries do not link testcontainers. One cluster serves a package; one isolated, migrated database serves a case:

 1var testCluster *storagetest.Cluster
 2
 3func TestMain(m *testing.M) {
 4    storagetest.Main(m, &testCluster)
 5}
 6
 7func TestUsers_CreateThenFind(t *testing.T) {
 8    db := testCluster.NewTestDB(t, migrations.Migrations, ".")
 9    users := domains.NewUsers(db)
10    // …
11}

storagetest.Main reports the reason and exits 0 when Docker is unavailable, so the suite is usable on machines without containers. Use NewTestDBWithConfig when a component needs a full storage.Config, for example to boot the runtime in an integration test.

River insertion and processing

QueueInsert is the web process's insert-only client; QueueProcessor owns the worker lifecycle in cmd/queue. Both take the shared connection.

1inserter, err := storage.NewQueueInsert(db, cfg.QueueInsert.Config)
2
3processor, err := storage.NewQueueProcessor(
4    db,
5    cfg.QueueWorker.Config,
6    storage.WithRiverWorkers(queue.NewWorkers(identity, tokens, transactional, marketing)),
7    storage.WithRiverPeriodicJobs(queue.PeriodicJobs()...),
8    storage.WithRiverLogger(tel.Logger()),
9)

QueueInsert.InTx(tx) returns a queue whose jobs commit with the transaction. DefaultQueueConfig() carries operational defaults but no processor queues; the generated worker configuration adds the default queue and the insertion configuration clears queues. QueueConfig.Clone() copies maps and slices before mutation, and Validate covers concurrency, queue names, schemas, retention, polling, timeouts and inconsistent rescue thresholds.

The framework email workers come from storage.AddEmailWorkers, together with storage.SendTransactionalEmailArgs, storage.SendMarketingEmailArgs and their worker constructors. Use them instead of writing another sender loop. See Queues and Email Package.

Failure behavior

Constructor failures are wrapped with a storage: prefix. Health failures close the partially created pool. AtomicOps propagates callback and commit errors and rolls back every non-committed path. Keep environment parsing in config/ so package construction always receives validated typed policy.