Database

apistock.dev/modules/postgres connects apps to PostgreSQL. Decisions: ADR-0005 (PostgreSQL, goose), ADR-0028 (Docker), ADR-0032 (hand-written SQL).

Connecting#

pool, err := postgres.Open(ctx, cfg.DatabaseURL, // config.Secret from DATABASE_URL
	postgres.WithMaxConns(10),
	postgres.WithApplicationName("acme-api"),
	postgres.WithTracerProvider(tel.TracerProvider()),
)
if err != nil {
	return err
}
cleanup.Add("postgres", func(context.Context) error { pool.Close(); return nil })
checker.Add(postgres.HealthCheck(pool)) // /readyz

Open pings the database, so a wrong URL fails at startup. Errors never include the URL, which may contain a password.

Repositories#

Repositories use hand-written SQL, one file per operation, like this:

internal/modules/users/repository/
├── store.go          UserStore{db postgres.DBTX}; NewUserStore(db)
├── errors.go         constraint names → domain errors
├── scan.go           shared row mapping (when several reads return the same columns)
├── insert_user.go    const insertUserSQL + (*UserStore).InsertUser
├── update_user.go
├── delete_user.go
├── select_user.go    SelectUserByID, SelectUsers
└── *_test.go         every method tested against a real database

store.go#

// UserStore persists users. It works on the pool or inside a transaction.
type UserStore struct {
	db postgres.DBTX
}

func NewUserStore(db postgres.DBTX) *UserStore { return &UserStore{db: db} }

postgres.DBTX is implemented by *pgxpool.Pool, *pgxpool.Conn, *pgx.Conn and pgx.Tx.

insert_user.go#

const insertUserSQL = `
	INSERT INTO users (id, name, email, password_hash, role, status)
	VALUES ($1, $2, $3, $4, $5, $6)
	RETURNING created_at, updated_at`

// InsertUser inserts a new user row.
func (s *UserStore) InsertUser(ctx context.Context, u *domain.User) error {
	err := s.db.QueryRow(ctx, insertUserSQL,
		u.ID, u.Name, u.Email, u.PasswordHash, u.Role, u.Status,
	).Scan(&u.CreatedAt, &u.UpdatedAt)
	if constraint, ok := postgres.UniqueViolation(err); ok && constraint == usersEmailKey {
		return domain.ErrDuplicateEmail
	}
	if err != nil {
		return fmt.Errorf("insert user: %v", err)
	}
	return nil
}

select_user.go#

const selectUserByIDSQL = `
	SELECT id, name, email, role, status, created_at, updated_at
	FROM users
	WHERE id = $1 AND deleted_at IS NULL`

func (s *UserStore) SelectUserByID(ctx context.Context, id string) (*domain.User, error) {
	var u domain.User
	err := s.db.QueryRow(ctx, selectUserByIDSQL, id).
		Scan(&u.ID, &u.Name, &u.Email, &u.Role, &u.Status, &u.CreatedAt, &u.UpdatedAt)
	if postgres.IsNoRows(err) {
		return nil, domain.ErrUserNotFound
	}
	if err != nil {
		return nil, fmt.Errorf("select user: %v", err)
	}
	return &u, nil
}

For wide or shared rows, scan into an unexported row struct with pgx.CollectRows(rows, pgx.RowToStructByName[userRow]) in scan.go, so columns are matched by name. Domain structs never get db tags.

delete_user.go#

const deleteUserSQL = `
	UPDATE users SET deleted_at = now(), updated_at = now()
	WHERE id = $1 AND deleted_at IS NULL`

func (s *UserStore) DeleteUser(ctx context.Context, id string) error {
	tag, err := s.db.Exec(ctx, deleteUserSQL, id)
	if err != nil {
		return fmt.Errorf("delete user: %v", err)
	}
	if tag.RowsAffected() == 0 {
		return domain.ErrUserNotFound
	}
	return nil
}

Rules#

• SQL lives in an unexported const next to the method that runs it; values are always placeholders, never concatenated.
• Translate expected driver conditions into domain errors; wrap everything else with %v so pgx errors never become part of the module's API (ADR-0018).
• A dynamic ORDER BY uses an allowlist of column names.

Error helpers#

Transactions#

err := postgres.InTx(ctx, pool, func(tx pgx.Tx) error {
	users := repository.NewUserStore(tx)
	if err := users.InsertUser(ctx, u); err != nil {
		return err // rolled back; the domain error is returned unchanged
	}
	_, err := jobsClient.InsertTx(ctx, tx, welcome.Args{UserID: u.ID}, nil) // job commits with the user
	return err
})

