The schema applies itself, and the setup script stops hiding failures

Migrations were run by hand and nothing recorded which had run, so
re-running the setup script against an existing database failed on the
first CREATE TABLE, and shipping a new migration gave an operator no way
to know whether an estate had it. A missed migration is not a startup
error - it is a query referencing a column that is not there, surfacing
later on whichever endpoint touches it first.

server/internal/migrate applies pending migrations at boot and refuses to
start against a schema it does not match. One transaction per file
holding both the DDL and the row that records it; an advisory lock so two
servers starting at once cannot both apply 008; checksums so an edited
migration is refused by name rather than silently skipped; numeric
ordering so 010 does not run before 009. `migrate -baseline N` adopts a
database built before any of this existed, because "the clients table
exists" does not say whether 007's index does.

Verified on the live database: adopted 001-007, applied 008.

008 adds two indexes on `purchases`, found by asking the database which
foreign keys had nothing behind them and then checking what queries the
table. The conversion report filters client_id + occurred_at, which is
exactly the estate-wide case with no site to narrow it.

run-local.sh had two bugs, both found by running it rather than reading
it: it reused a broker container whose bind mount pointed at a directory
that no longer existed, and it discarded stderr on the mosquitto_passwd
call, so under `set -e` it exited at step 5 with no output at all.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HViLj9gYNRtSr7YVZmW5sn

server/cmd/behavision-server/main.go

@@ -26,9 +26,11 @@ import (
 	"github.com/loyaly/behavision-server/internal/assistant"
 	"github.com/loyaly/behavision-server/internal/blob"
 	"github.com/loyaly/behavision-server/internal/ingest"
+	"github.com/loyaly/behavision-server/internal/migrate"
 	"github.com/loyaly/behavision-server/internal/web"
 	"github.com/loyaly/behavision-server/internal/secret"
 	"github.com/loyaly/behavision-server/internal/store"
+	"github.com/loyaly/behavision-server/migrations"
 )
 
 var version = "dev"
@@ -44,6 +46,13 @@ func main() {
 		}
 		return
 	}
+	if len(os.Args) > 1 && os.Args[1] == "migrate" {
+		if err := runMigrate(os.Args[2:]); err != nil {
+			fmt.Fprintln(os.Stderr, err)
+			os.Exit(1)
+		}
+		return
+	}
 	if err := run(); err != nil {
 		log.Fatalf("behavision-server: %v", err)
 	}
