Merge pull request 'Add the golib scaffold and the pg module' (#1) from benvin/initial-pg into main

Reviewed-on: #1

.gitignore

@@ -0,0 +1,2 @@
+cover.out
+cover.filtered.out

.pre-commit-config.yaml

@@ -0,0 +1,27 @@
+repos:
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v5.0.0
+    hooks:
+      - id: trailing-whitespace
+      - id: end-of-file-fixer
+      - id: check-yaml
+      - id: check-added-large-files
+      - id: check-merge-conflict
+
+  - repo: https://github.com/dnephin/pre-commit-golang
+    rev: v0.5.1
+    hooks:
+      - id: go-fmt
+      - id: go-mod-tidy
+
+  # golib has no root-level Go files (everything lives under a module
+  # directory), so the dnephin go-vet hook, which runs `go vet` at the repo
+  # root, fails with "no Go files". Vet the whole module instead.
+  - repo: local
+    hooks:
+      - id: go-vet
+        name: go vet
+        entry: go vet ./...
+        language: system
+        types: [go]
+        pass_filenames: false

.woodpecker/build.yaml

@@ -0,0 +1,20 @@
+when:
+  - event: pull_request
+
+steps:
+  - name: build
+    image: golang:1.25
+    commands:
+      # golib ships no binaries, so the build is a compile check over every
+      # package, including the test-only pgtest helper.
+      - make build
+    backend_options:
+      kubernetes:
+        serviceAccountName: golib-ci
+        resources:
+          requests:
+            memory: 512Mi
+            cpu: 1
+          limits:
+            memory: 2Gi
+            cpu: 2

.woodpecker/pre-commit.yaml

@@ -0,0 +1,49 @@
+when:
+  - event: pull_request
+
+steps:
+  - name: pre-commit
+    image: golang:1.25
+    commands:
+      - test -z "$(gofmt -l .)"
+      - go vet ./...
+    backend_options:
+      kubernetes:
+        serviceAccountName: golib-ci
+        resources:
+          requests:
+            memory: 512Mi
+            cpu: 1
+          limits:
+            memory: 2Gi
+            cpu: 2
+
+  - name: lint
+    image: golangci/golangci-lint:latest
+    commands:
+      - golangci-lint run ./...
+    backend_options:
+      kubernetes:
+        serviceAccountName: golib-ci
+        resources:
+          requests:
+            memory: 512Mi
+            cpu: 1
+          limits:
+            memory: 2Gi
+            cpu: 2
+
+  - name: hooks
+    image: git.unkin.net/unkin/almalinux9-gobuilder:20260606
+    commands:
+      - uvx pre-commit run --all-files
+    backend_options:
+      kubernetes:
+        serviceAccountName: golib-ci
+        resources:
+          requests:
+            memory: 512Mi
+            cpu: 1
+          limits:
+            memory: 2Gi
+            cpu: 2

.woodpecker/test.yaml

@@ -0,0 +1,20 @@
+when:
+  - event: pull_request
+
+steps:
+  - name: test
+    image: golang:1.25
+    commands:
+      # Coverage-gated unit tests. The container-backed integration tests skip
+      # themselves under -short; the Kubernetes runners have no Docker socket.
+      - make cover
+    backend_options:
+      kubernetes:
+        serviceAccountName: golib-ci
+        resources:
+          requests:
+            memory: 512Mi
+            cpu: 1
+          limits:
+            memory: 2Gi
+            cpu: 2

Makefile

@@ -0,0 +1,84 @@
+VERSION := $(shell git describe --tags --always --dirty 2>/dev/null || echo dev)
+
+# Minimum total statement coverage for the shipped packages. golib is small,
+# dependency-light and consumed by every service, so everything it exports is
+# expected to have a test.
+COVER_MIN := 90
+
+# Packages excluded from the coverage gate: pgtest exists only to be imported
+# from other repos' _test.go files, and is itself exercised by the
+# container-backed integration tests, which the gate deliberately does not run.
+COVER_EXCLUDE := /pg/pgtest/
+
+.PHONY: all build test test-all cover vet fmt lint tidy clean pre-commit patch minor major _tag
+
+all: build
+
+# Mirror the .woodpecker/pre-commit.yaml checks locally.
+pre-commit:
+	test -z "$$(gofmt -l .)"
+	go vet ./...
+
+# golib ships no binaries; building is a compile check over every package.
+build:
+	go build ./...
+
+# Unit tests. The container-backed integration tests skip themselves under
+# -short, which is how CI runs them: the Kubernetes runners have no Docker.
+test:
+	go test -race -count=1 -short ./...
+
+# Everything, including the integration tests. Needs a container runtime.
+test-all:
+	TESTCONTAINERS_RYUK_DISABLED=true go test -race -count=1 ./...
+
+# Coverage gate. Unit tests alone must clear COVER_MIN, so the gate runs
+# unchanged on a runner with no container runtime.
+cover:
+	go test -race -count=1 -short -coverprofile=cover.out ./...
+	@grep -Ev '$(COVER_EXCLUDE)' cover.out > cover.filtered.out
+	@go tool cover -func=cover.filtered.out | awk -v min=$(COVER_MIN) ' \
+		/^total:/ { \
+			seen = 1; pct = $$3; sub(/%/, "", pct); \
+			printf "total coverage: %s%% (minimum %d%%)\n", pct, min; \
+			if (pct + 0 < min + 0) { print "coverage below minimum" > "/dev/stderr"; exit 1 } \
+		} \
+		END { if (!seen) { print "no total in coverage output" > "/dev/stderr"; exit 1 } }'
+
+vet:
+	go vet ./...
+
+fmt:
+	gofmt -w .
+
+lint:
+	golangci-lint run ./...
+
+tidy:
+	go mod tidy
+
+clean:
+	rm -f cover.out cover.filtered.out
+
+# Bump helpers — read the latest semver tag and create the next one. Consumers
+# pin the resulting v* tag with `go get git.unkin.net/unkin/golib@vX.Y.Z`.
+_LATEST := $(shell git tag --sort=-v:refname | grep -E '^v[0-9]+\.[0-9]+\.[0-9]+$$' | head -1)
+_BASE   := $(if $(_LATEST),$(_LATEST),v0.0.0)
+_MAJ    := $(shell echo $(_BASE) | sed 's/^v//' | cut -d. -f1)
+_MIN    := $(shell echo $(_BASE) | sed 's/^v//' | cut -d. -f2)
+_PAT    := $(shell echo $(_BASE) | sed 's/^v//' | cut -d. -f3)
+
+patch:
+	@NEW=v$(_MAJ).$(_MIN).$(shell expr $(_PAT) + 1); \
+	git tag $$NEW && echo "Tagged $$NEW" && $(MAKE) _tag TAG=$$NEW
+
+minor:
+	@NEW=v$(_MAJ).$(shell expr $(_MIN) + 1).0; \
+	git tag $$NEW && echo "Tagged $$NEW" && $(MAKE) _tag TAG=$$NEW
+
+major:
+	@NEW=v$(shell expr $(_MAJ) + 1).0.0; \
+	git tag $$NEW && echo "Tagged $$NEW" && $(MAKE) _tag TAG=$$NEW
+
+_tag:
+	git push origin $(TAG)

README.md

@@ -1,3 +1,114 @@
 # golib
 
-Shared Go library for estate services: postgres, http service kit, vault and gitea clients
+Shared Go library for the unkin estate: the plumbing that was being copy-pasted
+between services, kept in one place with one set of tests.
+
+golib is a library only. It ships no binaries and no container images, holds no
+service configuration, and takes on dependencies grudgingly — every service that
+imports it inherits them.
+
+## Modules
+
+| Import | What it does |
+| --- | --- |
+| `git.unkin.net/unkin/golib/pg` | Postgres: DSN from the environment, pgxpool construction, and the estate's migration runner. |
+| `git.unkin.net/unkin/golib/pg/pgtest` | A throwaway Postgres container for a consumer's own `_test.go` files. Test-only. |
+
+### pg
+
+```go
+dsn, err := pg.DSNFromEnv("ENCAPI_")
+if err != nil {
+    return err
+}
+pool, err := pg.NewMigrated(ctx, dsn, migrations.FS, pg.MigrateOptions{
+    LockName: "encapi-migrations",
+    Logger:   log,
+})
+```
+
+`pg.DSNFromEnv(prefix)` resolves a connection string from the environment.
+Precedence, highest first:
+
+1. `<PREFIX>DATABASE_URL` — used verbatim.
+2. `DATABASE_URL` — likewise.
+3. `<PREFIX>DBHOST`, `<PREFIX>DBPORT`, `<PREFIX>DBUSER`, `<PREFIX>DBPASS`,
+   `<PREFIX>DBNAME`, `<PREFIX>DBSSL`.
+4. libpq's `PGHOST`, `PGPORT`, `PGUSER`, `PGPASSWORD`, `PGDATABASE`, `PGSSLMODE`.
+5. Defaults: `localhost`, `5432`, `sslmode=disable`.
+
+Levels 3 to 5 resolve per field, so a deployment can set the password from a
+secret and leave the rest to `PG*`. User and database name have no default —
+an unset one is an error naming the variables that were checked. An empty prefix
+reads the bare `DBHOST`/`DBPORT`/… names the estate's services already use, so
+the rendered DSN is unchanged from the `fmt.Sprintf` builders this replaces.
+
+`pg.New` opens a pgxpool and pings it, so an unreachable server fails at startup
+rather than on the first query. `pg.NewMigrated` does that and then migrates.
+
+`pg.Migrate(ctx, pool, fsys, opts)` applies the `.sql` files at the root of
+`fsys` in lexical filename order. Every replica calls it at startup:
+
+- The run holds a cluster-wide `pg_advisory_lock` keyed on FNV-1a/64 of
+  `opts.LockName`, on one dedicated pooled connection, because the lock is
+  session-scoped. Replicas that queue behind the winner find the set already
+  recorded and do nothing.
+- Each file is applied together with its `schema_migrations` row in a single
+  transaction, so a failure leaves neither a half-tracked migration nor a
+  tracking row that would skip it next time.
+- A file missing from `schema_migrations` is re-applied even if the live
+  database already has it, which is how a schema applied out of band is adopted.
+  Write migrations `IF NOT EXISTS`-guarded so that re-run is a no-op.
+- If the unlock does not land, the connection is discarded rather than returned
+  to the pool, so a session that may still hold the lock cannot be reused.
+
+`pg.LockKey(name)` exposes the derivation, so a service migrating off a
+hardcoded key can assert the two agree before switching over.
+
+### pgtest
+
+`pgtest` starts `postgres:17-alpine` via testcontainers. Import it only from
+`_test.go` files. The estate's CI runs on Kubernetes with no Docker socket, so
+container-backed tests must skip themselves under `-short`:
+
+```go
+func TestSomething(t *testing.T) {
+    ctx := context.Background()
+    dsn := pgtest.MustStartPostgres(ctx, t) // skips under -short, cleans up after
+    ...
+}
+```
+
+## Consuming
+
+golib is versioned with semver tags and consumed like any Go module. Pin a tag:
+
+```
+go get git.unkin.net/unkin/golib@v0.1.0
+```
+
+Nothing is released until a `v*` tag exists; `make patch`, `make minor` and
+`make major` cut and push the next one.
+
+Because everything shares one module path, a consumer that imports only `pg`
+still resolves golib's full dependency set in its module graph. That is the
+reason to keep the dependency list short, and the reason `pgtest`'s
+testcontainers dependency is the exception rather than the pattern.
+
+## Development
+
+```
+make build     # compile every package
+make test      # unit tests (-short: no container needed)
+make test-all  # everything, including the container-backed integration tests
+make cover     # unit tests with the coverage gate
+make lint      # golangci-lint
+```
+
+New code needs meaningful tests. `make cover` fails below **90% statement
+coverage**, measured over the shipped packages from the unit tests alone — the
+integration tests do not count towards it, so the bar has to be cleared without
+a database. `pg/pgtest` is excluded: it is test scaffolding for other repos, and
+is covered by the integration tests the gate does not run.
+
+CI runs `test`, `pre-commit` and `build` on every pull request.

go.mod

@@ -0,0 +1,67 @@
+module git.unkin.net/unkin/golib
+
+go 1.25.0
+
+require (
+	github.com/jackc/pgx/v5 v5.9.2
+	github.com/testcontainers/testcontainers-go v0.44.0
+	github.com/testcontainers/testcontainers-go/modules/postgres v0.44.0
+)
+
+require (
+	dario.cat/mergo v1.0.2 // indirect
+	github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c // indirect
+	github.com/Microsoft/go-winio v0.6.2 // indirect
+	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/containerd/errdefs v1.0.0 // indirect
+	github.com/containerd/errdefs/pkg v0.3.0 // indirect
+	github.com/containerd/log v0.1.0 // indirect
+	github.com/containerd/platforms v0.2.1 // indirect
+	github.com/cpuguy83/dockercfg v0.3.2 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/distribution/reference v0.6.0 // indirect
+	github.com/docker/go-connections v0.7.0 // indirect
+	github.com/docker/go-units v0.5.0 // indirect
+	github.com/ebitengine/purego v0.10.1 // indirect
+	github.com/felixge/httpsnoop v1.1.0 // indirect
+	github.com/go-logr/logr v1.4.3 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/go-ole/go-ole v1.3.0 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // indirect
+	github.com/klauspost/compress v1.18.6 // indirect
+	github.com/lufia/plan9stats v0.0.0-20260330125221-c963978e514e // indirect
+	github.com/magiconair/properties v1.8.10 // indirect
+	github.com/moby/docker-image-spec v1.3.1 // indirect
+	github.com/moby/go-archive v0.2.0 // indirect
+	github.com/moby/moby/api v1.55.0 // indirect
+	github.com/moby/moby/client v0.5.0 // indirect
+	github.com/moby/patternmatcher v0.6.1 // indirect
+	github.com/moby/sys/sequential v0.7.0 // indirect
+	github.com/moby/sys/user v0.4.0 // indirect
+	github.com/moby/sys/userns v0.1.0 // indirect
+	github.com/moby/term v0.5.2 // indirect
+	github.com/opencontainers/go-digest v1.0.0 // indirect
+	github.com/opencontainers/image-spec v1.1.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect
+	github.com/shirou/gopsutil/v4 v4.26.6 // indirect
+	github.com/sirupsen/logrus v1.9.4 // indirect
+	github.com/stretchr/testify v1.11.1 // indirect
+	github.com/tklauser/go-sysconf v0.4.0 // indirect
+	github.com/tklauser/numcpus v0.12.0 // indirect
+	github.com/yusufpapurcu/wmi v1.2.4 // indirect
+	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 // indirect
+	go.opentelemetry.io/otel v1.44.0 // indirect
+	go.opentelemetry.io/otel/metric v1.44.0 // indirect
+	go.opentelemetry.io/otel/trace v1.44.0 // indirect
+	golang.org/x/crypto v0.54.0 // indirect
+	golang.org/x/sync v0.22.0 // indirect
+	golang.org/x/sys v0.47.0 // indirect
+	golang.org/x/text v0.40.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)

go.sum

@@ -0,0 +1,158 @@
+dario.cat/mergo v1.0.2 h1:85+piFYR1tMbRrLcDwR18y4UKJ3aH1Tbzi24VRW1TK8=
+dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA=
+github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6 h1:He8afgbRMd7mFxO99hRNu+6tazq8nFF9lIwo9JFroBk=
+github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8=
+github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c h1:udKWzYgxTojEKWjV8V+WSxDXJ4NFATAsZjh8iIbsQIg=
+github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E=
+github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY=
+github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU=
+github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=
+github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG8PI=
+github.com/containerd/errdefs v1.0.0/go.mod h1:+YBYIdtsnF4Iw6nWZhJcqGSg/dwvV7tyJ/kCkyJ2k+M=
+github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151Xdx3ZPPE=
+github.com/containerd/errdefs/pkg v0.3.0/go.mod h1:NJw6s9HwNuRhnjJhM7pylWwMyAkmCQvQ4GpJHEqRLVk=
+github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I=
+github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3EhrzVo=
+github.com/containerd/platforms v0.2.1 h1:zvwtM3rz2YHPQsF2CHYM8+KtB5dvhISiXh5ZpSBQv6A=
+github.com/containerd/platforms v0.2.1/go.mod h1:XHCb+2/hzowdiut9rkudds9bE5yJ7npe7dG/wG+uFPw=
+github.com/cpuguy83/dockercfg v0.3.2 h1:DlJTyZGBDlXqUZ2Dk2Q3xHs/FtnooJJVaad2S9GKorA=
+github.com/cpuguy83/dockercfg v0.3.2/go.mod h1:sugsbF4//dDlL/i+S+rtpIWp+5h0BHJHfjj5/jFyUJc=
+github.com/creack/pty v1.1.24 h1:bJrF4RRfyJnbTJqzRLHzcGaZK1NeM5kTC9jGgovnR1s=
+github.com/creack/pty v1.1.24/go.mod h1:08sCNb52WyoAwi2QDyzUCTgcvVFhUzewun7wtTfvcwE=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk=
+github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E=
+github.com/docker/go-connections v0.7.0 h1:6SsRfJddP22WMrCkj19x9WKjEDTB+ahsdiGYf0mN39c=
+github.com/docker/go-connections v0.7.0/go.mod h1:no1qkHdjq7kLMGUXYAduOhYPSJxxvgWBh7ogVvptn3Q=
+github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
+github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
+github.com/ebitengine/purego v0.10.1 h1:dewVBCBT2GaMu1SrNTYxQhgQBethzfhiwvZiLGP/qyY=
+github.com/ebitengine/purego v0.10.1/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ=
+github.com/felixge/httpsnoop v1.1.0 h1:3YtUj32ZZkqZtt3sZZsClsymw/QDuVfpNhoA31zeORc=
+github.com/felixge/httpsnoop v1.1.0/go.mod h1:Zqxgdd+1Rkcz8euOqdr7lqgCRJztwr5hp9vDSi5UZCE=
+github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
+github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
+github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
+github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
+github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
+github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0=
+github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE=
+github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78=
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.9.2 h1:3ZhOzMWnR4yJ+RW1XImIPsD1aNSz4T4fyP7zlQb56hw=
+github.com/jackc/pgx/v5 v5.9.2/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
+github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
+github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
+github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao=
+github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
+github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
+github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
+github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
+github.com/lufia/plan9stats v0.0.0-20260330125221-c963978e514e h1:Q6MvJtQK/iRcRtzAscm/zF23XxJlbECiGPyRicsX+Ak=
+github.com/lufia/plan9stats v0.0.0-20260330125221-c963978e514e/go.mod h1:autxFIvghDt3jPTLoqZ9OZ7s9qTGNAWmYCjVFWPX/zg=
+github.com/magiconair/properties v1.8.10 h1:s31yESBquKXCV9a/ScB3ESkOjUYYv+X0rg8SYxI99mE=
+github.com/magiconair/properties v1.8.10/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0=
+github.com/mdelapenya/tlscert v0.2.0 h1:7H81W6Z/4weDvZBNOfQte5GpIMo0lGYEeWbkGp5LJHI=
+github.com/mdelapenya/tlscert v0.2.0/go.mod h1:O4njj3ELLnJjGdkN7M/vIVCpZ+Cf0L6muqOG4tLSl8o=
+github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0=
+github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo=
+github.com/moby/go-archive v0.2.0 h1:zg5QDUM2mi0JIM9fdQZWC7U8+2ZfixfTYoHL7rWUcP8=
+github.com/moby/go-archive v0.2.0/go.mod h1:mNeivT14o8xU+5q1YnNrkQVpK+dnNe/K6fHqnTg4qPU=
+github.com/moby/moby/api v1.55.0 h1:2/sexvQyqIWS8pRSCFddBfpW2qE7vR7FCL+vN8pxwMc=
+github.com/moby/moby/api v1.55.0/go.mod h1:+RQ6wluLwtYaTd1WnPLykIDPekkuyD/ROWQClE83pzs=
+github.com/moby/moby/client v0.5.0 h1:5XhyPk2fuOWf6RlSFa3MkIIgDZkF25xToXW8Q/BH7cc=
+github.com/moby/moby/client v0.5.0/go.mod h1:rcVpF8ncl9vo5gaIBdol6CnbEtSj1uxMvEV/UrykF/s=
+github.com/moby/patternmatcher v0.6.1 h1:qlhtafmr6kgMIJjKJMDmMWq7WLkKIo23hsrpR3x084U=
+github.com/moby/patternmatcher v0.6.1/go.mod h1:hDPoyOpDY7OrrMDLaYoY3hf52gNCR/YOUYxkhApJIxc=
+github.com/moby/sys/sequential v0.7.0 h1:ASQNGNROJSuOO6LL6bPHbKvuZu6NU8P4ldPWk31zj/8=
+github.com/moby/sys/sequential v0.7.0/go.mod h1:NfSTAp6V3fw4tmkD62PEcOKeZKquXT8VKCkf7aVR79o=
+github.com/moby/sys/user v0.4.0 h1:jhcMKit7SA80hivmFJcbB1vqmw//wU61Zdui2eQXuMs=
+github.com/moby/sys/user v0.4.0/go.mod h1:bG+tYYYJgaMtRKgEmuueC0hJEAZWwtIbZTB+85uoHjs=
+github.com/moby/sys/userns v0.1.0 h1:tVLXkFOxVu9A64/yh59slHVv9ahO9UIev4JZusOLG/g=
+github.com/moby/sys/userns v0.1.0/go.mod h1:IHUYgu/kao6N8YZlp9Cf444ySSvCmDlmzUcYfDHOl28=
+github.com/moby/term v0.5.2 h1:6qk3FJAFDs6i/q3W/pQ97SX192qKfZgGjCQqfCJkgzQ=
+github.com/moby/term v0.5.2/go.mod h1:d3djjFCrjnB+fl8NJux+EJzu0msscUP+f8it8hPkFLc=
+github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U=
+github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
+github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040=
+github.com/opencontainers/image-spec v1.1.1/go.mod h1:qpqAh3Dmcf36wStyyWU+kCeDgrGnAve2nCC8+7h8Q0M=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 h1:o4JXh1EVt9k/+g42oCprj/FisM4qX9L3sZB3upGN2ZU=
+github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE=
+github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
+github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
+github.com/shirou/gopsutil/v4 v4.26.6 h1:Mzr/npDtQC/xpeEuQKHZt8Zo9CmPvhTj8nkR8w5TLDs=
+github.com/shirou/gopsutil/v4 v4.26.6/go.mod h1:LZ6ewCSkBqUpvSOf+LsTGnRinC6iaNUNMGBtDkJBaLQ=
+github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w=
+github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4=
+github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+github.com/testcontainers/testcontainers-go v0.44.0 h1:/Fwh6HY1mIikhnm9e7HwoxGycx0lzRAE0f5VQpjFxzI=
+github.com/testcontainers/testcontainers-go v0.44.0/go.mod h1:IcnwQrYTO86xHXu5bvMaBH7ATlbS3Qn1M1QWW3c66rE=
+github.com/testcontainers/testcontainers-go/modules/postgres v0.44.0 h1:8fdv/9y3JMxjQ+ULAcOG8RtgeNu5t9XF9LolSXDuTwM=
+github.com/testcontainers/testcontainers-go/modules/postgres v0.44.0/go.mod h1:CFr2LncGYokw+OKjXcr8ARCKG1SaC2UEnGxFBovE86g=
+github.com/tklauser/go-sysconf v0.4.0 h1:7H0uAN+7RkwWRaxhYXDLqa5V3LPrJeV8wmD9dRUgPQU=
+github.com/tklauser/go-sysconf v0.4.0/go.mod h1:8mTNWyog7H+MpKijp4VmKJAd2bbYQ2zuUwkYRbUArPI=
+github.com/tklauser/numcpus v0.12.0 h1:NR85qdvHA9pFse3x3weVZ0r0ST8R6l5RHbZrlRaqob4=
+github.com/tklauser/numcpus v0.12.0/go.mod h1:ABHeXzJnr/qqwguhClkZKT1/8VABcYrsyUiUGobwWJg=
+github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0=
+github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
+go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
+go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
+go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 h1:8tvICD4vSTOOsNrsI4Ljf6C+6UKvpTEH5XY3JMoyPoo=
+go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0/go.mod h1:z9+yiacE0IHRqM4qFfkbt/JYlmYXgss8GY/jXoNuPJI=
+go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU=
+go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc=
+go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc=
+go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo=
+go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58=
+go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0=
+go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI=
+go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA=
+go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk=
+go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE=
+golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw=
+golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk=
+golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
+golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
+golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
+golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+golang.org/x/term v0.45.0 h1:NwWyBmoJCbfTHpxrWoZ9C6/VxOf7ic219I8xZZFdrf0=
+golang.org/x/term v0.45.0/go.mod h1:9aqxs0blBcrm/n0L9QW0aRVD+ktan8ssZromtqJC43w=
+golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
+golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gotest.tools/v3 v3.5.2 h1:7koQfIKdy+I8UTetycgUqXWSDwpgv193Ka+qRsmBY8Q=
+gotest.tools/v3 v3.5.2/go.mod h1:LtdLGcnqToBH83WByAAi/wiwSFCArdFIUV/xxN4pcjA=
+pgregory.net/rapid v1.2.0 h1:keKAYRcjm+e1F0oAuU5F5+YPAWcyxNNRK2wud503Gnk=
+pgregory.net/rapid v1.2.0/go.mod h1:PY5XlDGj0+V1FCq0o192FdRhpKHGTRIWBgqjDBTrq04=

pg/dsn.go

@@ -0,0 +1,125 @@
+package pg
+
+import (
+	"fmt"
+	"net/url"
+	"os"
+	"strconv"
+	"strings"
+)
+
+// Default connection settings applied when neither the prefixed nor the libpq
+// variable for a field is set. User and database have no default: a service
+// connecting to "postgres" as "postgres" is a bug, not a default.
+const (
+	defaultHost    = "localhost"
+	defaultPort    = 5432
+	defaultSSLMode = "disable"
+)
+
+// dsnField is one connection setting and the variable names it is read from.
+type dsnField struct {
+	// suffix is appended to the caller's prefix, e.g. "DBHOST".
+	suffix string
+	// libpq is the standard libpq variable for the same setting.
+	libpq string
+}
+
+var (
+	fieldHost = dsnField{"DBHOST", "PGHOST"}
+	fieldPort = dsnField{"DBPORT", "PGPORT"}
+	fieldUser = dsnField{"DBUSER", "PGUSER"}
+	fieldPass = dsnField{"DBPASS", "PGPASSWORD"}
+	fieldName = dsnField{"DBNAME", "PGDATABASE"}
+	fieldSSL  = dsnField{"DBSSL", "PGSSLMODE"}
+)
+
+// DSNFromEnv builds a libpq/pgx connection string from the environment.
+//
+// prefix is prepended to every custom variable name, so a service can namespace
+// its settings ("ENCAPI_" reads ENCAPI_DBHOST); an empty prefix reads the bare
+// names the estate's services already use (DBHOST, DBPORT, ...).
+//
+// Precedence, highest first:
+//
+//  1. <PREFIX>DATABASE_URL — returned verbatim, no parsing or validation.
+//  2. DATABASE_URL — likewise. Identical to 1 when prefix is empty.
+//  3. <PREFIX>DBHOST, <PREFIX>DBPORT, <PREFIX>DBUSER, <PREFIX>DBPASS,
+//     <PREFIX>DBNAME, <PREFIX>DBSSL.
+//  4. libpq's PGHOST, PGPORT, PGUSER, PGPASSWORD, PGDATABASE, PGSSLMODE.
+//  5. Built-in defaults: localhost, 5432, sslmode=disable.
+//
+// Levels 3 to 5 resolve per field, so a deployment may set DBPASS from a secret
+// and leave the rest to PG* variables. An unset variable and one set to the
+// empty string are treated alike, except for the password, where the empty
+// string is a legitimate value and only distinguishable from unset if set
+// explicitly — both render the same DSN, so the distinction does not matter.
+//
+// User and database name have no default: DSNFromEnv reports an error naming
+// the variables it looked at rather than connecting somewhere unintended.
+func DSNFromEnv(prefix string) (string, error) {
+	if v := os.Getenv(prefix + "DATABASE_URL"); v != "" {
+		return v, nil
+	}
+	if v := os.Getenv("DATABASE_URL"); v != "" {
+		return v, nil
+	}
+
+	host := lookup(prefix, fieldHost, defaultHost)
+	portStr := lookup(prefix, fieldPort, strconv.Itoa(defaultPort))
+	user := lookup(prefix, fieldUser, "")
+	pass := lookup(prefix, fieldPass, "")
+	name := lookup(prefix, fieldName, "")
+	ssl := lookup(prefix, fieldSSL, defaultSSLMode)
+
+	port, err := strconv.Atoi(portStr)
+	if err != nil {
+		return "", fmt.Errorf("invalid %s%s: %w", prefix, fieldPort.suffix, err)
+	}
+	if port < 1 || port > 65535 {
+		return "", fmt.Errorf("invalid %s%s: port %d out of range", prefix, fieldPort.suffix, port)
+	}
+	if user == "" {
+		return "", fmt.Errorf("no database user: set %s%s or %s", prefix, fieldUser.suffix, fieldUser.libpq)
+	}
+	if name == "" {
+		return "", fmt.Errorf("no database name: set %s%s or %s", prefix, fieldName.suffix, fieldName.libpq)
+	}
+
+	return DSN(host, port, user, pass, name, ssl), nil
+}
+
+// DSN renders the estate's standard connection string. User and password are
+// percent-escaped, so a password containing "@" or "/" does not truncate the
+// host; for values without reserved characters the result is byte-identical to
+// the fmt.Sprintf builders this replaces.
+func DSN(host string, port int, user, pass, name, sslMode string) string {
+	u := url.URL{
+		Scheme:   "postgres",
+		User:     url.UserPassword(user, pass),
+		Host:     hostPort(host, port),
+		Path:     "/" + name,
+		RawQuery: "sslmode=" + url.QueryEscape(sslMode),
+	}
+	return u.String()
+}
+
+// hostPort joins host and port, bracketing a bare IPv6 literal so the colons in
+// the address are not read as the port separator.
+func hostPort(host string, port int) string {
+	if host != "" && host[0] != '[' && strings.Contains(host, ":") {
+		return "[" + host + "]:" + strconv.Itoa(port)
+	}
+	return host + ":" + strconv.Itoa(port)
+}
+
+// lookup resolves one field: prefixed variable, then libpq variable, then def.
+func lookup(prefix string, f dsnField, def string) string {
+	if v := os.Getenv(prefix + f.suffix); v != "" {
+		return v
+	}
+	if v := os.Getenv(f.libpq); v != "" {
+		return v
+	}
+	return def
+}

pg/dsn_test.go

@@ -0,0 +1,236 @@
+package pg
+
+import (
+	"net/url"
+	"strings"
+	"testing"
+)
+
+// dsnVars is every variable DSNFromEnv reads, for both the bare and the
+// prefixed namespace used by these tests. Each case starts from all of them
+// unset so an inherited PGHOST on a developer's machine cannot change a result.
+var dsnVars = []string{
+	"DATABASE_URL", "DBHOST", "DBPORT", "DBUSER", "DBPASS", "DBNAME", "DBSSL",
+	"PGHOST", "PGPORT", "PGUSER", "PGPASSWORD", "PGDATABASE", "PGSSLMODE",
+	"APP_DATABASE_URL", "APP_DBHOST", "APP_DBPORT", "APP_DBUSER", "APP_DBPASS",
+	"APP_DBNAME", "APP_DBSSL",
+}
+
+// setEnv clears every variable DSNFromEnv consults, then sets the given ones.
+func setEnv(t *testing.T, env map[string]string) {
+	t.Helper()
+	for _, k := range dsnVars {
+		t.Setenv(k, "")
+	}
+	for k, v := range env {
+		t.Setenv(k, v)
+	}
+}
+
+func TestDSNFromEnv(t *testing.T) {
+	tests := []struct {
+		name   string
+		prefix string
+		env    map[string]string
+		want   string
+	}{
+		{
+			name:   "prefixed custom vars",
+			prefix: "APP_",
+			env: map[string]string{
+				"APP_DBHOST": "db.internal", "APP_DBPORT": "6432",
+				"APP_DBUSER": "app", "APP_DBPASS": "s3cret",
+				"APP_DBNAME": "appdb", "APP_DBSSL": "require",
+			},
+			want: "postgres://app:s3cret@db.internal:6432/appdb?sslmode=require",
+		},
+		{
+			name:   "bare custom vars with an empty prefix",
+			prefix: "",
+			env: map[string]string{
+				"DBHOST": "pg", "DBPORT": "5432", "DBUSER": "encapi",
+				"DBPASS": "encapi", "DBNAME": "encapi", "DBSSL": "disable",
+			},
+			// Byte-identical to the fmt.Sprintf builder this replaces.
+			want: "postgres://encapi:encapi@pg:5432/encapi?sslmode=disable",
+		},
+		{
+			name:   "libpq vars fill in",
+			prefix: "APP_",
+			env: map[string]string{
+				"PGHOST": "libpq.host", "PGPORT": "5433", "PGUSER": "pguser",
+				"PGPASSWORD": "pgpass", "PGDATABASE": "pgdb", "PGSSLMODE": "verify-full",
+			},
+			want: "postgres://pguser:pgpass@libpq.host:5433/pgdb?sslmode=verify-full",
+		},
+		{
+			name:   "prefixed vars beat libpq vars per field",
+			prefix: "APP_",
+			env: map[string]string{
+				"APP_DBPASS": "from-secret",
+				"PGHOST":     "libpq.host", "PGUSER": "pguser",
+				"PGPASSWORD": "ignored", "PGDATABASE": "pgdb",
+			},
+			want: "postgres://pguser:from-secret@libpq.host:5432/pgdb?sslmode=disable",
+		},
+		{
+			name:   "defaults for host, port and sslmode",
+			prefix: "",
+			env:    map[string]string{"DBUSER": "u", "DBNAME": "d"},
+			want:   "postgres://u:@localhost:5432/d?sslmode=disable",
+		},
+		{
+			name:   "prefixed DATABASE_URL passes through verbatim",
+			prefix: "APP_",
+			env: map[string]string{
+				"APP_DATABASE_URL": "postgres://who:cares@elsewhere/db?sslmode=require&application_name=x",
+				"APP_DBHOST":       "ignored", "APP_DBUSER": "ignored", "APP_DBNAME": "ignored",
+			},
+			want: "postgres://who:cares@elsewhere/db?sslmode=require&application_name=x",
+		},
+		{
+			name:   "bare DATABASE_URL wins over the field vars",
+			prefix: "APP_",
+			env: map[string]string{
+				"DATABASE_URL": "postgres://u:p@h:5432/d",
+				"APP_DBHOST":   "ignored", "APP_DBUSER": "ignored", "APP_DBNAME": "ignored",
+			},
+			want: "postgres://u:p@h:5432/d",
+		},
+		{
+			name:   "prefixed DATABASE_URL wins over the bare one",
+			prefix: "APP_",
+			env: map[string]string{
+				"APP_DATABASE_URL": "postgres://app@app-host/app",
+				"DATABASE_URL":     "postgres://bare@bare-host/bare",
+			},
+			want: "postgres://app@app-host/app",
+		},
+		{
+			name:   "reserved characters in the password are escaped",
+			prefix: "",
+			env: map[string]string{
+				"DBHOST": "h", "DBUSER": "u", "DBPASS": "p@ss/w:rd", "DBNAME": "d",
+			},
+			want: "postgres://u:p%40ss%2Fw%3Ard@h:5432/d?sslmode=disable",
+		},
+		{
+			name:   "IPv6 host is bracketed",
+			prefix: "",
+			env: map[string]string{
+				"DBHOST": "fd00::1", "DBUSER": "u", "DBNAME": "d",
+			},
+			want: "postgres://u:@[fd00::1]:5432/d?sslmode=disable",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			setEnv(t, tc.env)
+			got, err := DSNFromEnv(tc.prefix)
+			if err != nil {
+				t.Fatalf("DSNFromEnv: %v", err)
+			}
+			if got != tc.want {
+				t.Fatalf("DSNFromEnv = %q, want %q", got, tc.want)
+			}
+		})
+	}
+}
+
+// Every DSN this builds must survive the parse pgx will do on it.
+func TestDSNFromEnv_ResultParses(t *testing.T) {
+	setEnv(t, map[string]string{
+		"DBHOST": "fd00::1", "DBPORT": "6432", "DBUSER": "us er",
+		"DBPASS": "p@ss/w:rd", "DBNAME": "app", "DBSSL": "verify-full",
+	})
+	dsn, err := DSNFromEnv("")
+	if err != nil {
+		t.Fatalf("DSNFromEnv: %v", err)
+	}
+	u, err := url.Parse(dsn)
+	if err != nil {
+		t.Fatalf("parse %q: %v", dsn, err)
+	}
+	if u.Hostname() != "fd00::1" {
+		t.Errorf("host = %q, want fd00::1", u.Hostname())
+	}
+	if u.Port() != "6432" {
+		t.Errorf("port = %q, want 6432", u.Port())
+	}
+	if u.User.Username() != "us er" {
+		t.Errorf("user = %q, want %q", u.User.Username(), "us er")
+	}
+	pass, _ := u.User.Password()
+	if pass != "p@ss/w:rd" {
+		t.Errorf("password = %q, want %q", pass, "p@ss/w:rd")
+	}
+	if got := strings.TrimPrefix(u.Path, "/"); got != "app" {
+		t.Errorf("database = %q, want app", got)
+	}
+	if got := u.Query().Get("sslmode"); got != "verify-full" {
+		t.Errorf("sslmode = %q, want verify-full", got)
+	}
+}
+
+func TestDSNFromEnv_Errors(t *testing.T) {
+	tests := []struct {
+		name    string
+		env     map[string]string
+		wantSub string
+	}{
+		{
+			name:    "unparseable port",
+			env:     map[string]string{"DBPORT": "not-a-port", "DBUSER": "u", "DBNAME": "d"},
+			wantSub: "invalid DBPORT",
+		},
+		{
+			name:    "port out of range",
+			env:     map[string]string{"DBPORT": "70000", "DBUSER": "u", "DBNAME": "d"},
+			wantSub: "out of range",
+		},
+		{
+			name:    "no user",
+			env:     map[string]string{"DBNAME": "d"},
+			wantSub: "set DBUSER or PGUSER",
+		},
+		{
+			name:    "no database name",
+			env:     map[string]string{"DBUSER": "u"},
+			wantSub: "set DBNAME or PGDATABASE",
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			setEnv(t, tc.env)
+			got, err := DSNFromEnv("")
+			if err == nil {
+				t.Fatalf("expected an error, got DSN %q", got)
+			}
+			if !strings.Contains(err.Error(), tc.wantSub) {
+				t.Fatalf("error %q does not mention %q", err, tc.wantSub)
+			}
+		})
+	}
+}
+
+// The error names the prefixed variable the caller is expected to set, not the
+// bare one, or the message sends them looking for the wrong knob.
+func TestDSNFromEnv_ErrorNamesPrefixedVar(t *testing.T) {
+	setEnv(t, map[string]string{"APP_DBPORT": "x", "APP_DBUSER": "u", "APP_DBNAME": "d"})
+	_, err := DSNFromEnv("APP_")
+	if err == nil {
+		t.Fatal("expected an error")
+	}
+	if !strings.Contains(err.Error(), "APP_DBPORT") {
+		t.Fatalf("error %q does not name APP_DBPORT", err)
+	}
+}
+
+func TestDSN_EmptyPasswordMatchesLegacyFormat(t *testing.T) {
+	// The Sprintf builders rendered an unset password as an empty string
+	// between the colon and the "@"; keep that shape so DSNs do not churn.
+	if got, want := DSN("h", 5432, "u", "", "d", "disable"), "postgres://u:@h:5432/d?sslmode=disable"; got != want {
+		t.Fatalf("DSN = %q, want %q", got, want)
+	}
+}

