mesh-controller/internal/store/live_test.go

package store

import (
	"context"
	"fmt"
	"os"
	"strings"
	"sync"
	"testing"
	"testing/fstest"
	"time"

	"github.com/jackc/pgx/v5"
)

// These run against a real PostgreSQL. Not a fake, and for the reason novox/hq ADR 0017 gives
// where the host tests the real filesystem: what is being tested here *is* the database's
// behaviour — that DDL is transactional, that an advisory lock serialises, that a checksum
// mismatch is caught against a record the database actually kept. A fake would assert that the
// fake behaves as expected.
//
// `make check` raises one. Without it these skip, and say so rather than passing.

func admin(t *testing.T) string {
	t.Helper()
	dsn := os.Getenv("MESH_TEST_POSTGRES")
	if dsn == "" {
		t.Skip("no MESH_TEST_POSTGRES; run `make check` to raise one")
	}
	return dsn
}

// freshStore gives a test its own empty database.
//
// Its own, rather than a shared one cleaned between tests: these tests are about what a migration
// runner does to a schema, and a leftover table from a previous test is indistinguishable from
// the bug this whole package exists to catch.
func freshStore(t *testing.T) *Store {
	t.Helper()
	dsn := admin(t)
	name := fmt.Sprintf("test_%s_%d", strings.ToLower(strings.NewReplacer(
		"/", "_", "-", "_").Replace(t.Name())), time.Now().UnixNano()%1_000_000)
	if len(name) > 60 {
		name = name[:60]
	}

	conn, err := pgx.Connect(t.Context(), dsn)
	if err != nil {
		t.Fatalf("cannot reach the test PostgreSQL: %v", err)
	}
	if _, err := conn.Exec(t.Context(), "create database "+name); err != nil {
		t.Fatalf("cannot create %s: %v", name, err)
	}
	conn.Close(t.Context())

	t.Setenv(Variable("testing"), replaceDatabase(dsn, name))
	s, err := Open(t.Context(), "testing")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() {
		s.Close()
		c, err := pgx.Connect(context.Background(), dsn)
		if err != nil {
			return
		}
		defer c.Close(context.Background())
		_, _ = c.Exec(context.Background(), "drop database if exists "+name+" with (force)")
	})
	if err := s.Ready(t.Context(), 20*time.Second); err != nil {
		t.Fatal(err)
	}
	return s
}

func replaceDatabase(dsn, name string) string {
	cut := strings.LastIndex(dsn, "/")
	rest := ""
	if q := strings.Index(dsn[cut:], "?"); q >= 0 {
		rest = dsn[cut+q:]
	}
	return dsn[:cut] + "/" + name + rest
}

