From d59dcbe74e751e9fd6fba9649cb9cdd6c54f2dfd Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Mon, 31 Aug 2026 22:19:40 +1000 Subject: [PATCH] Add the golib scaffold and the pg module MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stand up the shared library with its first module: the Postgres plumbing every service currently copy-pastes — the DSN builder, the pool constructor, and the migration runner arrproxy proved out. - Add pg.DSNFromEnv, generalising the identical Sprintf builders in encapi, artifactapi and forgebot into one prefixed lookup with DATABASE_URL passthrough and libpq fallbacks. - Add pg.New and pg.NewMigrated, which ping before returning so an unreachable server fails at startup rather than on the first query. - Add pg.Migrate, lifting arrproxy's runner verbatim in semantics and generalising the hardcoded advisory-lock key to FNV-1a/64 of a caller-supplied name and the embedded set to an fs.FS. - Add pg/pgtest, unifying the encapi and artifactapi testcontainers helpers, with SkipIfShort so container-backed tests self-skip on the Docker-less Kubernetes runners. - Add the Makefile, README and pre-commit config, plus test, pre-commit and build pipelines on golib-ci. --- .gitignore | 2 + .pre-commit-config.yaml | 27 +++ .woodpecker/build.yaml | 20 ++ .woodpecker/pre-commit.yaml | 49 +++++ .woodpecker/test.yaml | 20 ++ Makefile | 84 +++++++ README.md | 113 +++++++++- go.mod | 67 ++++++ go.sum | 158 +++++++++++++ pg/dsn.go | 125 +++++++++++ pg/dsn_test.go | 236 ++++++++++++++++++++ pg/fakes_test.go | 173 +++++++++++++++ pg/integration_test.go | 186 ++++++++++++++++ pg/migrate.go | 243 ++++++++++++++++++++ pg/migrate_test.go | 428 ++++++++++++++++++++++++++++++++++++ pg/pg.go | 18 ++ pg/pgmigrator_test.go | 302 +++++++++++++++++++++++++ pg/pgtest/pgtest.go | 100 +++++++++ pg/pool.go | 44 ++++ pg/pool_test.go | 114 ++++++++++ 20 files changed, 2508 insertions(+), 1 deletion(-) create mode 100644 .gitignore create mode 100644 .pre-commit-config.yaml create mode 100644 .woodpecker/build.yaml create mode 100644 .woodpecker/pre-commit.yaml create mode 100644 .woodpecker/test.yaml create mode 100644 Makefile create mode 100644 go.mod create mode 100644 go.sum create mode 100644 pg/dsn.go create mode 100644 pg/dsn_test.go create mode 100644 pg/fakes_test.go create mode 100644 pg/integration_test.go create mode 100644 pg/migrate.go create mode 100644 pg/migrate_test.go create mode 100644 pg/pg.go create mode 100644 pg/pgmigrator_test.go create mode 100644 pg/pgtest/pgtest.go create mode 100644 pg/pool.go create mode 100644 pg/pool_test.go diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b74f33b --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +cover.out +cover.filtered.out diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..a3de0cf --- /dev/null +++ b/.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 diff --git a/.woodpecker/build.yaml b/.woodpecker/build.yaml new file mode 100644 index 0000000..4fa1c3d --- /dev/null +++ b/.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 diff --git a/.woodpecker/pre-commit.yaml b/.woodpecker/pre-commit.yaml new file mode 100644 index 0000000..97bac70 --- /dev/null +++ b/.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 diff --git a/.woodpecker/test.yaml b/.woodpecker/test.yaml new file mode 100644 index 0000000..19a297d --- /dev/null +++ b/.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 diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..37ad41d --- /dev/null +++ b/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) diff --git a/README.md b/README.md index 4eeff11..29fa7a4 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,114 @@ # golib -Shared Go library for estate services: postgres, http service kit, vault and gitea clients \ No newline at end of file +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. `DATABASE_URL` — used verbatim. +2. `DATABASE_URL` — likewise. +3. `DBHOST`, `DBPORT`, `DBUSER`, `DBPASS`, + `DBNAME`, `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. diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..05682a4 --- /dev/null +++ b/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 +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..a16f418 --- /dev/null +++ b/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= diff --git a/pg/dsn.go b/pg/dsn.go new file mode 100644 index 0000000..a97321c --- /dev/null +++ b/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. DATABASE_URL — returned verbatim, no parsing or validation. +// 2. DATABASE_URL — likewise. Identical to 1 when prefix is empty. +// 3. DBHOST, DBPORT, DBUSER, DBPASS, +// DBNAME, 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 +} diff --git a/pg/dsn_test.go b/pg/dsn_test.go new file mode 100644 index 0000000..4aed4f5 --- /dev/null +++ b/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) + } +} diff --git a/pg/fakes_test.go b/pg/fakes_test.go new file mode 100644 index 0000000..13579b7 --- /dev/null +++ b/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 +} diff --git a/pg/integration_test.go b/pg/integration_test.go new file mode 100644 index 0000000..5895f37 --- /dev/null +++ b/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 +} diff --git a/pg/migrate.go b/pg/migrate.go new file mode 100644 index 0000000..a5c8083 --- /dev/null +++ b/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 +} diff --git a/pg/migrate_test.go b/pg/migrate_test.go new file mode 100644 index 0000000..ce6ba26 --- /dev/null +++ b/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 } diff --git a/pg/pg.go b/pg/pg.go new file mode 100644 index 0000000..7aa1a9d --- /dev/null +++ b/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)) +} diff --git a/pg/pgmigrator_test.go b/pg/pgmigrator_test.go new file mode 100644 index 0000000..6ce8b4c --- /dev/null +++ b/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) + } +} diff --git a/pg/pgtest/pgtest.go b/pg/pgtest/pgtest.go new file mode 100644 index 0000000..7c4b9c8 --- /dev/null +++ b/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 +} diff --git a/pg/pool.go b/pg/pool.go new file mode 100644 index 0000000..0e62efc --- /dev/null +++ b/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 +} diff --git a/pg/pool_test.go b/pg/pool_test.go new file mode 100644 index 0000000..3a9b996 --- /dev/null +++ b/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") + } +} -- 2.47.3