pg/fakes_test.go

@@ -0,0 +1,173 @@
+package pg
+
+import (
+	"context"
+	"errors"
+	"io/fs"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+)
+
+// brokenFS fails every Open, so fs.ReadDir(".") fails.
+type brokenFS struct{}
+
+func (brokenFS) Open(string) (fs.File, error) { return nil, fs.ErrPermission }
+
+// missingFileFS lists a migration that cannot then be read, the shape a
+// mis-built embed or a racing file deletion produces.
+type missingFileFS struct{}
+
+func (missingFileFS) Open(string) (fs.File, error) { return nil, fs.ErrNotExist }
+
+func (missingFileFS) ReadDir(string) ([]fs.DirEntry, error) {
+	return []fs.DirEntry{fakeDirEntry{name: "0001_first.sql"}}, nil
+}
+
+type fakeDirEntry struct {
+	name string
+	dir  bool
+}
+
+func (e fakeDirEntry) Name() string { return e.name }
+func (e fakeDirEntry) IsDir() bool  { return e.dir }
+func (e fakeDirEntry) Type() fs.FileMode {
+	if e.dir {
+		return fs.ModeDir
+	}
+	return 0
+}
+func (e fakeDirEntry) Info() (fs.FileInfo, error) { return nil, errors.New("no info") }
+
+// execCall records one statement the migrator issued.
+type execCall struct {
+	sql  string
+	args []any
+}
+
+// fakeSession is a pgx connection that records statements instead of running
+// them. Embedding nothing: it implements the whole session interface.
+type fakeSession struct {
+	execs    []execCall
+	execErrs map[string]error
+
+	queries  []string
+	rows     *fakeRows
+	queryErr error
+
+	tx       *fakeTx
+	beginErr error
+}
+
+func (s *fakeSession) Exec(_ context.Context, sql string, args ...any) (pgconn.CommandTag, error) {
+	s.execs = append(s.execs, execCall{sql: sql, args: args})
+	return pgconn.NewCommandTag("SELECT 1"), s.execErrs[sql]
+}
+
+func (s *fakeSession) Query(_ context.Context, sql string, _ ...any) (pgx.Rows, error) {
+	s.queries = append(s.queries, sql)
+	if s.queryErr != nil {
+		return nil, s.queryErr
+	}
+	return s.rows, nil
+}
+
+func (s *fakeSession) Begin(context.Context) (pgx.Tx, error) {
+	if s.beginErr != nil {
+		return nil, s.beginErr
+	}
+	if s.tx == nil {
+		s.tx = &fakeTx{}
+	}
+	return s.tx, nil
+}
+
+// execSQL returns just the statements issued, in order.
+func (s *fakeSession) execSQL() []string {
+	out := make([]string, 0, len(s.execs))
+	for _, c := range s.execs {
+		out = append(out, c.sql)
+	}
+	return out
+}
+
+// fakeRows serves a fixed column of strings. The embedded interface supplies
+// the pgx.Rows methods the migrator never calls; calling one panics, which is
+// the intent — it would mean the migrator grew an untested dependency.
+type fakeRows struct {
+	pgx.Rows
+
+	values  []string
+	i       int
+	scanErr error
+	err     error
+	closed  bool
+}
+
+func (r *fakeRows) Next() bool {
+	if r.i >= len(r.values) {
+		return false
+	}
+	r.i++
+	return true
+}
+
+func (r *fakeRows) Scan(dest ...any) error {
+	if r.scanErr != nil {
+		return r.scanErr
+	}
+	if len(dest) != 1 {
+		return errors.New("expected exactly one scan destination")
+	}
+	p, ok := dest[0].(*string)
+	if !ok {
+		return errors.New("expected a *string scan destination")
+	}
+	*p = r.values[r.i-1]
+	return nil
+}
+
+func (r *fakeRows) Close()     { r.closed = true }
+func (r *fakeRows) Err() error { return r.err }
+
+// fakeTx records the statements and the terminal call of a transaction.
+type fakeTx struct {
+	pgx.Tx
+
+	execs     []execCall
+	execErrs  map[string]error
+	commitErr error
+	committed bool
+	rolled    int
+}
+
+func (t *fakeTx) Exec(_ context.Context, sql string, args ...any) (pgconn.CommandTag, error) {
+	t.execs = append(t.execs, execCall{sql: sql, args: args})
+	return pgconn.NewCommandTag("INSERT 0 1"), t.execErrs[sql]
+}
+
+func (t *fakeTx) Commit(context.Context) error {
+	if t.commitErr != nil {
+		return t.commitErr
+	}
+	t.committed = true
+	return nil
+}
+
+func (t *fakeTx) Rollback(context.Context) error {
+	t.rolled++
+	if t.committed {
+		// What pgx returns for a rollback after a successful commit; the
+		// deferred rollback in Apply must tolerate it.
+		return pgx.ErrTxClosed
+	}
+	return nil
+}
+
+func (t *fakeTx) execSQL() []string {
+	out := make([]string, 0, len(t.execs))
+	for _, c := range t.execs {
+		out = append(out, c.sql)
+	}
+	return out
+}

