diff --git a/.gitea/workflows/ci.yaml b/.gitea/workflows/ci.yaml index c575a92..b836380 100644 --- a/.gitea/workflows/ci.yaml +++ b/.gitea/workflows/ci.yaml @@ -45,6 +45,29 @@ jobs: - go-mod-cache:/go/pkg/mod - go-build-cache:/root/.cache/go-build - gobin-cache:/go/bin + + # The suite needs a real Postgres -- there is no in-memory Postgres the way there was + # an in-memory SQLite, so each test gets its own schema on a shared server instead. + # The job and the service share the dind bridge, so the service is reachable by its + # name rather than on localhost. + services: + postgres: + image: postgres:17-alpine + env: + POSTGRES_USER: terdut + POSTGRES_PASSWORD: terdut + POSTGRES_DB: terdut_test + options: >- + --health-cmd "pg_isready -U terdut -d terdut_test" + --health-interval 5s + --health-timeout 5s + --health-retries 12 + + env: + # `make test` fails without this rather than skipping, so a green job here means + # the tests actually ran against a database. + TERDUT_TEST_DSN: postgres://terdut:terdut@postgres:5432/terdut_test?sslmode=disable + steps: - name: Checkout env: diff --git a/CLAUDE.md b/CLAUDE.md index 2d91a9d..734a9ee 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -22,10 +22,16 @@ Two things about this repo specifically: `Ryuvia/charts` to this version. Added 2026-09-02; every release up to and including v0.9.3 was published with no CVE check at all. - **The wrapper chart has two `tag:` lines** — the app image and the python backup sidecar — - so `chart-bump` needs `--image "$IMAGE"` to know which one moves. + so `chart-bump` needs `--image "$IMAGE"` to know which one moves. That sidecar backs up + SQLite; the Postgres move (#2) retires it in favour of a `postgresql` CR with a k8up + `pg_dump` annotation, after which only the app image's tag is left. ## Checks +The tests need a Postgres: `make test-db` starts one and prints the DSN, `make test-db-stop` +removes it, and `TERDUT_TEST_DSN` is how both the Makefile and `ci.yaml`'s service container +point the suite at it. Without it the suite fails rather than skipping, on purpose. + `make fmt lint test helm-lint` **is** what the pipeline runs — `ci.yaml` and `release.yaml` call these targets rather than restating them, the way riksdata and rd-web do. A green gate here and a green pipeline are the same code, not two descriptions of it. `test` adds `-race`, diff --git a/Makefile b/Makefile index feb57dd..59c5a12 100644 --- a/Makefile +++ b/Makefile @@ -25,12 +25,45 @@ help: ## Show this help # # These three mirror .gitea/workflows/ci.yaml step for step, so a green `make fmt # lint test` here means the same thing CI means. The one deliberate difference is -# -race below. +# -race below. Both need a Postgres to test against; see test-db. + +# The suite needs a Postgres, because the server does: there is no in-memory +# Postgres the way there was an in-memory SQLite. TERDUT_TEST_DSN says where, and +# the tests fail rather than skip without it — a suite that quietly tests nothing +# is worse than one that does not run. `make test-db` starts a local one; +# ci.yaml runs the same thing as a service container. +TEST_DB_CONTAINER ?= terdut-test-db +TEST_DB_PORT ?= 5433 +TEST_DB_IMAGE ?= docker.io/library/postgres:17-alpine +export TERDUT_TEST_DSN ?= postgres://terdut:terdut@localhost:$(TEST_DB_PORT)/terdut_test?sslmode=disable .PHONY: test -test: ## Run the test suite +test: ## Run the test suite (needs TERDUT_TEST_DSN; see test-db) go test -race ./... +# podman, with docker as the fallback: this is a dev convenience, not part of the +# pipeline, where the database arrives as a service container instead. +.PHONY: test-db +test-db: ## Start a local Postgres for the tests + @runtime=$$(command -v podman || command -v docker); \ + if [ -z "$$runtime" ]; then echo "need podman or docker"; exit 1; fi; \ + $$runtime run -d --rm --name $(TEST_DB_CONTAINER) \ + -e POSTGRES_USER=terdut -e POSTGRES_PASSWORD=terdut -e POSTGRES_DB=terdut_test \ + -p $(TEST_DB_PORT):5432 $(TEST_DB_IMAGE) >/dev/null; \ + printf 'waiting for postgres'; \ + for i in $$(seq 1 60); do \ + if $$runtime exec $(TEST_DB_CONTAINER) pg_isready -U terdut -d terdut_test >/dev/null 2>&1; then \ + echo " ready: $(TERDUT_TEST_DSN)"; exit 0; \ + fi; \ + printf '.'; sleep 1; \ + done; \ + echo " timed out"; exit 1 + +.PHONY: test-db-stop +test-db-stop: ## Stop the local test Postgres + @runtime=$$(command -v podman || command -v docker); \ + $$runtime rm -f $(TEST_DB_CONTAINER) >/dev/null 2>&1 || true + # CI runs a bare `go test ./...`. This is stricter on purpose: the sweeper, the # notifier goroutine and the deadman sweep all touch the same single-connection # database, and a race there would surface as a flaky production incident rather @@ -54,16 +87,22 @@ fmt: ## Report unformatted files echo "gofmt needed:"; echo "$$unformatted"; gofmt -d .; exit 1; \ fi +# database.dsnSecret.name has no default and the deployment `required`s it: the +# chart provisions no database and cannot guess where the credentials live, so a +# render without it is meant to fail. Setting it here keeps the lint honest about +# what a working install needs. +HELM_LINT_SET = --set image.tag=v0.0.0 --set database.dsnSecret.name=terdut-db + .PHONY: helm-lint helm-lint: ## Lint and render the chart - helm lint $(HELM_CHART) --set image.tag=v0.0.0 + helm lint $(HELM_CHART) $(HELM_LINT_SET) helm template terdut-server $(HELM_CHART) --namespace terdut-server \ - --set image.tag=v0.0.0 >/dev/null + $(HELM_LINT_SET) >/dev/null @# networking.listener defaults to "", which attaches the route to every @# matching listener including plaintext HTTP. Production sets it, so the @# default render proves nothing about the path that actually ships. helm template terdut-server $(HELM_CHART) --namespace terdut-server \ - --set image.tag=v0.0.0 --set networking.listener=https-terdut >/dev/null + $(HELM_LINT_SET) --set networking.listener=https-terdut >/dev/null ## --- release --- diff --git a/README.md b/README.md index 338a82b..18c3dcf 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,7 @@ Incident management server for teams using Prometheus Alertmanager. - Alert and incident statistics, including MTTA and MTTR - Web UI for phones and desktops, served by the same binary - REST API with per-user API key authentication -- Single binary, SQLite storage — trivial to self-host +- Single binary plus a Postgres — straightforward to self-host --- @@ -89,11 +89,14 @@ the web UI (`/incidents/{id}`). ```bash docker build -t terdut-server . -docker run -p 8080:8080 -v $(pwd)/data:/data \ - -e TERDUT_DB_PATH=/data/terdut.db \ +docker run -p 8080:8080 \ + -e TERDUT_DB_DSN='postgres://terdut:secret@host.docker.internal:5432/terdut?sslmode=disable' \ terdut-server ``` +The server creates its own schema on startup and needs a reachable Postgres; it stores nothing on +disk, so there is no volume to mount. + ### Kubernetes A Helm chart is published from this repository as an OCI artifact, versioned in lockstep @@ -116,24 +119,30 @@ than an `Ingress`. TLS is terminated at the gateway, so the server itself never | `networking.listener` | `""` | Gateway listener (`sectionName`) to bind to. Empty attaches to every matching listener, **including plaintext HTTP** — set it to the HTTPS listener's name to serve TLS only | | `networking.servicePort` | `8080` | Port the route forwards to; keep in sync with `service.port` | | `bootstrap.enabled` | `true` | Runs a post-install hook that creates the first user and stores its API key in the `-admin-key` Secret. Already-bootstrapped servers are left alone | -| `backupSidecar.enabled` | `true` | Adds an idle `python` sidecar and the [k8up](https://k8up.io/) annotations that dump the database through it | +| `database.dsnSecret.name` | `""` | **Required.** Existing Secret holding the Postgres DSN. The chart provisions no database | +| `database.dsnSecret.key` | `dsn` | Key within that Secret | The API key travels in an `Authorization: Bearer` header, so set `networking.listener` whenever the hostname is reachable outside a trusted network. +#### The database + +The chart provisions no database: it takes a DSN from a Secret and expects a Postgres that already +exists. In this cluster the wrapper chart declares an `acid.zalan.do/v1 postgresql` CR and passes +the Secret the operator writes; anywhere else, any reachable Postgres 14+ will do. + +The server migrates its own schema on startup, so a new database only has to exist and be writable. + #### Backups -The server image is `FROM scratch` — the binary and nothing else — so there is no interpreter to -run a database dump in, and the database runs in WAL mode, where a file-level copy of the volume is -not crash-consistent. The chart therefore ships an idle `python:*-alpine` sidecar that shares the -data volume, and points k8up's `backupcommand` at it with `k8up.io/backupcommand-container`. Without -that annotation k8up execs into `.spec.containers[0]` and the dump fails. +Postgres is backed up where it runs, not from here. The database pod carries a +[k8up](https://k8up.io/) `k8up.io/backupcommand` annotation that streams a `pg_dump`, the same way +gitea and immich do in this cluster. -The dump is buffered and sanity-checked before its first byte reaches stdout, because k8up streams -stdout straight into Restic: a dump that dies partway is otherwise stored as a silently truncated -snapshot that k8up still reports as successful. - -Set `backupSidecar.enabled=false` if you back the volume up some other way. +This used to be the app's problem: the SQLite database lived on a PVC beside the server, the image +is `FROM scratch` with no interpreter to dump it, and WAL mode makes a file-level copy of the volume +non-crash-consistent — so the chart shipped an idle `python:*-alpine` sidecar purely to give k8up +somewhere to exec. The sidecar, the PVC and the `backupSidecar` values are all gone. --- @@ -142,7 +151,7 @@ Set `backupSidecar.enabled=false` if you back the volume up some other way. | Variable | Default | Description | |---|---|---| | `TERDUT_ADDR` | `:8080` | TCP address to listen on | -| `TERDUT_DB_PATH` | `terdut.db` | Path to the SQLite database file | +| `TERDUT_DB_DSN` | — | **Required.** Postgres connection string, e.g. `postgres://terdut:secret@localhost:5432/terdut?sslmode=require` | | `TERDUT_ARCHIVE_AFTER` | `168h` (7d) | How long a resolved alert or incident stays in the default list before being auto-archived | | `TERDUT_STALE_AFTER` | `6h` | How long a firing alert may go without a refreshing webhook before it is treated as resolved — **must exceed your Alertmanager `repeat_interval`** | | `TERDUT_DEADMAN_MATCHERS` | `alertname=Watchdog` | Which alerts are [dead man's switches](#dead-mans-switch). `;` separates matchers, `,` the label conditions within one, `=` is exact equality. Every matcher must name an `alertname` | @@ -662,6 +671,33 @@ averages over incidents that have actually been acknowledged or resolved, and ar --- +## Upgrading from SQLite + +Versions up to v0.10.2 stored everything in a SQLite file. From the Postgres release onwards +the server needs `TERDUT_DB_DSN` and keeps nothing on disk. + +The cutover is ordered — the server must not be running while the copy happens: + +```bash +# 1. Stop the old server, keeping its database file. +# 2. Create an empty Postgres database, then let the new binary build the schema: +TERDUT_DB_DSN='postgres://terdut:secret@localhost:5432/terdut?sslmode=disable' ./terdut & +# ...watch for "listening on", then stop it again. +# 3. Copy the data across: +go run -tags migrate ./scripts/sqlite-to-postgres.go \ + -sqlite /data/terdut.db \ + -dsn 'postgres://terdut:secret@localhost:5432/terdut?sslmode=disable' +# 4. Start the new server for good. +``` + +The copy preserves every id, so incidents keep their numbers and the timeline, alert +membership, outbox and ack tokens all still point where they did. It refuses a target that +already has rows, so a second run cannot double-insert. On Kubernetes, step 3 runs as a Job +with the same image against the PVC before it is removed. + +The script is deliberately temporary: it is the only thing left that needs the SQLite driver, +and both should be deleted once the installs that need them have migrated. + ## Upgrading to incidents The incidents release moves the workflow off alerts, which is a **breaking API @@ -712,15 +748,23 @@ There is no migration and no schema change. An existing open incident from a ## Development ```bash -go test ./... # run all tests +make test-db # start a local Postgres for the tests (podman or docker) +make test # run all tests go build ./... # compile all packages -go run ./cmd/terdut # run locally +go run ./cmd/terdut # run locally (needs TERDUT_DB_DSN) ``` +The tests need a real Postgres, because the server does — there is no in-memory Postgres the +way there was an in-memory SQLite. `TERDUT_TEST_DSN` says where it is, `make test-db` starts +one on port 5433 and prints the DSN, and `make test-db-stop` removes it. Each test gets its +own schema on that server, so tests cannot see each other's rows. An unset `TERDUT_TEST_DSN` +fails the suite rather than skipping it: a run that quietly tests nothing is worse than one +that does not run. + `make fmt lint test helm-lint` is the gate. It mirrors `.gitea/workflows/ci.yaml` step for step, so a green run here means a green pipeline — with one deliberate exception: `make test` adds `-race`, which CI does not. The sweeper, the notifier goroutine and the dead man's switch -sweep all touch the same single database connection, and a race between them would surface as +sweep all run concurrently against the same database, and a race between them would surface as a flaky incident in production rather than as a red build. The web UI lives in `internal/web/static/` as plain HTML, CSS and ES modules, @@ -776,7 +820,9 @@ Two things the release process needs to know about this repo: do not bump the wrapper chart to that version — it cannot unpublish anything. The image is `FROM scratch`, so trivy sees exactly one target, the Go binary and its module graph. - **The wrapper chart's `values.yaml` has two `tag:` lines** — the app image and the python - backup sidecar — so `chart-bump` is given `--image` to say which one moves. + backup sidecar — so `chart-bump` is given `--image` to say which one moves. The sidecar is + on its way out with SQLite: once the wrapper chart drops it and declares a `postgresql` CR + instead, there is one `tag:` line again, and `--image` becomes belt and braces. The wrapper chart must have **its own `version:` bumped in the same commit**. Flux reconciles with `reconcileStrategy: ChartVersion`, so a chart whose version did not change produces no diff --git a/charts/terdut-server/templates/deployment.yaml b/charts/terdut-server/templates/deployment.yaml index e3b6d62..d73326c 100644 --- a/charts/terdut-server/templates/deployment.yaml +++ b/charts/terdut-server/templates/deployment.yaml @@ -10,50 +10,15 @@ spec: selector: matchLabels: {{- include "terdut-server.selectorLabels" . | nindent 6 }} - # The data PVC is ReadWriteOnce, so a RollingUpdate deadlocks: the new pod - # cannot attach the volume until the old one releases it, and the old one is - # not torn down until the new one is ready. + # Recreate, not RollingUpdate, even though the PVC that forced it is gone: the + # sweeper and the notifier are unsynchronised singletons, and two replicas + # overlapping during a rollout would both page for the same incident. strategy: type: Recreate template: metadata: labels: {{- include "terdut-server.selectorLabels" . | nindent 8 }} - {{- if .Values.backupSidecar.enabled }} - annotations: - # Dumps the whole database: incidents, alerts, users, API key hashes, - # the schedule and the notification outbox. - # - # Runs in the `backup` sidecar, NOT in the app container: the server - # image is FROM scratch and has no interpreter at all. k8up execs into - # .spec.containers[0] unless told otherwise, hence the explicit - # k8up.io/backupcommand-container. - # - # Buffered and sanity-checked before the first byte reaches stdout: k8up - # streams stdout straight into restic, so a dump that dies partway is - # stored as a silently-truncated snapshot that k8up still reports as - # Succeeded. The check counts users rather than incidents -- incidents - # are swept and archived, so an empty incidents table is a legitimate - # state, whereas a database with no users never is. - # - # The connection is read-only but the mount is not: the database runs in - # WAL mode, and opening it mode=ro still needs write access to the -shm - # wal-index. - # - # chr(10), not '\n': k8up parses this annotation with go-shellquote. - k8up.io/backupcommand-container: backup - k8up.io/backupcommand: >- - python3 -c "import sqlite3, sys; - con = sqlite3.connect('file:/data/terdut.db?mode=ro', uri=True); - con.execute('BEGIN'); - users = con.execute('SELECT count(*) FROM users').fetchone()[0]; - out = chr(10).join(con.iterdump()) + chr(10); - (users > 0 and out.rstrip().endswith('COMMIT;')) - or sys.exit('terdut: db dump failed sanity checks'); - sys.stdout.write(out)" - k8up.io/file-extension: ".sql" - k8up.io/backup: "true" - {{- end }} spec: enableServiceLinks: false containers: @@ -67,8 +32,14 @@ spec: env: - name: TERDUT_ADDR value: ":{{ .Values.service.port }}" - - name: TERDUT_DB_PATH - value: "/data/terdut.db" + # The connection string, from a Secret: it carries the password. + # The wrapper chart points this at the Secret the Postgres operator + # writes for this database's role. + - name: TERDUT_DB_DSN + valueFrom: + secretKeyRef: + name: {{ required "database.dsnSecret.name is required" .Values.database.dsnSecret.name }} + key: {{ .Values.database.dsnSecret.key }} - name: TERDUT_STALE_AFTER value: "{{ .Values.sweeper.staleAfter }}" - name: TERDUT_ARCHIVE_AFTER @@ -96,9 +67,6 @@ spec: key: {{ .Values.notify.tokenSecret.key }} {{- end }} {{- end }} - volumeMounts: - - name: data - mountPath: /data livenessProbe: httpGet: path: /healthz @@ -110,25 +78,4 @@ spec: port: http initialDelaySeconds: 5 - {{- if .Values.backupSidecar.enabled }} - # Idle sidecar. It exists only so k8up has a container with a sqlite3 - # module to exec the backupcommand in. Mounted read-write on purpose: - # see the note on the backupcommand annotation above. - - name: backup - image: "{{ .Values.backupSidecar.image.repository }}:{{ .Values.backupSidecar.image.tag }}" - imagePullPolicy: {{ .Values.backupSidecar.image.pullPolicy }} - command: ["sleep", "infinity"] - volumeMounts: - - name: data - mountPath: /data - resources: - requests: - memory: "16Mi" - cpu: "10m" - limits: - memory: "64Mi" - {{- end }} - volumes: - - name: data - persistentVolumeClaim: - claimName: {{ .Release.Name }}-data + diff --git a/charts/terdut-server/templates/pvc.yaml b/charts/terdut-server/templates/pvc.yaml deleted file mode 100644 index 54ee5af..0000000 --- a/charts/terdut-server/templates/pvc.yaml +++ /dev/null @@ -1,13 +0,0 @@ ---- -apiVersion: v1 -kind: PersistentVolumeClaim -metadata: - name: {{ .Release.Name }}-data - namespace: {{ .Release.Namespace }} -spec: - storageClassName: {{ .Values.storage.storageClass | quote }} - accessModes: - - ReadWriteOnce - resources: - requests: - storage: {{ .Values.storage.size }} diff --git a/charts/terdut-server/values.yaml b/charts/terdut-server/values.yaml index 301d892..81f3f0a 100644 --- a/charts/terdut-server/values.yaml +++ b/charts/terdut-server/values.yaml @@ -11,9 +11,16 @@ image: tag: "latest" pullPolicy: IfNotPresent -storage: - size: 1Gi - storageClass: synology-iscsi +# Postgres connection, as a DSN in an existing Secret: +# postgres://user:password@host:5432/terdut?sslmode=require +# +# The chart provisions no database. In this cluster the wrapper chart declares an +# acid.zalan.do postgresql CR and points this at the Secret the operator writes; +# anywhere else, any reachable Postgres will do. +database: + dsnSecret: + name: "" + key: dsn service: type: ClusterIP @@ -88,17 +95,10 @@ notify: name: "" key: token -# The server image is FROM scratch — just the binary, with no shell, no sqlite3 -# and no python — so a k8up backupcommand cannot run in the app container. This -# idle sidecar shares the data volume and is selected with -# k8up.io/backupcommand-container. Only the stdlib sqlite3 module is used, so any -# python image works. -backupSidecar: - enabled: true - image: - repository: python - tag: "3.13-alpine" - pullPolicy: IfNotPresent +# Backups are no longer this chart's business. The SQLite database lived on a PVC +# beside the app, so it needed a sidecar with a sqlite3 module for k8up to exec a +# dump in; Postgres is backed up where it runs, through a k8up.io/backupcommand +# pg_dump annotation on the database pod itself. bootstrap: enabled: true diff --git a/cmd/terdut/main.go b/cmd/terdut/main.go index 2d14050..b6f2fa4 100644 --- a/cmd/terdut/main.go +++ b/cmd/terdut/main.go @@ -18,7 +18,7 @@ var version = "dev" func main() { cfg := config.Load() - database, err := db.Open(cfg.DBPath) + database, err := db.Open(cfg.DSN) if err != nil { log.Fatalf("open db: %v", err) } diff --git a/go.mod b/go.mod index 107dfe2..c0f55df 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,8 @@ go 1.25.9 require ( github.com/go-chi/chi/v5 v5.2.5 + github.com/jackc/pgerrcode v0.0.0-20250907135507-afb5586c32a6 + github.com/jackc/pgx/v5 v5.11.0 golang.org/x/crypto v0.55.0 modernc.org/sqlite v1.50.1 ) @@ -11,10 +13,15 @@ require ( require ( github.com/dustin/go-humanize v1.0.1 // 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/mattn/go-isatty v0.0.20 // indirect github.com/ncruces/go-strftime v1.0.0 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect + golang.org/x/sync v0.22.0 // indirect golang.org/x/sys v0.47.0 // indirect + golang.org/x/text v0.41.0 // indirect modernc.org/libc v1.72.3 // indirect modernc.org/mathutil v1.7.1 // indirect modernc.org/memory v1.11.0 // indirect diff --git a/go.sum b/go.sum index 8acc724..830fa1b 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,6 @@ +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/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= github.com/go-chi/chi/v5 v5.2.5 h1:Eg4myHZBjyvJmAFjFvWgrqDTXFyOzjj7YIm3L3mu6Ug= @@ -8,23 +11,46 @@ 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/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= +github.com/jackc/pgerrcode v0.0.0-20250907135507-afb5586c32a6 h1:D/V0gu4zQ3cL2WKeVNVM4r2gLxGGf6McLwgXzRTo2RQ= +github.com/jackc/pgerrcode v0.0.0-20250907135507-afb5586c32a6/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds= +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.11.0 h1:IzBBtyK9AHqf98cctWFifYSci2hgQR/cd56wB4p+ogg= +github.com/jackc/pgx/v5 v5.11.0/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/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= +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/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +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= golang.org/x/crypto v0.55.0 h1:+KWHjbgOaAQ66dh/YlkZKHlz9ZUlq61AFirAR9ntP8M= golang.org/x/crypto v0.55.0/go.mod h1:uq0V9dE/fzQuJtbnL+2EhWOE63vo164FY8xqEnV9xis= -golang.org/x/mod v0.33.0 h1:tHFzIWbBifEmbwtGz65eaWyGiGZatSrT9prnU8DbVL8= -golang.org/x/mod v0.33.0/go.mod h1:swjeQEj+6r7fODbD2cqrnje9PnziFuw4bmLbBZFrQ5w= -golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= -golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/mod v0.38.0 h1:MECBjubtXD7yj4HrhIUcywNaGeNVUdfVnxmPajOk4yk= +golang.org/x/mod v0.38.0/go.mod h1:V6Xz0pq8TQ3dGqVQ1FVHuelZpAL0uNhSkk9ogYP3c40= +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.6.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/tools v0.42.0 h1:uNgphsn75Tdz5Ji2q36v/nsFSfR/9BRFvqhGBaJGd5k= -golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0= +golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8= +golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M= +golang.org/x/tools v0.48.0 h1:3+hClM1aLL5mjMKm5ovokw9epgRXPuu2tILgismM6RE= +golang.org/x/tools v0.48.0/go.mod h1:08xX0orndb/F7jJxGDicx061tyd5pcMto75YMAXr6lk= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +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= modernc.org/cc/v4 v4.28.2 h1:3tQ0lf2ADtoby2EtSP+J7IE2SHwEJdP8ioR59wx7XpY= modernc.org/cc/v4 v4.28.2/go.mod h1:OnovgIhbbMXMu1aISnJ0wvVD1KnW+cAUJkIrAWh+kVI= modernc.org/ccgo/v4 v4.34.0 h1:yRLPFZieg532OT4rp4JFNIVcquwalMX26G95WQDqwCQ= diff --git a/internal/api/alertmanager.go b/internal/api/alertmanager.go index 4339a7b..2190e60 100644 --- a/internal/api/alertmanager.go +++ b/internal/api/alertmanager.go @@ -171,7 +171,7 @@ func upsertAlerts(ctx context.Context, tx *sql.Tx, deadman DeadmanConfig, alerts var prevStartsAt int64 existed := true switch err := tx.QueryRowContext(ctx, - "SELECT status, starts_at FROM alerts WHERE fingerprint = ?", a.Fingerprint, + "SELECT status, starts_at FROM alerts WHERE fingerprint = $1", a.Fingerprint, ).Scan(&prevStatus, &prevStartsAt); { case err == sql.ErrNoRows: existed = false @@ -212,8 +212,8 @@ func upsertAlerts(ctx context.Context, tx *sql.Tx, deadman DeadmanConfig, alerts INSERT INTO alerts (fingerprint, name, status, labels, annotations, starts_at, ends_at, generator_url, received_at, resolution_source) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(fingerprint) DO UPDATE SET + VALUES ($1, $2, $3, $4::jsonb, $5::jsonb, $6, $7, $8, $9, $10) + ON CONFLICT (fingerprint) DO UPDATE SET status = excluded.status, labels = excluded.labels, annotations = excluded.annotations, @@ -245,7 +245,7 @@ func upsertAlerts(ctx context.Context, tx *sql.Tx, deadman DeadmanConfig, alerts var curStatus string var curStartsAt int64 if err := tx.QueryRowContext(ctx, - "SELECT id, status, starts_at FROM alerts WHERE fingerprint = ?", a.Fingerprint, + "SELECT id, status, starts_at FROM alerts WHERE fingerprint = $1", a.Fingerprint, ).Scan(&id, &curStatus, &curStartsAt); err != nil { return nil, err } @@ -315,7 +315,7 @@ func incidentForGroup(ctx context.Context, tx *sql.Tx, notify NotifyConfig, payl var id int64 switch err := tx.QueryRowContext(ctx, - "SELECT id FROM incidents WHERE group_key = ? AND resolved_at IS NULL", groupKey, + "SELECT id FROM incidents WHERE group_key = $1 AND resolved_at IS NULL", groupKey, ).Scan(&id); { case err == nil: return id, nil @@ -349,15 +349,13 @@ func openIncident(ctx context.Context, q querier, notify NotifyConfig, groupKey, labelsJSON = []byte("{}") } - res, err := q.ExecContext(ctx, ` + var id int64 + err = q.QueryRowContext(ctx, ` INSERT INTO incidents (group_key, title, group_labels, status, severity, triggered_at, assigned_to) - VALUES (?, ?, ?, 'triggered', ?, ?, ?)`, + VALUES ($1, $2, $3::jsonb, 'triggered', $4, $5, $6) + RETURNING id`, groupKey, title, string(labelsJSON), severity, - time.Now().Unix(), onCall) - if err != nil { - return 0, err - } - id, err := res.LastInsertId() + time.Now().Unix(), onCall).Scan(&id) if err != nil { return 0, err } @@ -385,8 +383,9 @@ func openIncident(ctx context.Context, q querier, notify NotifyConfig, groupKey, // first time. Re-sends of an already-linked alert are silent. func linkAlert(ctx context.Context, tx *sql.Tx, incidentID, alertID int64) error { res, err := tx.ExecContext(ctx, ` - INSERT OR IGNORE INTO incident_alerts (incident_id, alert_id, added_at) - VALUES (?, ?, ?)`, incidentID, alertID, time.Now().Unix()) + INSERT INTO incident_alerts (incident_id, alert_id, added_at) + VALUES ($1, $2, $3) + ON CONFLICT (incident_id, alert_id) DO NOTHING`, incidentID, alertID, time.Now().Unix()) if err != nil { return err } diff --git a/internal/api/alerts.go b/internal/api/alerts.go index 1e44099..f9d666a 100644 --- a/internal/api/alerts.go +++ b/internal/api/alerts.go @@ -36,15 +36,13 @@ func handleListAlerts(db *sql.DB) http.HandlerFunc { q := r.URL.Query() where := []string{} - args := []any{} + args := &sqlArgs{} if status := q.Get("status"); status != "" { - where = append(where, "a.status = ?") - args = append(args, status) + where = append(where, "a.status = "+args.add(status)) } if name := q.Get("name"); name != "" { - where = append(where, "a.name = ?") - args = append(args, name) + where = append(where, "a.name = "+args.add(name)) } if archived := q.Get("archived"); archived == "true" { where = append(where, "a.archived_at IS NOT NULL") @@ -53,21 +51,18 @@ func handleListAlerts(db *sql.DB) http.HandlerFunc { } if incidentID := q.Get("incident_id"); incidentID != "" { if n, err := strconv.ParseInt(incidentID, 10, 64); err == nil { - where = append(where, "a.id IN (SELECT alert_id FROM incident_alerts WHERE incident_id = ?)") - args = append(args, n) + where = append(where, "a.id IN (SELECT alert_id FROM incident_alerts WHERE incident_id = "+args.add(n)+")") } } if from := q.Get("from"); from != "" { if t, err := time.Parse("2006-01-02", from); err == nil { - where = append(where, "a.received_at >= ?") - args = append(args, t.UTC().Unix()) + where = append(where, "a.received_at >= "+args.add(t.UTC().Unix())) } } if to := q.Get("to"); to != "" { if t, err := time.Parse("2006-01-02", to); err == nil { - where = append(where, "a.received_at < ?") - args = append(args, t.UTC().AddDate(0, 0, 1).Unix()) + where = append(where, "a.received_at < "+args.add(t.UTC().AddDate(0, 0, 1).Unix())) } } @@ -82,11 +77,10 @@ func handleListAlerts(db *sql.DB) http.HandlerFunc { if len(where) > 0 { clause = strings.Join(where, " AND ") } - args = append(args, limit) rows, err := db.QueryContext(r.Context(), - fmt.Sprintf("%s WHERE %s ORDER BY a.received_at DESC LIMIT ?", alertSelectFrom, clause), - args...) + fmt.Sprintf("%s WHERE %s ORDER BY a.received_at DESC LIMIT %s", alertSelectFrom, clause, args.add(limit)), + args.all()...) if err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return @@ -128,7 +122,7 @@ func handleGetAlert(db *sql.DB) http.HandlerFunc { // fetchAlert loads a single alert by ID using the shared query. func fetchAlert(ctx context.Context, db *sql.DB, id int64) (models.Alert, error) { - return scanAlert(db.QueryRowContext(ctx, alertSelectFrom+" WHERE a.id = ?", id)) + return scanAlert(db.QueryRowContext(ctx, alertSelectFrom+" WHERE a.id = $1", id)) } // scanner is satisfied by both *sql.Row and *sql.Rows. diff --git a/internal/api/api_test.go b/internal/api/api_test.go index 0e0334d..5092b69 100644 --- a/internal/api/api_test.go +++ b/internal/api/api_test.go @@ -13,7 +13,6 @@ import ( "time" "git.ryuvia.com/niklas/terdut-server/internal/api" - "git.ryuvia.com/niklas/terdut-server/internal/db" ) // ts wraps httptest.Server with a pre-bootstrapped API key. db is exposed so @@ -26,7 +25,7 @@ type ts struct { deadman api.DeadmanConfig } -// newTS builds a server over a fresh in-memory database. Notifications are off +// newTS builds a server over a fresh database. Notifications are off // unless a NotifyConfig is passed, so tests that predate them are unaffected. // Dead man's switches are off too — see newDeadmanTS. func newTS(t *testing.T, notify ...api.NotifyConfig) *ts { @@ -46,15 +45,9 @@ func newDeadmanTS(t *testing.T, deadman api.DeadmanConfig, notify ...api.NotifyC cfg = notify[0] } - database, err := db.Open(":memory:") - if err != nil { - t.Fatalf("open db: %v", err) - } - if err := db.Migrate(database); err != nil { - t.Fatalf("migrate: %v", err) - } + database := newTestDB(t) srv := httptest.NewServer(api.NewRouter(database, cfg, deadman)) - t.Cleanup(func() { srv.Close(); database.Close() }) + t.Cleanup(srv.Close) body, _ := json.Marshal(map[string]string{"username": "admin", "email": "admin@test.com"}) resp, err := http.Post(srv.URL+"/api/bootstrap", "application/json", bytes.NewReader(body)) @@ -84,7 +77,7 @@ func (s *ts) exec(t *testing.T, query string, args ...any) { func (s *ts) alertRow(t *testing.T, fingerprint string) (status string, source *string, archivedAt *int64) { t.Helper() err := s.db.QueryRow( - "SELECT status, resolution_source, archived_at FROM alerts WHERE fingerprint = ?", + "SELECT status, resolution_source, archived_at FROM alerts WHERE fingerprint = $1", fingerprint).Scan(&status, &source, &archivedAt) if err != nil { t.Fatalf("read alert %s: %v", fingerprint, err) @@ -96,7 +89,7 @@ func (s *ts) alertRow(t *testing.T, fingerprint string) (status string, source * func (s *ts) alertTimes(t *testing.T, fingerprint string) (startsAt, receivedAt int64) { t.Helper() err := s.db.QueryRow( - "SELECT starts_at, received_at FROM alerts WHERE fingerprint = ?", + "SELECT starts_at, received_at FROM alerts WHERE fingerprint = $1", fingerprint).Scan(&startsAt, &receivedAt) if err != nil { t.Fatalf("read alert times %s: %v", fingerprint, err) @@ -109,7 +102,7 @@ func (s *ts) alertEndsAt(t *testing.T, fingerprint string) *int64 { t.Helper() var endsAt *int64 if err := s.db.QueryRow( - "SELECT ends_at FROM alerts WHERE fingerprint = ?", fingerprint).Scan(&endsAt); err != nil { + "SELECT ends_at FROM alerts WHERE fingerprint = $1", fingerprint).Scan(&endsAt); err != nil { t.Fatalf("read ends_at %s: %v", fingerprint, err) } return endsAt @@ -556,7 +549,7 @@ func TestExpiry_StaleFiringAlert(t *testing.T) { postAlert(t, s, "stale1", "firing", time.Now().Add(-24*time.Hour).Format(time.RFC3339), zeroTime) // Age the last-seen timestamp past the staleness window. - s.exec(t, "UPDATE alerts SET received_at = ? WHERE fingerprint = 'stale1'", + s.exec(t, "UPDATE alerts SET received_at = $1 WHERE fingerprint = 'stale1'", time.Now().Add(-10*time.Hour).Unix()) sweep(t, s, 6*time.Hour) @@ -653,10 +646,10 @@ func TestWebhook_RefireUnarchivesAndClearsSource(t *testing.T) { postAlert(t, s, "refire1", "firing", time.Now().Add(-24*time.Hour).Format(time.RFC3339), zeroTime) // Expire it, then archive it. - s.exec(t, "UPDATE alerts SET received_at = ? WHERE fingerprint = 'refire1'", + s.exec(t, "UPDATE alerts SET received_at = $1 WHERE fingerprint = 'refire1'", time.Now().Add(-10*time.Hour).Unix()) sweep(t, s, 6*time.Hour) - s.exec(t, "UPDATE alerts SET archived_at = unixepoch() WHERE fingerprint = 'refire1'") + s.exec(t, "UPDATE alerts SET archived_at = FLOOR(EXTRACT(EPOCH FROM now()))::bigint WHERE fingerprint = 'refire1'") var alerts []map[string]any decode(t, s.req(t, http.MethodGet, "/api/alerts", nil), &alerts) @@ -714,7 +707,7 @@ func TestExpiry_EndsAtIsUpperBound(t *testing.T) { // No watermark: expires on the received_at heartbeat, so the sweeper has // nothing to go on but its own clock. postAlert(t, s, "ub-none", "firing", time.Now().Add(-24*time.Hour).Format(time.RFC3339), zeroTime) - s.exec(t, "UPDATE alerts SET received_at = ? WHERE fingerprint = 'ub-none'", + s.exec(t, "UPDATE alerts SET received_at = $1 WHERE fingerprint = 'ub-none'", time.Now().Add(-10*time.Hour).Unix()) // Stale watermark: expires on the ends_at branch, and that reported time @@ -768,7 +761,7 @@ func TestWebhook_ResendBumpsReceivedAt(t *testing.T) { // received_at has one-second granularity, so back-date it to make the bump // observable instead of sleeping out a second. aged := time.Now().Add(-2 * time.Hour).Unix() - s.exec(t, "UPDATE alerts SET received_at = ? WHERE fingerprint = 'beat1'", aged) + s.exec(t, "UPDATE alerts SET received_at = $1 WHERE fingerprint = 'beat1'", aged) // Identical re-send: same fingerprint, same startsAt, still firing. postAlert(t, s, "beat1", "firing", start, zeroTime) @@ -796,7 +789,7 @@ func TestWebhook_DiscardedRetryLeavesReceivedAtAlone(t *testing.T) { postAlert(t, s, "beat2", "resolved", start, time.Now().Format(time.RFC3339)) aged := time.Now().Add(-2 * time.Hour).Unix() - s.exec(t, "UPDATE alerts SET received_at = ? WHERE fingerprint = 'beat2'", aged) + s.exec(t, "UPDATE alerts SET received_at = $1 WHERE fingerprint = 'beat2'", aged) postAlert(t, s, "beat2", "firing", start, zeroTime) // stale retry, discarded diff --git a/internal/api/archiver.go b/internal/api/archiver.go index 2374f0f..b767b16 100644 --- a/internal/api/archiver.go +++ b/internal/api/archiver.go @@ -4,7 +4,7 @@ import ( "context" "database/sql" "log" - "strings" + "time" ) @@ -91,17 +91,18 @@ func expireStale(ctx context.Context, db *sql.DB, staleAfter time.Duration, skip return } - args := make([]any, 0, len(ids)+1) - args = append(args, resolutionExpiry) - for _, id := range ids { - args = append(args, id) + args := &sqlArgs{} + source := args.add(resolutionExpiry) + idList := make([]any, len(ids)) + for i, id := range ids { + idList[i] = id } if _, err := db.ExecContext(ctx, ` UPDATE alerts SET status = 'resolved', - resolution_source = ?, - ends_at = COALESCE(ends_at, unixepoch()) - WHERE id IN (`+placeholders(len(ids))+`)`, args...); err != nil { + resolution_source = `+source+`, + ends_at = COALESCE(ends_at, `+nowEpoch+`) + WHERE id IN (`+args.addList(idList)+`)`, args.all()...); err != nil { log.Printf("sweeper: expire stale: %v", err) return } @@ -124,14 +125,15 @@ func expireStale(ctx context.Context, db *sql.DB, staleAfter time.Duration, skip } // staleAlertIDs reads the ids in one go and closes the cursor before the caller -// writes: the pool is limited to a single connection, so an open read would -// block the update behind it. +// writes. Under SQLite's single connection an open read would have blocked the +// update outright; with a pool it is no longer a deadlock, but reading the set +// first still keeps the write off a cursor the same transaction is walking. func staleAlertIDs(ctx context.Context, db *sql.DB, now time.Time, staleAfter time.Duration) ([]int64, error) { rows, err := db.QueryContext(ctx, ` SELECT id FROM alerts WHERE status = 'firing' AND archived_at IS NULL - AND ((ends_at IS NOT NULL AND ends_at < ?) OR received_at < ?)`, + AND ((ends_at IS NOT NULL AND ends_at < $1) OR received_at < $2)`, now.Add(-expiryGrace).Unix(), now.Add(-staleAfter).Unix()) if err != nil { return nil, err @@ -206,10 +208,10 @@ func settledIncidentIDs(ctx context.Context, db *sql.DB) ([]int64, error) { func archiveResolved(ctx context.Context, db *sql.DB, archiveAfter time.Duration) { cutoff := time.Now().Add(-archiveAfter).Unix() res, err := db.ExecContext(ctx, - `UPDATE alerts SET archived_at = unixepoch() + `UPDATE alerts SET archived_at = `+nowEpoch+` WHERE status = 'resolved' AND archived_at IS NULL - AND COALESCE(ends_at, received_at) < ?`, cutoff) + AND COALESCE(ends_at, received_at) < $1`, cutoff) if err != nil { log.Printf("archiver: %v", err) return @@ -223,10 +225,10 @@ func archiveResolved(ctx context.Context, db *sql.DB, archiveAfter time.Duration func archiveResolvedIncidents(ctx context.Context, db *sql.DB, archiveAfter time.Duration) { cutoff := time.Now().Add(-archiveAfter).Unix() res, err := db.ExecContext(ctx, - `UPDATE incidents SET archived_at = unixepoch() + `UPDATE incidents SET archived_at = `+nowEpoch+` WHERE resolved_at IS NOT NULL AND archived_at IS NULL - AND resolved_at < ?`, cutoff) + AND resolved_at < $1`, cutoff) if err != nil { log.Printf("archiver: incidents: %v", err) return @@ -235,8 +237,3 @@ func archiveResolvedIncidents(ctx context.Context, db *sql.DB, archiveAfter time log.Printf("archiver: archived %d resolved incident(s)", n) } } - -// placeholders builds "?, ?, …" for an IN clause of n values. -func placeholders(n int) string { - return strings.TrimSuffix(strings.Repeat("?, ", n), ", ") -} diff --git a/internal/api/auth.go b/internal/api/auth.go index 30f4a67..47e3733 100644 --- a/internal/api/auth.go +++ b/internal/api/auth.go @@ -163,7 +163,7 @@ func handleLogin(db *sql.DB, limiter *loginLimiter, publicURL string) http.Handl var userID int64 var hash sql.NullString err := db.QueryRowContext(r.Context(), - "SELECT id, password_hash FROM users WHERE username = ?", username, + "SELECT id, password_hash FROM users WHERE username = $1", username, ).Scan(&userID, &hash) if err != nil && !errors.Is(err, sql.ErrNoRows) { respond(w, http.StatusInternalServerError, errResp("internal error")) @@ -190,7 +190,7 @@ func handleLogin(db *sql.DB, limiter *loginLimiter, publicURL string) http.Handl now := time.Now() if _, err := db.ExecContext(r.Context(), ` INSERT INTO sessions (token_hash, user_id, created_at, last_seen_at, expires_at, user_agent) - VALUES (?, ?, ?, ?, ?, ?)`, + VALUES ($1, $2, $3, $4, $5, $6)`, tokenHash, userID, now.Unix(), now.Unix(), now.Add(sessionTTL).Unix(), r.UserAgent()); err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return @@ -225,7 +225,7 @@ func handleLogout(db *sql.DB, publicURL string) http.HandlerFunc { return } if c, err := r.Cookie(sessionCookie); err == nil && c.Value != "" { - db.ExecContext(r.Context(), "DELETE FROM sessions WHERE token_hash = ?", hashToken(c.Value)) + db.ExecContext(r.Context(), "DELETE FROM sessions WHERE token_hash = $1", hashToken(c.Value)) } http.SetCookie(w, &http.Cookie{ Name: sessionCookie, @@ -257,7 +257,7 @@ func handleMe(db *sql.DB) http.HandlerFunc { } var hash sql.NullString db.QueryRowContext(r.Context(), - "SELECT password_hash FROM users WHERE id = ?", caller.ID).Scan(&hash) + "SELECT password_hash FROM users WHERE id = $1", caller.ID).Scan(&hash) respond(w, http.StatusOK, meResponse{User: user, HasPassword: hash.Valid}) } } @@ -293,7 +293,7 @@ func handleSetPassword(db *sql.DB) http.HandlerFunc { var existing sql.NullString err = db.QueryRowContext(r.Context(), - "SELECT password_hash FROM users WHERE id = ?", id).Scan(&existing) + "SELECT password_hash FROM users WHERE id = $1", id).Scan(&existing) if errors.Is(err, sql.ErrNoRows) { respond(w, http.StatusNotFound, errResp("user not found")) return @@ -324,13 +324,13 @@ func handleSetPassword(db *sql.DB) http.HandlerFunc { defer tx.Rollback() if _, err := tx.ExecContext(r.Context(), - "UPDATE users SET password_hash = ? WHERE id = ?", hash, id); err != nil { + "UPDATE users SET password_hash = $1 WHERE id = $2", hash, id); err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return } keep, _ := sessionFromContext(r.Context()) // zero when changed with an API key if _, err := tx.ExecContext(r.Context(), - "DELETE FROM sessions WHERE user_id = ? AND id != ?", id, keep); err != nil { + "DELETE FROM sessions WHERE user_id = $1 AND id != $2", id, keep); err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return } @@ -345,7 +345,7 @@ func handleSetPassword(db *sql.DB) http.HandlerFunc { // purgeSessions deletes sessions that have expired, from the sweeper. func purgeSessions(ctx context.Context, db *sql.DB) { res, err := db.ExecContext(ctx, - "DELETE FROM sessions WHERE expires_at < ?", time.Now().Unix()) + "DELETE FROM sessions WHERE expires_at < $1", time.Now().Unix()) if err != nil { log.Printf("sweeper: purge sessions: %v", err) return diff --git a/internal/api/auth_test.go b/internal/api/auth_test.go index 7240fc8..f2af701 100644 --- a/internal/api/auth_test.go +++ b/internal/api/auth_test.go @@ -11,7 +11,6 @@ import ( "testing" "git.ryuvia.com/niklas/terdut-server/internal/api" - "git.ryuvia.com/niklas/terdut-server/internal/db" ) const adminPassword = "correct horse battery" @@ -301,15 +300,9 @@ func TestSetPassword_EndsOtherSessionsButNotThisOne(t *testing.T) { } func TestBootstrap_WithPassword(t *testing.T) { - database, err := db.Open(":memory:") - if err != nil { - t.Fatal(err) - } - if err := db.Migrate(database); err != nil { - t.Fatal(err) - } + database := newTestDB(t) srv := httptest.NewServer(api.NewRouter(database, api.NotifyConfig{}, api.DeadmanConfig{})) - t.Cleanup(func() { srv.Close(); database.Close() }) + t.Cleanup(srv.Close) body := `{"username":"admin","email":"a@test.com","password":"` + adminPassword + `"}` resp, err := http.Post(srv.URL+"/api/bootstrap", "application/json", strings.NewReader(body)) diff --git a/internal/api/deadman.go b/internal/api/deadman.go index d54447c..e1e524d 100644 --- a/internal/api/deadman.go +++ b/internal/api/deadman.go @@ -225,19 +225,21 @@ func sweepDeadman(ctx context.Context, db *sql.DB, cfg DeadmanConfig, notify Not // deadmanAlerts loads every alert row that a matcher claims. The candidate query // is narrowed by alertname so it rides alerts_name_idx; the rest of the matching // happens in Go, which keeps one implementation of the rules. The rows are read -// in full before the caller writes, because the pool holds a single connection. +// in full before the caller writes, so the writes do not run against an open +// cursor over the same table. func deadmanAlerts(ctx context.Context, db *sql.DB, cfg DeadmanConfig) ([]deadmanAlert, error) { names := cfg.names() - args := make([]any, 0, len(names)) - for _, n := range names { - args = append(args, n) + args := &sqlArgs{} + nameList := make([]any, len(names)) + for i, n := range names { + nameList[i] = n } rows, err := db.QueryContext(ctx, ` SELECT id, fingerprint, labels, status, received_at FROM alerts - WHERE name IN (`+placeholders(len(names))+`) - AND archived_at IS NULL`, args...) + WHERE name IN (`+args.addList(nameList)+`) + AND archived_at IS NULL`, args.all()...) if err != nil { return nil, err } @@ -277,8 +279,8 @@ func deadmanDied(ctx context.Context, db *sql.DB, cfg DeadmanConfig, notify Noti var lastTriggered, open int64 if err := db.QueryRowContext(ctx, ` SELECT COALESCE(MAX(triggered_at), 0), - COALESCE(SUM(resolved_at IS NULL), 0) - FROM incidents WHERE group_key = ?`, + COUNT(*) FILTER (WHERE resolved_at IS NULL) + FROM incidents WHERE group_key = $1`, sw.groupKey()).Scan(&lastTriggered, &open); err != nil { return err } @@ -299,9 +301,9 @@ func deadmanDied(ctx context.Context, db *sql.DB, cfg DeadmanConfig, notify Noti if _, err := tx.ExecContext(ctx, ` UPDATE alerts SET status = 'resolved', - resolution_source = ?, - ends_at = COALESCE(ends_at, unixepoch()) - WHERE id = ? AND status = 'firing'`, resolutionDeadman, sw.id); err != nil { + resolution_source = $1, + ends_at = COALESCE(ends_at, `+nowEpoch+`) + WHERE id = $2 AND status = 'firing'`, resolutionDeadman, sw.id); err != nil { return err } } @@ -341,7 +343,7 @@ func deadmanRecovered(ctx context.Context, db *sql.DB, sw deadmanAlert) error { var incidentID int64 switch err := db.QueryRowContext(ctx, ` SELECT id FROM incidents - WHERE group_key = ? AND resolved_at IS NULL`, sw.groupKey()).Scan(&incidentID); { + WHERE group_key = $1 AND resolved_at IS NULL`, sw.groupKey()).Scan(&incidentID); { case err == sql.ErrNoRows: return nil case err != nil: @@ -356,8 +358,8 @@ func deadmanRecovered(ctx context.Context, db *sql.DB, sw deadmanAlert) error { if _, err := tx.ExecContext(ctx, ` UPDATE incidents - SET status = 'resolved', resolved_at = ?, resolution_source = ? - WHERE id = ? AND resolved_at IS NULL`, + SET status = 'resolved', resolved_at = $1, resolution_source = $2 + WHERE id = $3 AND resolved_at IS NULL`, time.Now().Unix(), incidentResolutionRecovered, incidentID); err != nil { return err } diff --git a/internal/api/deadman_test.go b/internal/api/deadman_test.go index 6806f37..d9cecfb 100644 --- a/internal/api/deadman_test.go +++ b/internal/api/deadman_test.go @@ -51,7 +51,7 @@ func heartbeat(t *testing.T, s *ts, fingerprint string, labels map[string]string // sweeper reads. There is no fake clock in this package. func silence(t *testing.T, s *ts, fingerprint string, ago time.Duration) { t.Helper() - s.exec(t, "UPDATE alerts SET received_at = ? WHERE fingerprint = ?", + s.exec(t, "UPDATE alerts SET received_at = $1 WHERE fingerprint = $2", time.Now().Add(-ago).Unix(), fingerprint) } @@ -61,9 +61,12 @@ func silence(t *testing.T, s *ts, fingerprint string, ago time.Duration) { func ageIncidents(t *testing.T, s *ts, ago time.Duration) { t.Helper() past := time.Now().Add(-ago).Unix() + // $2 is cast explicitly: with NULL in the other branch Postgres has nothing + // to infer the parameter's type from and defaults it to text, which the + // bigint column then refuses. s.exec(t, `UPDATE incidents - SET triggered_at = ?, - resolved_at = CASE WHEN resolved_at IS NULL THEN NULL ELSE ? END`, + SET triggered_at = $1, + resolved_at = CASE WHEN resolved_at IS NULL THEN NULL ELSE $2::bigint END`, past, past) } @@ -72,7 +75,7 @@ func incidentByGroup(t *testing.T, s *ts, groupKey string) (id int64, status, se t.Helper() err := s.db.QueryRow(` SELECT id, status, COALESCE(severity, ''), resolution_source - FROM incidents WHERE group_key = ? ORDER BY id DESC LIMIT 1`, + FROM incidents WHERE group_key = $1 ORDER BY id DESC LIMIT 1`, groupKey).Scan(&id, &status, &severity, &source) if err != nil { t.Fatalf("read incident for group %s: %v", groupKey, err) diff --git a/internal/api/helpers.go b/internal/api/helpers.go index d2a040d..48bfc70 100644 --- a/internal/api/helpers.go +++ b/internal/api/helpers.go @@ -2,9 +2,67 @@ package api import ( "encoding/json" + "errors" "net/http" + "strconv" + "strings" + + "github.com/jackc/pgerrcode" + "github.com/jackc/pgx/v5/pgconn" ) +// sqlArgs accumulates query arguments and hands back the placeholder for each. +// +// Postgres numbers its placeholders, so a dynamically assembled WHERE clause has +// to keep its $1, $2, … in step with the order of the values — which SQLite's +// positional `?` did for free. Handing out the placeholder and storing the value +// in one call is what keeps them in step: a filter can be added, removed or +// reordered without renumbering anything by hand. +type sqlArgs struct{ vals []any } + +// add stores v and returns the placeholder that refers to it. +func (a *sqlArgs) add(v any) string { + a.vals = append(a.vals, v) + return "$" + strconv.Itoa(len(a.vals)) +} + +// addList stores every value and returns their placeholders as "$1, $2, …", +// ready to drop into an IN (…) clause. Returns an empty string for no values, +// which no caller should reach: `IN ()` is a syntax error in Postgres as it was +// in SQLite, so callers check for an empty set before building the query. +func (a *sqlArgs) addList(vs []any) string { + parts := make([]string, len(vs)) + for i, v := range vs { + parts[i] = a.add(v) + } + return strings.Join(parts, ", ") +} + +// all returns the accumulated values, to be passed straight to Query or Exec. +func (a *sqlArgs) all() []any { return a.vals } + +// nowEpoch is the SQL expression for "now, as unix seconds", matching how every +// timestamp in this schema is stored. SQLite spelled it unixepoch(). +// +// FLOOR, not a bare cast: EXTRACT returns fractional seconds and casting to +// bigint rounds half up, so a row written at .6 of a second would claim a +// timestamp one second in the future — off by one against the time.Now().Unix() +// the Go side stamps, which is what the expiry tests measure. +const nowEpoch = "FLOOR(EXTRACT(EPOCH FROM now()))::bigint" + +// isUniqueViolation reports whether err is a broken unique constraint, which +// callers turn into 409 Conflict rather than 500. +// +// Postgres reports it as SQLSTATE 23505 on a typed error; the SQLite driver this +// replaced only put "UNIQUE constraint failed" in the message, which is why the +// check used to be a substring match. Matching the code means a renamed +// constraint or a translated message cannot quietly turn a conflict back into a +// 500. +func isUniqueViolation(err error) bool { + var pgErr *pgconn.PgError + return errors.As(err, &pgErr) && pgErr.Code == pgerrcode.UniqueViolation +} + func respond(w http.ResponseWriter, status int, v any) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) diff --git a/internal/api/incident_store.go b/internal/api/incident_store.go index cc60a25..26def15 100644 --- a/internal/api/incident_store.go +++ b/internal/api/incident_store.go @@ -95,7 +95,7 @@ func unixPtr(sec *int64) *time.Time { } func fetchIncident(ctx context.Context, q querier, id int64) (models.Incident, error) { - return scanIncident(q.QueryRowContext(ctx, incidentSelectFrom+" WHERE i.id = ?", id)) + return scanIncident(q.QueryRowContext(ctx, incidentSelectFrom+" WHERE i.id = $1", id)) } // logEvent appends one entry to an incident's timeline. A nil userID means the @@ -103,7 +103,7 @@ func fetchIncident(ctx context.Context, q querier, id int64) (models.Incident, e func logEvent(ctx context.Context, q querier, incidentID int64, evType string, userID, alertID *int64, detail *string) error { _, err := q.ExecContext(ctx, ` INSERT INTO incident_events (incident_id, type, user_id, alert_id, detail, created_at) - VALUES (?, ?, ?, ?, ?, ?)`, + VALUES ($1, $2, $3, $4, $5, $6)`, incidentID, evType, userID, alertID, detail, time.Now().Unix()) return err } @@ -118,7 +118,7 @@ func todayUTC() string { func currentOnCall(ctx context.Context, q querier) (*int64, error) { var userID int64 err := q.QueryRowContext(ctx, - "SELECT user_id FROM schedule_entries WHERE date = ?", todayUTC()).Scan(&userID) + "SELECT user_id FROM schedule_entries WHERE date = $1", todayUTC()).Scan(&userID) if err == sql.ErrNoRows { return nil, nil } @@ -154,10 +154,10 @@ func severityRank(s string) int { // in the queue while the work is still open. func refreshSeverity(ctx context.Context, q querier, incidentID int64) error { rows, err := q.QueryContext(ctx, ` - SELECT json_extract(a.labels, '$.'||?) + SELECT a.labels ->> $1 FROM incident_alerts ia JOIN alerts a ON a.id = ia.alert_id - WHERE ia.incident_id = ?`, severityLabel, incidentID) + WHERE ia.incident_id = $2`, severityLabel, incidentID) if err != nil { return err } @@ -185,9 +185,9 @@ func refreshSeverity(ctx context.Context, q querier, incidentID int64) error { // The comparison lives in SQL so an unrelated concurrent update cannot be // clobbered by a stale read. _, err = q.ExecContext(ctx, ` - UPDATE incidents SET severity = ? - WHERE id = ? - AND (severity IS NULL OR `+severityRankSQL("severity")+` < ?)`, + UPDATE incidents SET severity = $1 + WHERE id = $2 + AND (severity IS NULL OR `+severityRankSQL("severity")+` < $3)`, best, incidentID, severityRank(best)) return err } @@ -210,9 +210,9 @@ func resolveIfSettled(ctx context.Context, q querier, incidentID int64) (bool, e res, err := q.ExecContext(ctx, ` UPDATE incidents SET status = 'resolved', - resolved_at = ?, - resolution_source = ? - WHERE id = ? + resolved_at = $1, + resolution_source = $2 + WHERE id = $3 AND resolved_at IS NULL -- An incident with no members yet is mid-creation, not settled. AND EXISTS (SELECT 1 FROM incident_alerts ia WHERE ia.incident_id = incidents.id) @@ -245,8 +245,8 @@ func resolveIfSettled(ctx context.Context, q querier, incidentID int64) (bool, e func acknowledgeIncident(ctx context.Context, q querier, incidentID, userID int64) (bool, error) { res, err := q.ExecContext(ctx, ` UPDATE incidents - SET status = 'acknowledged', acknowledged_by = ?, acknowledged_at = ? - WHERE id = ? AND resolved_at IS NULL`, + SET status = 'acknowledged', acknowledged_by = $1, acknowledged_at = $2 + WHERE id = $3 AND resolved_at IS NULL`, userID, time.Now().Unix(), incidentID) if err != nil { return false, err @@ -266,7 +266,7 @@ func openIncidentForAlert(ctx context.Context, q querier, alertID int64) (int64, SELECT i.id FROM incident_alerts ia JOIN incidents i ON i.id = ia.incident_id - WHERE ia.alert_id = ? AND i.resolved_at IS NULL`, alertID).Scan(&id) + WHERE ia.alert_id = $1 AND i.resolved_at IS NULL`, alertID).Scan(&id) if err == sql.ErrNoRows { return 0, nil } diff --git a/internal/api/incidents.go b/internal/api/incidents.go index 0720a9b..3369676 100644 --- a/internal/api/incidents.go +++ b/internal/api/incidents.go @@ -17,13 +17,12 @@ func handleListIncidents(db *sql.DB) http.HandlerFunc { q := r.URL.Query() where := []string{} - args := []any{} + args := &sqlArgs{} // Without an explicit status the queue shows open work, which is what an // on-call person opens the tool to see. if status := q.Get("status"); status != "" { - where = append(where, "i.status = ?") - args = append(args, status) + where = append(where, "i.status = "+args.add(status)) } else { where = append(where, "i.resolved_at IS NULL") } @@ -36,33 +35,27 @@ func handleListIncidents(db *sql.DB) http.HandlerFunc { // A snooze expires by simply falling into the past; nothing sweeps it. if q.Get("snoozed") == "true" { - where = append(where, "i.snoozed_until > ?") - args = append(args, time.Now().Unix()) + where = append(where, "i.snoozed_until > "+args.add(time.Now().Unix())) } else { - where = append(where, "(i.snoozed_until IS NULL OR i.snoozed_until <= ?)") - args = append(args, time.Now().Unix()) + where = append(where, "(i.snoozed_until IS NULL OR i.snoozed_until <= "+args.add(time.Now().Unix())+")") } if severity := q.Get("severity"); severity != "" { - where = append(where, "i.severity = ?") - args = append(args, severity) + where = append(where, "i.severity = "+args.add(severity)) } if assignee := q.Get("assigned_to"); assignee != "" { if n, err := strconv.ParseInt(assignee, 10, 64); err == nil { - where = append(where, "i.assigned_to = ?") - args = append(args, n) + where = append(where, "i.assigned_to = "+args.add(n)) } } if from := q.Get("from"); from != "" { if t, err := time.Parse("2006-01-02", from); err == nil { - where = append(where, "i.triggered_at >= ?") - args = append(args, t.UTC().Unix()) + where = append(where, "i.triggered_at >= "+args.add(t.UTC().Unix())) } } if to := q.Get("to"); to != "" { if t, err := time.Parse("2006-01-02", to); err == nil { - where = append(where, "i.triggered_at < ?") - args = append(args, t.UTC().AddDate(0, 0, 1).Unix()) + where = append(where, "i.triggered_at < "+args.add(t.UTC().AddDate(0, 0, 1).Unix())) } } @@ -77,12 +70,11 @@ func handleListIncidents(db *sql.DB) http.HandlerFunc { if q.Get("sort") == "severity" { order = severityRankSQL("i.severity") + " DESC, i.triggered_at DESC" } - args = append(args, limit) rows, err := db.QueryContext(r.Context(), - fmt.Sprintf("%s WHERE %s ORDER BY %s LIMIT ?", - incidentSelectFrom, strings.Join(where, " AND "), order), - args...) + fmt.Sprintf("%s WHERE %s ORDER BY %s LIMIT %s", + incidentSelectFrom, strings.Join(where, " AND "), order, args.add(limit)), + args.all()...) if err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return @@ -158,7 +150,7 @@ func handleIncidentTimeline(db *sql.DB) http.HandlerFunc { e.alert_id, e.detail, e.created_at FROM incident_events e LEFT JOIN users u ON u.id = e.user_id - WHERE e.incident_id = ? + WHERE e.incident_id = $1 ORDER BY e.created_at ASC, e.id ASC`, id) if err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) @@ -214,7 +206,7 @@ func handleIncidentUnacknowledge(db *sql.DB) http.HandlerFunc { user, _ := userFromContext(r.Context()) if !updateOpenIncident(w, r, db, id, `UPDATE incidents SET status = 'triggered', acknowledged_by = NULL, acknowledged_at = NULL - WHERE id = ? AND resolved_at IS NULL`, id) { + WHERE id = $1 AND resolved_at IS NULL`, id) { return } if err := logEvent(r.Context(), db, id, evUnacknowledged, &user.ID, nil, nil); err != nil { @@ -237,8 +229,8 @@ func handleIncidentResolve(db *sql.DB) http.HandlerFunc { } user, _ := userFromContext(r.Context()) if !updateOpenIncident(w, r, db, id, - `UPDATE incidents SET status = 'resolved', resolved_at = ?, resolution_source = ? - WHERE id = ? AND resolved_at IS NULL`, + `UPDATE incidents SET status = 'resolved', resolved_at = $1, resolution_source = $2 + WHERE id = $3 AND resolved_at IS NULL`, time.Now().Unix(), incidentResolutionManual, id) { return } @@ -269,13 +261,13 @@ func handleIncidentAssign(db *sql.DB) http.HandlerFunc { } var exists int if err := db.QueryRowContext(r.Context(), - "SELECT 1 FROM users WHERE id = ?", req.UserID).Scan(&exists); err != nil { + "SELECT 1 FROM users WHERE id = $1", req.UserID).Scan(&exists); err != nil { respond(w, http.StatusNotFound, errResp("user not found")) return } if !updateOpenIncident(w, r, db, id, - "UPDATE incidents SET assigned_to = ? WHERE id = ? AND resolved_at IS NULL", + "UPDATE incidents SET assigned_to = $1 WHERE id = $2 AND resolved_at IS NULL", req.UserID, id) { return } @@ -333,7 +325,7 @@ func handleIncidentSnooze(db *sql.DB) http.HandlerFunc { user, _ := userFromContext(r.Context()) if !updateOpenIncident(w, r, db, id, - "UPDATE incidents SET snoozed_until = ? WHERE id = ? AND resolved_at IS NULL", + "UPDATE incidents SET snoozed_until = $1 WHERE id = $2 AND resolved_at IS NULL", until.Unix(), id) { return } @@ -354,7 +346,7 @@ func handleIncidentUnsnooze(db *sql.DB) http.HandlerFunc { } user, _ := userFromContext(r.Context()) if !updateOpenIncident(w, r, db, id, - "UPDATE incidents SET snoozed_until = NULL WHERE id = ? AND resolved_at IS NULL", id) { + "UPDATE incidents SET snoozed_until = NULL WHERE id = $1 AND resolved_at IS NULL", id) { return } if err := logEvent(r.Context(), db, id, evUnsnoozed, &user.ID, nil, nil); err != nil { @@ -372,7 +364,7 @@ func handleIncidentArchive(db *sql.DB) http.HandlerFunc { return } res, err := db.ExecContext(r.Context(), - "UPDATE incidents SET archived_at = unixepoch() WHERE id = ?", id) + "UPDATE incidents SET archived_at = "+nowEpoch+" WHERE id = $1", id) if err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return @@ -392,7 +384,7 @@ func handleIncidentUnarchive(db *sql.DB) http.HandlerFunc { return } res, err := db.ExecContext(r.Context(), - "UPDATE incidents SET archived_at = NULL WHERE id = ?", id) + "UPDATE incidents SET archived_at = NULL WHERE id = $1", id) if err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return @@ -430,14 +422,15 @@ func handleCreateNote(db *sql.DB) http.HandlerFunc { user, _ := userFromContext(r.Context()) now := time.Now() - res, err := db.ExecContext(r.Context(), ` + var eventID int64 + err := db.QueryRowContext(r.Context(), ` INSERT INTO incident_events (incident_id, type, user_id, detail, created_at) - VALUES (?, ?, ?, ?, ?)`, id, evNote, user.ID, req.Content, now.Unix()) + VALUES ($1, $2, $3, $4, $5) + RETURNING id`, id, evNote, user.ID, req.Content, now.Unix()).Scan(&eventID) if err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return } - eventID, _ := res.LastInsertId() respond(w, http.StatusCreated, models.IncidentEvent{ ID: eventID, @@ -468,7 +461,7 @@ func handleDeleteNote(db *sql.DB) http.HandlerFunc { user, _ := userFromContext(r.Context()) res, err := db.ExecContext(r.Context(), ` DELETE FROM incident_events - WHERE id = ? AND incident_id = ? AND type = ? AND user_id = ?`, + WHERE id = $1 AND incident_id = $2 AND type = $3 AND user_id = $4`, eventID, id, evNote, user.ID) if err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) @@ -498,7 +491,7 @@ func incidentIDParam(w http.ResponseWriter, r *http.Request) (int64, bool) { func incidentExists(w http.ResponseWriter, r *http.Request, db *sql.DB, id int64) bool { var exists int if err := db.QueryRowContext(r.Context(), - "SELECT 1 FROM incidents WHERE id = ?", id).Scan(&exists); err != nil { + "SELECT 1 FROM incidents WHERE id = $1", id).Scan(&exists); err != nil { respond(w, http.StatusNotFound, errResp("incident not found")) return false } @@ -538,7 +531,7 @@ func respondIncident(w http.ResponseWriter, r *http.Request, db *sql.DB, id int6 func incidentAlerts(r *http.Request, db *sql.DB, id int64) ([]models.Alert, error) { rows, err := db.QueryContext(r.Context(), alertSelectFrom+` JOIN incident_alerts m ON m.alert_id = a.id - WHERE m.incident_id = ? + WHERE m.incident_id = $1 ORDER BY a.received_at DESC`, id) if err != nil { return nil, err diff --git a/internal/api/incidents_test.go b/internal/api/incidents_test.go index 4820503..04757d1 100644 --- a/internal/api/incidents_test.go +++ b/internal/api/incidents_test.go @@ -4,14 +4,10 @@ import ( "context" "fmt" "net/http" - "os" - "path/filepath" - "sort" "testing" "time" "git.ryuvia.com/niklas/terdut-server/internal/api" - "git.ryuvia.com/niklas/terdut-server/internal/db" ) // amAlert builds one alert of a webhook payload. @@ -288,7 +284,7 @@ func TestExpiry_CascadesToIncidentResolution(t *testing.T) { s := newTS(t) postAlert(t, s, "fp-exp", "firing", time.Now().Add(-24*time.Hour).Format(time.RFC3339), zeroTime) - s.exec(t, "UPDATE alerts SET received_at = ? WHERE fingerprint = 'fp-exp'", + s.exec(t, "UPDATE alerts SET received_at = $1 WHERE fingerprint = 'fp-exp'", time.Now().Add(-10*time.Hour).Unix()) sweep(t, s, 6*time.Hour) @@ -611,7 +607,7 @@ func TestSweeper_ArchivesResolvedIncidents(t *testing.T) { }) s.req(t, http.MethodPost, "/api/incidents/1/resolve", nil).Body.Close() - s.exec(t, "UPDATE incidents SET resolved_at = ? WHERE id = 1", + s.exec(t, "UPDATE incidents SET resolved_at = $1 WHERE id = 1", time.Now().Add(-30*24*time.Hour).Unix()) api.Sweep(context.Background(), s.db, 7*24*time.Hour, 6*time.Hour, s.deadman, s.notify) @@ -654,8 +650,9 @@ func TestStats_Incidents(t *testing.T) { } // An empty window is a report of zero, not a failure. SUM over no rows is NULL -// in SQLite, which used to come back as a 500 the moment every incident was -// archived — the state a quiet installation settles into. +// in Postgres as it was in SQLite, and that used to come back as a 500 the +// moment every incident was archived — the state a quiet installation settles +// into. func TestStats_IncidentsEmptyWindowIsZeroNotAnError(t *testing.T) { s := newTS(t) @@ -722,101 +719,6 @@ func TestStats_IncidentsNullMTTAWhenNothingAcknowledged(t *testing.T) { } } -// --------------------------------------------------------------------------- -// Migration backfill -// --------------------------------------------------------------------------- - -// An upgrade must not drop the acknowledgements and comments people already -// have, so 008 is replayed here over a database left at 007. -func TestMigration_BackfillCarriesAckAndComments(t *testing.T) { - database, err := db.Open(":memory:") - if err != nil { - t.Fatalf("open db: %v", err) - } - t.Cleanup(func() { database.Close() }) - - files, err := filepath.Glob("../db/migrations/*.sql") - if err != nil || len(files) == 0 { - t.Fatalf("find migrations: %v", err) - } - sort.Strings(files) - - var incidentsMigration string - for _, f := range files { - if filepath.Base(f) >= "008" { - incidentsMigration = f - break - } - data, err := os.ReadFile(f) - if err != nil { - t.Fatalf("read %s: %v", f, err) - } - if _, err := database.Exec(string(data)); err != nil { - t.Fatalf("apply %s: %v", f, err) - } - } - if incidentsMigration == "" { - t.Fatal("008 migration not found") - } - - // A database as it would look on the old schema: an acknowledged firing - // alert with a comment on it. - if _, err := database.Exec(` - INSERT INTO users (id, username, email) VALUES (1, 'admin', 'admin@test.com'); - INSERT INTO alerts (id, fingerprint, name, status, labels, annotations, - starts_at, received_at, acknowledged_by, acknowledged_at) - VALUES (1, 'legacy-fp', 'LegacyAlert', 'firing', - '{"severity":"warning"}', '{}', 1000, 1000, 1, 1500); - INSERT INTO alert_comments (alert_id, user_id, content, created_at) - VALUES (1, 1, 'legacy comment', 1600);`); err != nil { - t.Fatalf("seed pre-008 data: %v", err) - } - - data, err := os.ReadFile(incidentsMigration) - if err != nil { - t.Fatalf("read 008: %v", err) - } - if _, err := database.Exec(string(data)); err != nil { - t.Fatalf("apply 008: %v", err) - } - - var status, groupKey string - var ackBy int64 - var severity string - if err := database.QueryRow( - "SELECT status, group_key, acknowledged_by, severity FROM incidents WHERE id = 1", - ).Scan(&status, &groupKey, &ackBy, &severity); err != nil { - t.Fatalf("read backfilled incident: %v", err) - } - if status != "acknowledged" { - t.Errorf("expected the ack to carry over as status, got %q", status) - } - if groupKey != "backfill:legacy-fp" { - t.Errorf("unexpected group_key %q", groupKey) - } - if ackBy != 1 { - t.Errorf("expected acknowledged_by 1, got %d", ackBy) - } - if severity != "warning" { - t.Errorf("expected severity carried from labels, got %q", severity) - } - - var notes int - if err := database.QueryRow( - "SELECT COUNT(*) FROM incident_events WHERE type = 'note' AND detail = 'legacy comment'", - ).Scan(¬es); err != nil { - t.Fatalf("count notes: %v", err) - } - if notes != 1 { - t.Errorf("expected the comment to become a note, got %d", notes) - } - - // And the columns that caused the ack-survives-a-re-fire bug are gone. - if _, err := database.Exec("SELECT acknowledged_by FROM alerts"); err == nil { - t.Error("expected alerts.acknowledged_by to be dropped") - } -} - func contains(haystack []string, needle string) bool { for _, s := range haystack { if s == needle { diff --git a/internal/api/middleware.go b/internal/api/middleware.go index 0a3a17c..e9e90af 100644 --- a/internal/api/middleware.go +++ b/internal/api/middleware.go @@ -70,7 +70,7 @@ func AuthMiddleware(db *sql.DB) func(http.Handler) http.Handler { func apiKeyUser(ctx context.Context, db *sql.DB, token string) (int64, bool) { var keyID, userID int64 err := db.QueryRowContext(ctx, - "SELECT id, user_id FROM api_keys WHERE key_hash = ?", hashToken(token), + "SELECT id, user_id FROM api_keys WHERE key_hash = $1", hashToken(token), ).Scan(&keyID, &userID) if err != nil { return 0, false @@ -78,7 +78,7 @@ func apiKeyUser(ctx context.Context, db *sql.DB, token string) (int64, bool) { // best-effort; don't fail the request if this update fails db.ExecContext(ctx, - "UPDATE api_keys SET last_used_at = ? WHERE id = ?", + "UPDATE api_keys SET last_used_at = $1 WHERE id = $2", time.Now().Unix(), keyID) return userID, true } @@ -91,7 +91,7 @@ func sessionUser(ctx context.Context, db *sql.DB, token string) (sessionID, user var lastSeen int64 err := db.QueryRowContext(ctx, ` SELECT id, user_id, last_seen_at FROM sessions - WHERE token_hash = ? AND expires_at > ?`, + WHERE token_hash = $1 AND expires_at > $2`, hashToken(token), now.Unix()).Scan(&sessionID, &userID, &lastSeen) if err != nil { return 0, 0, false @@ -99,7 +99,7 @@ func sessionUser(ctx context.Context, db *sql.DB, token string) (sessionID, user if now.Sub(time.Unix(lastSeen, 0)) > sessionTouchEvery { db.ExecContext(ctx, - "UPDATE sessions SET last_seen_at = ?, expires_at = ? WHERE id = ?", + "UPDATE sessions SET last_seen_at = $1, expires_at = $2 WHERE id = $3", now.Unix(), now.Add(sessionTTL).Unix(), sessionID) } return sessionID, userID, true @@ -111,7 +111,7 @@ func serveAs(w http.ResponseWriter, r *http.Request, next http.Handler, db *sql. var u models.User var createdUnix int64 if err := db.QueryRowContext(r.Context(), - "SELECT id, username, email, created_at FROM users WHERE id = ?", userID, + "SELECT id, username, email, created_at FROM users WHERE id = $1", userID, ).Scan(&u.ID, &u.Username, &u.Email, &createdUnix); err != nil { respond(w, http.StatusUnauthorized, errResp("unauthorized")) return diff --git a/internal/api/notifier.go b/internal/api/notifier.go index 7bf50d4..c341183 100644 --- a/internal/api/notifier.go +++ b/internal/api/notifier.go @@ -151,19 +151,20 @@ func enqueueReminders(ctx context.Context, db *sql.DB, cfg NotifyConfig) { JOIN incidents i ON i.id = n.incident_id WHERE n.id = (SELECT MAX(id) FROM notifications WHERE incident_id = n.incident_id) AND n.sent_at IS NOT NULL - AND n.created_at <= ? + AND n.created_at <= $1 AND i.resolved_at IS NULL AND i.archived_at IS NULL AND i.status = 'triggered' - AND (i.snoozed_until IS NULL OR i.snoozed_until <= ?)`, + AND (i.snoozed_until IS NULL OR i.snoozed_until <= $2)`, now.Add(-cfg.RepeatEvery).Unix(), now.Unix()) if err != nil { log.Printf("notifier: find reminders: %v", err) return } - // Collected before inserting: the pool is limited to a single connection, so - // an open cursor would block the writes behind it. + // Collected before inserting, rather than written while walking the cursor: + // the inserts below are what this query selects on, and a cursor reading its + // own writes is a hazard whatever the pool size. var pending []due for rows.Next() { var d due @@ -217,7 +218,7 @@ func deliverPending(ctx context.Context, db *sql.DB, cfg NotifyConfig) { continue } if _, err := db.ExecContext(ctx, - "UPDATE notifications SET sent_at = ?, attempts = attempts + 1, last_error = NULL WHERE id = ?", + "UPDATE notifications SET sent_at = $1, attempts = attempts + 1, last_error = NULL WHERE id = $2", time.Now().Unix(), n.id); err != nil { log.Printf("notifier: mark sent %d: %v", n.id, err) } @@ -240,10 +241,10 @@ func pendingNotifications(ctx context.Context, db *sql.DB) ([]outboxRow, error) SELECT id, incident_id, user_id, topic, kind, attempts FROM notifications WHERE sent_at IS NULL - AND send_after <= ? - AND attempts < ? + AND send_after <= $1 + AND attempts < $2 ORDER BY id - LIMIT ?`, time.Now().Unix(), notifyMaxAttempts, notifyBatch) + LIMIT $3`, time.Now().Unix(), notifyMaxAttempts, notifyBatch) if err != nil { return nil, err } @@ -269,7 +270,7 @@ func pendingNotifications(ctx context.Context, db *sql.DB) ([]outboxRow, error) func markFailed(ctx context.Context, db *sql.DB, n outboxRow, cause error) { next := time.Now().Add(retryDelay(n.attempts)).Unix() if _, err := db.ExecContext(ctx, - "UPDATE notifications SET attempts = attempts + 1, send_after = ?, last_error = ? WHERE id = ?", + "UPDATE notifications SET attempts = attempts + 1, send_after = $1, last_error = $2 WHERE id = $3", next, cause.Error(), n.id); err != nil { log.Printf("notifier: mark failed %d: %v", n.id, err) } @@ -307,7 +308,7 @@ func deliver(ctx context.Context, db *sql.DB, cfg NotifyConfig, n outboxRow) err SELECT COUNT(*) FROM incident_alerts ia JOIN alerts a ON a.id = ia.alert_id - WHERE ia.incident_id = ? AND a.status = 'firing'`, n.incidentID).Scan(&firing); err != nil { + WHERE ia.incident_id = $1 AND a.status = 'firing'`, n.incidentID).Scan(&firing); err != nil { return fmt.Errorf("count firing: %w", err) } @@ -475,7 +476,7 @@ func enqueueNotification(ctx context.Context, q querier, incidentID int64, userI now := time.Now().Unix() _, err := q.ExecContext(ctx, ` INSERT INTO notifications (incident_id, user_id, topic, kind, created_at, send_after) - VALUES (?, ?, ?, ?, ?, ?)`, incidentID, userID, topic, kind, now, now) + VALUES ($1, $2, $3, $4, $5, $6)`, incidentID, userID, topic, kind, now, now) return err } @@ -489,7 +490,7 @@ func notifyTarget(ctx context.Context, q querier, cfg NotifyConfig, onCall *int6 if onCall != nil { var t *string err := q.QueryRowContext(ctx, - "SELECT ntfy_topic FROM users WHERE id = ?", *onCall).Scan(&t) + "SELECT ntfy_topic FROM users WHERE id = $1", *onCall).Scan(&t) if err == nil && t != nil && *t != "" { return *t, onCall } @@ -522,7 +523,7 @@ func enqueueResolved(ctx context.Context, q querier, incidentID int64) error { var userID *int64 err := q.QueryRowContext(ctx, ` SELECT topic, user_id FROM notifications - WHERE incident_id = ? ORDER BY id DESC LIMIT 1`, incidentID).Scan(&topic, &userID) + WHERE incident_id = $1 ORDER BY id DESC LIMIT 1`, incidentID).Scan(&topic, &userID) if err == sql.ErrNoRows { return nil } diff --git a/internal/api/notify_ack.go b/internal/api/notify_ack.go index f624138..a23c43f 100644 --- a/internal/api/notify_ack.go +++ b/internal/api/notify_ack.go @@ -28,7 +28,7 @@ func issueAckToken(ctx context.Context, q querier, incidentID, userID int64) (st now := time.Now() if _, err := q.ExecContext(ctx, ` INSERT INTO incident_ack_tokens (token_hash, incident_id, user_id, created_at, expires_at) - VALUES (?, ?, ?, ?, ?)`, + VALUES ($1, $2, $3, $4, $5)`, hash, incidentID, userID, now.Unix(), now.Add(ackTokenTTL).Unix()); err != nil { return "", err } @@ -51,7 +51,7 @@ func handleNotifyAck(db *sql.DB) http.HandlerFunc { var incidentID, userID int64 err := db.QueryRowContext(r.Context(), ` SELECT incident_id, user_id FROM incident_ack_tokens - WHERE token_hash = ? AND expires_at > ?`, + WHERE token_hash = $1 AND expires_at > $2`, hash, time.Now().Unix()).Scan(&incidentID, &userID) if err != nil { // Unknown and expired get the same answer, so the endpoint cannot be @@ -87,7 +87,7 @@ func handleNotifyAck(db *sql.DB) http.HandlerFunc { // fires in practice. func purgeAckTokens(ctx context.Context, db *sql.DB) { res, err := db.ExecContext(ctx, - "DELETE FROM incident_ack_tokens WHERE expires_at < ?", time.Now().Unix()) + "DELETE FROM incident_ack_tokens WHERE expires_at < $1", time.Now().Unix()) if err != nil { log.Printf("sweeper: purge ack tokens: %v", err) return diff --git a/internal/api/notify_test.go b/internal/api/notify_test.go index cb661ef..7b29a2f 100644 --- a/internal/api/notify_test.go +++ b/internal/api/notify_test.go @@ -125,7 +125,7 @@ func (s *ts) countNotifications(t *testing.T, kind string) int { query := "SELECT COUNT(*) FROM notifications" args := []any{} if kind != "" { - query += " WHERE kind = ?" + query += " WHERE kind = $1" args = append(args, kind) } if err := s.db.QueryRow(query, args...).Scan(&n); err != nil { @@ -326,7 +326,7 @@ func TestNotify_ExhaustedRetriesAreRecordedOnce(t *testing.T) { // One pass per attempt, each made due by clearing the backoff the last one set. for i := 0; i < 10; i++ { s.sweepNotify(t) - s.exec(t, "UPDATE notifications SET send_after = ? WHERE sent_at IS NULL", + s.exec(t, "UPDATE notifications SET send_after = $1 WHERE sent_at IS NULL", time.Now().Add(-time.Second).Unix()) } @@ -415,7 +415,7 @@ func TestNotify_AckRejectsExpiredToken(t *testing.T) { // Age the token past its TTL. The token's inputs are wall-clock timestamps, // so this is the same trick the sweeper tests use. - s.exec(t, "UPDATE incident_ack_tokens SET expires_at = ?", time.Now().Add(-time.Minute).Unix()) + s.exec(t, "UPDATE incident_ack_tokens SET expires_at = $1", time.Now().Add(-time.Minute).Unix()) resp, err := http.Post(s.URL+path, "application/json", nil) if err != nil { @@ -436,7 +436,7 @@ func TestNotify_SweepPurgesExpiredAckTokens(t *testing.T) { fireCritical(t, s) s.sweepNotify(t) - s.exec(t, "UPDATE incident_ack_tokens SET expires_at = ?", time.Now().Add(-time.Minute).Unix()) + s.exec(t, "UPDATE incident_ack_tokens SET expires_at = $1", time.Now().Add(-time.Minute).Unix()) api.Sweep(context.Background(), s.db, 168*time.Hour, 6*time.Hour, s.deadman, s.notify) @@ -457,7 +457,7 @@ func TestNotify_SweepPurgesExpiredAckTokens(t *testing.T) { // reminder as due. func (s *ts) ageNotifications(t *testing.T, by time.Duration) { t.Helper() - s.exec(t, "UPDATE notifications SET created_at = ? WHERE sent_at IS NOT NULL", + s.exec(t, "UPDATE notifications SET created_at = $1 WHERE sent_at IS NOT NULL", time.Now().Add(-by).Unix()) } @@ -693,7 +693,7 @@ func TestNotify_FailedDeliveryRetriesWithBackoff(t *testing.T) { // Once due and once ntfy recovers, it goes out. f.failWith(http.StatusOK) - s.exec(t, "UPDATE notifications SET send_after = ? WHERE id = 1", time.Now().Add(-time.Second).Unix()) + s.exec(t, "UPDATE notifications SET send_after = $1 WHERE id = 1", time.Now().Add(-time.Second).Unix()) s.sweepNotify(t) if err := s.db.QueryRow("SELECT sent_at FROM notifications WHERE id = 1").Scan(&sentAt); err != nil { @@ -715,7 +715,7 @@ func TestNotify_UnsentNotificationBlocksReminders(t *testing.T) { fireCritical(t, s) s.sweepNotify(t) - s.exec(t, "UPDATE notifications SET created_at = ?", time.Now().Add(-time.Hour).Unix()) + s.exec(t, "UPDATE notifications SET created_at = $1", time.Now().Add(-time.Hour).Unix()) s.sweepNotify(t) if got := s.countNotifications(t, "reminder"); got != 0 { diff --git a/internal/api/schedule.go b/internal/api/schedule.go index a7aafad..9e96501 100644 --- a/internal/api/schedule.go +++ b/internal/api/schedule.go @@ -45,7 +45,7 @@ func handleCreateSchedule(db *sql.DB) http.HandlerFunc { // Verify the user exists. var exists int - if err := db.QueryRowContext(r.Context(), "SELECT 1 FROM users WHERE id = ?", req.UserID).Scan(&exists); err != nil { + if err := db.QueryRowContext(r.Context(), "SELECT 1 FROM users WHERE id = $1", req.UserID).Scan(&exists); err != nil { respond(w, http.StatusNotFound, errResp("user not found")) return } @@ -64,14 +64,14 @@ func handleCreateSchedule(db *sql.DB) http.HandlerFunc { for _, d := range req.Dates { if req.Replace { if _, err := tx.ExecContext(r.Context(), - "DELETE FROM schedule_entries WHERE date = ?", d); err != nil { + "DELETE FROM schedule_entries WHERE date = $1", d); err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return } } if _, err := tx.ExecContext(r.Context(), - "INSERT INTO schedule_entries (user_id, date) VALUES (?, ?)", req.UserID, d); err != nil { - if strings.Contains(err.Error(), "UNIQUE constraint failed") { + "INSERT INTO schedule_entries (user_id, date) VALUES ($1, $2)", req.UserID, d); err != nil { + if isUniqueViolation(err) { respond(w, http.StatusConflict, errResp("date already assigned: "+d+" (pass replace to take it)")) return @@ -139,7 +139,7 @@ func handleDeleteSchedule(db *sql.DB) http.HandlerFunc { respond(w, http.StatusBadRequest, errResp("invalid schedule id")) return } - res, err := db.ExecContext(r.Context(), "DELETE FROM schedule_entries WHERE id = ?", id) + res, err := db.ExecContext(r.Context(), "DELETE FROM schedule_entries WHERE id = $1", id) if err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return @@ -162,7 +162,7 @@ func handleCurrentSchedule(db *sql.DB) http.HandlerFunc { SELECT s.id, s.user_id, u.username, s.date, s.created_at FROM schedule_entries s JOIN users u ON u.id = s.user_id - WHERE s.date = ?`, today).Scan(&e.ID, &e.UserID, &e.Username, &e.Date, &ts) + WHERE s.date = $1`, today).Scan(&e.ID, &e.UserID, &e.Username, &e.Date, &ts) if err == sql.ErrNoRows { respond(w, http.StatusNotFound, errResp("no one is on call today")) return @@ -180,14 +180,12 @@ func handleCurrentSchedule(db *sql.DB) http.HandlerFunc { // from and to are YYYY-MM-DD strings; an empty string means unbounded on that side. func scheduleRange(ctx context.Context, db *sql.DB, from, to string) ([]models.ScheduleEntry, error) { where := []string{} - args := []any{} + args := &sqlArgs{} if from != "" { - where = append(where, "s.date >= ?") - args = append(args, from) + where = append(where, "s.date >= "+args.add(from)) } if to != "" { - where = append(where, "s.date <= ?") - args = append(args, to) + where = append(where, "s.date <= "+args.add(to)) } clause := "1=1" @@ -200,7 +198,7 @@ func scheduleRange(ctx context.Context, db *sql.DB, from, to string) ([]models.S FROM schedule_entries s JOIN users u ON u.id = s.user_id WHERE `+clause+` - ORDER BY s.date ASC`, args...) + ORDER BY s.date ASC`, args.all()...) if err != nil { return nil, err } diff --git a/internal/api/stats.go b/internal/api/stats.go index 2572994..01f0ce6 100644 --- a/internal/api/stats.go +++ b/internal/api/stats.go @@ -21,7 +21,7 @@ func handleStatsAlerts(db *sql.DB) http.HandlerFunc { SELECT COUNT(*), COALESCE(SUM(CASE WHEN status = 'firing' THEN 1 ELSE 0 END), 0), COALESCE(SUM(CASE WHEN status = 'resolved' THEN 1 ELSE 0 END), 0) - FROM alerts WHERE %s`, where), args..., + FROM alerts WHERE %s`, where), args.all()..., ).Scan(&total, &firing, &resolved) if err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) @@ -47,15 +47,13 @@ func handleStatsTop(db *sql.DB) http.HandlerFunc { limit = n } } - args = append(args, limit) - rows, err := db.QueryContext(r.Context(), fmt.Sprintf(` SELECT name, COUNT(*) AS cnt FROM alerts WHERE %s GROUP BY name ORDER BY cnt DESC - LIMIT ?`, where), args...) + LIMIT %s`, where, args.add(limit)), args.all()...) if err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return @@ -84,12 +82,12 @@ func handleStatsByHour(db *sql.DB) http.HandlerFunc { where, args := statsFilter(r.URL.Query(), "received_at") rows, err := db.QueryContext(r.Context(), fmt.Sprintf(` - SELECT CAST(strftime('%%H', datetime(received_at, 'unixepoch')) AS INTEGER) AS hr, + SELECT EXTRACT(HOUR FROM to_timestamp(received_at) AT TIME ZONE 'UTC')::int AS hr, COUNT(*) AS cnt FROM alerts WHERE %s GROUP BY hr - ORDER BY hr ASC`, where), args...) + ORDER BY hr ASC`, where), args.all()...) if err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return @@ -123,14 +121,15 @@ func handleStatsByDay(db *sql.DB) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { where, args := statsFilter(r.URL.Query(), "received_at") - // SQLite strftime('%w') → 0=Sunday … 6=Saturday + // Postgres EXTRACT(DOW …) → 0=Sunday … 6=Saturday, the same numbering + // SQLite's strftime('%w') returned, so the frontend needs no change. rows, err := db.QueryContext(r.Context(), fmt.Sprintf(` - SELECT CAST(strftime('%%w', datetime(received_at, 'unixepoch')) AS INTEGER) AS dow, + SELECT EXTRACT(DOW FROM to_timestamp(received_at) AT TIME ZONE 'UTC')::int AS dow, COUNT(*) AS cnt FROM alerts WHERE %s GROUP BY dow - ORDER BY dow ASC`, where), args...) + ORDER BY dow ASC`, where), args.all()...) if err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return @@ -184,7 +183,7 @@ func handleStatsIncidents(db *sql.DB) http.HandlerFunc { THEN acknowledged_at - triggered_at END), AVG(CASE WHEN resolved_at IS NOT NULL THEN resolved_at - triggered_at END) - FROM incidents WHERE %s`, where), args..., + FROM incidents WHERE %s`, where), args.all()..., ).Scan(&total, &triggered, &acknowledged, &resolved, &mtta, &mttr) if err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) @@ -207,18 +206,17 @@ func handleStatsIncidents(db *sql.DB) http.HandlerFunc { // statsFilter builds a WHERE clause and args from optional ?from and ?to query // params, filtering on timeCol. Archived rows are always excluded, matching the // default list views. -func statsFilter(q url.Values, timeCol string) (where string, args []any) { +func statsFilter(q url.Values, timeCol string) (where string, args *sqlArgs) { + args = &sqlArgs{} clauses := []string{"archived_at IS NULL"} if from := q.Get("from"); from != "" { if t, err := time.Parse("2006-01-02", from); err == nil { - clauses = append(clauses, timeCol+" >= ?") - args = append(args, t.UTC().Unix()) + clauses = append(clauses, timeCol+" >= "+args.add(t.UTC().Unix())) } } if to := q.Get("to"); to != "" { if t, err := time.Parse("2006-01-02", to); err == nil { - clauses = append(clauses, timeCol+" < ?") - args = append(args, t.UTC().AddDate(0, 0, 1).Unix()) + clauses = append(clauses, timeCol+" < "+args.add(t.UTC().AddDate(0, 0, 1).Unix())) } } return strings.Join(clauses, " AND "), args diff --git a/internal/api/testdb_test.go b/internal/api/testdb_test.go new file mode 100644 index 0000000..d8f4c9a --- /dev/null +++ b/internal/api/testdb_test.go @@ -0,0 +1,100 @@ +package api_test + +import ( + "database/sql" + "fmt" + "net/url" + "os" + "strings" + "testing" + + "git.ryuvia.com/niklas/terdut-server/internal/db" +) + +// Tests run against a real Postgres, because the server does. SQLite's +// ":memory:" gave every test a private database for free; Postgres has no +// equivalent, so isolation is bought with a schema per test. +// +// A schema rather than a database: CREATE DATABASE copies a template on disk and +// costs a hundred milliseconds or so each time, while CREATE SCHEMA plus the one +// baseline migration is a few, and the suite runs a few hundred of them. Each +// test's pool is pinned to its own schema through search_path, so two tests +// cannot see each other's rows even though they share a server. +// +// TERDUT_TEST_DSN must point at a database the test role may create schemas in: +// +// postgres://terdut:terdut@localhost:5432/terdut_test?sslmode=disable +// +// `make test-db` starts one locally; ci.yaml runs one as a service container. +// An unset DSN fails rather than skips, deliberately — a suite that quietly +// tests nothing is worse than one that does not run. +const testDSNEnv = "TERDUT_TEST_DSN" + +var schemaSeq int + +// newTestDB returns a migrated database private to this test, and drops it +// afterwards. +func newTestDB(t *testing.T) *sql.DB { + t.Helper() + + dsn := os.Getenv(testDSNEnv) + if dsn == "" { + t.Fatalf("%s is not set: these tests need Postgres.\n"+ + "Run `make test-db` for a local one, then\n"+ + " export %s=postgres://terdut:terdut@localhost:5432/terdut_test?sslmode=disable", + testDSNEnv, testDSNEnv) + } + + schemaSeq++ + schema := fmt.Sprintf("test_%d_%d", os.Getpid(), schemaSeq) + + admin, err := sql.Open("pgx", dsn) + if err != nil { + t.Fatalf("connect to %s: %v", testDSNEnv, err) + } + defer admin.Close() + if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil { + t.Fatalf("create schema %s: %v", schema, err) + } + + database, err := db.Open(withSearchPath(dsn, schema)) + if err != nil { + t.Fatalf("open db: %v", err) + } + if err := db.Migrate(database); err != nil { + t.Fatalf("migrate: %v", err) + } + + t.Cleanup(func() { + database.Close() + cleanup, err := sql.Open("pgx", dsn) + if err != nil { + return + } + defer cleanup.Close() + if _, err := cleanup.Exec("DROP SCHEMA " + schema + " CASCADE"); err != nil { + t.Logf("drop schema %s: %v", schema, err) + } + }) + + return database +} + +// withSearchPath pins a DSN to one schema, so every connection the pool opens +// lands there and nothing has to qualify a table name. +// +// Handles both DSN spellings: a postgres:// URL, and libpq's keyword/value form. +func withSearchPath(dsn, schema string) string { + opt := "-csearch_path=" + schema + + if strings.HasPrefix(dsn, "postgres://") || strings.HasPrefix(dsn, "postgresql://") { + u, err := url.Parse(dsn) + if err == nil { + q := u.Query() + q.Set("options", opt) + u.RawQuery = q.Encode() + return u.String() + } + } + return dsn + " options='" + opt + "'" +} diff --git a/internal/api/users.go b/internal/api/users.go index 8338949..e685589 100644 --- a/internal/api/users.go +++ b/internal/api/users.go @@ -56,27 +56,26 @@ func handleBootstrap(db *sql.DB) http.HandlerFunc { return } - res, err := db.ExecContext(r.Context(), - "INSERT INTO users (username, email, password_hash) VALUES (?, ?, ?)", - req.Username, req.Email, passwordHash) - if err != nil { + var userID int64 + if err := db.QueryRowContext(r.Context(), + "INSERT INTO users (username, email, password_hash) VALUES ($1, $2, $3) RETURNING id", + req.Username, req.Email, passwordHash).Scan(&userID); err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return } - userID, _ := res.LastInsertId() raw, hash, err := randomToken() if err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return } - keyRes, err := db.ExecContext(r.Context(), - "INSERT INTO api_keys (user_id, key_hash, name) VALUES (?, ?, ?)", userID, hash, "bootstrap") - if err != nil { + var keyID int64 + if err := db.QueryRowContext(r.Context(), + "INSERT INTO api_keys (user_id, key_hash, name) VALUES ($1, $2, $3) RETURNING id", + userID, hash, "bootstrap").Scan(&keyID); err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return } - keyID, _ := keyRes.LastInsertId() user, _ := fetchUser(r.Context(), db, userID) key := models.APIKey{ID: keyID, UserID: userID, Name: "bootstrap", Key: raw, CreatedAt: user.CreatedAt} @@ -124,17 +123,17 @@ func handleCreateUser(db *sql.DB) http.HandlerFunc { return } - res, err := db.ExecContext(r.Context(), - "INSERT INTO users (username, email) VALUES (?, ?)", req.Username, req.Email) - if err != nil { - if strings.Contains(err.Error(), "UNIQUE constraint failed") { + var id int64 + if err := db.QueryRowContext(r.Context(), + "INSERT INTO users (username, email) VALUES ($1, $2) RETURNING id", + req.Username, req.Email).Scan(&id); err != nil { + if isUniqueViolation(err) { respond(w, http.StatusConflict, errResp("username or email already exists")) return } respond(w, http.StatusInternalServerError, errResp("internal error")) return } - id, _ := res.LastInsertId() user, _ := fetchUser(r.Context(), db, id) respond(w, http.StatusCreated, user) } @@ -165,7 +164,7 @@ func handleSetNotifyTarget(db *sql.DB) http.HandlerFunc { } res, err := db.ExecContext(r.Context(), - "UPDATE users SET ntfy_topic = ? WHERE id = ?", topic, id) + "UPDATE users SET ntfy_topic = $1 WHERE id = $2", topic, id) if err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return @@ -191,7 +190,7 @@ func handleDeleteUser(db *sql.DB) http.HandlerFunc { respond(w, http.StatusBadRequest, errResp("invalid user id")) return } - res, err := db.ExecContext(r.Context(), "DELETE FROM users WHERE id = ?", id) + res, err := db.ExecContext(r.Context(), "DELETE FROM users WHERE id = $1", id) if err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return @@ -226,7 +225,7 @@ func handleCreateAPIKey(db *sql.DB) http.HandlerFunc { } var exists int - if err := db.QueryRowContext(r.Context(), "SELECT 1 FROM users WHERE id = ?", userID).Scan(&exists); err != nil { + if err := db.QueryRowContext(r.Context(), "SELECT 1 FROM users WHERE id = $1", userID).Scan(&exists); err != nil { respond(w, http.StatusNotFound, errResp("user not found")) return } @@ -236,13 +235,13 @@ func handleCreateAPIKey(db *sql.DB) http.HandlerFunc { respond(w, http.StatusInternalServerError, errResp("internal error")) return } - res, err := db.ExecContext(r.Context(), - "INSERT INTO api_keys (user_id, key_hash, name) VALUES (?, ?, ?)", userID, hash, req.Name) - if err != nil { + var keyID int64 + if err := db.QueryRowContext(r.Context(), + "INSERT INTO api_keys (user_id, key_hash, name) VALUES ($1, $2, $3) RETURNING id", + userID, hash, req.Name).Scan(&keyID); err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return } - keyID, _ := res.LastInsertId() key := models.APIKey{ID: keyID, UserID: userID, Name: req.Name, Key: raw, CreatedAt: time.Now().UTC()} respond(w, http.StatusCreated, key) } @@ -262,7 +261,7 @@ func handleDeleteAPIKey(db *sql.DB) http.HandlerFunc { } res, err := db.ExecContext(r.Context(), - "DELETE FROM api_keys WHERE id = ? AND user_id = ?", keyID, userID) + "DELETE FROM api_keys WHERE id = $1 AND user_id = $2", keyID, userID) if err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return @@ -293,7 +292,7 @@ func fetchUser(ctx context.Context, db *sql.DB, id int64) (models.User, error) { var u models.User var ts int64 err := db.QueryRowContext(ctx, - "SELECT id, username, email, created_at, ntfy_topic FROM users WHERE id = ?", id). + "SELECT id, username, email, created_at, ntfy_topic FROM users WHERE id = $1", id). Scan(&u.ID, &u.Username, &u.Email, &ts, &u.NtfyTopic) if err != nil { return u, err diff --git a/internal/config/config.go b/internal/config/config.go index 590fdc4..a148623 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -6,8 +6,15 @@ import ( ) type Config struct { - Addr string - DBPath string + Addr string + + // DSN is the Postgres connection string, e.g. + // postgres://terdut:secret@host:5432/terdut?sslmode=require. Required: + // unlike the SQLite path it replaced there is no sensible default, and a + // server that silently came up against the wrong database would be worse + // than one that refuses to start. + DSN string + ArchiveAfter time.Duration // StaleAfter is how long a firing alert may go without a refreshing webhook @@ -59,10 +66,6 @@ func Load() Config { if addr == "" { addr = ":8080" } - dbPath := os.Getenv("TERDUT_DB_PATH") - if dbPath == "" { - dbPath = "terdut.db" - } deadmanMatchers := os.Getenv("TERDUT_DEADMAN_MATCHERS") if deadmanMatchers == "" { deadmanMatchers = "alertname=Watchdog" @@ -73,7 +76,7 @@ func Load() Config { } return Config{ Addr: addr, - DBPath: dbPath, + DSN: os.Getenv("TERDUT_DB_DSN"), ArchiveAfter: duration("TERDUT_ARCHIVE_AFTER", 7*24*time.Hour), StaleAfter: duration("TERDUT_STALE_AFTER", 6*time.Hour), diff --git a/internal/db/db.go b/internal/db/db.go index 9ec4054..c696df4 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -7,24 +7,32 @@ import ( "io/fs" "sort" "strings" + "time" - _ "modernc.org/sqlite" + _ "github.com/jackc/pgx/v5/stdlib" ) //go:embed migrations var migrationsFS embed.FS -func Open(path string) (*sql.DB, error) { - db, err := sql.Open("sqlite", path) +// Open connects to Postgres. dsn is a libpq connection string or URL, e.g. +// postgres://terdut:secret@localhost:5432/terdut?sslmode=disable. +// +// The pool is modest on purpose: this server's concurrency comes from a handful +// of HTTP handlers plus two background loops, and a cloud-native-pg instance +// sized for it has a low max_connections. It is still a pool, unlike the single +// connection SQLite forced, so the notifier no longer blocks a webhook. +func Open(dsn string) (*sql.DB, error) { + if dsn == "" { + return nil, fmt.Errorf("empty DSN: set TERDUT_DB_DSN") + } + db, err := sql.Open("pgx", dsn) if err != nil { return nil, err } - // SQLite does not support concurrent writers; a single connection avoids locking errors. - db.SetMaxOpenConns(1) - if _, err := db.Exec("PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON;"); err != nil { - db.Close() - return nil, fmt.Errorf("set pragmas: %w", err) - } + db.SetMaxOpenConns(10) + db.SetMaxIdleConns(5) + db.SetConnMaxLifetime(time.Hour) if err := db.Ping(); err != nil { db.Close() return nil, fmt.Errorf("ping: %w", err) @@ -32,10 +40,16 @@ func Open(path string) (*sql.DB, error) { return db, nil } +// Migrate applies every embedded migration that has not been applied yet, in +// filename order, recording each in schema_migrations. +// +// Each file runs inside a transaction, which SQLite's version did not do: a +// migration that failed half way used to leave the schema in whatever state it +// had reached. Postgres has transactional DDL, so the rollback is real. func Migrate(db *sql.DB) error { if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS schema_migrations ( version TEXT PRIMARY KEY, - applied_at DATETIME DEFAULT CURRENT_TIMESTAMP + applied_at BIGINT NOT NULL DEFAULT FLOOR(EXTRACT(EPOCH FROM now()))::bigint )`); err != nil { return fmt.Errorf("create schema_migrations: %w", err) } @@ -55,7 +69,7 @@ func Migrate(db *sql.DB) error { for _, name := range files { var count int - if err := db.QueryRow("SELECT COUNT(*) FROM schema_migrations WHERE version = ?", name).Scan(&count); err != nil { + if err := db.QueryRow("SELECT COUNT(*) FROM schema_migrations WHERE version = $1", name).Scan(&count); err != nil { return fmt.Errorf("check migration %s: %w", name, err) } if count > 0 { @@ -67,13 +81,28 @@ func Migrate(db *sql.DB) error { return fmt.Errorf("read migration %s: %w", name, err) } - if _, err := db.Exec(string(data)); err != nil { - return fmt.Errorf("apply migration %s: %w", name, err) - } - - if _, err := db.Exec("INSERT INTO schema_migrations (version) VALUES (?)", name); err != nil { - return fmt.Errorf("record migration %s: %w", name, err) + if err := applyMigration(db, name, string(data)); err != nil { + return err } } return nil } + +func applyMigration(db *sql.DB, name, body string) error { + tx, err := db.Begin() + if err != nil { + return fmt.Errorf("begin migration %s: %w", name, err) + } + defer tx.Rollback() + + if _, err := tx.Exec(body); err != nil { + return fmt.Errorf("apply migration %s: %w", name, err) + } + if _, err := tx.Exec("INSERT INTO schema_migrations (version) VALUES ($1)", name); err != nil { + return fmt.Errorf("record migration %s: %w", name, err) + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit migration %s: %w", name, err) + } + return nil +} diff --git a/internal/db/migrations/001_baseline.sql b/internal/db/migrations/001_baseline.sql new file mode 100644 index 0000000..2cfb54e --- /dev/null +++ b/internal/db/migrations/001_baseline.sql @@ -0,0 +1,182 @@ +-- The Postgres baseline: the schema as it stood at the end of the SQLite line, +-- in one file rather than ten. +-- +-- The ten SQLite migrations are in git history up to the commit that introduced +-- this one, and they replay against nothing here: their shape was incremental +-- (columns added, then dropped again in 008) and 008's backfill rewrote data +-- that a Postgres install never had. An existing SQLite database is carried over +-- by scripts/sqlite-to-postgres.go, which copies rows into this schema. +-- +-- Two conventions inherited deliberately: +-- +-- * Timestamps are BIGINT unix seconds, not timestamptz. Everything in Go +-- already speaks epochs, and converting was a second change riding along +-- with the port. Worth revisiting on its own. +-- +-- * Ids are GENERATED BY DEFAULT, not ALWAYS, so the migration script can +-- insert rows with their original ids and keep every foreign key intact. +-- setval at the end of the copy puts the sequences past them. + +CREATE TABLE users ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + username TEXT NOT NULL UNIQUE, + email TEXT NOT NULL UNIQUE, + created_at BIGINT NOT NULL DEFAULT FLOOR(EXTRACT(EPOCH FROM now()))::bigint, + -- Where this user's notifications go. NULL means they get none; incidents + -- assigned to them fall back to the configured fallback topic. + ntfy_topic TEXT, + -- NULL means the user has no password and can only use API keys. + password_hash TEXT +); + +CREATE TABLE api_keys ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + key_hash TEXT NOT NULL UNIQUE, + name TEXT NOT NULL, + created_at BIGINT NOT NULL DEFAULT FLOOR(EXTRACT(EPOCH FROM now()))::bigint, + last_used_at BIGINT +); + +-- A session is a browser's credential, the cookie counterpart of an API key: +-- only the hash of the token is stored. expires_at slides forward while the +-- session is in use, so an on-call phone stays signed in. +CREATE TABLE sessions ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + token_hash TEXT NOT NULL UNIQUE, + user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + created_at BIGINT NOT NULL, + last_seen_at BIGINT NOT NULL, + expires_at BIGINT NOT NULL, + user_agent TEXT +); + +CREATE INDEX idx_sessions_user ON sessions(user_id); + +-- The machine-owned signal record: what Alertmanager says is true right now. +-- Workflow state lives on incidents, never here, because the webhook upsert owns +-- these rows and would overwrite it. +CREATE TABLE alerts ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + fingerprint TEXT NOT NULL UNIQUE, + name TEXT NOT NULL, + status TEXT NOT NULL CHECK (status IN ('firing', 'resolved')), + labels JSONB NOT NULL DEFAULT '{}'::jsonb, + annotations JSONB NOT NULL DEFAULT '{}'::jsonb, + starts_at BIGINT NOT NULL, + ends_at BIGINT, + generator_url TEXT NOT NULL DEFAULT '', + received_at BIGINT NOT NULL DEFAULT FLOOR(EXTRACT(EPOCH FROM now()))::bigint, + archived_at BIGINT, + -- Why the alert left the firing state: 'alertmanager' when a resolved + -- webhook set it, 'expiry' when the sweeper inferred it from staleness. + resolution_source TEXT +); + +CREATE INDEX alerts_status_idx ON alerts(status); +CREATE INDEX alerts_name_idx ON alerts(name); +CREATE INDEX alerts_received_at_idx ON alerts(received_at DESC); +CREATE INDEX alerts_archived_at_idx ON alerts(archived_at); + +CREATE TABLE schedule_entries ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + date TEXT NOT NULL UNIQUE, -- YYYY-MM-DD; one person per day + created_at BIGINT NOT NULL DEFAULT FLOOR(EXTRACT(EPOCH FROM now()))::bigint +); + +CREATE INDEX schedule_entries_date_idx ON schedule_entries(date); + +-- The human work item: what people acknowledge, assign, snooze, discuss and +-- resolve. Correlation uses Alertmanager's own groupKey, so incidents follow the +-- group_by routing tree the operator already tuned. +CREATE TABLE incidents ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + group_key TEXT NOT NULL, -- Alertmanager groupKey, opaque + title TEXT NOT NULL, -- rendered from group_labels + group_labels JSONB NOT NULL DEFAULT '{}'::jsonb, + status TEXT NOT NULL CHECK (status IN ('triggered', 'acknowledged', 'resolved')), + severity TEXT, -- highest `severity` label across firing members + triggered_at BIGINT NOT NULL, + acknowledged_by BIGINT REFERENCES users(id) ON DELETE SET NULL, + acknowledged_at BIGINT, + assigned_to BIGINT REFERENCES users(id) ON DELETE SET NULL, + snoozed_until BIGINT, + resolved_at BIGINT, + resolution_source TEXT, -- 'alerts' | 'manual' + archived_at BIGINT +); + +-- Load-bearing: at most one OPEN incident per group_key. This is what makes +-- "resolved incident + a new alert occurrence = a new incident" work, and it is +-- the constraint the webhook's find-or-open lookup relies on. +CREATE UNIQUE INDEX incidents_open_group_key_idx ON incidents(group_key) WHERE resolved_at IS NULL; +CREATE INDEX incidents_status_idx ON incidents(status); +CREATE INDEX incidents_triggered_at_idx ON incidents(triggered_at DESC); +CREATE INDEX incidents_archived_at_idx ON incidents(archived_at); + +-- Membership is historical, not a pointer on alerts: one alert row (one +-- fingerprint) resolves and re-fires over time and belongs to a different +-- incident each occurrence. +CREATE TABLE incident_alerts ( + incident_id BIGINT NOT NULL REFERENCES incidents(id) ON DELETE CASCADE, + alert_id BIGINT NOT NULL REFERENCES alerts(id) ON DELETE CASCADE, + added_at BIGINT NOT NULL DEFAULT FLOOR(EXTRACT(EPOCH FROM now()))::bigint, + PRIMARY KEY (incident_id, alert_id) +); + +CREATE INDEX incident_alerts_alert_id_idx ON incident_alerts(alert_id); + +-- The timeline. Append-only, and the only history this server keeps: alert rows +-- are mutated in place, so without this there is no record that anything +-- happened. Notes are events too, so one query renders the whole story. +CREATE TABLE incident_events ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + incident_id BIGINT NOT NULL REFERENCES incidents(id) ON DELETE CASCADE, + -- triggered | alert_added | alert_resolved | acknowledged | unacknowledged + -- | assigned | snoozed | unsnoozed | resolved | note | notified | notify_failed + type TEXT NOT NULL, + user_id BIGINT REFERENCES users(id) ON DELETE SET NULL, -- NULL = the server acted + alert_id BIGINT REFERENCES alerts(id) ON DELETE SET NULL, + detail TEXT, + created_at BIGINT NOT NULL DEFAULT FLOOR(EXTRACT(EPOCH FROM now()))::bigint +); + +CREATE INDEX incident_events_incident_idx ON incident_events(incident_id, created_at); + +-- Delivery is an outbox rather than an inline HTTP call: a POST made while +-- holding the webhook's transaction would hold a connection open across a +-- network round trip. The webhook inserts a row; the notifier goroutine +-- delivers it. +CREATE TABLE notifications ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + incident_id BIGINT NOT NULL REFERENCES incidents(id) ON DELETE CASCADE, + -- Nullable: a notification sent to the fallback topic belongs to nobody, + -- because nobody was on call when the incident opened. + user_id BIGINT REFERENCES users(id) ON DELETE SET NULL, + topic TEXT NOT NULL, -- resolved at enqueue: who was on call then + kind TEXT NOT NULL CHECK (kind IN ('triggered', 'reminder', 'resolved')), + created_at BIGINT NOT NULL, + send_after BIGINT NOT NULL, -- retry backoff watermark + attempts BIGINT NOT NULL DEFAULT 0, + sent_at BIGINT, + last_error TEXT -- kept after the last attempt, for debugging +); + +-- The delivery loop's only query: what is due and still unsent. +CREATE INDEX notifications_pending_idx ON notifications(send_after) WHERE sent_at IS NULL; +-- Reminders and resolved notices both look up an incident's newest row. +CREATE INDEX notifications_incident_idx ON notifications(incident_id, id DESC); + +-- A notification body is stored on the ntfy server and cached on the device, so +-- a real API key must never appear in one. Each delivery mints its own token +-- instead: one incident, one action, one day. +CREATE TABLE incident_ack_tokens ( + token_hash TEXT PRIMARY KEY, -- SHA-256 of the raw token, as with api_keys + incident_id BIGINT NOT NULL REFERENCES incidents(id) ON DELETE CASCADE, + user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + created_at BIGINT NOT NULL, + expires_at BIGINT NOT NULL +); + +CREATE INDEX incident_ack_tokens_expires_idx ON incident_ack_tokens(expires_at); diff --git a/internal/db/migrations/001_foundation.sql b/internal/db/migrations/001_foundation.sql deleted file mode 100644 index 747d30a..0000000 --- a/internal/db/migrations/001_foundation.sql +++ /dev/null @@ -1,2 +0,0 @@ --- Stage 1 foundation. No tables yet; subsequent migrations add schema. -SELECT 1; diff --git a/internal/db/migrations/002_users_api_keys.sql b/internal/db/migrations/002_users_api_keys.sql deleted file mode 100644 index 5e5a9df..0000000 --- a/internal/db/migrations/002_users_api_keys.sql +++ /dev/null @@ -1,15 +0,0 @@ -CREATE TABLE users ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - username TEXT NOT NULL UNIQUE, - email TEXT NOT NULL UNIQUE, - created_at INTEGER NOT NULL DEFAULT (strftime('%s', 'now')) -); - -CREATE TABLE api_keys ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, - key_hash TEXT NOT NULL UNIQUE, - name TEXT NOT NULL, - created_at INTEGER NOT NULL DEFAULT (strftime('%s', 'now')), - last_used_at INTEGER -); diff --git a/internal/db/migrations/003_alerts.sql b/internal/db/migrations/003_alerts.sql deleted file mode 100644 index 016e265..0000000 --- a/internal/db/migrations/003_alerts.sql +++ /dev/null @@ -1,16 +0,0 @@ -CREATE TABLE alerts ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - fingerprint TEXT NOT NULL UNIQUE, - name TEXT NOT NULL, - status TEXT NOT NULL CHECK(status IN ('firing', 'resolved')), - labels TEXT NOT NULL DEFAULT '{}', - annotations TEXT NOT NULL DEFAULT '{}', - starts_at INTEGER NOT NULL, - ends_at INTEGER, - generator_url TEXT NOT NULL DEFAULT '', - received_at INTEGER NOT NULL DEFAULT (strftime('%s', 'now')) -); - -CREATE INDEX alerts_status_idx ON alerts(status); -CREATE INDEX alerts_name_idx ON alerts(name); -CREATE INDEX alerts_received_at_idx ON alerts(received_at DESC); diff --git a/internal/db/migrations/004_alert_ack_comments.sql b/internal/db/migrations/004_alert_ack_comments.sql deleted file mode 100644 index 775bb64..0000000 --- a/internal/db/migrations/004_alert_ack_comments.sql +++ /dev/null @@ -1,12 +0,0 @@ -ALTER TABLE alerts ADD COLUMN acknowledged_by INTEGER REFERENCES users(id) ON DELETE SET NULL; -ALTER TABLE alerts ADD COLUMN acknowledged_at INTEGER; - -CREATE TABLE alert_comments ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - alert_id INTEGER NOT NULL REFERENCES alerts(id) ON DELETE CASCADE, - user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, - content TEXT NOT NULL, - created_at INTEGER NOT NULL DEFAULT (strftime('%s', 'now')) -); - -CREATE INDEX alert_comments_alert_id_idx ON alert_comments(alert_id); diff --git a/internal/db/migrations/005_schedule.sql b/internal/db/migrations/005_schedule.sql deleted file mode 100644 index 15cdc16..0000000 --- a/internal/db/migrations/005_schedule.sql +++ /dev/null @@ -1,8 +0,0 @@ -CREATE TABLE schedule_entries ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, - date TEXT NOT NULL UNIQUE, -- YYYY-MM-DD; one person per day - created_at INTEGER NOT NULL DEFAULT (strftime('%s', 'now')) -); - -CREATE INDEX schedule_entries_date_idx ON schedule_entries(date); diff --git a/internal/db/migrations/006_alert_archive.sql b/internal/db/migrations/006_alert_archive.sql deleted file mode 100644 index 5217138..0000000 --- a/internal/db/migrations/006_alert_archive.sql +++ /dev/null @@ -1,2 +0,0 @@ -ALTER TABLE alerts ADD COLUMN archived_at INTEGER; -CREATE INDEX alerts_archived_at_idx ON alerts(archived_at); diff --git a/internal/db/migrations/007_alert_resolution_source.sql b/internal/db/migrations/007_alert_resolution_source.sql deleted file mode 100644 index dda6611..0000000 --- a/internal/db/migrations/007_alert_resolution_source.sql +++ /dev/null @@ -1,4 +0,0 @@ --- Records why an alert left the firing state: 'alertmanager' when a resolved --- webhook set it, 'expiry' when the sweeper inferred it from staleness. --- NULL for firing alerts and for rows that predate this migration. -ALTER TABLE alerts ADD COLUMN resolution_source TEXT; diff --git a/internal/db/migrations/008_incidents.sql b/internal/db/migrations/008_incidents.sql deleted file mode 100644 index c81f7a6..0000000 --- a/internal/db/migrations/008_incidents.sql +++ /dev/null @@ -1,123 +0,0 @@ --- Splits the single alerts row into two objects, the way an incident management --- tool needs them: alerts stay the machine-owned signal record that Alertmanager --- writes, and incidents become the human work item people acknowledge, assign, --- snooze, discuss and resolve. --- --- Correlation uses Alertmanager's own groupKey, so incidents follow the group_by --- routing tree the operator already tuned rather than a second grouping scheme --- invented here. - -CREATE TABLE incidents ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - group_key TEXT NOT NULL, -- Alertmanager groupKey, opaque - title TEXT NOT NULL, -- rendered from group_labels - group_labels TEXT NOT NULL DEFAULT '{}', -- JSON - status TEXT NOT NULL CHECK(status IN ('triggered', 'acknowledged', 'resolved')), - severity TEXT, -- highest `severity` label across firing members - triggered_at INTEGER NOT NULL, - acknowledged_by INTEGER REFERENCES users(id) ON DELETE SET NULL, - acknowledged_at INTEGER, - assigned_to INTEGER REFERENCES users(id) ON DELETE SET NULL, - snoozed_until INTEGER, - resolved_at INTEGER, - resolution_source TEXT, -- 'alerts' | 'manual' - archived_at INTEGER -); - --- Load-bearing: at most one OPEN incident per group_key. This is what makes --- "resolved incident + a new alert occurrence = a new incident" work, and it is --- the constraint the webhook's find-or-open lookup relies on. -CREATE UNIQUE INDEX incidents_open_group_key_idx ON incidents(group_key) WHERE resolved_at IS NULL; -CREATE INDEX incidents_status_idx ON incidents(status); -CREATE INDEX incidents_triggered_at_idx ON incidents(triggered_at DESC); -CREATE INDEX incidents_archived_at_idx ON incidents(archived_at); - --- Membership is historical, not a pointer on alerts: one alert row (one --- fingerprint) resolves and re-fires over time and belongs to a different --- incident each occurrence. -CREATE TABLE incident_alerts ( - incident_id INTEGER NOT NULL REFERENCES incidents(id) ON DELETE CASCADE, - alert_id INTEGER NOT NULL REFERENCES alerts(id) ON DELETE CASCADE, - added_at INTEGER NOT NULL DEFAULT (strftime('%s', 'now')), - PRIMARY KEY (incident_id, alert_id) -); - -CREATE INDEX incident_alerts_alert_id_idx ON incident_alerts(alert_id); - --- The timeline. Append-only, and the only history this server keeps: alert rows --- are mutated in place, so without this there is no record that anything --- happened. Notes are events too, so one query renders the whole story. -CREATE TABLE incident_events ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - incident_id INTEGER NOT NULL REFERENCES incidents(id) ON DELETE CASCADE, - -- triggered | alert_added | alert_resolved | acknowledged | unacknowledged - -- | assigned | snoozed | unsnoozed | resolved | note - type TEXT NOT NULL, - user_id INTEGER REFERENCES users(id) ON DELETE SET NULL, -- NULL = the server acted - alert_id INTEGER REFERENCES alerts(id) ON DELETE SET NULL, - detail TEXT, - created_at INTEGER NOT NULL DEFAULT (strftime('%s', 'now')) -); - -CREATE INDEX incident_events_incident_idx ON incident_events(incident_id, created_at); - --- --------------------------------------------------------------------------- --- Backfill --- --- Every pre-existing alert gets its own incident, archived ones included, so no --- acknowledgement and no comment is orphaned. There is no historical groupKey to --- correlate on, hence one incident per fingerprint under a 'backfill:' prefix --- that can never collide with a real Alertmanager groupKey. --- --------------------------------------------------------------------------- - -INSERT INTO incidents (group_key, title, group_labels, status, severity, triggered_at, - acknowledged_by, acknowledged_at, assigned_to, - resolved_at, resolution_source, archived_at) -SELECT 'backfill:' || a.fingerprint, - a.name, - json_object('alertname', a.name), - CASE WHEN a.status = 'resolved' THEN 'resolved' - WHEN a.acknowledged_by IS NOT NULL THEN 'acknowledged' - ELSE 'triggered' END, - json_extract(a.labels, '$.severity'), - a.starts_at, - a.acknowledged_by, - a.acknowledged_at, - a.acknowledged_by, - CASE WHEN a.status = 'resolved' THEN COALESCE(a.ends_at, a.received_at) END, - CASE WHEN a.status = 'resolved' THEN 'alerts' END, - a.archived_at -FROM alerts a; - -INSERT INTO incident_alerts (incident_id, alert_id, added_at) -SELECT i.id, a.id, a.starts_at -FROM alerts a -JOIN incidents i ON i.group_key = 'backfill:' || a.fingerprint; - -INSERT INTO incident_events (incident_id, type, alert_id, created_at) -SELECT i.id, 'triggered', ia.alert_id, i.triggered_at -FROM incidents i JOIN incident_alerts ia ON ia.incident_id = i.id; - -INSERT INTO incident_events (incident_id, type, user_id, created_at) -SELECT i.id, 'acknowledged', i.acknowledged_by, i.acknowledged_at -FROM incidents i WHERE i.acknowledged_at IS NOT NULL; - -INSERT INTO incident_events (incident_id, type, created_at) -SELECT i.id, 'resolved', i.resolved_at -FROM incidents i WHERE i.resolved_at IS NOT NULL; - -INSERT INTO incident_events (incident_id, type, user_id, alert_id, detail, created_at) -SELECT ia.incident_id, 'note', c.user_id, c.alert_id, c.content, c.created_at -FROM alert_comments c -JOIN incident_alerts ia ON ia.alert_id = c.alert_id; - --- --------------------------------------------------------------------------- --- Workflow state now lives on incidents only. Leaving these behind would keep --- the bug they caused: the webhook upsert owns the alerts row and never cleared --- the acknowledgement, so a re-fire days later still read as acknowledged. --- --------------------------------------------------------------------------- - -DROP TABLE alert_comments; - -ALTER TABLE alerts DROP COLUMN acknowledged_by; -ALTER TABLE alerts DROP COLUMN acknowledged_at; diff --git a/internal/db/migrations/009_notifications.sql b/internal/db/migrations/009_notifications.sql deleted file mode 100644 index 22e97ce..0000000 --- a/internal/db/migrations/009_notifications.sql +++ /dev/null @@ -1,44 +0,0 @@ --- Adds push notification delivery, so an incident reaches the person on call --- instead of waiting to be discovered. --- --- Delivery is an outbox rather than an inline HTTP call: the pool is limited to --- a single connection, so a POST made while holding the webhook's transaction --- would stall every other request behind it. The webhook inserts a row; the --- notifier goroutine delivers it. - -CREATE TABLE notifications ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - incident_id INTEGER NOT NULL REFERENCES incidents(id) ON DELETE CASCADE, - -- Nullable: a notification sent to the fallback topic belongs to nobody, - -- because nobody was on call when the incident opened. - user_id INTEGER REFERENCES users(id) ON DELETE SET NULL, - topic TEXT NOT NULL, -- resolved at enqueue: who was on call then - kind TEXT NOT NULL CHECK(kind IN ('triggered', 'reminder', 'resolved')), - created_at INTEGER NOT NULL, - send_after INTEGER NOT NULL, -- retry backoff watermark - attempts INTEGER NOT NULL DEFAULT 0, - sent_at INTEGER, - last_error TEXT -- kept after the last attempt, for debugging -); - --- The delivery loop's only query: what is due and still unsent. -CREATE INDEX notifications_pending_idx ON notifications(send_after) WHERE sent_at IS NULL; --- Reminders and resolved notices both look up an incident's newest row. -CREATE INDEX notifications_incident_idx ON notifications(incident_id, id DESC); - --- A notification body is stored on the ntfy server and cached on the device, so --- a real API key must never appear in one. Each delivery mints its own token --- instead: one incident, one action, one day. -CREATE TABLE incident_ack_tokens ( - token_hash TEXT PRIMARY KEY, -- SHA-256 of the raw token, as with api_keys - incident_id INTEGER NOT NULL REFERENCES incidents(id) ON DELETE CASCADE, - user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, - created_at INTEGER NOT NULL, - expires_at INTEGER NOT NULL -); - -CREATE INDEX incident_ack_tokens_expires_idx ON incident_ack_tokens(expires_at); - --- Where this user's notifications go. NULL means they get none; incidents --- assigned to them fall back to the configured fallback topic. -ALTER TABLE users ADD COLUMN ntfy_topic TEXT; diff --git a/internal/db/migrations/010_web_auth.sql b/internal/db/migrations/010_web_auth.sql deleted file mode 100644 index c8b685e..0000000 --- a/internal/db/migrations/010_web_auth.sql +++ /dev/null @@ -1,18 +0,0 @@ --- A password is what lets a person sign in to the web UI. NULL means the user --- has none and can only use API keys, which is every user created before this. -ALTER TABLE users ADD COLUMN password_hash TEXT; - --- A session is a browser's credential, the cookie counterpart of an API key: --- only the hash of the token is stored. expires_at slides forward while the --- session is in use, so an on-call phone stays signed in. -CREATE TABLE sessions ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - token_hash TEXT NOT NULL UNIQUE, - user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, - created_at INTEGER NOT NULL, - last_seen_at INTEGER NOT NULL, - expires_at INTEGER NOT NULL, - user_agent TEXT -); - -CREATE INDEX idx_sessions_user ON sessions(user_id); diff --git a/scripts/sqlite-to-postgres.go b/scripts/sqlite-to-postgres.go new file mode 100644 index 0000000..9260220 --- /dev/null +++ b/scripts/sqlite-to-postgres.go @@ -0,0 +1,198 @@ +//go:build migrate + +// Command sqlite-to-postgres copies a terdut SQLite database into a freshly +// migrated Postgres one. It exists for exactly one upgrade — the one that moved +// this server off SQLite — and should be deleted once the installs that need it +// have run it. The modernc.org/sqlite dependency goes with it. +// +// Build-tagged so the dependency stays out of the server binary and out of a +// plain `go build ./...`: +// +// go run -tags migrate ./scripts/sqlite-to-postgres.go \ +// -sqlite /path/to/terdut.db \ +// -dsn 'postgres://terdut:secret@localhost:5432/terdut?sslmode=disable' +// +// The Postgres side must already have the schema: start the new server once +// against an empty database, let it migrate, stop it, then run this. The copy +// refuses to touch a database that already has rows, so a second run cannot +// double-insert. +// +// Ids are preserved, which is what keeps every foreign key — incident_alerts, +// incident_events, notifications, the ack tokens — pointing at the same rows it +// pointed at before. The identity sequences are moved past the copied ids at the +// end, so the first row the server writes afterwards does not collide. +package main + +import ( + "database/sql" + "flag" + "fmt" + "log" + "strings" + + _ "github.com/jackc/pgx/v5/stdlib" + _ "modernc.org/sqlite" +) + +// tables are copied parents first: every foreign key points at a table earlier +// in this list. +var tables = []struct { + name string + columns []string + // jsonb marks columns that were TEXT in SQLite and are jsonb in Postgres, + // so the insert can cast them. + jsonb []string + // sequence is the identity sequence to advance afterwards, empty when the + // table has no generated id. + sequence string +}{ + {name: "users", columns: []string{"id", "username", "email", "created_at", "ntfy_topic", "password_hash"}, sequence: "users_id_seq"}, + {name: "api_keys", columns: []string{"id", "user_id", "key_hash", "name", "created_at", "last_used_at"}, sequence: "api_keys_id_seq"}, + {name: "sessions", columns: []string{"id", "token_hash", "user_id", "created_at", "last_seen_at", "expires_at", "user_agent"}, sequence: "sessions_id_seq"}, + {name: "alerts", columns: []string{"id", "fingerprint", "name", "status", "labels", "annotations", "starts_at", "ends_at", "generator_url", "received_at", "archived_at", "resolution_source"}, jsonb: []string{"labels", "annotations"}, sequence: "alerts_id_seq"}, + {name: "schedule_entries", columns: []string{"id", "user_id", "date", "created_at"}, sequence: "schedule_entries_id_seq"}, + {name: "incidents", columns: []string{"id", "group_key", "title", "group_labels", "status", "severity", "triggered_at", "acknowledged_by", "acknowledged_at", "assigned_to", "snoozed_until", "resolved_at", "resolution_source", "archived_at"}, jsonb: []string{"group_labels"}, sequence: "incidents_id_seq"}, + {name: "incident_alerts", columns: []string{"incident_id", "alert_id", "added_at"}}, + {name: "incident_events", columns: []string{"id", "incident_id", "type", "user_id", "alert_id", "detail", "created_at"}, sequence: "incident_events_id_seq"}, + {name: "notifications", columns: []string{"id", "incident_id", "user_id", "topic", "kind", "created_at", "send_after", "attempts", "sent_at", "last_error"}, sequence: "notifications_id_seq"}, + {name: "incident_ack_tokens", columns: []string{"token_hash", "incident_id", "user_id", "created_at", "expires_at"}}, +} + +func main() { + sqlitePath := flag.String("sqlite", "", "path to the existing terdut SQLite database") + dsn := flag.String("dsn", "", "Postgres DSN of the migrated, empty database") + flag.Parse() + + if *sqlitePath == "" || *dsn == "" { + log.Fatal("both -sqlite and -dsn are required") + } + + src, err := sql.Open("sqlite", *sqlitePath) + if err != nil { + log.Fatalf("open sqlite: %v", err) + } + defer src.Close() + + dst, err := sql.Open("pgx", *dsn) + if err != nil { + log.Fatalf("open postgres: %v", err) + } + defer dst.Close() + + if err := dst.Ping(); err != nil { + log.Fatalf("ping postgres: %v", err) + } + if err := assertEmpty(dst); err != nil { + log.Fatalf("%v", err) + } + + // One transaction for the whole copy: a run that dies half way leaves the + // target as it found it, rather than a partial database someone has to + // recognise as partial. + tx, err := dst.Begin() + if err != nil { + log.Fatalf("begin: %v", err) + } + defer tx.Rollback() //nolint:errcheck + + for _, t := range tables { + n, err := copyTable(src, tx, t.name, t.columns, t.jsonb) + if err != nil { + log.Fatalf("copy %s: %v", t.name, err) + } + log.Printf("%-20s %d row(s)", t.name, n) + } + + for _, t := range tables { + if t.sequence == "" { + continue + } + if err := advanceSequence(tx, t.sequence, t.name); err != nil { + log.Fatalf("advance %s: %v", t.sequence, err) + } + } + + if err := tx.Commit(); err != nil { + log.Fatalf("commit: %v", err) + } + log.Print("done") +} + +// assertEmpty refuses a target that already holds data, so running this twice +// cannot duplicate anything. +func assertEmpty(dst *sql.DB) error { + for _, t := range tables { + var n int64 + if err := dst.QueryRow("SELECT COUNT(*) FROM " + t.name).Scan(&n); err != nil { + return fmt.Errorf("count %s (has the new server migrated this database?): %w", t.name, err) + } + if n > 0 { + return fmt.Errorf("%s already has %d row(s): the target must be empty", t.name, n) + } + } + return nil +} + +func copyTable(src *sql.DB, tx *sql.Tx, table string, columns, jsonb []string) (int64, error) { + rows, err := src.Query("SELECT " + strings.Join(columns, ", ") + " FROM " + table) + if err != nil { + return 0, err + } + defer rows.Close() + + insert := "INSERT INTO " + table + " (" + strings.Join(columns, ", ") + ") VALUES (" + + strings.Join(valuePlaceholders(columns, jsonb), ", ") + ")" + + stmt, err := tx.Prepare(insert) + if err != nil { + return 0, err + } + defer stmt.Close() + + var copied int64 + for rows.Next() { + // Scanning into any lets the SQLite driver decide each column's Go type + // and hands it straight back to pgx, which is all this needs: the column + // types match on both sides, apart from the JSON casts above. + values := make([]any, len(columns)) + targets := make([]any, len(columns)) + for i := range values { + targets[i] = &values[i] + } + if err := rows.Scan(targets...); err != nil { + return copied, err + } + if _, err := stmt.Exec(values...); err != nil { + return copied, fmt.Errorf("insert row %d: %w", copied+1, err) + } + copied++ + } + return copied, rows.Err() +} + +// valuePlaceholders numbers the placeholders, casting the columns that became +// jsonb: pgx sends a Go string as text, and Postgres will not assign text to a +// jsonb column without being told. +func valuePlaceholders(columns, jsonb []string) []string { + isJSON := make(map[string]bool, len(jsonb)) + for _, c := range jsonb { + isJSON[c] = true + } + out := make([]string, len(columns)) + for i, c := range columns { + out[i] = fmt.Sprintf("$%d", i+1) + if isJSON[c] { + out[i] += "::jsonb" + } + } + return out +} + +// advanceSequence puts an identity sequence past the largest copied id. Without +// it the first insert after the migration would reuse id 1. +func advanceSequence(tx *sql.Tx, sequence, table string) error { + _, err := tx.Exec(fmt.Sprintf( + `SELECT setval('%s', COALESCE((SELECT MAX(id) FROM %s), 0) + 1, false)`, + sequence, table)) + return err +}