@@ -79,6 +88,29 @@ func run() error {
 	st.UseLogger(logger)
 	logger.Print("database connected")
 
+	// Applied here, not by hand, because an upgrade of this product is "copy
+	// the new binary and restart it". A migration an operator has to remember
+	// to run is a migration that does not get run, and its failure is not a
+	// startup error - it is a query referencing a column that is not there,
+	// surfacing later on whichever endpoint touches it first.
+	//
+	// Refusing to start on failure is deliberate: a server running against a
+	// schema it does not match writes wrong data, and wrong data outlives the
+	// outage that stopping causes.
+	if os.Getenv("BEHAVISION_SKIP_MIGRATE") == "1" {
+		logger.Print("WARN BEHAVISION_SKIP_MIGRATE=1 - schema not checked")
+	} else {
+		applied, err := migrate.Apply(ctx, st.Pool(), migrations.FS)
+		if err != nil {
+			return fmt.Errorf("schema: %w", err)
+		}
+		if len(applied) == 0 {
+			logger.Print("schema up to date")
+		} else {
+			logger.Printf("schema: applied %s", strings.Join(applied, ", "))
+		}
+	}
+
 	// Without the key the server still ingests events and serves reports; only
 	// enrolment fails, and it fails with a message naming the missing variable.
 	// Refusing to start would take a working estate down over a feature that

server/cmd/behavision-server/migrate.go

@@ -0,0 +1,107 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/loyaly/behavision-server/internal/migrate"
+	"github.com/loyaly/behavision-server/internal/store"
+	"github.com/loyaly/behavision-server/migrations"
+)
+
+// runMigrate is the operator's view of the schema.
+//
+// The server applies pending migrations itself at boot, so this exists for the
+// two things it cannot do: SAY what a database is on without changing it, and
+// adopt a database whose schema predates the tracking table.
+func runMigrate(args []string) error {
+	fs := flag.NewFlagSet("migrate", flag.ExitOnError)
+	status := fs.Bool("status", false, "print what this database has recorded and exit")
+	baseline := fs.Int("baseline", 0,
+		"record migrations up to this version as applied WITHOUT running them")
+	fs.Usage = func() {
+		fmt.Fprint(os.Stderr, `usage: behavision-server migrate [flags]
+
+  no flags     apply everything this database has not recorded
+  -status      print the recorded versions and what is pending
+  -baseline N  adopt a database built before migration tracking existed:
+               record 1..N as applied without running them. Use it once, on a
+               database whose schema you know already matches those files.
+
+DATABASE_URL is required.
+`)
+		fs.PrintDefaults()
+	}
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+
+	dsn := os.Getenv("DATABASE_URL")
+	if dsn == "" {
+		return fmt.Errorf("DATABASE_URL is required")
+	}
+	ctx := context.Background()
+	st, err := store.Open(ctx, dsn)
+	if err != nil {
+		return err
+	}
+	defer st.Close()
+
+	files, err := migrate.Load(migrations.FS)
+	if err != nil {
+		return err
+	}
+
+	if *baseline > 0 {
+		marked, err := migrate.Baseline(ctx, st.Pool(), migrations.FS, *baseline)
+		if err != nil {
+			return err
+		}
+		if len(marked) == 0 {
+			fmt.Println("nothing to adopt - every migration up to that version was already recorded")
+			return nil
+		}
+		fmt.Printf("adopted without running: %s\n", strings.Join(marked, ", "))
+		fmt.Println("run `migrate` (no flags) to apply anything after that.")
+		return nil
+	}
+
+	if *status {
+		done, err := migrate.Status(ctx, st.Pool())
+		if err != nil {
+			return err
+		}
+		seen := map[int]migrate.Record{}
+		for _, r := range done {
+			seen[r.Version] = r
+		}
+		for _, f := range files {
+			r, ok := seen[f.Version]
+			switch {
+			case !ok:
+				fmt.Printf("  pending   %s\n", f.Name)
+			case r.Baselined:
+				fmt.Printf("  adopted   %s  (recorded, never run here)\n", f.Name)
+			case r.Checksum != f.Checksum:
+				fmt.Printf("  CHANGED   %s  (the file differs from what was applied)\n", f.Name)
+			default:
+				fmt.Printf("  applied   %s\n", f.Name)
+			}
+		}
+		return nil
+	}
+
+	applied, err := migrate.Apply(ctx, st.Pool(), migrations.FS)
+	if err != nil {
+		return err
+	}
+	if len(applied) == 0 {
+		fmt.Println("schema up to date")
+		return nil
+	}
+	fmt.Printf("applied: %s\n", strings.Join(applied, ", "))
+	return nil
+}

server/internal/migrate/migrate.go