pg/integration_test.go

@@ -0,0 +1,186 @@
+package pg_test
+
+import (
+	"context"
+	"net/url"
+	"strings"
+	"testing"
+	"testing/fstest"
+	"time"
+
+	"git.unkin.net/unkin/golib/pg"
+	"git.unkin.net/unkin/golib/pg/pgtest"
+)
+
+// migrations is a two-file set exercising the things the runner promises:
+// multi-statement files (simple protocol) and IF NOT EXISTS re-runnability.
+var migrations = fstest.MapFS{
+	"0001_widgets.sql": {Data: []byte(`
+CREATE TABLE IF NOT EXISTS widgets (id BIGSERIAL PRIMARY KEY, name TEXT NOT NULL);
+CREATE INDEX IF NOT EXISTS widgets_name_idx ON widgets (name);
+`)},
+	"0002_gadgets.sql": {Data: []byte(`CREATE TABLE IF NOT EXISTS gadgets (id BIGSERIAL PRIMARY KEY);`)},
+	"notes.md":         {Data: []byte("ignored")},
+}
+
+func testCtx(t *testing.T) context.Context {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
+	t.Cleanup(cancel)
+	return ctx
+}
+
+// The full path against a real server: connect, migrate, and confirm the schema
+// and the bookkeeping table both landed.
+func TestNewMigrated_AgainstRealPostgres(t *testing.T) {
+	ctx := testCtx(t)
+	dsn := pgtest.MustStartPostgres(ctx, t)
+
+	pool, err := pg.NewMigrated(ctx, dsn, migrations, pg.MigrateOptions{LockName: "golib-pg-integration"})
+	if err != nil {
+		t.Fatalf("NewMigrated: %v", err)
+	}
+	t.Cleanup(pool.Close)
+
+	for _, table := range []string{"widgets", "gadgets", "schema_migrations"} {
+		var exists bool
+		err := pool.QueryRow(ctx,
+			`SELECT EXISTS (SELECT 1 FROM information_schema.tables
+			                WHERE table_schema = 'public' AND table_name = $1)`, table).Scan(&exists)
+		if err != nil {
+			t.Fatalf("check %s: %v", table, err)
+		}
+		if !exists {
+			t.Errorf("table %s was not created", table)
+		}
+	}
+
+	// The second statement of the multi-statement file must have run too.
+	var idx bool
+	if err := pool.QueryRow(ctx,
+		`SELECT EXISTS (SELECT 1 FROM pg_indexes WHERE indexname = 'widgets_name_idx')`).Scan(&idx); err != nil {
+		t.Fatalf("check index: %v", err)
+	}
+	if !idx {
+		t.Error("the second statement of the multi-statement migration did not run")
+	}
+
+	var versions []string
+	rows, err := pool.Query(ctx, "SELECT version FROM schema_migrations ORDER BY version")
+	if err != nil {
+		t.Fatalf("read schema_migrations: %v", err)
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var v string
+		if err := rows.Scan(&v); err != nil {
+			t.Fatalf("scan: %v", err)
+		}
+		versions = append(versions, v)
+	}
+	if err := rows.Err(); err != nil {
+		t.Fatalf("rows: %v", err)
+	}
+	want := []string{"0001_widgets.sql", "0002_gadgets.sql"}
+	if len(versions) != len(want) {
+		t.Fatalf("recorded versions %v, want %v", versions, want)
+	}
+	for i := range want {
+		if versions[i] != want[i] {
+			t.Fatalf("recorded versions %v, want %v", versions, want)
+		}
+	}
+}
+
+// Every replica calls Migrate at startup. Running it concurrently against one
+// server must apply the set exactly once and leave no lock held.
+func TestMigrate_ConcurrentRepliesConverge(t *testing.T) {
+	ctx := testCtx(t)
+	dsn := pgtest.MustStartPostgres(ctx, t)
+
+	pool, err := pg.New(ctx, dsn, nil)
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	t.Cleanup(pool.Close)
+
+	const replicas = 4
+	errs := make(chan error, replicas)
+	for range replicas {
+		go func() {
+			errs <- pg.Migrate(ctx, pool, migrations, pg.MigrateOptions{LockName: "golib-pg-integration"})
+		}()
+	}
+	for range replicas {
+		if err := <-errs; err != nil {
+			t.Fatalf("Migrate: %v", err)
+		}
+	}
+
+	var n int
+	if err := pool.QueryRow(ctx, "SELECT count(*) FROM schema_migrations").Scan(&n); err != nil {
+		t.Fatalf("count versions: %v", err)
+	}
+	if n != 2 {
+		t.Fatalf("schema_migrations has %d rows, want 2", n)
+	}
+
+	// Nothing may still hold the migration lock once every replica has finished.
+	var held bool
+	if err := pool.QueryRow(ctx,
+		`SELECT EXISTS (SELECT 1 FROM pg_locks WHERE locktype = 'advisory' AND objid IS NOT NULL AND granted)`,
+	).Scan(&held); err != nil {
+		t.Fatalf("check locks: %v", err)
+	}
+	if held {
+		t.Error("an advisory lock is still held after every replica finished")
+	}
+}
+
+// DSNFromEnv's output must be something pgx can actually connect with.
+func TestDSNFromEnv_ConnectsToRealPostgres(t *testing.T) {
+	ctx := testCtx(t)
+	dsn := pgtest.MustStartPostgres(ctx, t)
+
+	t.Setenv("DATABASE_URL", "")
+	t.Setenv("APP_DATABASE_URL", "")
+
+	cfg, err := parseDSN(dsn)
+	if err != nil {
+		t.Fatalf("parse container DSN: %v", err)
+	}
+	t.Setenv("APP_DBHOST", cfg.host)
+	t.Setenv("APP_DBPORT", cfg.port)
+	t.Setenv("APP_DBUSER", cfg.user)
+	t.Setenv("APP_DBPASS", cfg.pass)
+	t.Setenv("APP_DBNAME", cfg.name)
+	t.Setenv("APP_DBSSL", "disable")
+
+	built, err := pg.DSNFromEnv("APP_")
+	if err != nil {
+		t.Fatalf("DSNFromEnv: %v", err)
+	}
+	pool, err := pg.New(ctx, built, nil)
+	if err != nil {
+		t.Fatalf("New(%q): %v", built, err)
+	}
+	pool.Close()
+}
+
+// dsnParts is the container DSN split back into the fields DSNFromEnv reads.
+type dsnParts struct{ host, port, user, pass, name string }
+
+func parseDSN(dsn string) (dsnParts, error) {
+	u, err := url.Parse(dsn)
+	if err != nil {
+		return dsnParts{}, err
+	}
+	pass, _ := u.User.Password()
+	return dsnParts{
+		host: u.Hostname(),
+		port: u.Port(),
+		user: u.User.Username(),
+		pass: pass,
+		name: strings.TrimPrefix(u.Path, "/"),
+	}, nil
+}

