Database (pgx)

pkg/db/drivers/pgx manages a PostgreSQL connection pool using jackc/pgx.

Setup

lakta.NewRuntime(
    config.NewModule(
        config.WithConfigDirs(".", "./config"),
        config.WithArgs(os.Args[1:]),
    ),
    tint.NewModule(),
    slog.NewModule(),
    pgx.NewModule(),
    myapp.NewModule(),
)

Consuming the connection

func (m *MyModule) Init(ctx context.Context) error {
    pool := do.MustInvoke[*pgxpool.Pool](lakta.GetInjector(ctx))
    m.pool = pool
    return nil
}

Multi-instance

pgx.NewModule(pgx.WithName("primary")),
pgx.NewModule(pgx.WithName("replica")),

modules:
  db:
    pgx:
      primary:
        dsn: "postgres://user:pass@primary:5432/mydb"
      replica:
        dsn: "postgres://user:pass@replica:5432/mydb"

Migrations

Schema migrations are driven by pressly/goose v3. Embed the .sql files and pass the filesystem with WithMigrations:

//go:embed migrations/*.sql
var migrationsFS embed.FS

pgx.NewModule(pgx.WithMigrations(migrationsFS)),

By default migrations do not run at startup. Enable the dev-time on-start path with config or pgx.WithMigrationsRunOnStart(true):

modules:
  db:
    pgx:
      default:
        dsn: "postgres://user:pass@localhost:5432/mydb"
        migrations:
          run_on_start: false        # apply pending migrations during StartAsync
          table: "schema_migrations" # goose history table
          dir: "migrations"          # sub-path within the embedded FS
          lock: "advisory"           # "advisory" (replica-safe) or "none"
          allow_missing: false       # apply out-of-order migrations

For production, apply migrations out-of-band (e.g. an init container or a migrate subcommand) via RunMigrations, which opens its own pool and runs goose Up:

if err := pgx.RunMigrations(ctx, &cfg, migrationsFS); err != nil {
    return err
}

The advisory lock lets concurrent on-start runs across replicas serialize on a Postgres session advisory lock so each migration applies exactly once. Migrations containing non-transactional DDL (e.g. CREATE INDEX CONCURRENTLY) should use RunMigrations rather than run_on_start.

Transactions

Q(ctx) returns a Querier — the active transaction when one is live in ctx, otherwise the pool. Repos call it without knowing whether they run inside a transaction:

func (r *Repo) Insert(ctx context.Context, name string) error {
    _, err := pgx.Q(ctx).Exec(ctx, "INSERT INTO users (name) VALUES ($1)", name)
    return err
}

WithTx runs a function inside a transaction, committing on success and rolling back on error (or panic):

err := pgx.WithTx(ctx, func(ctx context.Context) error {
    if err := repo.Insert(ctx, "alice"); err != nil {
        return err // rolls back
    }
    return repo.Insert(ctx, "bob")
}, pgx.WithIsolation(pgx.Serializable), pgx.WithReadOnly())

Nesting WithTx inside a live transaction runs the inner scope in a SAVEPOINT; WithNewTx forces a fresh top-level transaction on the pool instead. QNamed(ctx, name) resolves a named instance.

Each transaction carries a done flag: once it commits or rolls back, Q(ctx) falls back to the pool rather than handing out the finished transaction. This guards against a goroutine detached inside WithTx (which copies the context values under context.WithoutCancel) accidentally using a completed transaction — a spawned goroutine correctly gets the pool, never the dead transaction.

Configuration Reference

Config path: modules.db.pgx.<name>

dsnrequiredstring

DSN is the database connection string used to configure the database connection

envLAKTA_MODULES__DB__PGX__<NAME>__DSN

max_open_connsint32default: 10

maxOpenConns specifies the maximum number of open connections to the database. It maps to the "max_open_conns" configuration

envLAKTA_MODULES__DB__PGX__<NAME>__MAX_OPEN_CONNS

log_levelenumstringdefault: info

logLevel specifies the logging level for database operations, supporting values like trace, debug, info, warn, error, none

envLAKTA_MODULES__DB__PGX__<NAME>__LOG_LEVEL

health_checkbool

healthCheck enables or disables the database health check mechanism

envLAKTA_MODULES__DB__PGX__<NAME>__HEALTH_CHECK

min_connsint32

minConns is the minimum number of idle connections kept in the pool

envLAKTA_MODULES__DB__PGX__<NAME>__MIN_CONNS

max_conn_lifetimetime.Durationdefault: 1h0m0s

maxConnLifetime is the maximum age of a connection before it is closed

envLAKTA_MODULES__DB__PGX__<NAME>__MAX_CONN_LIFETIME

max_conn_idle_timetime.Durationdefault: 30m0s

maxConnIdleTime is the maximum idle time before a connection is closed

envLAKTA_MODULES__DB__PGX__<NAME>__MAX_CONN_IDLE_TIME

health_check_periodtime.Durationdefault: 1m0s

healthCheckPeriod is how often the pool checks idle connection health

envLAKTA_MODULES__DB__PGX__<NAME>__HEALTH_CHECK_PERIOD

statement_timeouttime.Durationdefault: 30s

statementTimeout sets the per-statement timeout (Postgres statement_timeout). Zero disables it

envLAKTA_MODULES__DB__PGX__<NAME>__STATEMENT_TIMEOUT

migrationsmigrations configures goose-driven schema migrations for this instance

migrations.run_on_startbool

runOnStart applies pending migrations during StartAsync. Default false —

envLAKTA_MODULES__DB__PGX__<NAME>__MIGRATIONS__RUN_ON_START

migrations.tablestringdefault: schema_migrations

table is the migration history table name

envLAKTA_MODULES__DB__PGX__<NAME>__MIGRATIONS__TABLE

migrations.dirstringdefault: migrations

dir is the sub-path within the embedded FS that holds the .sql files

envLAKTA_MODULES__DB__PGX__<NAME>__MIGRATIONS__DIR

migrations.lockenumstringdefault: advisory

lock selects the on-start locking strategy: "advisory" uses a Postgres

envLAKTA_MODULES__DB__PGX__<NAME>__MIGRATIONS__LOCK

migrations.allow_missingbool

allowMissing applies out-of-order (missing) migrations instead of erroring

envLAKTA_MODULES__DB__PGX__<NAME>__MIGRATIONS__ALLOW_MISSING