@@ -0,0 +1,285 @@
+// Package migrate applies the schema and records what it applied.
+//
+// Until this existed the migrations were run by hand - `psql < 001.sql` for
+// each file, in order, by whoever remembered - and nothing anywhere recorded
+// which had run. Three consequences, all of which had already happened:
+//
+//   - Re-running them against an existing database fails on the first CREATE
+//     TABLE, so the setup script only worked once.
+//   - Adding migration 008 to a release gave the operator no way to know
+//     whether a given estate had it. The failure of a missed migration is not
+//     a startup error; it is a query referencing a column that is not there,
+//     surfacing on whichever endpoint touches it first.
+//   - A half-applied migration - the file interrupted midway - left a schema
+//     nothing could describe.
+//
+// So: every file runs inside ONE transaction together with the row that
+// records it. Either both happen or neither does, and a database can always
+// say which version it is on.
+package migrate
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"io/fs"
+	"sort"
+	"strconv"
+	"strings"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// lockID namespaces the advisory lock this package takes. Two servers starting
+// at the same moment - which is the normal shape of a rolling restart - must
+// not both decide migration 008 is pending and both run it.
+const lockID int64 = 7623094512340001
+
+// File is one migration on disk.
+type File struct {
+	Version  int
+	Name     string
+	SQL      string
+	Checksum string
+}
+
+// Record is one migration as the database remembers it.
+type Record struct {
+	Version   int
+	Name      string
+	Checksum  string
+	Baselined bool
+}
+
+// Load reads and orders the migrations, rejecting anything it cannot place.
+func Load(src fs.FS) ([]File, error) {
+	entries, err := fs.Glob(src, "*.sql")
+	if err != nil {
+		return nil, err
+	}
+	seen := map[int]string{}
+	out := make([]File, 0, len(entries))
+	for _, name := range entries {
+		version, err := versionOf(name)
+		if err != nil {
+			return nil, err
+		}
+		if other, dup := seen[version]; dup {
+			// Two files claiming one version means the order they run in is
+			// whatever the filesystem felt like, which is not an order.
+			return nil, fmt.Errorf("migrations %s and %s share version %d",
+				other, name, version)
+		}
+		seen[version] = name
+		body, err := fs.ReadFile(src, name)
+		if err != nil {
+			return nil, err
+		}
+		sum := sha256.Sum256(body)
+		out = append(out, File{
+			Version:  version,
+			Name:     name,
+			SQL:      string(body),
+			Checksum: hex.EncodeToString(sum[:]),
+		})
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].Version < out[j].Version })
+	return out, nil
+}
+
+func versionOf(name string) (int, error) {
+	prefix, _, ok := strings.Cut(name, "_")
+	if !ok {
+		return 0, fmt.Errorf("migration %q is not named NNN_description.sql", name)
+	}
+	n, err := strconv.Atoi(prefix)
+	if err != nil || n <= 0 {
+		return 0, fmt.Errorf("migration %q does not start with a version number", name)
+	}
+	return n, nil
+}
+
+// Apply runs every migration the database has not recorded, in order.
+//
+// Returns the names it applied, so a caller can log "nothing to do" rather
+// than staying silent about work it did not need to do.
+func Apply(ctx context.Context, pool *pgxpool.Pool, src fs.FS) ([]string, error) {
+	files, err := Load(src)
+	if err != nil {
+		return nil, err
+	}
+
+	conn, err := pool.Acquire(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer conn.Release()
+
+	// Taken on this one connection and held for the whole run: a second server
+	// booting waits here rather than racing us, and the lock is released with
+	// the connection even if this process is killed.
+	if _, err := conn.Exec(ctx, `SELECT pg_advisory_lock($1)`, lockID); err != nil {
+		return nil, fmt.Errorf("migration lock: %w", err)
+	}
+	defer func() {
+		_, _ = conn.Exec(context.WithoutCancel(ctx),
+			`SELECT pg_advisory_unlock($1)`, lockID)
+	}()
+
+	if err := ensureTable(ctx, conn.Conn()); err != nil {
+		return nil, err
+	}
+	done, err := recorded(ctx, conn.Conn())
+	if err != nil {
+		return nil, err
+	}
+
+	var applied []string
+	for _, f := range files {
+		prev, ok := done[f.Version]
+		if ok {
+			// An already-applied file that has since been edited. The database
+			// does NOT contain what the repository says it does, and running
+			// the new text now would apply half of it twice. Refuse and say so:
+			// the fix is a new migration, never an edited one.
+			if prev.Checksum != f.Checksum && !prev.Baselined {
+				return applied, fmt.Errorf(
+					"migration %s has changed since it was applied "+
+						"(recorded %s, file %s) - add a new migration instead of editing one",
+					f.Name, short(prev.Checksum), short(f.Checksum))
+			}
+			continue
+		}
+		if err := applyOne(ctx, conn.Conn(), f); err != nil {
+			return applied, err
+		}
+		applied = append(applied, f.Name)
+	}
+	return applied, nil
+}
+
+// applyOne runs one file and records it in the SAME transaction. A migration
+// that ran but was not recorded would run again on the next start; one that was
+// recorded but did not run leaves a schema missing a column nothing will ever
+// add.
+func applyOne(ctx context.Context, conn *pgx.Conn, f File) error {
+	tx, err := conn.Begin(ctx)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
+
+	if _, err := tx.Exec(ctx, f.SQL); err != nil {
+		return fmt.Errorf("migration %s: %w", f.Name, err)
+	}
+	if _, err := tx.Exec(ctx,
+		`INSERT INTO schema_migrations (version, name, checksum) VALUES ($1, $2, $3)`,
+		f.Version, f.Name, f.Checksum); err != nil {
+		return fmt.Errorf("recording migration %s: %w", f.Name, err)
+	}
+	return tx.Commit(ctx)
+}
+
+// Baseline records migrations up to and including `through` as applied WITHOUT
+// running them.
+//
+// For the one case that cannot be handled automatically: a database built
+// before this package existed, whose schema is already there and whose history
+// is not. Guessing is not an option - "the clients table exists" does not say
+// whether migration 007's index does - so an operator states it, once, and it
+// is recorded as a baseline rather than as a real application.
+func Baseline(ctx context.Context, pool *pgxpool.Pool, src fs.FS, through int) ([]string, error) {
+	files, err := Load(src)
+	if err != nil {
+		return nil, err
+	}
+	conn, err := pool.Acquire(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer conn.Release()
+	if err := ensureTable(ctx, conn.Conn()); err != nil {
+		return nil, err
+	}
+	var marked []string
+	for _, f := range files {
+		if f.Version > through {
+			continue
+		}
+		tag, err := conn.Exec(ctx,
+			`INSERT INTO schema_migrations (version, name, checksum, baselined)
+			 VALUES ($1, $2, $3, true) ON CONFLICT (version) DO NOTHING`,
+			f.Version, f.Name, f.Checksum)
+		if err != nil {
+			return marked, err
+		}
+		if tag.RowsAffected() == 1 {
+			marked = append(marked, f.Name)
+		}
+	}
+	return marked, nil
+}
+
+// Status reports what the database has recorded, oldest first.
+func Status(ctx context.Context, pool *pgxpool.Pool) ([]Record, error) {
+	conn, err := pool.Acquire(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer conn.Release()
+	if err := ensureTable(ctx, conn.Conn()); err != nil {
+		return nil, err
+	}
+	done, err := recorded(ctx, conn.Conn())
+	if err != nil {
+		return nil, err
+	}
+	out := make([]Record, 0, len(done))
+	for _, r := range done {
+		out = append(out, r)
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].Version < out[j].Version })
+	return out, nil
+}
+
+func ensureTable(ctx context.Context, conn *pgx.Conn) error {
+	_, err := conn.Exec(ctx, `
+		CREATE TABLE IF NOT EXISTS schema_migrations (
+			version    integer PRIMARY KEY,
+			name       text NOT NULL,
+			checksum   text NOT NULL,
+			applied_at timestamptz NOT NULL DEFAULT now(),
+			-- true when the row records a migration that was NOT run here,
+			-- because the schema predates this table. Kept distinct so an
+			-- adopted database never looks like one this code built.
+			baselined  boolean NOT NULL DEFAULT false
+		)`)
+	return err
+}
+
+func recorded(ctx context.Context, conn *pgx.Conn) (map[int]Record, error) {
+	rows, err := conn.Query(ctx,
+		`SELECT version, name, checksum, baselined FROM schema_migrations`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	out := map[int]Record{}
+	for rows.Next() {
+		var r Record
+		if err := rows.Scan(&r.Version, &r.Name, &r.Checksum, &r.Baselined); err != nil {
+			return nil, err
+		}
+		out[r.Version] = r
+	}
+	return out, rows.Err()
+}
+
+func short(sum string) string {
+	if len(sum) > 12 {
+		return sum[:12]
+	}
+	return sum
+}

server/internal/migrate/migrate_test.go

@@ -0,0 +1,73 @@
+package migrate
+
+import (
+	"strings"
+	"testing"
+	"testing/fstest"
+)
+
+func TestMigrationsRunInNumericOrderNotAlphabetical(t *testing.T) {
+	// The bug this prevents: at ten migrations, "10_x.sql" sorts before
+	// "9_x.sql" as text, so the tenth would run before the ninth and the
+	// failure would be a column that does not exist yet - on the day the
+	// project happens to reach double figures.
+	src := fstest.MapFS{
+		"002_b.sql": {Data: []byte("select 2")},
+		"010_j.sql": {Data: []byte("select 10")},
+		"009_i.sql": {Data: []byte("select 9")},
+		"001_a.sql": {Data: []byte("select 1")},
+	}
+	files, err := Load(src)
+	if err != nil {
+		t.Fatalf("load: %v", err)
+	}
+	got := make([]int, len(files))
+	for i, f := range files {
+		got[i] = f.Version
+	}
+	want := []int{1, 2, 9, 10}
+	for i := range want {
+		if got[i] != want[i] {
+			t.Fatalf("order was %v, want %v", got, want)
+		}
+	}
+}
+
+func TestTwoFilesCannotShareAVersion(t *testing.T) {
+	// Two people adding "008" on separate branches is the ordinary way this
+	// happens. Whichever runs first is then decided by the filesystem, which is
+	// not an order, and one of them silently never runs at all.
+	_, err := Load(fstest.MapFS{
+		"008_one.sql": {Data: []byte("select 1")},
+		"008_two.sql": {Data: []byte("select 2")},
+	})
+	if err == nil {
+		t.Fatal("duplicate versions were accepted")
+	}
+	if !strings.Contains(err.Error(), "share version 8") {
+		t.Fatalf("error does not name the collision: %v", err)
+	}
+}
+
+func TestAFileThatIsNotNumberedIsRejected(t *testing.T) {
+	for _, name := range []string{"schema.sql", "abc_x.sql", "0_zero.sql"} {
+		if _, err := Load(fstest.MapFS{name: {Data: []byte("select 1")}}); err == nil {
+			t.Fatalf("%s was accepted as a migration", name)
+		}
+	}
+}
+
+func TestTheChecksumChangesWithTheContent(t *testing.T) {
+	// What makes an edited-after-applying migration detectable at all.
+	a, err := Load(fstest.MapFS{"001_a.sql": {Data: []byte("select 1")}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	b, err := Load(fstest.MapFS{"001_a.sql": {Data: []byte("select 2")}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if a[0].Checksum == b[0].Checksum {
+		t.Fatal("two different migrations hashed the same")
+	}
+}

server/internal/store/store.go

@@ -63,6 +63,14 @@ func Open(ctx context.Context, dsn string) (*Store, error) {
 
 func (s *Store) Close() { s.pool.Close() }
 
+// Pool exposes the connection pool for the schema migrator.
+//
+// Deliberately narrow in intent: the migrator has to run arbitrary DDL and
+// take an advisory lock, neither of which belongs behind a typed store method.
+// Nothing else should reach for this - a query that lives out here is a query
+// nothing tenant-scopes.
+func (s *Store) Pool() *pgxpool.Pool { return s.pool }
+
 func (s *Store) Ping(ctx context.Context) error { return s.pool.Ping(ctx) }
 
 // ResolveSite maps an authenticated MQTT username to a provisioned tenant.

server/migrations/008_purchase_indexes.sql

@@ -0,0 +1,27 @@
+-- Two indexes on `purchases`, both for queries that already exist.
+--
+-- Found by asking the database which foreign keys had no index behind them and
+-- then checking what actually queries the table, rather than by adding indexes
+-- on principle: every one of them costs a write on the path that records a
+-- sale.
+--
+-- 1. The conversion report filters `client_id` + `occurred_at`, with the site
+--    optional - an owner comparing shops is the whole reason that report
+--    exists, and that is precisely the case with no site to narrow it. The
+--    existing purchases_site_time_idx cannot serve it. Today the table has a
+--    handful of rows and a sequential scan is free; purchases is the table
+--    that grows with a shop's trade, so this is the one that stops being free.
+--
+-- 2. purchases.visit_id is a foreign key with nothing behind it. Every delete
+--    of a visit has to prove no purchase references it, which without an index
+--    is a full scan per row - and erasing a customer deletes their visits.
+
+BEGIN;
+
+CREATE INDEX IF NOT EXISTS purchases_client_time_idx
+    ON purchases (client_id, occurred_at DESC);
+
+CREATE INDEX IF NOT EXISTS purchases_visit_idx
+    ON purchases (visit_id);
+
+COMMIT;

server/migrations/embed.go

@@ -0,0 +1,15 @@
+// Package migrations carries the schema files themselves.
+//
+// It exists only so `go:embed` can reach them: embed cannot see outside its own
+// package directory, and moving the .sql files into some internal/ folder would
+// break every path that documents them - RUN.md, run-local.sh, the test that
+// tells you how to build a database by hand.
+package migrations
+
+import "embed"
+
+// FS holds every migration, named NNN_description.sql. The number is the
+// version and must be unique; the rest is for humans.
+//
+//go:embed *.sql
+var FS embed.FS