func TestTheSchemaIsAppliedAndRecorded(t *testing.T) {
	s := freshStore(t)
	migrations := []Migration{{Number: 1, Name: "people", SQL: "create table person (id int)", Checksum: "a"}}

	done, err := s.Migrate(t.Context(), migrations)
	if err != nil {
		t.Fatal(err)
	}
	if len(done) != 1 {
		t.Fatalf("applied %d migrations, expected 1", len(done))
	}

	// Read back from the system rather than trusting the return value (novox/hq ADR 0018).
	var exists bool
	if err := s.Pool().QueryRow(t.Context(),
		`select exists (select 1 from information_schema.tables where table_name = 'person')`,
	).Scan(&exists); err != nil {
		t.Fatal(err)
	}
	if !exists {
		t.Error("Migrate reported success and the table is not there")
	}

	applied, err := s.AppliedMigrations(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	if len(applied) != 1 || applied[0].Checksum != "a" {
		t.Errorf("the record says %+v", applied)
	}
}

func TestRunningTwiceChangesNothing(t *testing.T) {
	// The bootstrap runs this, and so does every restart of the control plane. A second run that
	// re-applied the schema would fail on the first `create table`, so a control plane would come
	// up exactly once.
	s := freshStore(t)
	migrations := []Migration{{Number: 1, Name: "people", SQL: "create table person (id int)", Checksum: "a"}}

	if _, err := s.Migrate(t.Context(), migrations); err != nil {
		t.Fatal(err)
	}
	done, err := s.Migrate(t.Context(), migrations)
	if err != nil {
		t.Fatalf("the second run failed: %v", err)
	}
	if len(done) != 0 {
		t.Errorf("the second run applied %d migrations", len(done))
	}
}

func TestAFailedMigrationLeavesNothingBehind(t *testing.T) {
	// Note what this does and does not defend. PostgreSQL wraps a multi-statement simple query in
	// an implicit transaction of its own, so this passes with this package's transaction removed
	// — it was checked, and it did. What it defends is the database and driver behaviour relied
	// on: a driver sending each statement separately would break it, and nothing else would say
	// so. The property that belongs to this code is the next test.
	s := freshStore(t)
	migrations := []Migration{{
		Number: 1, Name: "half", Checksum: "a",
		SQL: `create table kept (id int);
		      create table broken (id int) this is not sql;`,
	}}

	if _, err := s.Migrate(t.Context(), migrations); err == nil {
		t.Fatal("a migration with a syntax error reported success")
	}

	var tables int
	if err := s.Pool().QueryRow(t.Context(),
		`select count(*) from information_schema.tables where table_name in ('kept','broken')`,
	).Scan(&tables); err != nil {
		t.Fatal(err)
	}
	if tables != 0 {
		t.Errorf("%d table(s) survived a failed migration; it must be all or nothing", tables)
	}
}

func TestASchemaChangeAndItsRecordCommitTogether(t *testing.T) {
	// This is what the explicit transaction is for, and all it is for.
	//
	// Split the change from the row saying it happened, and a schema moves with nothing recording
	// it — so the next run finds the migration outstanding and applies it to a database that
	// already has it. The failure surfaces as a broken migration rather than as a lost record.
	//
	// The real case is the process dying between the two, which a test cannot arrange. Standing
	// in for it: a migration that makes its own record impossible to write.
	s := freshStore(t)
	if _, err := s.AppliedMigrations(t.Context()); err != nil {
		t.Fatal(err)
	}

	migrations := []Migration{{
		Number: 1, Name: "hostile", Checksum: "a",
		SQL: "create table kept (id int); drop table migration;",
	}}
	if _, err := s.Migrate(t.Context(), migrations); err == nil {
		t.Fatal("a migration whose record could not be written reported success")
	}

	var kept, ledger bool
	if err := s.Pool().QueryRow(t.Context(),
		`select exists (select 1 from information_schema.tables where table_name = 'kept'),
		        exists (select 1 from information_schema.tables where table_name = 'migration')`,
	).Scan(&kept, &ledger); err != nil {
		t.Fatal(err)
	}
	if kept {
		t.Error("the schema change survived although nothing recorded it; the next run would " +
			"apply it again, to a database that already has it")
	}
	if !ledger {
		t.Error("the migration record did not come back with the rollback")
	}
}

func TestAChangedMigrationIsRefusedAgainstARealRecord(t *testing.T) {
	// The same refusal as the unit test, against a record PostgreSQL actually kept — which is
	// what the guard protects, and the unit test can only model.
	s := freshStore(t)
	first := []Migration{{Number: 1, Name: "people", SQL: "create table person (id int)", Checksum: "a"}}
	if _, err := s.Migrate(t.Context(), first); err != nil {
		t.Fatal(err)
	}

	edited := []Migration{{Number: 1, Name: "people", SQL: "create table person (id bigint)", Checksum: "b"}}
	if _, err := s.Migrate(t.Context(), edited); err == nil {
		t.Fatal("a migration edited after it ran was accepted")
	}
}

func TestTwoRunnersDoNotRaceEachOther(t *testing.T) {
	// A restart during a slow migration produces exactly this: two copies of the control plane
	// migrating one database. Without the advisory lock both read an empty record, both decide
	// everything is outstanding, and the second fails on `create table` — which looks like a
	// broken migration rather than a race.
	s := freshStore(t)
	migrations := []Migration{{
		Number: 1, Name: "slow", Checksum: "a",
		SQL: "create table slow (id int); select pg_sleep(0.4);",
	}}

	var wg sync.WaitGroup
	results := make([]error, 2)
	counts := make([]int, 2)
	for i := range results {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			done, err := s.Migrate(context.Background(), migrations)
			results[i], counts[i] = err, len(done)
		}(i)
	}
	wg.Wait()

	for i, err := range results {
		if err != nil {
			t.Errorf("runner %d failed: %v", i, err)
		}
	}
	if counts[0]+counts[1] != 1 {
		t.Errorf("the migration was applied %d times between two runners; exactly one should have "+
			"done the work and the other should have found nothing to do", counts[0]+counts[1])
	}
}

func TestLoadedMigrationsApplyToARealDatabase(t *testing.T) {
	// A migration that parses and does not run is the failure this catches. The unit tests read
	// files and check names; nothing there executes SQL.
	s := freshStore(t)
	migrations, err := LoadMigrations(fstest.MapFS{
		"migrations/0001-first.sql":  {Data: []byte("create table a (id int);")},
		"migrations/0002-second.sql": {Data: []byte("alter table a add column b text;")},
	}, "migrations")
	if err != nil {
		t.Fatal(err)
	}
	done, err := s.Migrate(t.Context(), migrations)
	if err != nil {
		t.Fatal(err)
	}
	if len(done) != 2 || done[0].Number != 1 || done[1].Number != 2 {
		t.Fatalf("applied %+v", done)
	}
}

func TestReadyRefusesADatabaseThatWillNotAnswer(t *testing.T) {
	// Ready is what stands between the bootstrap and a control plane that starts against a
	// database still coming up. It has to give up rather than block for ever, and it has to fail
	// when nothing is there.
	admin(t)
	t.Setenv(Variable("testing"), "postgres://nobody@127.0.0.1:1/nothing")
	s, err := Open(t.Context(), "testing")
	if err != nil {
		t.Fatal(err)
	}
	defer s.Close()

	start := time.Now()
	if err := s.Ready(t.Context(), 1*time.Second); err == nil {
		t.Fatal("Ready returned success against a port with nothing on it")
	}
	if elapsed := time.Since(start); elapsed > 10*time.Second {
		t.Errorf("Ready took %s to give up on a 1s budget", elapsed)
	}
}