pg/migrate.go

@@ -0,0 +1,243 @@
+package pg
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"hash/fnv"
+	"io/fs"
+	"log/slog"
+	"sort"
+	"strings"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// schemaMigrationsDDL creates the version table itself, outside the tracked
+// set: it is step zero of every run and is never recorded as a migration.
+const schemaMigrationsDDL = `CREATE TABLE IF NOT EXISTS schema_migrations (
+    version    TEXT PRIMARY KEY,
+    applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
+)`
+
+// MigrateOptions configures one migration run.
+type MigrateOptions struct {
+	// LockName names the cluster-wide advisory lock replicas contend for.
+	// Every process migrating the same database must pass the same name, and
+	// two databases sharing a Postgres cluster should not: the lock is per
+	// cluster, not per database. Required.
+	LockName string
+
+	// Logger receives one line per applied migration. Nil discards them.
+	Logger *slog.Logger
+}
+
+// LockKey derives the pg_advisory_lock key for a lock name as FNV-1a/64 of the
+// name, reinterpreted as int64. It is exported so a service migrating off a
+// hardcoded key can assert the two agree before switching over.
+func LockKey(name string) int64 {
+	h := fnv.New64a()
+	// hash.Hash.Write never returns an error.
+	_, _ = h.Write([]byte(name))
+	return int64(h.Sum64())
+}
+
+// migrator is the slice of Postgres the migration runner drives. It keeps the
+// ordering, locking and bookkeeping logic testable without a live database.
+type migrator interface {
+	// Lock blocks until this process holds the cluster-wide migration lock.
+	Lock(ctx context.Context) error
+	// Unlock releases it.
+	Unlock(ctx context.Context) error
+	// Discard throws away the underlying session so a lock that could not be
+	// released dies with the connection instead of being returned to the pool.
+	Discard(ctx context.Context)
+	// EnsureVersionTable creates schema_migrations if it is missing.
+	EnsureVersionTable(ctx context.Context) error
+	// AppliedVersions returns the versions already recorded.
+	AppliedVersions(ctx context.Context) (map[string]bool, error)
+	// Apply runs one migration's SQL and records its version in a single
+	// transaction, so a failure leaves neither behind.
+	Apply(ctx context.Context, version, sql string) error
+}
+
+// migrationNames returns the .sql files in fsys in version (lexical) order.
+func migrationNames(fsys fs.FS) ([]string, error) {
+	entries, err := fs.ReadDir(fsys, ".")
+	if err != nil {
+		return nil, fmt.Errorf("read migrations: %w", err)
+	}
+	var names []string
+	for _, e := range entries {
+		if !e.IsDir() && strings.HasSuffix(e.Name(), ".sql") {
+			names = append(names, e.Name())
+		}
+	}
+	if len(names) == 0 {
+		return nil, errors.New("no migrations found")
+	}
+	sort.Strings(names)
+	return names, nil
+}
+
+// runMigrations applies every migration in fsys that schema_migrations does not
+// already record, in version order, while holding the advisory lock. Replicas
+// starting at the same time queue on the lock and then find nothing to do.
+//
+// A file absent from schema_migrations is re-run even if the live database
+// already has it, which is how a schema applied out of band before adopting
+// this runner is picked up. Migrations are therefore expected to be
+// IF NOT EXISTS-guarded, so such a re-run is a no-op that only lands the
+// missing tracking row.
+func runMigrations(ctx context.Context, m migrator, fsys fs.FS, log *slog.Logger) error {
+	names, err := migrationNames(fsys)
+	if err != nil {
+		return err
+	}
+	if err := m.Lock(ctx); err != nil {
+		return fmt.Errorf("acquire migration lock: %w", err)
+	}
+	defer func() {
+		if err := m.Unlock(ctx); err != nil {
+			// The lock is session-scoped: if the unlock did not land we cannot
+			// know the session dropped it, so kill the session rather than let a
+			// still-locked connection back into the pool.
+			log.Warn("release migration lock, discarding connection", "err", err)
+			m.Discard(ctx)
+		}
+	}()
+
+	if err := m.EnsureVersionTable(ctx); err != nil {
+		return fmt.Errorf("ensure schema_migrations: %w", err)
+	}
+	applied, err := m.AppliedVersions(ctx)
+	if err != nil {
+		return fmt.Errorf("read applied migrations: %w", err)
+	}
+
+	var n int
+	for _, name := range names {
+		if applied[name] {
+			continue
+		}
+		body, err := fs.ReadFile(fsys, name)
+		if err != nil {
+			return fmt.Errorf("read migration %s: %w", name, err)
+		}
+		if err := m.Apply(ctx, name, string(body)); err != nil {
+			return fmt.Errorf("apply migration %s: %w", name, err)
+		}
+		log.Info("applied migration", "version", name)
+		n++
+	}
+	if n == 0 {
+		log.Info("schema up to date")
+	}
+	return nil
+}
+
+// session is the slice of a pgx connection the migrator drives. *pgxpool.Conn
+// satisfies it; naming it keeps the SQL testable without a live database.
+type session interface {
+	Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
+	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
+	Begin(ctx context.Context) (pgx.Tx, error)
+}
+
+// pgMigrator runs migrations on one dedicated pooled connection: the advisory
+// lock is session-scoped, so lock, apply and unlock must share a connection.
+type pgMigrator struct {
+	conn session
+	key  int64
+	// discard closes the physical connection; the pool destroys a closed
+	// connection on Release instead of reusing it.
+	discard func(context.Context)
+}
+
+func (p *pgMigrator) Lock(ctx context.Context) error {
+	_, err := p.conn.Exec(ctx, "SELECT pg_advisory_lock($1)", p.key)
+	return err
+}
+
+func (p *pgMigrator) Unlock(ctx context.Context) error {
+	_, err := p.conn.Exec(ctx, "SELECT pg_advisory_unlock($1)", p.key)
+	return err
+}
+
+func (p *pgMigrator) Discard(ctx context.Context) { p.discard(ctx) }
+
+func (p *pgMigrator) EnsureVersionTable(ctx context.Context) error {
+	_, err := p.conn.Exec(ctx, schemaMigrationsDDL)
+	return err
+}
+
+func (p *pgMigrator) AppliedVersions(ctx context.Context) (map[string]bool, error) {
+	rows, err := p.conn.Query(ctx, "SELECT version FROM schema_migrations")
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	applied := map[string]bool{}
+	for rows.Next() {
+		var v string
+		if err := rows.Scan(&v); err != nil {
+			return nil, err
+		}
+		applied[v] = true
+	}
+	if err := rows.Err(); err != nil {
+		// A read that died part-way must not look like a complete set, or the
+		// caller would skip migrations it has not actually applied.
+		return nil, err
+	}
+	return applied, nil
+}
+
+func (p *pgMigrator) Apply(ctx context.Context, version, sql string) error {
+	tx, err := p.conn.Begin(ctx)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+	// Zero-arg Exec uses the simple protocol, so a multi-statement file runs.
+	if _, err := tx.Exec(ctx, sql); err != nil {
+		return err
+	}
+	if _, err := tx.Exec(ctx, "INSERT INTO schema_migrations (version) VALUES ($1)", version); err != nil {
+		return err
+	}
+	return tx.Commit(ctx)
+}
+
+// Migrate brings the database behind pool up to date with the .sql files at the
+// root of fsys, applied in lexical filename order. It is safe to call from
+// every replica at once: the run holds a cluster-wide advisory lock derived
+// from opts.LockName, and replicas that queue behind the winner find the set
+// already recorded and do nothing.
+//
+// Migrations run on one dedicated pooled connection, because the advisory lock
+// is session-scoped. Each file is applied together with its schema_migrations
+// row in a single transaction, so a failure part-way through leaves neither the
+// half-applied file nor a tracking row that would skip it next time.
+func Migrate(ctx context.Context, pool *pgxpool.Pool, fsys fs.FS, opts MigrateOptions) error {
+	if opts.LockName == "" {
+		return errors.New("pg: MigrateOptions.LockName is required")
+	}
+	conn, err := pool.Acquire(ctx)
+	if err != nil {
+		return fmt.Errorf("acquire migration connection: %w", err)
+	}
+	defer conn.Release()
+	return migrateSession(ctx, conn, func(ctx context.Context) { _ = conn.Conn().Close(ctx) }, fsys, opts)
+}
+
+// migrateSession runs the set on an already-acquired connection.
+func migrateSession(ctx context.Context, conn session, discard func(context.Context), fsys fs.FS, opts MigrateOptions) error {
+	m := &pgMigrator{conn: conn, key: LockKey(opts.LockName), discard: discard}
+	if err := runMigrations(ctx, m, fsys, logger(opts.Logger)); err != nil {
+		return fmt.Errorf("migrate schema: %w", err)
+	}
+	return nil
+}