• InTx commits when the function returns nil and rolls back on an error or a panic (the panic continues).
• Build tx-bound stores inside the function; never keep tx after it returns or put it in a context (ADR-0030).
• InTxWithOptions(ctx, pool, pgx.TxOptions{IsoLevel: pgx.Serializable}, fn); retry the whole function when IsRetryable(err).
• In generated apps, use cases reach transactions through a TxManager port and never import pgx (ADR-0022).

Migrations#

• Goose SQL files in db/migrations, named <timestamp>_<description>.sql, with -- +goose Up sections.
• One ordered history for the app's tables and apistock module tables (modules ship theirs as settings.Migrations, jobs.Migrations, auditpg.Migrations, and they are copied in).
• Released migrations are never edited; changes are new files, forward-only.
• aps gen migration <name> creates an empty one that runs after the existing ones (CLI guide). Write its SQL before running cmd/migrate: an empty migration is recorded as applied.
• cmd/migrate applies them with postgres.Migrate(ctx, pool, migrations.FS), then River's migrations with jobs.Migrate. Apps never migrate at startup (ADR-0017).
• postgres.Migrate takes a PostgreSQL advisory lock, so concurrent migrators apply each migration once.
• postgres.Migrations(ctx, pool, fsys) reports Current, Latest and Pending without changing anything.

Running PostgreSQL locally#

Ports bind to 127.0.0.1 only, so the development password is never reachable from the network.

Commands#

Each needs the environment: aps dev provides it, or set -a; . ./.env; set +a.

Rollbacks. There are no down migrations (ADR-0005). To undo a released migration, write a new one that reverses it. Locally, before a migration is shared, edit it and reset with docker compose down -v. A database's schema and data are restored from backups, not by running migrations backwards.

Schema of a Full app#

The tables a new single-tenant Full app creates, by migration. Library modules own their tables' shape; the app owns its modules' tables. Every ID is text with a type prefix (usr_…), and every timestamp is timestamptz in UTC.

A multi-tenant app also has orgs, org_members and org_invitations (…orgs.sql), and its projects belong to an organisation (org_id, created_by) rather than a user.

Relationships#

auth_users ─┬─< auth_sessions            ON DELETE CASCADE
            ├─< auth_codes               ON DELETE CASCADE
            ├─< auth_user_roles          ON DELETE CASCADE
            ├── auth_totp (1:1)          ON DELETE CASCADE
            ├─< auth_recovery_codes      ON DELETE CASCADE
            ├─< auth_mfa_challenges ─< auth_webauthn_ceremonies
            ├─< auth_passkeys            ON DELETE CASCADE
            ├─< auth_identities          ON DELETE CASCADE, UNIQUE (provider, subject)
            └─< projects (owner_id)      ON DELETE CASCADE

multi-tenant:
orgs ─┬─< org_members >── auth_users     PRIMARY KEY (org_id, user_id)
      ├─< org_invitations
      └─< projects (org_id)              ON DELETE CASCADE, UNIQUE (org_id, id)

Settings, job definitions, audit events and release instances have no foreign keys to users: audit history outlives the accounts it mentions. Deleting a user is soft first (deleted_at); the auth_cleanup job purges accounts after auth.deleted_account_retention, and the cascades remove their rows.

Indexes worth knowing#

Transactions and concurrency#

• Use cases that write more than one row do it in one transaction through their TxManager port (postgres.InTx underneath): a row and its audit event (auditpg.RecordTx), a user and their verification code, a row and a job (jobs.Client.InsertTx).
• Updates to resources, settings and job definitions carry a version; the UPDATE … WHERE version = $n fails with a conflict when someone changed the row first, so concurrent edits never overwrite each other silently.
• Reads that must not change before a write lock the row (SELECT … FOR UPDATE) inside the transaction.
• Single-use values (codes, challenges, nonces, OAuth states) are consumed with a conditional UPDATE … WHERE consumed_at IS NULL, so two concurrent uses can't both succeed, on any number of instances.
• Migrations take an advisory lock; River elects a single leader for periodic jobs.

Testing with pgtest#

func TestInsertUser(t *testing.T) {
	pool := pgtest.New(t, pgtest.WithMigrations(migrations.FS))
	store := repository.NewUserStore(pool)
	// ...
}

Options: WithMigrations(fsys) (applied once into a template per distinct set of files), WithMaxConns(n) (default 4). Databases are dropped when the test ends. See local development for the environment variables.