cf_postgres

README

caerus-framework-postgresql

Caerus Framework PostgreSQL Component. Wraps a pgx connection pool (pgxpool), verifies connectivity at Init (fail-fast), and closes it at Shutdown. Registers in the data initialization stage.

Not an ORM — the ops chassis for Postgres

This module owns the connection & ops chassis, not your queries. Write SQL with sqlc or pgx against Pool() (and WithinTx for transactions); this component owns everything around it:

• pool lifecycle: Init fail-fast ping, Shutdown, Health (/readyz)
• live reload / reconnect from a configuration source (WithConfigSource)
• migrate Job + CLI (WithMigrations / Migrate / framework job flag)
• day-2 ops: pool metrics, WithQueryTracer hooks, TLS file rotation, timeouts

Non-goals: ActiveRecord, auto-migrate from structs, Repository[T], hiding Pool().

Service checklist

For every new service using this component:

• Health wired — observability /readyz auto-discovers HealthProvider
• Schema via WithMigrations + --postgresql.job=migrate (or WithEmbeddedMigrations for //go:embed FS) as a K8s Job before the Deployment; serving pods omit WithMigrateOnInit
• WithConfigSource("…") for file-based config (External Secrets rotation)
• WithName + GetByName when a process needs primary + replica
• Confirm postgresql_pool_* on /metrics (pool saturation visible)

Wiring

Two wiring shapes. Prefer the app-owned shape.

App-owned consumer (golden — demoapp pattern)

main declares postgres as chassis. The app holds *CFPostgres and calls Pool() per use (never copy the pool at Init — reload/reconnect swap it).

fw := cf.New(&cf.FrameworkOptions{
	Logs:          &cf.LogsSettings{Format: "json", Level: "info", ConfigSource: "logs"},
	Observability: &cf.ObservabilitySettings{Bind: ":9090", ConfigSource: "observability"},
	Components: []cf.CaerusComponent{
		cf_postgres.New(cf_postgres.WithConfigSource("postgresql", "config/postgresql.json")),
		app.New(),
	},
})

type App struct {
	pg *cf_postgres.CFPostgres
}

func (a *App) GetDependencies() []string {
	return []string{cf_postgres.ComponentName}
}

func (a *App) Init(ctx context.Context, fw *cf.CaerusFramework) error {
	pg, ok := cf.Get[*cf_postgres.CFPostgres](fw)
	if !ok {
		return errors.New("app: postgresql missing")
	}
	a.pg = pg
	return nil
}

func (a *App) note(ctx context.Context, id int) (string, error) {
	var note string
	err := a.pg.Pool().QueryRow(ctx, "SELECT note FROM notes WHERE id = $1", id).Scan(&note)
	return note, err
}

A store or sqlc helper is the same idea: hold *CFPostgres, never a *pgxpool.Pool from Init.

type Store struct {
	pg *cf_postgres.CFPostgres
}

func New(pg *cf_postgres.CFPostgres) *Store { return &Store{pg: pg} }

func (s *Store) queries() *db.Queries {
	return db.New(s.pg.Pool()) // sqlc New per call — cheap; pool is live
}

Wrong: store.New(pg.Pool()) at Init, or q := db.New(pg.Pool()) kept on the
       struct. After reload/reconnect the component closes that pool.
Right: store.New(pg) with *CFPostgres; Pool() / db.New(pg.Pool()) per query.

Simple main-level wiring

fw := cf.New()
fw.AddComponent(cf_logs.New(cf_logs.WithWriter(os.Stdout)))
fw.AddComponent(cf_postgres.New(
	cf_postgres.WithHost("127.0.0.1"),
	cf_postgres.WithDatabase("mydb"),
))

Then cf.MustGet[*cf_postgres.CFPostgres](fw). Still Pool() per use.

With degraded_mode, Init can succeed while Postgres is down. A background loop retries ping/rebuild until the server is up. Health stays not-ready unless health_when_degraded is ready.

Usage

After fw.Run (or in any component whose stage runs after data), call Pool() on the component for each query. Do not keep the *pgxpool.Pool value from Init.

pg := cf.MustGet[*cf_postgres.CFPostgres](fw)

var note string
err := pg.Pool().QueryRow(ctx, "SELECT note FROM notes WHERE id = $1", 42).Scan(&note)

The pool provides connection reuse, health-checked idle connections, and safe concurrency.

Transactions

Use WithinTx for multi-statement atomic operations — commit on nil, rollback on error or a canceled context:

pg := cf.MustGet[*cf_postgres.CFPostgres](fw)
err := pg.WithinTx(ctx, func(tx pgx.Tx) error {
    if _, err := tx.Exec(ctx, "INSERT INTO a (id) VALUES ($1)", 1); err != nil {
        return err
    }
    _, err := tx.Exec(ctx, "INSERT INTO b (id) VALUES ($1)", 1)
    return err
})

For more control (pgx.TxOptions, savepoints) use pool.Begin(ctx) and pgx.Tx directly — WithinTx is the common case, not a transaction API.

Options

Option	Description
WithConfig(PostgresConfig)	connection config loaded from the configuration component; non-zero fields override option-set defaults
WithPoolConfig(*pgxpool.Config)	full pgxpool.Config (deep-copied); call before convenience setters you want overridden
WithConnString(dsn)	full connection string (DSN or URL), parsed via pgxpool.ParseConfig; bad DSN fails at Init
WithHost(h)	server host (default 127.0.0.1)
WithPort(p)	server port (default 5432)
WithUser(u)	role to connect as (default: current OS user)
WithPassword(p)	authentication password
WithDatabase(d)	database name (default: the user name)
WithSSLMode(m)	disable, prefer, require, verify-ca, verify-full (default prefer); unknown mode fails at Init
WithApplicationName(n)	sets the application_name runtime parameter
WithStatementTimeout(d)	per-connection statement_timeout runtime parameter (ms); 0 = unset
WithLockTimeout(d)	per-connection lock_timeout runtime parameter (ms); 0 = unset
WithTLSRootCAFile(path)	PEM CA bundle to verify the server (mTLS/private CA); re-read on every connect/reload
WithTLSClientCertFile(cert, key)	PEM client cert + key for mTLS; re-read on every connect/reload
WithMaxConns(n)	pool max connections (default max(4, GOMAXPROCS))
WithMinConns(n)	pool min connections (default 0)
WithMaxConnLifetime(d)	max connection age (default 1h)
WithMaxConnIdleTime(d)	max idle time (default 30m)
WithHealthCheckPeriod(d)	idle health-check interval (default 1m)
WithConnectTimeout(d)	per-attempt connect timeout (default 0 = none, libpq default)
WithPingTimeout(d)	Init connectivity-ping timeout (default 5s)
WithDegradedMode(bool)	when true, Init may succeed without a live ping/pool (default off / hard-fail)
WithHealthWhenDegraded("not_ready"|"ready")	/readyz while degraded: default not_ready; ready is break-glass LB traffic
WithMigrations(fsys, opts...)	configure a migration FS (already rooted at the .up.sql/.down.sql dir) for Migrate / the framework job flag
WithEmbeddedMigrations(fsys, dir, opts...)	like WithMigrations but takes the //go:embed FS + directory and resolves the sub-FS internally (mismatched dir panics at construction)
WithMigrateOnInit()	Init calls Migrate (local/single-replica only)
WithMigrationsTable(name)	migrations tracking table (default schema_migrations)
WithName(name)	custom component name for multiple instances (default "postgresql")
WithQueryTracer(pgx.QueryTracer)	pgx query tracer around every Query/QueryRow/Exec; survives pool rebuilds on config reload
WithLogger(*slog.Logger)	explicit logger override; defaults to the framework logs component's logger (re-delivered on logs Reconfigure), falling back to slog.Default()

Query tracing

WithQueryTracer exposes pgx's own hook seam (TraceQueryStart / TraceQueryEnd, invoked around every Query, QueryRow, and Exec), so you can attach tracing or slow-query logging without importing an instrumentation library into this module — the tracer interface lives in pgx. An OpenTelemetry example (the otel import lives in your app, not here):

import (
    "context"

    "github.com/jackc/pgx/v5"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/trace"
)

type ctxKey struct{}

type spanTracer struct{ tracer string }

func (t spanTracer) TraceQueryStart(ctx context.Context, conn *pgx.Conn,
    data pgx.TraceQueryStartData) context.Context {
    ctx, span := otel.Tracer(t.tracer).Start(ctx, "postgres:"+data.SQL)
    return context.WithValue(ctx, ctxKey{}, span)
}

func (t spanTracer) TraceQueryEnd(ctx context.Context, conn *pgx.Conn, data pgx.TraceQueryEndData) {
    span, ok := ctx.Value(ctxKey{}).(trace.Span)
    if !ok {
        return
    }
    if data.Err != nil {
        span.RecordError(data.Err)
    }
    span.End()
}

p := cf_postgres.New(cf_postgres.WithConnString("postgres://..."),
    cf_postgres.WithQueryTracer(spanTracer{tracer: "postgres"}))

The tracer is part of the option base config, so a WithConfigSource reload rebuilds the pool with the same tracer. Pgx calls a single tracer per query; chain several with a small composite type. BatchTracer / ConnectTracer / pool acquire tracing are secondary and not exposed — use WithPoolConfig if you need them.

Multiple instances

A read replica is a second Postgres that replays the primary’s WAL. It is not a second schema you migrate, and it is not “Postgres but faster.” Use it when some queries can be seconds behind (exports, dashboards). Auth-style “I just saved, show me the row” stays on the primary.

Caerus already decided how that looks in process: two components, WithName, two config files, GetByName, Pool() per use. Do not put replica_dsn on one PostgresConfig or make Pool() pick a server. Health, metrics, migrate jobs, and TLS reload are per Name(). A silent split is how on-call cannot tell which DSN died.

GetDependencies lists those component names (what Name() returns), not a source nickname. When two instances exist, cf.Get[*cf_postgres.CFPostgres] is ambiguous and returns false — use GetByName.

pri := cf_postgres.New(
    cf_postgres.WithName("primary"),
    cf_postgres.WithConfigSource("postgresql", "config/postgresql.json"),
)
rep := cf_postgres.New(
    cf_postgres.WithName("replica"),
    cf_postgres.WithConfigSource("postgresql-replica", "config/postgresql-replica.json"),
)
fw.AddComponent(pri)
fw.AddComponent(rep)

primary := cf.MustGetByName[*cf_postgres.CFPostgres](fw, "primary")
replica := cf.MustGetByName[*cf_postgres.CFPostgres](fw, "replica")
err := replica.Pool().QueryRow(ctx, "SELECT 1").Scan(&n)

Writes, transactions, and read-your-writes: primary.Pool(). Stale-OK reads: replica.Pool(). Only primary runs --postgresql.job=migrate (or --postgresql.primary.job=migrate when WithName("primary")). The replica is a streaming copy.

flowchart LR
  W[Writes and tx] --> P[primary Pool]
  R[Stale-OK reads] --> S[replica Pool]
  P --> DB[(Primary)]
  S --> RP[(Streaming replica)]
  DB -->|WAL| RP

Readiness: two process shapes

/readyz is red if any HealthProvider fails. A replica in the API process that cannot ping takes checkout out of the Service even when primary is fine.

Path A — Replica in the API pod (optional reads on the user request): replica degraded_mode + health_when_degraded: ready so a dead replica does not drain traffic; metrics must scream; checkout must not require the replica.

Path B — Replica only on reporting/worker pods (usual when those reads are not checkout): API registers primary only. Reporting pods register the replica. Each process’s /readyz matches that process’s job.

Mix-ups that show up in postmortems

These are identity mistakes, not “Postgres is slow.”

1. Magic Pool() — one component, two DSNs, driver or wrapper routes SELECTs. Logs and /readyz name a single postgresql. You cannot tell which host failed.
2. Read-your-writes on the replica — user saves, next GET hits the replica before WAL apply. Looks like data loss. CI often has zero lag, so tests stay green. Keep “I just wrote this” on primary.
3. /readyz AND’s the replica — replica blip pages as “the app is down.” Use Path A (degraded-ready replica) or Path B (replica not in the API graph).
4. Writes or migrate on the replica — cannot execute … in a read-only transaction, or --postgresql.replica.job=migrate treating the standby as a second schema. Migrate primary only. sqlc writes use primary.Pool().
5. DSN / endpoint swap — copy-paste leaves both files on the writer, or after failover the “reader” CNAME is now the writer. The component is still named replica but it is the primary (or both names hit one host). Check host on postgresql_info / connect logs per component label; do not trust the filename alone.

Wrong: replica_dsn on one struct; store.New(pg.Pool()) snapshot; checkout GetDependencies includes replica with default not-ready Health.
Right: two named *CFPostgres; Pool() per use; readyz matches the Deployment’s job.

Migrations

Files use the golang-migrate layout (<version>_<name>.up.sql / .down.sql), may contain multiple statements, and run under an advisory lock against the same pool TLS/credentials as the app. "Already up to date" is success.

Production policy (Job flag, not Init)

Environment	Pattern
K8s / multi-replica	Same image; Job args --postgresql.job=migrate (the flag is declared by this module on its own configuration source — see Same binary: --postgresql.job=migrate). Serving Deployment omits WithMigrateOnInit (keep WithMigrations so the Job can migrate).
Local / single-replica	WithMigrations + WithMigrateOnInit() so Init migrates after ping.

Do not use WithMigrateOnInit on every replica of a Deployment. See Dirty migrations when a Job crashes mid-apply.

Dirty migrations

golang-migrate tracks (version, dirty) in the migrations table (default schema_migrations). While an up is in progress the row is dirty. If the process dies mid-version, further Up calls fail until an operator repairs state — Caerus does not auto-force.

Situation	What happens
Concurrent migrate Jobs / Init	Serialized by golang-migrate’s Postgres advisory lock on that database
Job re-run when already current	Success (ErrNoChange / “up to date”) — safe and expected
Crash mid-version	Dirty set → later migrate Jobs fail closed until repaired

Why no auto-force: dirty means “we do not know whether version N’s SQL finished.” Clearing the flag without inspecting the database can mark a half-applied schema as done. Silent heal is how replicas diverge.

Operator runbook (default):

1. Fail the release / stop serving until fixed (Job failed; Deployment should not have started).
2. Inspect DB + Job logs: did objects from version N exist?
3. Either finish the up manually, or undo partial objects (hand SQL / careful use of the matching .down.sql). golang-migrate does not auto-run downs on crash.
4. migrate force VERSION to the true clean version (clears dirty).
5. Re-run myapp --postgresql.job=migrate (or --postgresql.<name>.job=migrate).

Reducing how often this hurts: keep versions small; prefer expand/contract and idempotent ups; avoid irreversible DDL in the middle of a multi-statement version. Prod rollback of a bad up is usually restore/PITR or a forward fix, not a casual Down.

Not in this module: automatic dirty policies (force to N or N−1 and retry). If a product ever adds that, it must be an explicit opt-in with a documented assumption that ups are safe to retry — never the default Migrate path.

Same binary: --postgresql.job=migrate

Wire postgres once with WithMigrations. The module declares the job flag on its own configuration source (Source.Job: cf.JobSpec{Flag: "postgresql.job", Tasks: ["migrate"]}), so configuration parses/validates the value like any other setting. The flag names the instance and the value names the task (--postgresql.job=migrate, or --postgresql.orders.job=migrate for a WithName("orders") instance). Jobs are CLI-only — the task never flows from env or file. RunWithSignals asks configuration (via cf.JobSource) and runs only the core plus the named postgres component's RunJob (migrate → Migrate), then exits — no serve ceremony:

import "embed"

//go:embed migrations
var migrations embed.FS

func main() {
	ctx := context.Background()

	fw := cf.New(&cf.FrameworkOptions{
		Components: []cf.CaerusComponent{
			cf_postgres.New(
				cf_postgres.WithConfigSource("postgresql", "config.yaml"),
				cf_postgres.WithEmbeddedMigrations(migrations, "migrations"),
				// cf_postgres.WithMigrateOnInit(), // local only
			),
			// … logs/configuration/observability are auto-registered core …
			app, // Runnable
		},
	})

	if err := fw.RunWithSignals(ctx,
		cf.WithShutdownTimeout(15*time.Second),
	); err != nil {
		log.Fatal(err)
	}
}

Run it as myapp --postgresql.job=migrate (K8s Job), or call fw.Migrate(ctx, "postgresql") directly from a migrate subcommand in a multi-tool binary.

Invocation	Behavior
myapp --postgresql.job=migrate	Initialize (core + named postgres) → RunJob("migrate") → Shutdown → exit
myapp --postgresql.orders.job=migrate	Same, on the WithName("orders") instance
myapp (serve)	Normal RunWithSignals; no migrate unless WithMigrateOnInit

Migrate(ctx) (explicit API)

if err := postgres.Migrate(ctx); err != nil {
	log.Fatal(err)
}

Requires Init (live pool) and a migrations FS (WithMigrations or WithEmbeddedMigrations). Prefer the framework job flag (--postgresql.job=migrate) or fw.Migrate(ctx, target) for Jobs.

Helm / GitOps Job before Deployment

• batch/v1 Job (or Helm pre-install/pre-upgrade hook, or Argo sync-wave -1).
• Same container image; args: ["--postgresql.job=migrate"] (or equivalent).
• Same DB secret/config as the API.
• restartPolicy: Never; fail the release if the Job fails.
• Deployment pods start only after the Job succeeds; readiness is /readyz.

Migration files under fsys:

migrations/
  000001_init.up.sql
  000001_init.down.sql

Shared databases: give each service its own tracking table via WithMigrationsTable (e.g. orders_migrations). Default is schema_migrations.

Configuration

Drive connection settings via caerus-framework-configuration (file → env → DSN). PostgresConfig has json/yaml/env tags. Durations are in seconds.

The module is self-sufficient: WithConfigSource(name, path) registers its own Source[PostgresConfig] with the configuration component (via cf.ConfigSourceRegistrar, run by the framework during argv absorption). The default EnvPrefix is the uppercase source name ("postgresql" → "POSTGRESQL_"); override with WithSourceEnvPrefix("POSTGRES_"). POSTGRES_DSN is overlaid in AfterLoad inside the module. main only points the instance at where config lives:

postgres := cf_postgres.New(
	cf_postgres.WithConfigSource("postgresql", "config.yaml"), // Init + OnConfigReload reconnect
)

For low-level control (custom AfterLoad, format, env prefix), register the source manually instead:

conf := cf_configuration.New()
_ = fw.AddComponent(conf)
_ = cf_configuration.AddSource(conf, cf_configuration.Source[cf_postgres.PostgresConfig]{
	Name:      "postgresql",
	Path:      "config.yaml", // optional if EnvPrefix set
	Format:    cf_configuration.FormatYAML,
	Owner:     cf_postgres.ComponentName,
	EnvPrefix: "POSTGRES_",
	AfterLoad: func(c *cf_postgres.PostgresConfig) error {
		if dsn := os.Getenv("POSTGRES_DSN"); dsn != "" {
			return cf_postgres.OverlayDSN(c, dsn) // wins over file+env fields
		}
		return nil
	},
})

postgres := cf_postgres.New(
	cf_postgres.WithConfigSource("postgresql", ""), // bind by name only
)

Helpers: ParseDSN / OverlayDSN for postgres:// URLs and keyword DSNs. ParseDSN errors never include the raw DSN (pgx would interpolate the password). Password is tagged secret:"redact": LogArgs prints [redacted], connect/reload logs use password_set only. Do not log the config struct. WithConfigSource implements ConfigReloader: on file reload (or cfg.Reload), builds a new pool, pings, swaps, closes the old pool; on failure keeps the previous pool. In Kubernetes prefer file-mounted secrets for rotation; use env/DSN for local and CI.

Fail-fast behaviour (default)

Init creates the pool, pings the server, and applies pending migrations. If the connection is refused, the ping times out, or a migration fails, Init returns an error and startup aborts before any dependent component runs. Pool() returns nil before Init or after Shutdown.

DegradedMode (optional break-glass)

Not automatic. Default remains hard Init. Set degraded_mode: true (or WithDegradedMode(true)) when the process must finish Initialize even if Postgres is unreachable. Migrate-on-Init and migrate Jobs still need a live pool — DegradedMode is for serve/break-glass shapes, not for skipping schema work.

{
  "host": "postgres",
  "port": 5432,
  "user": "app",
  "password": "…",
  "database": "app",
  "degraded_mode": true,
  "health_when_degraded": "not_ready"
}

Setting	Meaning
degraded_mode	Init may succeed without a successful ping/create; logs/metrics scream (degraded_unreachable, degraded_mode_uses_total). Default off.
health_when_degraded: "not_ready"	Default — Health still fails → /readyz 503 while disconnected.
health_when_degraded: "ready"	Break-glass — Health returns nil while down so LB may send traffic. Use deliberately; prefer not lying about DB-backed routes on the same pod.

DegradedMode answers “may Initialize finish?” — it does not mean the database is healthy. Hot reload of the postgresql source can rebuild the pool when the file updates; env alone does not wake a running process.

Observability

CFPostgres implements cf.HealthProvider: Health(ctx) pings the pool, so the observability component's /readyz endpoint reflects real database connectivity. Before Init or after Shutdown (nil pool) it reports unhealthy. After DegradedMode without a live ping, behaviour follows health_when_degraded.

It also implements cf.MetricsProvider: while connected it contributes postgresql_info plus pool gauges and counters to /metrics (all labeled database, user, host, port, component = Name()):

Metric	Type	Meaning
postgresql_degraded_unreachable	gauge	1 when running without a successful ping (DegradedMode / lost connectivity)
postgresql_degraded_mode_uses_total	counter	times Init continued after failed ping/create under DegradedMode
postgresql_pool_idle	gauge	idle connections
postgresql_pool_total	gauge	total connections
postgresql_pool_max	gauge	pool maximum
postgresql_pool_acquired	gauge	currently acquired
postgresql_pool_constructing	gauge	dials still in flight
postgresql_pool_acquire_total	counter	cumulative successful acquires
postgresql_pool_empty_acquire_total	counter	acquires that had to wait
postgresql_pool_canceled_acquire_total	counter	acquires canceled by context
postgresql_pool_acquire_duration_seconds	counter	cumulative time in successful acquires
postgresql_pool_empty_acquire_wait_seconds	counter	cumulative wait while the pool was empty
postgresql_pool_new_conns_total	counter	new connections opened
postgresql_pool_max_lifetime_destroy_total	counter	closed for MaxConnLifetime
postgresql_pool_max_idle_destroy_total	counter	closed for MaxConnIdleTime

Scrape postgresql_pool_acquired / max for pool saturation. When saturation looks fine but latency is not, use wait time:

rate(postgresql_pool_empty_acquire_wait_seconds[5m]) and rate(postgresql_pool_empty_acquire_wait_seconds[5m]) / rate(postgresql_pool_empty_acquire_total[5m]) (mean wait when the pool was empty). Churn: rate(postgresql_pool_new_conns_total[5m]) vs rate(postgresql_pool_max_lifetime_destroy_total[5m]) / rate(postgresql_pool_max_idle_destroy_total[5m]). Counters reset when the pool is rebuilt (reload / reconnect). Before Init or after Shutdown it reports nothing (lazy pickup).

Tests

Unit tests cover the component contract without a server. Integration tests are gated on POSTGRES_ADDR (or POSTGRES_DSN for full control):

POSTGRES_ADDR=127.0.0.1:5433 POSTGRES_PASSWORD=secret go test -race ./...

License

Apache License 2.0 — see LICENSE.

Documentation

Overview

Package cf_postgres provides the caerus-framework PostgreSQL component. It wraps a pgx/v5 connection pool, verifies connectivity with a fail-fast ping at Init, and exposes the pool to dependent components via Pool().

Index

• Constants
• func OverlayDSN(cfg *PostgresConfig, dsn string) error
• type CFPostgres
  • func New(opts ...Option) *CFPostgres
  • func (c *CFPostgres) GetDependencies() []string
  • func (c *CFPostgres) GetInitOrderStage() cf.Stage
  • func (c *CFPostgres) Health(ctx context.Context) error
  • func (c *CFPostgres) Init(ctx context.Context, fw *cf.CaerusFramework) error
  • func (c *CFPostgres) Metrics() []cf_observability.Metric
  • func (c *CFPostgres) Migrate(ctx context.Context) error
  • func (c *CFPostgres) Name() string
  • func (c *CFPostgres) OnConfigReload(source string, cfg any)
  • func (c *CFPostgres) Pool() *pgxpool.Pool
  • func (c *CFPostgres) RegisterConfigSources(conf any) error
  • func (c *CFPostgres) RunJob(ctx context.Context, task string) error
  • func (c *CFPostgres) Shutdown(ctx context.Context) error
  • func (c *CFPostgres) WithinTx(ctx context.Context, fn func(pgx.Tx) error) error
• type MigrationOption
  • func WithMigrationsTable(name string) MigrationOption
• type Option
  • func WithApplicationName(name string) Option
  • func WithConfig(cfg PostgresConfig) Option
  • func WithConfigSource(name, path string, opts ...SourceOption) Option
  • func WithConnString(dsn string) Option
  • func WithConnectTimeout(d time.Duration) Option
  • func WithDatabase(db string) Option
  • func WithDegradedMode(enabled bool) Option
  • func WithEmbeddedMigrations(fsys embed.FS, dir string, opts ...MigrationOption) Option
  • func WithHealthCheckPeriod(d time.Duration) Option
  • func WithHealthWhenDegraded(policy string) Option
  • func WithHost(host string) Option
  • func WithLockTimeout(d time.Duration) Option
  • func WithLogger(logger *slog.Logger) Option
  • func WithMaxConnIdleTime(d time.Duration) Option
  • func WithMaxConnLifetime(d time.Duration) Option
  • func WithMaxConns(n int32) Option
  • func WithMigrateOnInit() Option
  • func WithMigrations(fsys fs.FS, opts ...MigrationOption) Option
  • func WithMinConns(n int32) Option
  • func WithName(name string) Option
  • func WithPassword(password string) Option
  • func WithPingTimeout(d time.Duration) Option
  • func WithPoolConfig(cfg *pgxpool.Config) Option
  • func WithPort(port int) Option
  • func WithQueryTracer(t pgx.QueryTracer) Option
  • func WithSSLMode(mode string) Option
  • func WithStatementTimeout(d time.Duration) Option
  • func WithTLSClientCertFile(certPath, keyPath string) Option
  • func WithTLSRootCAFile(path string) Option
  • func WithUser(user string) Option
• type PostgresConfig
  • func ParseDSN(dsn string) (PostgresConfig, error)
• type SourceOption
  • func WithSourceEnvPrefix(prefix string) SourceOption
  • func WithSourceFormat(f cf_configuration.Format) SourceOption

Constants

const (
	// ComponentName is the framework component name for the postgresql
	// component. It is the identifier other components use in GetDependencies
	// to require postgres.
	ComponentName = "postgresql"

	// ComponentStage is the stage data-layer components initialize in. It is
	// not a built-in bootstrap stage; AddComponent registers it automatically
	// the first time a component declares it.
	ComponentStage = cf.Stage("data")
)

Functions

func OverlayDSN

func OverlayDSN(cfg *PostgresConfig, dsn string) error

OverlayDSN merges connection fields from dsn into cfg. DSN-derived fields win over existing values (file/env). Pool sizing from the DSN is applied only when the parsed pool sets a non-zero value.

Types

type CFPostgres

type CFPostgres struct {
	// contains filtered or unexported fields
}

CFPostgres is the caerus-framework-postgresql component. It wraps a pgx connection pool, verifies connectivity at Init, and closes it at Shutdown.

func New

func New(opts ...Option) *CFPostgres

New creates a postgresql component. The pool is created and pinged at Init, not here. Invalid user-facing options (bad DSN, unknown ssl_mode) are deferred to Init rather than panicking; only an unparseable built-in default connection string panics (internal invariant).

func (*CFPostgres) GetDependencies

func (c *CFPostgres) GetDependencies() []string

GetDependencies implements cf.Dependencies. The component logs through the framework logs component, and depends on configuration when WithConfigSource is set.

func (*CFPostgres) GetInitOrderStage

func (c *CFPostgres) GetInitOrderStage() cf.Stage

GetInitOrderStage implements cf.CaerusComponent.

func (*CFPostgres) Health

func (c *CFPostgres) Health(ctx context.Context) error

Health implements cf.HealthProvider. It pings the pool, so the observability component's readiness endpoint reflects real database connectivity. A nil pool (before Init or after Shutdown) is unhealthy. After DegradedMode with a failed ping (or nil pool from a failed create), behaviour follows health_when_degraded (default not_ready → still unhealthy for /readyz).

func (*CFPostgres) Init

func (c *CFPostgres) Init(ctx context.Context, fw *cf.CaerusFramework) error

Init implements cf.CaerusComponent. It creates the pgx connection pool and verifies connectivity with a ping. By default a broken database fails startup (fail-fast). With DegradedMode, a failed ping keeps the pool (or a failed create leaves Pool nil) and lets Initialize continue (metrics/logs scream; Health stays honest unless health_when_degraded=ready).

func (*CFPostgres) Metrics

func (c *CFPostgres) Metrics() []cf_observability.Metric

Metrics implements cf_observability.MetricsProvider. Before Init or after Shutdown it returns nil. After Init (including DegradedMode without a live ping) it always returns samples so degrade/unreachable state is visible.

Pool gauges come from pgxpool.Stat on every scrape when the pool is non-nil; *_total counters are cumulative for the pool's lifetime and reset only when the pool is rebuilt (config reload / reconnect). The "component" label carries Name() so named primary/replica instances are distinguishable on /metrics.

func (*CFPostgres) Migrate

func (c *CFPostgres) Migrate(ctx context.Context) error

Migrate applies pending up migrations using the live pool and the filesystem configured via WithMigrations. Requires a successful Init. "Already up to date" (migrate.ErrNoChange) is success. Concurrent callers are serialized by golang-migrate's advisory lock.

Production: prefer the framework job flag --postgresql.job=migrate from the same binary; omit WithMigrateOnInit on serving Deployments so Init does not migrate.

func (*CFPostgres) Name

func (c *CFPostgres) Name() string

Name implements cf.CaerusComponent. Name implements cf.CaerusComponent. Returns the custom name set via WithName, or the default ComponentName ("postgresql") if no custom name was set.

func (*CFPostgres) OnConfigReload

func (c *CFPostgres) OnConfigReload(source string, cfg any)

OnConfigReload implements cf.ConfigReloader. It rebuilds the pool from the bound configuration source (file → env → DSN already applied by configuration). The fresh value is delivered as cfg but the pool is rebuilt from the source so the translation stays in one place. On failure the previous pool is kept (last-good).

func (*CFPostgres) Pool

func (c *CFPostgres) Pool() *pgxpool.Pool

Pool returns the pgx connection pool. It is non-nil after a successful Init and nil before Init or after Shutdown.

func (*CFPostgres) RegisterConfigSources

func (c *CFPostgres) RegisterConfigSources(conf any) error

RegisterConfigSources implements cf.ConfigSourceRegistrar. The framework calls it during argv absorption; it registers this component's configuration source (name, path, env prefix, format, Owner and the POSTGRES_DSN AfterLoad overlay) with the configuration component. No-op when no source is bound.

func (*CFPostgres) RunJob

func (c *CFPostgres) RunJob(ctx context.Context, task string) error

RunJob implements cf.JobRunner. The only supported task is "migrate", which it dispatches to CFPostgres.Migrate; any other task is an error (the framework validates the task against the source's declared set before this runs, so this is the last line of defense).

func (*CFPostgres) Shutdown

func (c *CFPostgres) Shutdown(ctx context.Context) error

Shutdown implements cf.CaerusComponent. It closes the pgx pool; further use of Pool() after shutdown returns nil.

func (*CFPostgres) WithinTx

func (c *CFPostgres) WithinTx(ctx context.Context, fn func(pgx.Tx) error) error

WithinTx runs fn inside a single transaction, committing when it returns nil and rolling back otherwise (including on an error from fn or a canceled context). Use it for multi-statement operations that must be atomic — the framework keeps the pool lifecycle; this helper keeps the commit/rollback boilerplate out of every caller. Requires a successful Init.

type MigrationOption

type MigrationOption func(*migrationConfig)

MigrationOption configures how WithMigrations applies schema migrations.

func WithMigrationsTable

func WithMigrationsTable(name string) MigrationOption

WithMigrationsTable sets the table golang-migrate uses to track applied versions (default "schema_migrations"). Pick a distinct name when several services share one database.

type Option

type Option func(*options)

Option configures the postgresql component at construction time.

func WithApplicationName

func WithApplicationName(name string) Option

WithApplicationName sets the application_name runtime parameter, identifying this process's connections to the server (default: unset, so the server-side default applies).

func WithConfig

func WithConfig(cfg PostgresConfig) Option

WithConfig sets a static connection configuration snapshot. Non-zero fields of cfg override the values set by the convenience options. Prefer WithConfigSource when using caerus-framework-configuration with hot-reload.

cfg, err := cf_configuration.Lookup[cf_postgres.PostgresConfig](conf, "postgresql")
p := cf_postgres.New(cf_postgres.WithConfig(*cfg))

func WithConfigSource

func WithConfigSource(name, path string, opts ...SourceOption) Option

WithConfigSource binds this component to a named configuration source and registers that source with the configuration component (via the framework's ConfigSourceRegistrar pass during argv absorption). The module owns the Source: the config type, default EnvPrefix, the POSTGRES_DSN AfterLoad overlay and its Owner (Name(), so named instances reload correctly). main only points the instance at where the config lives.

cf_postgres.New(cf_postgres.WithConfigSource("postgresql", "config/postgresql.json"))
cf_postgres.New(cf_postgres.WithConfigSource("orders", "/etc/app/orders.yaml",
    cf_postgres.WithSourceFormat(cf_configuration.FormatYAML)))

A path of "" registers an env-only (fileless) source when the EnvPrefix is non-empty. The path CLI override stays --<source-name> (ParseFlags). Declares a dependency on "configuration".

func WithConnString

func WithConnString(dsn string) Option

WithConnString sets the full connection string (DSN or URL), parsed via pgxpool.ParseConfig. A bad DSN is recorded and returned from Init (New does not panic). Later convenience setters override individual fields.

func WithConnectTimeout

func WithConnectTimeout(d time.Duration) Option

WithConnectTimeout sets how long a single connection attempt may take (default 0 = no timeout, the pgx/libpq default).

func WithDatabase

func WithDatabase(db string) Option

WithDatabase sets the database name (default: the user name).

func WithDegradedMode

func WithDegradedMode(enabled bool) Option

WithDegradedMode allows Init to succeed when the connectivity ping (or pool create) fails. Default is hard-fail. Degraded mode screams in logs/metrics; Health still fails ping unless HealthWhenDegraded is "ready".

func WithEmbeddedMigrations

func WithEmbeddedMigrations(fsys embed.FS, dir string, opts ...MigrationOption) Option

WithEmbeddedMigrations is WithMigrations for go:embed: it takes the embedded filesystem plus the directory holding the golang-migrate files and resolves the sub-filesystem internally. embed guarantees the directory exists at build time, so the only failure mode is a mismatched dir string — a programmer error that panics loudly:

//go:embed migrations
var migrations embed.FS

p := cf_postgres.New(
	cf_postgres.WithEmbeddedMigrations(migrations, "migrations"),
)

func WithHealthCheckPeriod

func WithHealthCheckPeriod(d time.Duration) Option

WithHealthCheckPeriod sets how often the pool health-checks idle connections (default 1m).

func WithHealthWhenDegraded

func WithHealthWhenDegraded(policy string) Option

WithHealthWhenDegraded sets Health() behaviour while unreachable after DegradedMode: "not_ready" (default) or "ready" (break-glass LB traffic).

func WithHost

func WithHost(host string) Option

WithHost sets the server host (default "127.0.0.1").

func WithLockTimeout

func WithLockTimeout(d time.Duration) Option

WithLockTimeout sets the per-connection lock_timeout (0 = unset). The value is applied as a runtime parameter in milliseconds on every connection, so a blocked row lock aborts instead of waiting forever.

func WithLogger

func WithLogger(logger *slog.Logger) Option

WithLogger overrides the logger used for component diagnostics. By default the component logs through the framework logs component (declared in GetDependencies); WithLogger is an explicit override for tests and embedded use and wins over the framework logger. slog.Default() remains the fallback only when neither is available.

func WithMaxConnIdleTime

func WithMaxConnIdleTime(d time.Duration) Option

WithMaxConnIdleTime sets how long an idle connection may live before being closed (default 30m).

func WithMaxConnLifetime

func WithMaxConnLifetime(d time.Duration) Option

WithMaxConnLifetime sets how long a connection may live before being closed and replaced (default 1h).

func WithMaxConns

func WithMaxConns(n int32) Option

WithMaxConns sets the pool's maximum number of connections (default: max(4, runtime.GOMAXPROCS)).

func WithMigrateOnInit

func WithMigrateOnInit() Option

WithMigrateOnInit makes Init call CFPostgres.Migrate after the connectivity ping (fail-fast). Use for local/single-replica only. Production should keep WithMigrations (so the framework job flag works) but omit WithMigrateOnInit on the serving Deployment, and run the Job with --postgresql.job=migrate instead.

func WithMigrations

func WithMigrations(fsys fs.FS, opts ...MigrationOption) Option

WithMigrations configures the migration filesystem (and optional tracking table) used by CFPostgres.Migrate. It does not migrate by itself at Init; combine with WithMigrateOnInit for local single-process apps, or use the framework job flag --postgresql.job=migrate so the same binary can run as a K8s Job without a separate cmd/migrate.

fsys must be rooted at the directory containing golang-migrate files (<version>_<name>.up.sql / .down.sql). For go:embed use WithEmbeddedMigrations, which takes the embedded filesystem and directory and resolves the sub-filesystem for you.

func WithMinConns

func WithMinConns(n int32) Option

WithMinConns sets the pool's minimum number of connections (default 0).

func WithName

func WithName(name string) Option

WithName sets a custom component name, allowing multiple postgres instances in the same process. The default name is "postgresql" (ComponentName). Use this when you need multiple postgres clients (e.g., primary and replica) in one binary. Retrieve named instances with GetByName[*CFPostgres](fw, "primary").

func WithPassword

func WithPassword(password string) Option

WithPassword sets the authentication password.

func WithPingTimeout

func WithPingTimeout(d time.Duration) Option

WithPingTimeout sets how long Init waits for the connectivity ping before failing (default 5s).

func WithPoolConfig

func WithPoolConfig(cfg *pgxpool.Config) Option

WithPoolConfig sets the full pgxpool.Config; the config is deep-copied, so later mutation by the caller does not affect the component and instances never share state. Convenience setters (WithConnString, WithHost, WithPort, WithUser, WithPassword, WithDatabase, WithSSLMode, WithMaxConns, ...) override the matching fields, so call them after WithPoolConfig if you combine them. Pass a config returned by pgxpool.ParseConfig when you need TLS roots, runtime params, or other pgx-only settings.

func WithPort

func WithPort(port int) Option

WithPort sets the server port (default 5432).

func WithQueryTracer

func WithQueryTracer(t pgx.QueryTracer) Option

WithQueryTracer sets the pgx query tracer, invoked around every Query, QueryRow, and Exec call. Use it for observability hooks (otel spans, slow-query logging) without importing an instrumentation library into this module — the tracer interface lives in pgx. It is part of the option base config, so it survives pool rebuilds on configuration reload. Apps that need several tracers chain them with a small composite type; pgx calls a single tracer per query.

func WithSSLMode

func WithSSLMode(mode string) Option

WithSSLMode sets the sslmode (disable, prefer, require, verify-ca, verify-full). The default from the base connection string is "prefer". The TLS config and fallback chain are re-derived from the current host/port, so sslmode="prefer" still attempts plaintext as a fallback on the same address. An unknown mode is recorded and returned from Init (New does not panic): silently ignoring an explicit TLS requirement would be worse than failing startup.

func WithStatementTimeout

func WithStatementTimeout(d time.Duration) Option

WithStatementTimeout sets the per-connection statement_timeout (0 = unset). The value is applied as a runtime parameter in milliseconds on every connection, guarding against runaway queries.

func WithTLSClientCertFile

func WithTLSClientCertFile(certPath, keyPath string) Option

WithTLSClientCertFile configures the client PEM certificate and key pair for mTLS. Both paths must be non-empty. Re-read on every pool build, so rotated Secret mounts are picked up on the next configuration reload.

func WithTLSRootCAFile

func WithTLSRootCAFile(path string) Option

WithTLSRootCAFile configures a PEM CA bundle used to verify the server certificate (mutual trust for sslmode verify-ca/verify-full, or a private CA for require). The file is re-read on every pool build, so a rotated Secret mount is picked up on the next configuration reload. Requires sslmode require/verify-ca/verify-full.

func WithUser

func WithUser(user string) Option

WithUser sets the role to connect as (default: current OS user).

type PostgresConfig

type PostgresConfig struct {
	Host                 string `json:"host,omitempty" yaml:"host,omitempty" env:"HOST"`
	Port                 int    `json:"port,omitempty" yaml:"port,omitempty" env:"PORT"`
	User                 string `json:"user,omitempty" yaml:"user,omitempty" env:"USER"`
	Password             string `json:"password,omitempty" yaml:"password,omitempty" env:"PASSWORD" secret:"redact"`
	Database             string `json:"database,omitempty" yaml:"database,omitempty" env:"DATABASE"`
	SSLMode              string `json:"ssl_mode,omitempty" yaml:"ssl_mode,omitempty" env:"SSL_MODE"`
	MaxConns             int32  `json:"max_conns,omitempty" yaml:"max_conns,omitempty" env:"MAX_CONNS"`
	MinConns             int32  `json:"min_conns,omitempty" yaml:"min_conns,omitempty" env:"MIN_CONNS"`
	MaxConnLifetimeSec   int32  `json:"max_conn_lifetime_sec,omitempty" yaml:"max_conn_lifetime_sec,omitempty" env:"MAX_CONN_LIFETIME_SEC"`
	MaxConnIdleTimeSec   int32  `json:"max_conn_idle_time_sec,omitempty" yaml:"max_conn_idle_time_sec,omitempty" env:"MAX_CONN_IDLE_TIME_SEC"`
	HealthCheckPeriodSec int32  `json:"health_check_period_sec,omitempty" yaml:"health_check_period_sec,omitempty" env:"HEALTH_CHECK_PERIOD_SEC"`
	ConnectTimeoutSec    int32  `json:"connect_timeout_sec,omitempty" yaml:"connect_timeout_sec,omitempty" env:"CONNECT_TIMEOUT_SEC"`
	// ApplicationName sets the application_name runtime parameter on every
	// connection (parity with WithApplicationName, which wins when both set).
	ApplicationName string `json:"application_name,omitempty" yaml:"application_name,omitempty" env:"APPLICATION_NAME"`
	// StatementTimeoutSec and LockTimeoutSec are per-connection statement/lock
	// timeouts (Postgres milliseconds are derived from the seconds value; 0 =
	// unset). Guards against runaway queries and stuck advisory locks.
	StatementTimeoutSec int32 `json:"statement_timeout_sec,omitempty" yaml:"statement_timeout_sec,omitempty" env:"STATEMENT_TIMEOUT_SEC"`
	LockTimeoutSec      int32 `json:"lock_timeout_sec,omitempty" yaml:"lock_timeout_sec,omitempty" env:"LOCK_TIMEOUT_SEC"`
	// TLSRootCAFile, TLSClientCertFile, TLSClientKeyFile point at PEM files
	// (typically External Secrets / Secret mounts) used to build the TLS
	// config on every connect and reload — rotation is picked up without a
	// process restart. Requires an ssl_mode of require/verify-ca/verify-full.
	TLSRootCAFile     string `json:"tls_root_ca_file,omitempty" yaml:"tls_root_ca_file,omitempty" env:"TLS_ROOT_CA_FILE"`
	TLSClientCertFile string `json:"tls_client_cert_file,omitempty" yaml:"tls_client_cert_file,omitempty" env:"TLS_CLIENT_CERT_FILE"`
	TLSClientKeyFile  string `json:"tls_client_key_file,omitempty" yaml:"tls_client_key_file,omitempty" env:"TLS_CLIENT_KEY_FILE"`
	// DegradedMode — when true, a failed Init ping (or pool create) does not
	// abort the process. The pool is kept when create succeeded so later
	// reconnect can work; metrics/logs scream. Default off (pointer so
	// omitted ≠ explicit false). Off by default (hard Init).
	DegradedMode *bool `json:"degraded_mode,omitempty" yaml:"degraded_mode,omitempty" env:"DEGRADED_MODE"`
	// HealthWhenDegraded: "not_ready" (default) or "ready". Controls Health()
	// (and thus /readyz) while the pool cannot ping after a degraded Init
	// or while disconnected. "ready" is break-glass: send LB traffic anyway.
	HealthWhenDegraded string `json:"health_when_degraded,omitempty" yaml:"health_when_degraded,omitempty" env:"HEALTH_WHEN_DEGRADED"`
}

PostgresConfig is the file/env-drivable connection configuration. Load it through the configuration component (caerus-framework-configuration) and pass it via WithConfig; both JSON and YAML tags are provided. Durations are in seconds.

func ParseDSN

func ParseDSN(dsn string) (PostgresConfig, error)

ParseDSN parses a PostgreSQL connection URL or keyword/value DSN into a PostgresConfig. Supported forms are those accepted by pgxpool.ParseConfig (e.g. postgres://user:pass@host:5432/db?sslmode=require).

type SourceOption

type SourceOption func(*sourceOptions)

SourceOption configures the self-registered configuration source created by WithConfigSource.

func WithSourceEnvPrefix

func WithSourceEnvPrefix(prefix string) SourceOption

WithSourceEnvPrefix sets the environment overlay prefix for the source (default: the uppercase source name with "-" replaced by "_", plus "_" — "valkey-cache" → "VALKEY_CACHE_"). An empty prefix disables env overlay.

func WithSourceFormat

func WithSourceFormat(f cf_configuration.Format) SourceOption

WithSourceFormat forces the file format instead of inferring it from the path extension (".yaml"/".yml" → YAML; anything else JSON).