pg/migrate_test.go

@@ -0,0 +1,428 @@
+package pg
+
+import (
+	"context"
+	"errors"
+	"hash/fnv"
+	"io"
+	"log/slog"
+	"reflect"
+	"strings"
+	"sync"
+	"testing"
+	"testing/fstest"
+)
+
+func testLogger() *slog.Logger {
+	return slog.New(slog.NewTextHandler(io.Discard, nil))
+}
+
+// testFS is a three-file migration set in deliberately unsorted map order.
+func testFS() fstest.MapFS {
+	return fstest.MapFS{
+		"0002_second.sql": {Data: []byte("CREATE TABLE IF NOT EXISTS b ();")},
+		"0001_first.sql":  {Data: []byte("CREATE TABLE IF NOT EXISTS a ();")},
+		"0003_third.sql":  {Data: []byte("CREATE TABLE IF NOT EXISTS c ();")},
+		"README.md":       {Data: []byte("not a migration")},
+	}
+}
+
+var testNames = []string{"0001_first.sql", "0002_second.sql", "0003_third.sql"}
+
+// fakeDB stands in for Postgres: the advisory lock is a mutex, the version
+// table an in-memory set, and a migration is "run" by recording its version.
+type fakeDB struct {
+	lock sync.Mutex // the advisory lock: one holder at a time, cluster-wide
+
+	mu           sync.Mutex
+	held         bool
+	tableCreated bool
+	applied      []string
+	applyCalls   []string
+	bodies       []string
+	failOn       string
+	unlockErr    error
+	discards     int
+	violations   []string
+}
+
+type fakeMigrator struct{ db *fakeDB }
+
+func (f *fakeMigrator) Lock(ctx context.Context) error {
+	f.db.lock.Lock()
+	f.db.mu.Lock()
+	defer f.db.mu.Unlock()
+	f.db.held = true
+	return nil
+}
+
+func (f *fakeMigrator) Unlock(ctx context.Context) error {
+	f.db.mu.Lock()
+	f.db.held = false
+	err := f.db.unlockErr
+	f.db.mu.Unlock()
+	f.db.lock.Unlock()
+	return err
+}
+
+func (f *fakeMigrator) Discard(ctx context.Context) {
+	f.db.mu.Lock()
+	defer f.db.mu.Unlock()
+	f.db.discards++
+}
+
+// requireHeld records any access made without the advisory lock; every schema
+// read or write must happen inside the locked section.
+func (f *fakeMigrator) requireHeld(op string) {
+	if !f.db.held {
+		f.db.violations = append(f.db.violations, op)
+	}
+}
+
+func (f *fakeMigrator) EnsureVersionTable(ctx context.Context) error {
+	f.db.mu.Lock()
+	defer f.db.mu.Unlock()
+	f.requireHeld("EnsureVersionTable")
+	f.db.tableCreated = true
+	return nil
+}
+
+func (f *fakeMigrator) AppliedVersions(ctx context.Context) (map[string]bool, error) {
+	f.db.mu.Lock()
+	defer f.db.mu.Unlock()
+	f.requireHeld("AppliedVersions")
+	if !f.db.tableCreated {
+		return nil, errors.New("schema_migrations does not exist")
+	}
+	out := map[string]bool{}
+	for _, v := range f.db.applied {
+		out[v] = true
+	}
+	return out, nil
+}
+
+func (f *fakeMigrator) Apply(ctx context.Context, version, sql string) error {
+	f.db.mu.Lock()
+	defer f.db.mu.Unlock()
+	f.requireHeld("Apply")
+	f.db.applyCalls = append(f.db.applyCalls, version)
+	f.db.bodies = append(f.db.bodies, sql)
+	if sql == "" {
+		return errors.New("empty migration body")
+	}
+	// A failing migration rolls back, so neither the SQL nor the version row lands.
+	if version == f.db.failOn {
+		return errors.New("boom")
+	}
+	f.db.applied = append(f.db.applied, version)
+	return nil
+}
+
+func (f *fakeDB) snapshot() (applied, calls, violations []string) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return append([]string(nil), f.applied...),
+		append([]string(nil), f.applyCalls...),
+		append([]string(nil), f.violations...)
+}
+
+func (f *fakeDB) discardCount() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.discards
+}
+
+// arrproxy pinned its advisory lock key as a literal derived from
+// FNV-1a/64("arrproxy-migrations"). LockKey must reproduce it exactly, or
+// adopting this package would silently stop excluding the old deployment.
+func TestLockKey_MatchesTheEstatesPinnedKey(t *testing.T) {
+	const arrproxyKey int64 = 7816645656172167846
+	if got := LockKey("arrproxy-migrations"); got != arrproxyKey {
+		t.Fatalf("LockKey(%q) = %d, want %d", "arrproxy-migrations", got, arrproxyKey)
+	}
+}
+
+func TestLockKey_IsFNV1a64(t *testing.T) {
+	for _, name := range []string{"", "encapi-migrations", "a much longer lock name"} {
+		h := fnv.New64a()
+		if _, err := h.Write([]byte(name)); err != nil {
+			t.Fatalf("hash: %v", err)
+		}
+		if got, want := LockKey(name), int64(h.Sum64()); got != want {
+			t.Errorf("LockKey(%q) = %d, want %d", name, got, want)
+		}
+	}
+}
+
+// Distinct names must not collide, or two services sharing a cluster would
+// serialise against each other by accident.
+func TestLockKey_DistinctNamesDistinctKeys(t *testing.T) {
+	if LockKey("encapi-migrations") == LockKey("artifactapi-migrations") {
+		t.Fatal("two different lock names produced the same key")
+	}
+}
+
+func TestMigrationNames_SortedSQLOnly(t *testing.T) {
+	got, err := migrationNames(testFS())
+	if err != nil {
+		t.Fatalf("migrationNames: %v", err)
+	}
+	if !reflect.DeepEqual(got, testNames) {
+		t.Fatalf("got %v, want %v", got, testNames)
+	}
+}
+
+func TestMigrationNames_IgnoresDirectories(t *testing.T) {
+	fsys := fstest.MapFS{
+		"0001_first.sql":     {Data: []byte("SELECT 1;")},
+		"sub.sql/keep.sql":   {Data: []byte("SELECT 1;")},
+		"0002_second.sql.gz": {Data: []byte("SELECT 1;")},
+	}
+	got, err := migrationNames(fsys)
+	if err != nil {
+		t.Fatalf("migrationNames: %v", err)
+	}
+	if want := []string{"0001_first.sql"}; !reflect.DeepEqual(got, want) {
+		t.Fatalf("got %v, want %v", got, want)
+	}
+}
+
+func TestMigrationNames_EmptySetIsAnError(t *testing.T) {
+	if _, err := migrationNames(fstest.MapFS{}); err == nil {
+		t.Fatal("expected an error for an empty migration set")
+	}
+}
+
+func TestMigrationNames_UnreadableFSIsAnError(t *testing.T) {
+	if _, err := migrationNames(brokenFS{}); err == nil {
+		t.Fatal("expected an error when the migration directory cannot be read")
+	}
+}
+
+func TestRunMigrations_FreshDatabaseAppliesAllInOrder(t *testing.T) {
+	db := &fakeDB{}
+	if err := runMigrations(context.Background(), &fakeMigrator{db: db}, testFS(), testLogger()); err != nil {
+		t.Fatalf("runMigrations: %v", err)
+	}
+	applied, _, violations := db.snapshot()
+	if !reflect.DeepEqual(applied, testNames) {
+		t.Fatalf("applied %v, want %v", applied, testNames)
+	}
+	if !db.tableCreated {
+		t.Error("schema_migrations was not created")
+	}
+	if len(violations) != 0 {
+		t.Errorf("schema accessed without the advisory lock: %v", violations)
+	}
+}
+
+// The runner must hand Apply the file's bytes, not just its name.
+func TestRunMigrations_PassesFileBodies(t *testing.T) {
+	db := &fakeDB{}
+	fsys := testFS()
+	if err := runMigrations(context.Background(), &fakeMigrator{db: db}, fsys, testLogger()); err != nil {
+		t.Fatalf("runMigrations: %v", err)
+	}
+	db.mu.Lock()
+	defer db.mu.Unlock()
+	for i, name := range testNames {
+		if want := string(fsys[name].Data); db.bodies[i] != want {
+			t.Errorf("body %d = %q, want %q", i, db.bodies[i], want)
+		}
+	}
+}
+
+func TestRunMigrations_SecondRunIsANoOp(t *testing.T) {
+	db := &fakeDB{}
+	ctx := context.Background()
+	for i := range 2 {
+		if err := runMigrations(ctx, &fakeMigrator{db: db}, testFS(), testLogger()); err != nil {
+			t.Fatalf("run %d: %v", i, err)
+		}
+	}
+	applied, calls, _ := db.snapshot()
+	if !reflect.DeepEqual(applied, testNames) {
+		t.Fatalf("applied %v, want %v", applied, testNames)
+	}
+	if len(calls) != len(testNames) {
+		t.Fatalf("Apply called %d times across two runs, want %d", len(calls), len(testNames))
+	}
+}
+
+// A file missing from schema_migrations is re-applied even when the rest of the
+// set is recorded: that is how a schema applied out of band is adopted.
+func TestRunMigrations_AppliesOnlyTheMissingVersion(t *testing.T) {
+	db := &fakeDB{tableCreated: true, applied: []string{testNames[0], testNames[2]}}
+	if err := runMigrations(context.Background(), &fakeMigrator{db: db}, testFS(), testLogger()); err != nil {
+		t.Fatalf("runMigrations: %v", err)
+	}
+	_, calls, _ := db.snapshot()
+	if want := testNames[1:2]; !reflect.DeepEqual(calls, want) {
+		t.Fatalf("Apply calls %v, want %v", calls, want)
+	}
+}
+
+// Replicas starting together queue on the advisory lock: the first applies the
+// set, the rest find it already recorded and do nothing.
+func TestRunMigrations_ConcurrentStartersSerialize(t *testing.T) {
+	db := &fakeDB{}
+	ctx := context.Background()
+	var wg sync.WaitGroup
+	errs := make([]error, 4)
+	for i := range errs {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			errs[i] = runMigrations(ctx, &fakeMigrator{db: db}, testFS(), testLogger())
+		}()
+	}
+	wg.Wait()
+	for i, err := range errs {
+		if err != nil {
+			t.Fatalf("starter %d: %v", i, err)
+		}
+	}
+	applied, calls, violations := db.snapshot()
+	if !reflect.DeepEqual(applied, testNames) {
+		t.Fatalf("applied %v, want %v", applied, testNames)
+	}
+	if len(calls) != len(testNames) {
+		t.Fatalf("Apply called %d times, want %d (one starter should have applied)", len(calls), len(testNames))
+	}
+	if len(violations) != 0 {
+		t.Errorf("schema accessed without the advisory lock: %v", violations)
+	}
+}
+
+// A migration that fails mid-set aborts the run, leaves earlier versions
+// recorded, and never reaches later ones.
+func TestRunMigrations_FailureStopsAndKeepsEarlierVersions(t *testing.T) {
+	db := &fakeDB{failOn: testNames[1]}
+	err := runMigrations(context.Background(), &fakeMigrator{db: db}, testFS(), testLogger())
+	if err == nil {
+		t.Fatal("expected the failing migration to abort the run")
+	}
+	if !strings.Contains(err.Error(), testNames[1]) {
+		t.Errorf("error %q does not name the failing migration", err)
+	}
+	applied, calls, _ := db.snapshot()
+	if !reflect.DeepEqual(applied, testNames[:1]) {
+		t.Fatalf("applied %v, want %v", applied, testNames[:1])
+	}
+	if !reflect.DeepEqual(calls, testNames[:2]) {
+		t.Fatalf("Apply calls %v, want %v (later migrations must not run)", calls, testNames[:2])
+	}
+	// The lock must be released even on failure, or every later pod deadlocks.
+	if !db.lock.TryLock() {
+		t.Fatal("advisory lock still held after a failed run")
+	}
+	db.lock.Unlock()
+}
+
+// A run that cannot take the lock must not touch the schema.
+func TestRunMigrations_LockFailureAborts(t *testing.T) {
+	db := &fakeDB{}
+	m := &failingLockMigrator{fakeMigrator{db: db}}
+	if err := runMigrations(context.Background(), m, testFS(), testLogger()); err == nil {
+		t.Fatal("expected a lock failure to abort the run")
+	}
+	applied, calls, _ := db.snapshot()
+	if len(applied) != 0 || len(calls) != 0 {
+		t.Fatalf("migrations ran without the lock: applied %v, calls %v", applied, calls)
+	}
+	if db.tableCreated {
+		t.Error("schema_migrations was created without the lock")
+	}
+}
+
+// A run that cannot see its bookkeeping table fails instead of re-applying blind.
+func TestRunMigrations_ReadAppliedFailureAborts(t *testing.T) {
+	// tableCreated stays false because EnsureVersionTable is a no-op here, so
+	// the fake's AppliedVersions reports the table as missing.
+	db := &fakeDB{}
+	m := &noTableMigrator{fakeMigrator{db: db}}
+	err := runMigrations(context.Background(), m, testFS(), testLogger())
+	if err == nil {
+		t.Fatal("expected the run to fail when applied versions cannot be read")
+	}
+	if !strings.Contains(err.Error(), "read applied migrations") {
+		t.Fatalf("error %q does not identify the failing step", err)
+	}
+	_, calls, _ := db.snapshot()
+	if len(calls) != 0 {
+		t.Fatalf("migrations applied without reading the version table: %v", calls)
+	}
+}
+
+func TestRunMigrations_EnsureVersionTableFailureAborts(t *testing.T) {
+	db := &fakeDB{}
+	m := &failingEnsureMigrator{fakeMigrator{db: db}}
+	err := runMigrations(context.Background(), m, testFS(), testLogger())
+	if err == nil {
+		t.Fatal("expected the run to fail when schema_migrations cannot be created")
+	}
+	if !strings.Contains(err.Error(), "ensure schema_migrations") {
+		t.Fatalf("error %q does not identify the failing step", err)
+	}
+	// Even this early failure must release the lock.
+	if !db.lock.TryLock() {
+		t.Fatal("advisory lock still held")
+	}
+	db.lock.Unlock()
+}
+
+func TestRunMigrations_UnreadableMigrationAborts(t *testing.T) {
+	db := &fakeDB{}
+	// A directory entry that ReadDir reports as a file but ReadFile cannot open.
+	err := runMigrations(context.Background(), &fakeMigrator{db: db}, missingFileFS{}, testLogger())
+	if err == nil {
+		t.Fatal("expected an unreadable migration to abort the run")
+	}
+	if !strings.Contains(err.Error(), "read migration") {
+		t.Fatalf("error %q does not identify the failing step", err)
+	}
+}
+
+// An unlock that does not land leaves a session that may still hold the lock, so
+// the connection is discarded rather than returned to the pool. The migrations
+// themselves already succeeded, so the run still reports success.
+func TestRunMigrations_UnlockFailureDiscardsConnection(t *testing.T) {
+	db := &fakeDB{unlockErr: errors.New("connection reset")}
+	if err := runMigrations(context.Background(), &fakeMigrator{db: db}, testFS(), testLogger()); err != nil {
+		t.Fatalf("runMigrations: %v", err)
+	}
+	if got := db.discardCount(); got != 1 {
+		t.Fatalf("Discard called %d times after a failed unlock, want 1", got)
+	}
+	applied, _, _ := db.snapshot()
+	if !reflect.DeepEqual(applied, testNames) {
+		t.Fatalf("applied %v, want %v", applied, testNames)
+	}
+}
+
+// A clean unlock keeps the connection: discarding every migration connection
+// would churn the pool on every start.
+func TestRunMigrations_CleanUnlockKeepsConnection(t *testing.T) {
+	db := &fakeDB{}
+	if err := runMigrations(context.Background(), &fakeMigrator{db: db}, testFS(), testLogger()); err != nil {
+		t.Fatalf("runMigrations: %v", err)
+	}
+	if got := db.discardCount(); got != 0 {
+		t.Fatalf("Discard called %d times after a clean unlock, want 0", got)
+	}
+}
+
+type failingLockMigrator struct{ fakeMigrator }
+
+func (f *failingLockMigrator) Lock(ctx context.Context) error { return errors.New("lock unavailable") }
+
+type failingEnsureMigrator struct{ fakeMigrator }
+
+func (f *failingEnsureMigrator) EnsureVersionTable(ctx context.Context) error {
+	return errors.New("permission denied")
+}
+
+type noTableMigrator struct{ fakeMigrator }
+
+// EnsureVersionTable silently does nothing, so AppliedVersions then fails.
+func (f *noTableMigrator) EnsureVersionTable(ctx context.Context) error { return nil }

pg/pg.go

@@ -0,0 +1,18 @@
+// Package pg holds the estate's shared Postgres plumbing: environment-driven
+// DSN construction, pgxpool construction, and the migration runner every
+// service uses to bring its own schema up to date at startup.
+package pg
+
+import (
+	"io"
+	"log/slog"
+)
+
+// logger returns log, or a logger that discards everything when log is nil, so
+// callers may leave the option unset.
+func logger(log *slog.Logger) *slog.Logger {
+	if log != nil {
+		return log
+	}
+	return slog.New(slog.NewTextHandler(io.Discard, nil))
+}

pg/pgmigrator_test.go

@@ -0,0 +1,302 @@
+package pg
+
+import (
+	"context"
+	"errors"
+	"reflect"
+	"strings"
+	"testing"
+)
+
+const testLockName = "golib-test-migrations"
+
+func newTestMigrator(s *fakeSession) (*pgMigrator, *int) {
+	discards := 0
+	m := &pgMigrator{
+		conn:    s,
+		key:     LockKey(testLockName),
+		discard: func(context.Context) { discards++ },
+	}
+	return m, &discards
+}
+
+// The lock and unlock must name the same key, and it must be the derived one:
+// a mismatch here is a deadlock or a lock nobody else respects.
+func TestPgMigrator_LockUnlockUseTheDerivedKey(t *testing.T) {
+	s := &fakeSession{}
+	m, _ := newTestMigrator(s)
+	ctx := context.Background()
+
+	if err := m.Lock(ctx); err != nil {
+		t.Fatalf("Lock: %v", err)
+	}
+	if err := m.Unlock(ctx); err != nil {
+		t.Fatalf("Unlock: %v", err)
+	}
+
+	want := []string{"SELECT pg_advisory_lock($1)", "SELECT pg_advisory_unlock($1)"}
+	if got := s.execSQL(); !reflect.DeepEqual(got, want) {
+		t.Fatalf("statements %v, want %v", got, want)
+	}
+	key := LockKey(testLockName)
+	for _, c := range s.execs {
+		if len(c.args) != 1 || c.args[0] != key {
+			t.Fatalf("%q got args %v, want [%d]", c.sql, c.args, key)
+		}
+	}
+}
+
+func TestPgMigrator_LockAndUnlockPropagateErrors(t *testing.T) {
+	lockErr := errors.New("lock failed")
+	unlockErr := errors.New("unlock failed")
+	s := &fakeSession{execErrs: map[string]error{
+		"SELECT pg_advisory_lock($1)":   lockErr,
+		"SELECT pg_advisory_unlock($1)": unlockErr,
+	}}
+	m, _ := newTestMigrator(s)
+	if err := m.Lock(context.Background()); !errors.Is(err, lockErr) {
+		t.Errorf("Lock error = %v, want %v", err, lockErr)
+	}
+	if err := m.Unlock(context.Background()); !errors.Is(err, unlockErr) {
+		t.Errorf("Unlock error = %v, want %v", err, unlockErr)
+	}
+}
+
+func TestPgMigrator_DiscardClosesTheSession(t *testing.T) {
+	m, discards := newTestMigrator(&fakeSession{})
+	m.Discard(context.Background())
+	if *discards != 1 {
+		t.Fatalf("discard called %d times, want 1", *discards)
+	}
+}
+
+func TestPgMigrator_EnsureVersionTableRunsTheDDL(t *testing.T) {
+	s := &fakeSession{}
+	m, _ := newTestMigrator(s)
+	if err := m.EnsureVersionTable(context.Background()); err != nil {
+		t.Fatalf("EnsureVersionTable: %v", err)
+	}
+	if got := s.execSQL(); !reflect.DeepEqual(got, []string{schemaMigrationsDDL}) {
+		t.Fatalf("statements %v, want the schema_migrations DDL", got)
+	}
+	// Re-running the DDL must be harmless, so it has to be IF NOT EXISTS.
+	if !strings.Contains(schemaMigrationsDDL, "IF NOT EXISTS") {
+		t.Error("schema_migrations DDL is not IF NOT EXISTS-guarded")
+	}
+}
+
+func TestPgMigrator_EnsureVersionTablePropagatesErrors(t *testing.T) {
+	want := errors.New("permission denied")
+	s := &fakeSession{execErrs: map[string]error{schemaMigrationsDDL: want}}
+	m, _ := newTestMigrator(s)
+	if err := m.EnsureVersionTable(context.Background()); !errors.Is(err, want) {
+		t.Fatalf("error = %v, want %v", err, want)
+	}
+}
+
+func TestPgMigrator_AppliedVersionsReadsAndClosesRows(t *testing.T) {
+	rows := &fakeRows{values: []string{"0001_first.sql", "0002_second.sql"}}
+	s := &fakeSession{rows: rows}
+	m, _ := newTestMigrator(s)
+
+	got, err := m.AppliedVersions(context.Background())
+	if err != nil {
+		t.Fatalf("AppliedVersions: %v", err)
+	}
+	want := map[string]bool{"0001_first.sql": true, "0002_second.sql": true}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("versions %v, want %v", got, want)
+	}
+	if !rows.closed {
+		t.Error("rows were not closed")
+	}
+	if wantQ := []string{"SELECT version FROM schema_migrations"}; !reflect.DeepEqual(s.queries, wantQ) {
+		t.Fatalf("queries %v, want %v", s.queries, wantQ)
+	}
+}
+
+func TestPgMigrator_AppliedVersionsEmptyTable(t *testing.T) {
+	s := &fakeSession{rows: &fakeRows{}}
+	m, _ := newTestMigrator(s)
+	got, err := m.AppliedVersions(context.Background())
+	if err != nil {
+		t.Fatalf("AppliedVersions: %v", err)
+	}
+	if len(got) != 0 {
+		t.Fatalf("versions %v, want an empty set", got)
+	}
+}
+
+func TestPgMigrator_AppliedVersionsErrors(t *testing.T) {
+	queryErr := errors.New("relation does not exist")
+	scanErr := errors.New("bad column type")
+	rowsErr := errors.New("connection reset mid-read")
+
+	tests := []struct {
+		name string
+		sess *fakeSession
+		want error
+	}{
+		{"query fails", &fakeSession{queryErr: queryErr}, queryErr},
+		{"scan fails", &fakeSession{rows: &fakeRows{values: []string{"x"}, scanErr: scanErr}}, scanErr},
+		// A read that dies part-way must not be reported as a complete set, or
+		// already-applied migrations would be re-run.
+		{"rows.Err after iteration", &fakeSession{rows: &fakeRows{values: []string{"x"}, err: rowsErr}}, rowsErr},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			m, _ := newTestMigrator(tc.sess)
+			got, err := m.AppliedVersions(context.Background())
+			if !errors.Is(err, tc.want) {
+				t.Fatalf("error = %v, want %v", err, tc.want)
+			}
+			if got != nil {
+				t.Fatalf("versions = %v, want nil on error", got)
+			}
+		})
+	}
+}
+
+// One transaction carries both the migration body and its tracking row, so a
+// crash can never leave the schema changed but unrecorded.
+func TestPgMigrator_ApplyRunsBodyAndVersionRowInOneTx(t *testing.T) {
+	const body = "CREATE TABLE IF NOT EXISTS a ();"
+	tx := &fakeTx{}
+	s := &fakeSession{tx: tx}
+	m, _ := newTestMigrator(s)
+
+	if err := m.Apply(context.Background(), "0001_first.sql", body); err != nil {
+		t.Fatalf("Apply: %v", err)
+	}
+	want := []string{body, "INSERT INTO schema_migrations (version) VALUES ($1)"}
+	if got := tx.execSQL(); !reflect.DeepEqual(got, want) {
+		t.Fatalf("statements %v, want %v", got, want)
+	}
+	// The body must be sent with no arguments so pgx uses the simple protocol
+	// and a multi-statement file runs.
+	if len(tx.execs[0].args) != 0 {
+		t.Errorf("migration body sent with args %v, want none", tx.execs[0].args)
+	}
+	if got := tx.execs[1].args; len(got) != 1 || got[0] != "0001_first.sql" {
+		t.Errorf("version row args %v, want [0001_first.sql]", got)
+	}
+	if !tx.committed {
+		t.Error("transaction was not committed")
+	}
+	if tx.rolled != 1 {
+		t.Errorf("Rollback called %d times, want 1 (the no-op deferred rollback)", tx.rolled)
+	}
+}
+
+func TestPgMigrator_ApplyRollsBackOnFailure(t *testing.T) {
+	const body = "CREATE TABLE oops ();"
+	bodyErr := errors.New("syntax error")
+	insertErr := errors.New("duplicate key")
+	const insert = "INSERT INTO schema_migrations (version) VALUES ($1)"
+
+	tests := []struct {
+		name      string
+		execErrs  map[string]error
+		commitErr error
+		want      error
+		wantExecs int
+	}{
+		{"body fails", map[string]error{body: bodyErr}, nil, bodyErr, 1},
+		{"version row fails", map[string]error{insert: insertErr}, nil, insertErr, 2},
+		{"commit fails", nil, errors.New("commit failed"), nil, 2},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			tx := &fakeTx{execErrs: tc.execErrs, commitErr: tc.commitErr}
+			s := &fakeSession{tx: tx}
+			m, _ := newTestMigrator(s)
+
+			err := m.Apply(context.Background(), "0001_first.sql", body)
+			if err == nil {
+				t.Fatal("expected Apply to fail")
+			}
+			if tc.want != nil && !errors.Is(err, tc.want) {
+				t.Fatalf("error = %v, want %v", err, tc.want)
+			}
+			if len(tx.execs) != tc.wantExecs {
+				t.Errorf("%d statements ran, want %d", len(tx.execs), tc.wantExecs)
+			}
+			if tx.committed {
+				t.Error("transaction was committed despite the failure")
+			}
+			if tx.rolled == 0 {
+				t.Error("transaction was not rolled back")
+			}
+		})
+	}
+}
+
+func TestPgMigrator_ApplyPropagatesBeginFailure(t *testing.T) {
+	want := errors.New("cannot start transaction")
+	m, _ := newTestMigrator(&fakeSession{beginErr: want})
+	if err := m.Apply(context.Background(), "0001_first.sql", "SELECT 1;"); !errors.Is(err, want) {
+		t.Fatalf("error = %v, want %v", err, want)
+	}
+}
+
+// migrateSession is the whole runner over a real pgMigrator: lock, DDL, read,
+// apply each file, unlock — in that order, on one session.
+func TestMigrateSession_FullRunOnOneSession(t *testing.T) {
+	tx := &fakeTx{}
+	s := &fakeSession{rows: &fakeRows{}, tx: tx}
+	discards := 0
+	err := migrateSession(context.Background(), s,
+		func(context.Context) { discards++ },
+		testFS(),
+		MigrateOptions{LockName: testLockName})
+	if err != nil {
+		t.Fatalf("migrateSession: %v", err)
+	}
+
+	want := []string{
+		"SELECT pg_advisory_lock($1)",
+		schemaMigrationsDDL,
+		"SELECT pg_advisory_unlock($1)",
+	}
+	if got := s.execSQL(); !reflect.DeepEqual(got, want) {
+		t.Fatalf("session statements %v, want %v", got, want)
+	}
+	// Every file's body plus its version row, all through the one fake tx.
+	if got, wantN := len(tx.execs), 2*len(testNames); got != wantN {
+		t.Fatalf("%d statements in transactions, want %d", got, wantN)
+	}
+	if discards != 0 {
+		t.Errorf("connection discarded %d times after a clean run, want 0", discards)
+	}
+}
+
+func TestMigrateSession_WrapsRunnerErrors(t *testing.T) {
+	s := &fakeSession{execErrs: map[string]error{"SELECT pg_advisory_lock($1)": errors.New("no lock")}}
+	err := migrateSession(context.Background(), s, func(context.Context) {}, testFS(),
+		MigrateOptions{LockName: testLockName})
+	if err == nil {
+		t.Fatal("expected an error")
+	}
+	if !strings.Contains(err.Error(), "migrate schema") {
+		t.Fatalf("error %q is not wrapped with the migration context", err)
+	}
+}
+
+// An unlock that fails must reach the discard func through the real migrator,
+// not just the fake one the runner tests use.
+func TestMigrateSession_UnlockFailureDiscards(t *testing.T) {
+	s := &fakeSession{
+		rows:     &fakeRows{},
+		tx:       &fakeTx{},
+		execErrs: map[string]error{"SELECT pg_advisory_unlock($1)": errors.New("gone")},
+	}
+	discards := 0
+	err := migrateSession(context.Background(), s, func(context.Context) { discards++ },
+		testFS(), MigrateOptions{LockName: testLockName})
+	if err != nil {
+		t.Fatalf("migrateSession: %v", err)
+	}
+	if discards != 1 {
+		t.Fatalf("connection discarded %d times, want 1", discards)
+	}
+}

pg/pgtest/pgtest.go

@@ -0,0 +1,100 @@
+// Package pgtest starts a throwaway Postgres container for integration-style
+// tests. Import it only from _test.go files so testcontainers never reaches a
+// production binary.
+//
+// The estate's CI runs on Kubernetes with no Docker socket, so tests that need
+// a container must skip themselves under -short. SkipIfShort does exactly that,
+// and StartPostgres reports a plain error when no container runtime is
+// reachable, leaving it to the caller whether that is a skip or a failure.
+package pgtest
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/testcontainers/testcontainers-go"
+	tcpostgres "github.com/testcontainers/testcontainers-go/modules/postgres"
+	"github.com/testcontainers/testcontainers-go/wait"
+)
+
+// Image is the Postgres the estate targets; every service runs the same major.
+const Image = "postgres:17-alpine"
+
+const (
+	database = "pgtest"
+	username = "pgtest"
+	password = "pgtest123"
+)
+
+// startTimeout bounds the container's readiness wait.
+const startTimeout = 60 * time.Second
+
+func init() {
+	// The Ryuk reaper container cannot start in every environment (rootless
+	// podman, restricted CI). Callers get an explicit terminate func instead,
+	// so disable Ryuk unless the environment has deliberately enabled it.
+	if _, ok := os.LookupEnv("TESTCONTAINERS_RYUK_DISABLED"); !ok {
+		_ = os.Setenv("TESTCONTAINERS_RYUK_DISABLED", "true")
+	}
+}
+
+// SkipIfShort skips the test under -short. Container-backed tests call it
+// first, which is what keeps `go test -short ./...` green with no Docker.
+func SkipIfShort(t *testing.T) {
+	t.Helper()
+	if testing.Short() {
+		t.Skip("skipping container-backed test in short mode")
+	}
+}
+
+// StartPostgres launches a Postgres container and returns its DSN plus a
+// terminate func the caller must run, even on failure paths.
+func StartPostgres(ctx context.Context) (dsn string, terminate func(), err error) {
+	c, err := tcpostgres.Run(ctx,
+		Image,
+		tcpostgres.WithDatabase(database),
+		tcpostgres.WithUsername(username),
+		tcpostgres.WithPassword(password),
+		testcontainers.WithWaitStrategy(
+			// Postgres opens the port, runs its init scripts, then restarts, so
+			// wait for the readiness log twice to avoid connection resets.
+			wait.ForLog("database system is ready to accept connections").
+				WithOccurrence(2).
+				WithStartupTimeout(startTimeout),
+		),
+	)
+	if err != nil {
+		return "", nil, fmt.Errorf("start postgres container: %w", err)
+	}
+	terminate = func() { _ = c.Terminate(ctx) }
+
+	host, err := c.Host(ctx)
+	if err != nil {
+		terminate()
+		return "", nil, fmt.Errorf("container host: %w", err)
+	}
+	port, err := c.MappedPort(ctx, "5432/tcp")
+	if err != nil {
+		terminate()
+		return "", nil, fmt.Errorf("container port: %w", err)
+	}
+	dsn = fmt.Sprintf("postgres://%s:%s@%s:%s/%s?sslmode=disable",
+		username, password, host, port.Port(), database)
+	return dsn, terminate, nil
+}
+
+// MustStartPostgres is StartPostgres for a TestMain-less test: it skips under
+// -short, fails the test if the container cannot start, and registers cleanup.
+func MustStartPostgres(ctx context.Context, t *testing.T) string {
+	t.Helper()
+	SkipIfShort(t)
+	dsn, terminate, err := StartPostgres(ctx)
+	if err != nil {
+		t.Fatalf("start postgres: %v", err)
+	}
+	t.Cleanup(terminate)
+	return dsn
+}

pg/pool.go

@@ -0,0 +1,44 @@
+package pg
+
+import (
+	"context"
+	"fmt"
+	"io/fs"
+	"log/slog"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// New opens a pgxpool against dsn and verifies it can reach the server before
+// returning. pgxpool connects lazily, so without the ping a bad address only
+// surfaces on the first query, long after startup has reported success.
+//
+// The caller owns the pool and must Close it. log may be nil.
+func New(ctx context.Context, dsn string, log *slog.Logger) (*pgxpool.Pool, error) {
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		return nil, fmt.Errorf("connect postgres: %w", err)
+	}
+	if err := pool.Ping(ctx); err != nil {
+		pool.Close()
+		return nil, fmt.Errorf("ping postgres: %w", err)
+	}
+	logger(log).Debug("postgres pool ready", "host", pool.Config().ConnConfig.Host)
+	return pool, nil
+}
+
+// NewMigrated opens a pool and brings the schema up to date before returning
+// it, so a service cannot start serving against a half-migrated database. A
+// migration failure closes the pool and returns the error; callers treat it as
+// fatal.
+func NewMigrated(ctx context.Context, dsn string, fsys fs.FS, opts MigrateOptions) (*pgxpool.Pool, error) {
+	pool, err := New(ctx, dsn, opts.Logger)
+	if err != nil {
+		return nil, err
+	}
+	if err := Migrate(ctx, pool, fsys, opts); err != nil {
+		pool.Close()
+		return nil, err
+	}
+	return pool, nil
+}

pg/pool_test.go

@@ -0,0 +1,114 @@
+package pg
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"testing/fstest"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// deadDSN points at a port nothing listens on, so connecting fails immediately
+// and locally: no container, no network, no waiting.
+const deadDSN = "postgres://u:p@127.0.0.1:1/d?sslmode=disable&connect_timeout=2"
+
+func shortCtx(t *testing.T) context.Context {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	t.Cleanup(cancel)
+	return ctx
+}
+
+func TestNew_RejectsAnUnparseableDSN(t *testing.T) {
+	pool, err := New(shortCtx(t), "://not a dsn", nil)
+	if err == nil {
+		pool.Close()
+		t.Fatal("expected an error for an unparseable DSN")
+	}
+	if !strings.Contains(err.Error(), "connect postgres") {
+		t.Fatalf("error %q does not identify the failing step", err)
+	}
+}
+
+// pgxpool connects lazily, so New must ping: without it a wrong address only
+// surfaces on the first query, long after startup reported success.
+func TestNew_PingsBeforeReturning(t *testing.T) {
+	pool, err := New(shortCtx(t), deadDSN, nil)
+	if err == nil {
+		pool.Close()
+		t.Fatal("expected New to fail against an unreachable server")
+	}
+	if !strings.Contains(err.Error(), "ping postgres") {
+		t.Fatalf("error %q does not identify the failing step", err)
+	}
+}
+
+func TestNewMigrated_PropagatesConnectFailure(t *testing.T) {
+	pool, err := NewMigrated(shortCtx(t), deadDSN, testFS(), MigrateOptions{LockName: testLockName})
+	if err == nil {
+		pool.Close()
+		t.Fatal("expected NewMigrated to fail against an unreachable server")
+	}
+	if !strings.Contains(err.Error(), "ping postgres") {
+		t.Fatalf("error %q does not identify the failing step", err)
+	}
+}
+
+// LockName is what makes replicas exclude each other; defaulting it would let a
+// caller silently share a key with an unrelated service, so it is required.
+func TestMigrate_RequiresALockName(t *testing.T) {
+	pool, err := pgxpool.New(shortCtx(t), deadDSN)
+	if err != nil {
+		t.Fatalf("pgxpool.New: %v", err)
+	}
+	t.Cleanup(pool.Close)
+
+	err = Migrate(shortCtx(t), pool, testFS(), MigrateOptions{})
+	if err == nil {
+		t.Fatal("expected Migrate to reject an empty LockName")
+	}
+	if !strings.Contains(err.Error(), "LockName") {
+		t.Fatalf("error %q does not name the missing option", err)
+	}
+}
+
+func TestMigrate_ReportsAcquireFailure(t *testing.T) {
+	pool, err := pgxpool.New(shortCtx(t), deadDSN)
+	if err != nil {
+		t.Fatalf("pgxpool.New: %v", err)
+	}
+	t.Cleanup(pool.Close)
+
+	err = Migrate(shortCtx(t), pool, testFS(), MigrateOptions{LockName: testLockName})
+	if err == nil {
+		t.Fatal("expected Migrate to fail when no connection can be acquired")
+	}
+	if !strings.Contains(err.Error(), "acquire migration connection") {
+		t.Fatalf("error %q does not identify the failing step", err)
+	}
+}
+
+// An empty migration set is a build mistake, not an empty schema; it must fail
+// before any connection work rather than reporting a successful no-op run.
+func TestMigrateSession_EmptySetIsAnError(t *testing.T) {
+	err := migrateSession(context.Background(), &fakeSession{}, func(context.Context) {},
+		fstest.MapFS{}, MigrateOptions{LockName: testLockName})
+	if err == nil {
+		t.Fatal("expected an empty migration set to be an error")
+	}
+}
+
+// A nil Logger is the common case for a library caller; it must not panic.
+func TestLogger_NilIsDiscarding(t *testing.T) {
+	if logger(nil) == nil {
+		t.Fatal("logger(nil) returned nil")
+	}
+	logger(nil).Info("this must not panic")
+
+	custom := testLogger()
+	if logger(custom) != custom {
+		t.Fatal("logger replaced the caller's logger")
+	}
+}