package ops

import (
	"database/sql"
	"os"
	"path/filepath"
	"slices"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"

	_ "modernc.org/sqlite"
)

// TestConfigGenSkipsHibernatingRoutes pins Eleven's two-replicators lesson in
// executable form: a route with Hibernate=true must never appear in the fleet
// litestream config, because the activator's embedded Litestream owns that
// database's replication — a second replicator on the same WAL would fight it
// for checkpointing.
func TestConfigGenSkipsHibernatingRoutes(t *testing.T) {
	dir := t.TempDir()
	reg, err := registry.Open(filepath.Join(dir, "registry.db"))
	if err != nil {
		t.Fatal(err)
	}

	// A unit tenant's data lives OUTSIDE dataDir entirely (its systemd
	// StateDirectory, e.g. /var/lib/carlos-app/<host>) — a separate temp dir
	// here stands in for that, deliberately not under dir, so this test
	// actually exercises FleetDBs' DBPath-based discovery instead of
	// accidentally passing because the file happened to sit where the glob
	// already looks.
	unitDir := t.TempDir()

	awake := filepath.Join(dir, "a.example.com.db")
	asleep := filepath.Join(dir, "b.example.com.db")
	unitTenant := filepath.Join(unitDir, "state.db")
	for _, p := range []string{awake, asleep, unitTenant} {
		if err := os.WriteFile(p, []byte("x"), 0o644); err != nil {
			t.Fatal(err)
		}
	}
	if err := reg.Upsert(registry.Route{Host: "a.example.com", DBPath: awake}); err != nil {
		t.Fatal(err)
	}
	if err := reg.Upsert(registry.Route{Host: "b.example.com", DBPath: asleep, Hibernate: true}); err != nil {
		t.Fatal(err)
	}
	// A unit-backed hibernating tenant (carlos-app@) with an operator-set
	// -db: its wake/sleep is systemctl and the activator never replicates
	// it, so this box-wide replicator must own its database — found via
	// DBPath, not the dataDir glob (its file isn't under dataDir at all).
	if err := reg.Upsert(registry.Route{Host: "c.example.com", DBPath: unitTenant,
		Hibernate: true, Backing: registry.BackingUnit}); err != nil {
		t.Fatal(err)
	}

	dbs, err := FleetDBs(reg, dir)
	if err != nil {
		t.Fatal(err)
	}
	cfg := LitestreamConfig(dbs, "bucket-x", "fleet", "eu-west-1")
	if !strings.Contains(cfg, awake) {
		t.Fatalf("awake route's DB missing from config:\n%s", cfg)
	}
	if strings.Contains(cfg, asleep) {
		t.Fatalf("hibernating route's DB must be excluded (activator owns its replication):\n%s", cfg)
	}
	if !strings.Contains(cfg, unitTenant) {
		t.Fatalf("unit-backed hibernating route's DB (outside dataDir) missing (nobody else replicates it):\n%s", cfg)
	}
	// The registry itself lives in the data dir and is not anyone's hibernating
	// instance — it must be replicated.
	if !strings.Contains(cfg, filepath.Join(dir, "registry.db")) {
		t.Fatalf("registry.db missing from config:\n%s", cfg)
	}
	if !strings.Contains(cfg, "GENERATED") {
		t.Fatalf("generated-file warning missing:\n%s", cfg)
	}
	if !strings.Contains(cfg, "region: eu-west-1") {
		t.Fatalf("replica region missing:\n%s", cfg)
	}
}

// TestFleetDBsDiscoversNestedDatabases pins the isolated-exec companion to
// the flat glob above (design §7): a route's DBPath being "claimed" only
// FILTERS what the glob already found. For a non-unit route it never adds
// a path by itself, so a nested row's database, under
// <dataDir>/instances/<id>/<host>.db, would stay invisible to replication
// forever without the nested glob to find it in the first place.
func TestFleetDBsDiscoversNestedDatabases(t *testing.T) {
	dir := t.TempDir()
	reg, err := registry.Open(filepath.Join(dir, "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	host := "nested.example.com"
	nested := provision.NestedDBPath(dir, host)
	if err := os.MkdirAll(provision.InstanceDir(dir, host), 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(nested, []byte("x"), 0o644); err != nil {
		t.Fatal(err)
	}
	// Not hibernating: a resident, activator-managed nested row is exactly
	// as replicable by this box-wide pass as a flat one.
	if err := reg.Upsert(registry.Route{Host: host, DBPath: nested, Socket: provision.NestedSocket(host)}); err != nil {
		t.Fatal(err)
	}

	dbs, err := FleetDBs(reg, dir)
	if err != nil {
		t.Fatal(err)
	}
	if !slices.Contains(dbs, nested) {
		t.Fatalf("nested database missing from FleetDBs: %v, want %s", dbs, nested)
	}
}

// TestFleetDBsUnitTenantWithoutDBPathIsInvisible pins the honest half of the
// I1 fix: a BackingUnit route added without -db has no discoverable database
// path (the platform enforces no filename convention inside a unit tenant's
// state dir), so FleetDBs must not guess one — it must simply not appear in
// the generated config. This is the documented gap operators must close by
// hand (or cli.go's -hibernate requirement closes it for them); it must
// never masquerade as covered.
func TestFleetDBsUnitTenantWithoutDBPathIsInvisible(t *testing.T) {
	dir := t.TempDir()
	reg, err := registry.Open(filepath.Join(dir, "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	if err := reg.Upsert(registry.Route{Host: "no-db.example.com", Backing: registry.BackingUnit}); err != nil {
		t.Fatal(err)
	}

	dbs, err := FleetDBs(reg, dir)
	if err != nil {
		t.Fatal(err)
	}
	for _, p := range dbs {
		if strings.Contains(p, "no-db") {
			t.Fatalf("FleetDBs invented a path for a unit tenant with no --db: %v", dbs)
		}
	}
}

// TestFleetDBsDedupesUnitPathUnderDataDir: a unit tenant's DBPath can, in
// principle, sit inside dataDir (nothing forbids it) — FleetDBs must not
// list it twice just because both the glob and the DBPath pass found it.
func TestFleetDBsDedupesUnitPathUnderDataDir(t *testing.T) {
	dir := t.TempDir()
	reg, err := registry.Open(filepath.Join(dir, "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	p := filepath.Join(dir, "d.example.com.db")
	if err := os.WriteFile(p, []byte("x"), 0o644); err != nil {
		t.Fatal(err)
	}
	if err := reg.Upsert(registry.Route{Host: "d.example.com", DBPath: p, Backing: registry.BackingUnit}); err != nil {
		t.Fatal(err)
	}

	dbs, err := FleetDBs(reg, dir)
	if err != nil {
		t.Fatal(err)
	}
	count := 0
	for _, got := range dbs {
		if got == p {
			count++
		}
	}
	if count != 1 {
		t.Fatalf("FleetDBs listed %s %d times, want exactly once: %v", p, count, dbs)
	}
}

// The round-trip is the contract that lets restore-verify read the generated
// config without a YAML parser: whatever LitestreamConfig writes,
// ParseLitestreamDBs must recover exactly.
func TestConfigRoundTrip(t *testing.T) {
	in := []string{"/data/registry.db", "/data/a.example.com.db", "/data/b.example.com.db"}
	got := ParseLitestreamDBs(LitestreamConfig(in, "b", "p", "eu-west-1"))
	if len(got) != len(in) {
		t.Fatalf("round-trip lost databases: in %d, out %d (%v)", len(in), len(got), got)
	}
	want := map[string]bool{}
	for _, p := range in {
		want[p] = true
	}
	for _, p := range got {
		if !want[p] {
			t.Fatalf("round-trip invented path %q", p)
		}
	}
}

func TestParseLitestreamDBsIgnoresNoise(t *testing.T) {
	got := ParseLitestreamDBs("# comment\ndbs:\n  - path: /data/x.db\n    replicas:\n      - url: s3://b/p/x.db\n        region: eu-west-1\n\n")
	if len(got) != 1 || got[0] != "/data/x.db" {
		t.Fatalf("ParseLitestreamDBs = %#v, want [/data/x.db]", got)
	}
	if got := ParseLitestreamDBs(""); len(got) != 0 {
		t.Fatalf("ParseLitestreamDBs(\"\") = %#v, want none", got)
	}
}

// Re-running the generator over an unchanged fleet must produce a
// byte-identical file — that is what lets the `litestream` verb report
// "unchanged" and skip a restart-worthy config write.
func TestConfigChangeDetection(t *testing.T) {
	dbs := []string{"/data/b.db", "/data/a.db", "/data/registry.db"}
	first := LitestreamConfig(dbs, "bucket", "prefix", "eu-west-1")
	second := LitestreamConfig([]string{"/data/registry.db", "/data/a.db", "/data/b.db"}, "bucket", "prefix", "eu-west-1")
	if first != second {
		t.Fatalf("same fleet produced different configs:\n--- first\n%s\n--- second\n%s", first, second)
	}
}

func TestCheckRestoredDBRejectsEmptySchema(t *testing.T) {
	dir := t.TempDir()

	good := filepath.Join(dir, "good.db")
	db, err := sql.Open("sqlite", good)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec("CREATE TABLE t (id INTEGER PRIMARY KEY); INSERT INTO t VALUES (1)"); err != nil {
		t.Fatal(err)
	}
	db.Close()
	if err := CheckRestoredDB(good); err != nil {
		t.Fatalf("CheckRestoredDB(good) = %v, want nil", err)
	}

	// A schema-less database restores "successfully" but would be useless in a
	// disaster — the check must refuse it.
	empty := filepath.Join(dir, "empty.db")
	db, err = sql.Open("sqlite", empty)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec("PRAGMA user_version=1"); err != nil { // force file creation
		t.Fatal(err)
	}
	db.Close()
	if err := CheckRestoredDB(empty); err == nil {
		t.Fatal("CheckRestoredDB accepted a database with no tables")
	}

	// Garbage bytes must fail loudly, not pass as an open-able file.
	junk := filepath.Join(dir, "junk.db")
	if err := os.WriteFile(junk, []byte("this is not a sqlite database at all"), 0o644); err != nil {
		t.Fatal(err)
	}
	if err := CheckRestoredDB(junk); err == nil {
		t.Fatal("CheckRestoredDB accepted a non-SQLite file")
	}
}

// The console's own SQLite — accounts, ownership, sessions, the deployment's
// Keymail credentials — lives one directory BELOW dataDir, where FleetDBs'
// flat <data>/*.db glob cannot see it. Losing it loses who everyone is, so it
// is included explicitly.
func TestFleetDBsIncludesConsoleDB(t *testing.T) {
	dir := t.TempDir()
	reg, err := registry.Open(filepath.Join(dir, "registry.db"))
	if err != nil {
		t.Fatal(err)
	}

	// Absent console: nothing invented, and no error.
	dbs, err := FleetDBs(reg, dir)
	if err != nil {
		t.Fatal(err)
	}
	for _, p := range dbs {
		if strings.Contains(p, "console") {
			t.Fatalf("FleetDBs listed a console DB on a box that has none: %v", dbs)
		}
	}

	consoleDB := filepath.Join(dir, "console", "console.db")
	if err := os.MkdirAll(filepath.Dir(consoleDB), 0o750); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(consoleDB, []byte("x"), 0o600); err != nil {
		t.Fatal(err)
	}

	dbs, err = FleetDBs(reg, dir)
	if err != nil {
		t.Fatal(err)
	}
	if !slices.Contains(dbs, consoleDB) {
		t.Fatalf("console DB missing from the replication set: %v", dbs)
	}
}

// CONSOLE_DB is the console's own env var for its database; a box that moved
// it must not silently drop out of replication.
func TestFleetDBsHonoursConsoleDBEnv(t *testing.T) {
	dir := t.TempDir()
	reg, err := registry.Open(filepath.Join(dir, "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	elsewhere := filepath.Join(t.TempDir(), "moved-console.db")
	if err := os.WriteFile(elsewhere, []byte("x"), 0o600); err != nil {
		t.Fatal(err)
	}
	t.Setenv("CONSOLE_DB", elsewhere)

	dbs, err := FleetDBs(reg, dir)
	if err != nil {
		t.Fatal(err)
	}
	if !slices.Contains(dbs, elsewhere) {
		t.Fatalf("relocated console DB missing from the replication set: %v", dbs)
	}
}

// The generated config must state the recovery window rather than inherit
// litestream's 24h snapshot retention by silence. A day is the shortest
// useful window there is: a bad migration noticed on Tuesday is unrecoverable
// if it landed on Sunday.
func TestConfigPinsSnapshotRetention(t *testing.T) {
	cfg := LitestreamConfig([]string{"/data/registry.db"}, "b", "p", "eu-west-1")
	if !strings.Contains(cfg, "snapshot:") {
		t.Fatalf("generated config states no snapshot policy — it inherits a 24h window silently:\n%s", cfg)
	}
	if !strings.Contains(cfg, "retention: "+SnapshotRetention.String()) {
		t.Fatalf("generated config does not pin retention to %s:\n%s", SnapshotRetention, cfg)
	}
	if SnapshotRetention <= 24*time.Hour {
		t.Fatalf("SnapshotRetention = %s, which is no better than litestream's default", SnapshotRetention)
	}
	// And the round-trip the restore-verify verb depends on must survive it.
	if got := ParseLitestreamDBs(cfg); len(got) != 1 || got[0] != "/data/registry.db" {
		t.Fatalf("ParseLitestreamDBs broke on the retention stanza: %v", got)
	}
}

// A config that STATES a snapshot policy the installed binary silently
// ignores is worse than one that states nothing: it reads as a decision
// someone made. litestream 0.3.13 (still on the flagship, verified 2026-08-15)
// parses the stanza and drops it on the floor, and 0.5 accepts unknown keys
// just as quietly — so parsing proves nothing and the VERSION is the only
// signal there is.
func TestSnapshotPolicySupported(t *testing.T) {
	cases := []struct {
		out  string
		want bool
	}{
		{"v0.3.13", false}, // the flagship's binary, dated 2023
		{"0.3.13", false},  // some builds print no leading v
		{"v0.4.9", false},  // still pre-snapshot-policy
		{"0.5.15", true},   // tito's binary, and the repo's pin
		{"v0.5.14", true},  // the vendored library version
		{"v1.0.0", true},   // a future major must not regress to "unsupported"
		{"litestream v0.5.15\n", true},
		{"", false}, // unknown is treated as unsupported: warn, never assume
		{"not-a-version", false},
	}
	for _, c := range cases {
		if got := SnapshotPolicySupported(c.out); got != c.want {
			t.Errorf("SnapshotPolicySupported(%q) = %v, want %v", c.out, got, c.want)
		}
	}
}

// The drill must hold ONE restored database at a time, not all of them.
//
// This is not tidiness. carlos-restore-verify.service runs PrivateTmp=yes, so
// its temp dir is a tmpfs whose pages are charged to the unit's own cgroup —
// accumulating every restored copy there spends MemoryMax on files. Measured
// on carlos-tito-ie 2026-08-15: 54 databases totalling 247M against
// MemoryMax=256M, killed by the OOM killer at "256.2M memory peak" after ~40
// clean restores. Deleting each copy once it has been checked bounds the peak
// at the LARGEST SINGLE database instead of their sum.
func TestRestoreVerifyHoldsOneRestoreAtATime(t *testing.T) {
	tmp := t.TempDir()
	srcDir := t.TempDir()
	cfgPath := filepath.Join(t.TempDir(), "litestream.yml")
	// Real source databases: the freshness check reads their headers, and a
	// config naming files that do not exist is itself a finding now.
	var dbs []string
	for _, name := range []string{"a.db", "b.db", "c.db"} {
		p := filepath.Join(srcDir, name)
		writeCounterDB(t, p, 3)
		dbs = append(dbs, p)
	}
	if err := os.WriteFile(cfgPath, []byte(LitestreamConfig(dbs, "b", "p", "eu-west-1")), 0o644); err != nil {
		t.Fatal(err)
	}

	maxSeen := 0
	prev := restoreAndCheckFn
	restoreAndCheckFn = func(_, _, dbPath, tmpDir string) error {
		// Stand in for a restore of a CURRENT replica: a byte copy of the
		// source, so its change counter matches and freshness passes.
		body, err := os.ReadFile(dbPath)
		if err != nil {
			return err
		}
		if err := os.WriteFile(filepath.Join(tmpDir, filepath.Base(dbPath)), body, 0o600); err != nil {
			return err
		}
		ents, err := os.ReadDir(tmpDir)
		if err != nil {
			return err
		}
		if len(ents) > maxSeen {
			maxSeen = len(ents)
		}
		return nil
	}
	t.Cleanup(func() { restoreAndCheckFn = prev })

	if err := RestoreVerify("litestream", cfgPath, tmp); err != nil {
		t.Fatalf("RestoreVerify: %v", err)
	}
	if maxSeen > 1 {
		t.Fatalf("as many as %d restored copies were held at once — the drill spends MemoryMax on tmpfs files", maxSeen)
	}
	left, err := os.ReadDir(tmp)
	if err != nil {
		t.Fatal(err)
	}
	if len(left) != 0 {
		t.Fatalf("RestoreVerify left %d file(s) behind", len(left))
	}
}

// --- replica freshness ---------------------------------------------------
//
// The drill proves a replica RESTORES. It has never proved the replica is
// CURRENT — and those come apart exactly when it matters. On 2026-08-11 the
// flagship's litestream died with "permission denied" writing its shadow WAL
// into a DynamicUser tenant's state dir; had nobody noticed by hand, the daily
// drill would have gone on passing indefinitely, restoring a replica frozen at
// the moment of death. A valid backup of last week is not a backup.

func writeCounterDB(t *testing.T, path string, writes int) {
	t.Helper()
	db, err := sql.Open("sqlite", path)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	// DELETE journalling: the header change counter advances per commit
	// without needing a checkpoint, which keeps the fixture honest about what
	// it is testing.
	if _, err := db.Exec("PRAGMA journal_mode=DELETE"); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec("CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY)"); err != nil {
		t.Fatal(err)
	}
	for i := 0; i < writes; i++ {
		// A fresh rowid each call, so re-seeding an existing file adds
		// transactions rather than colliding on the primary key.
		if _, err := db.Exec("INSERT INTO t (id) VALUES (NULL)"); err != nil {
			t.Fatal(err)
		}
	}
}

func TestChangeCounterAdvancesWithWrites(t *testing.T) {
	dir := t.TempDir()
	a := filepath.Join(dir, "a.db")
	writeCounterDB(t, a, 1)
	first, err := ChangeCounter(a)
	if err != nil {
		t.Fatal(err)
	}
	writeCounterDB(t, a, 5)
	second, err := ChangeCounter(a)
	if err != nil {
		t.Fatal(err)
	}
	if second <= first {
		t.Fatalf("change counter did not advance across writes: %d then %d", first, second)
	}

	junk := filepath.Join(dir, "junk.db")
	if err := os.WriteFile(junk, []byte("nope"), 0o600); err != nil {
		t.Fatal(err)
	}
	if _, err := ChangeCounter(junk); err == nil {
		t.Fatal("ChangeCounter accepted a file that is not a SQLite database")
	}
}

func TestReplicaFreshness(t *testing.T) {
	dir := t.TempDir()
	local := filepath.Join(dir, "local.db")
	behind := filepath.Join(dir, "behind.db")
	writeCounterDB(t, behind, 1) // the replica, frozen early
	writeCounterDB(t, local, 9)  // the source, written since

	long := time.Now().Add(-2 * time.Hour)
	recent := time.Now()

	// Quiet since well before the drill, replica still behind: the replicator
	// has had every chance to catch up and has not. That is the alarm.
	if v, detail := replicaFreshness(local, behind, long); v != freshnessStale {
		t.Fatalf("a replica behind a long-quiet source = %v (%s), want stale", v, detail)
	}
	// Written moments ago: the replica being a transaction or two behind is
	// ordinary lag, not a fault. Never cry wolf here — a drill that goes red
	// on healthy churn is a drill people stop reading.
	if v, _ := replicaFreshness(local, behind, recent); v != freshnessOK {
		t.Fatalf("a replica behind a just-written source = %v, want ok (still catching up)", v)
	}
	// Caught up.
	if v, _ := replicaFreshness(local, local, long); v != freshnessOK {
		t.Fatalf("an up-to-date replica = %v, want ok", v)
	}
	// The source is gone: a config naming a database that no longer exists is
	// its own finding — the replica can never advance again.
	if v, _ := replicaFreshness(filepath.Join(dir, "absent.db"), behind, long); v != freshnessNoSource {
		t.Fatalf("a missing source = %v, want no-source", v)
	}
	// Unreadable header: say nothing rather than invent a verdict.
	junk := filepath.Join(dir, "junk.db")
	if err := os.WriteFile(junk, []byte("nope"), 0o600); err != nil {
		t.Fatal(err)
	}
	if v, _ := replicaFreshness(junk, behind, long); v != freshnessUnknown {
		t.Fatalf("an unreadable source = %v, want unknown", v)
	}
}

// The whole point, end to end: a replica that restores perfectly but has
// stopped tracking its source must FAIL the run. Before this, such a replica
// was reported "ok" every night for as long as the replicator stayed dead.
func TestRestoreVerifyFailsOnAStaleReplica(t *testing.T) {
	srcDir, replicaDir := t.TempDir(), t.TempDir()
	cfgPath := filepath.Join(t.TempDir(), "litestream.yml")

	live := filepath.Join(srcDir, "live.db")
	writeCounterDB(t, live, 2)
	// Snapshot the replica HERE — this stands for the moment the replicator
	// died — then let the source move on without it.
	frozen, err := os.ReadFile(live)
	if err != nil {
		t.Fatal(err)
	}
	writeCounterDB(t, live, 20)
	// And make the source quiet for longer than the grace window, so this is
	// unambiguously "replication stopped" rather than "still catching up".
	old := time.Now().Add(-2 * time.Hour)
	if err := os.Chtimes(live, old, old); err != nil {
		t.Fatal(err)
	}

	if err := os.WriteFile(cfgPath, []byte(LitestreamConfig([]string{live}, "b", "p", "eu-west-1")), 0o644); err != nil {
		t.Fatal(err)
	}
	prev := restoreAndCheckFn
	restoreAndCheckFn = func(_, _, dbPath, tmpDir string) error {
		return os.WriteFile(filepath.Join(tmpDir, filepath.Base(dbPath)), frozen, 0o600)
	}
	t.Cleanup(func() { restoreAndCheckFn = prev })

	err = RestoreVerify("litestream", cfgPath, replicaDir)
	if err == nil {
		t.Fatal("a replica frozen at the moment its replicator died passed the drill")
	}
	if !strings.Contains(err.Error(), "STOPPED TRACKING") {
		t.Fatalf("error does not name the failure: %v", err)
	}
}

// A unit tenant whose app manages its OWN sub-instances keeps a TREE of
// databases, not one file: kass on the flagship has registry.db plus an
// instances/ directory that grows as the tenant creates instances. Naming a
// single -db path cannot express that, which is why the flagship's config is
// still hand-maintained ("carlos ops litestream takes over once nested layouts
// are supported") — and why it had already drifted, missing one instance
// database that the tenant created after the list was written.
//
// So: a BackingUnit route whose -db names a DIRECTORY replicates every *.db
// beneath it. Exec routes are untouched by this — the activator consumes their
// DBPath verbatim (Restore, Seed, the --db flag), so a directory there would
// be meaningless.
func TestFleetDBsWalksAUnitTenantDirectory(t *testing.T) {
	dir := t.TempDir()
	reg, err := registry.Open(filepath.Join(dir, "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	state := t.TempDir()
	nested := filepath.Join(state, "instances")
	if err := os.MkdirAll(nested, 0o750); err != nil {
		t.Fatal(err)
	}
	top := filepath.Join(state, "registry.db")
	inst1 := filepath.Join(nested, "5vepbveytb8y.db")
	inst2 := filepath.Join(nested, "eafu5brhjs26.db")
	for _, p := range []string{top, inst1, inst2} {
		if err := os.WriteFile(p, []byte("x"), 0o600); err != nil {
			t.Fatal(err)
		}
	}
	// Litestream's own shadow-WAL directory sits beside each database and is
	// full of files that are not databases. Replicating a replicator's
	// bookkeeping would be absurd, and dot-directories are its convention.
	meta := filepath.Join(state, ".registry.db-litestream", "generations", "abc", "wal")
	if err := os.MkdirAll(meta, 0o750); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(filepath.Join(meta, "00000000.db"), []byte("x"), 0o600); err != nil {
		t.Fatal(err)
	}
	// Sidecars and empty placeholders are not databases either.
	for _, p := range []string{top + "-wal", top + "-shm", filepath.Join(state, "empty.db")} {
		if err := os.WriteFile(p, nil, 0o600); err != nil {
			t.Fatal(err)
		}
	}

	if err := reg.Upsert(registry.Route{Host: "home.kass.training", Backing: registry.BackingUnit, DBPath: state}); err != nil {
		t.Fatal(err)
	}

	dbs, err := FleetDBs(reg, dir)
	if err != nil {
		t.Fatal(err)
	}
	for _, want := range []string{top, inst1, inst2} {
		if !slices.Contains(dbs, want) {
			t.Fatalf("tree database %s missing: %v", want, dbs)
		}
	}
	for _, unwanted := range []string{meta, top + "-wal", top + "-shm", filepath.Join(state, "empty.db")} {
		if slices.Contains(dbs, unwanted) {
			t.Fatalf("FleetDBs swept up %s: %v", unwanted, dbs)
		}
	}
	for _, got := range dbs {
		if strings.Contains(got, "-litestream") {
			t.Fatalf("FleetDBs descended into a litestream meta directory: %s", got)
		}
	}
}

// The tree rule must not leak to exec routes: their DBPath is consumed
// verbatim by the activator, so a directory there is a mistake, not a tree.
func TestFleetDBsDoesNotWalkExecRouteDBPath(t *testing.T) {
	dir := t.TempDir()
	reg, err := registry.Open(filepath.Join(dir, "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	state := t.TempDir()
	if err := os.WriteFile(filepath.Join(state, "inner.db"), []byte("x"), 0o600); err != nil {
		t.Fatal(err)
	}
	if err := reg.Upsert(registry.Route{Host: "exec.example.com", DBPath: state}); err != nil {
		t.Fatal(err)
	}

	dbs, err := FleetDBs(reg, dir)
	if err != nil {
		t.Fatal(err)
	}
	if slices.Contains(dbs, filepath.Join(state, "inner.db")) {
		t.Fatalf("an exec route's directory DBPath was walked: %v", dbs)
	}
}

// Cutting a box over from a hand-maintained config to a generated one is the
// moment a database can silently stop being replicated: the new list is
// derived from the registry, and anything the registry does not know about
// simply is not in it. Nobody would notice — replication of a dropped database
// just stops, and the drill only checks what the config names.
//
// So the generator refuses to shrink the set unless told to.
func TestCoverageShortfall(t *testing.T) {
	existing := LitestreamConfig([]string{"/data/a.db", "/data/b.db", "/data/registry.db"}, "b", "p", "eu-west-1")

	// A superset is fine — that is the ordinary case (console.db arriving).
	if missing := CoverageShortfall(existing, []string{"/data/a.db", "/data/b.db", "/data/registry.db", "/data/console/console.db"}); len(missing) != 0 {
		t.Fatalf("a superset was reported as a shortfall: %v", missing)
	}
	// Identical is fine.
	if missing := CoverageShortfall(existing, []string{"/data/registry.db", "/data/b.db", "/data/a.db"}); len(missing) != 0 {
		t.Fatalf("an identical set was reported as a shortfall: %v", missing)
	}
	// Dropping one is the thing to catch, and the caller must be told WHICH.
	missing := CoverageShortfall(existing, []string{"/data/a.db", "/data/registry.db"})
	if len(missing) != 1 || missing[0] != "/data/b.db" {
		t.Fatalf("CoverageShortfall = %v, want [/data/b.db]", missing)
	}
	// No existing config (a fresh box) can never be a shortfall.
	if missing := CoverageShortfall("", []string{"/data/a.db"}); len(missing) != 0 {
		t.Fatalf("a fresh box reported a shortfall: %v", missing)
	}
}

// Two databases with the same basename must not replicate to the same place.
//
// Every SQLite-backed tenant on the fleet names its database registry.db or
// data.db; the box's own registry is /data/registry.db. Deriving the replica
// URL from filepath.Base collapses all of them onto one S3 key, and two
// litestream replicators writing one prefix interleave their WAL into a
// replica that restores as neither database. Tree discovery makes the
// collision the normal case rather than a curiosity: every tenant tree has a
// registry.db in it.
func TestReplicaURLsAreUniquePerPath(t *testing.T) {
	cfg := LitestreamConfig([]string{
		"/data/registry.db",
		"/var/lib/carlos-app/home.kass.training/registry.db",
		"/var/lib/carlos-app/other.example.com/registry.db",
	}, "bucket-x", "fleet", "eu-west-1")

	var urls []string
	for _, line := range strings.Split(cfg, "\n") {
		if u, ok := strings.CutPrefix(strings.TrimSpace(line), "- url: "); ok {
			urls = append(urls, u)
		}
	}
	if len(urls) != 3 {
		t.Fatalf("expected 3 replica urls, got %d:\n%s", len(urls), cfg)
	}
	seen := map[string]bool{}
	for _, u := range urls {
		if seen[u] {
			t.Fatalf("two databases share the replica url %q — their WALs would interleave:\n%s", u, cfg)
		}
		seen[u] = true
	}
	// And the key should still be legible to whoever is doing the restore.
	if !strings.Contains(cfg, "home.kass.training") {
		t.Fatalf("replica keys lost the path that identifies them:\n%s", cfg)
	}
}

// The dataDir glob takes every *.db it finds, which means the box-wide
// replicator will happily stream anything that lands there — including things
// that are not tenant databases at all. The flagship has a 410M
// harbourmail-purge-backup.db sitting in /data: a leftover from a purge, owned
// by root, matching the glob exactly. Today it is not replicated only because
// that box's config is hand-written; the moment the generator takes over it
// would join the replication set and start costing storage forever.
//
// So the glob proposes and the REGISTRY disposes: a file under dataDir is
// replicated when some route claims it (or it is the box's own registry or
// console). Anything else is skipped and NAMED — never silently, because a
// skipped database that someone did want is exactly the failure this whole
// area keeps producing.
func TestFleetDBsSkipsUnclaimedFilesAndSaysSo(t *testing.T) {
	dir := t.TempDir()
	reg, err := registry.Open(filepath.Join(dir, "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	claimed := filepath.Join(dir, "app.example.com.db")
	stray := filepath.Join(dir, "harbourmail-purge-backup.db")
	for _, p := range []string{claimed, stray} {
		if err := os.WriteFile(p, []byte("x"), 0o600); err != nil {
			t.Fatal(err)
		}
	}
	if err := reg.Upsert(registry.Route{Host: "app.example.com", DBPath: claimed}); err != nil {
		t.Fatal(err)
	}

	dbs, skipped, err := FleetDBsWithSkipped(reg, dir, nil)
	if err != nil {
		t.Fatal(err)
	}
	if !slices.Contains(dbs, claimed) {
		t.Fatalf("a route's own database was skipped: %v", dbs)
	}
	if !slices.Contains(dbs, filepath.Join(dir, "registry.db")) {
		t.Fatalf("the box registry was skipped: %v", dbs)
	}
	if slices.Contains(dbs, stray) {
		t.Fatalf("an unclaimed file joined the replication set: %v", dbs)
	}
	if !slices.Contains(skipped, stray) {
		t.Fatalf("the unclaimed file was dropped SILENTLY — skipped = %v", skipped)
	}
}

// A host with a layout migration journal is between two directories, so
// the generator claims neither of its database paths: its file is named
// among the skipped rather than replicated from a place it is leaving.
func TestFleetDBsSkipsAHostTheLayoutPassHolds(t *testing.T) {
	dir := t.TempDir()
	reg, err := registry.Open(filepath.Join(dir, "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	held := filepath.Join(dir, "held.example.com.db")
	kept := filepath.Join(dir, "kept.example.com.db")
	for _, p := range []string{held, kept} {
		if err := os.WriteFile(p, []byte("x"), 0o600); err != nil {
			t.Fatal(err)
		}
	}
	for _, rt := range []registry.Route{
		{Host: "held.example.com", DBPath: held},
		{Host: "kept.example.com", DBPath: kept},
	} {
		if err := reg.Upsert(rt); err != nil {
			t.Fatal(err)
		}
	}
	journal := provision.LayoutJournalPath(dir, "held.example.com")
	if err := os.MkdirAll(filepath.Dir(journal), 0o700); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(journal, []byte("{}"), 0o600); err != nil {
		t.Fatal(err)
	}

	dbs, skipped, err := FleetDBsWithSkipped(reg, dir, LayoutJournalPresent(dir))
	if err != nil {
		t.Fatal(err)
	}
	if slices.Contains(dbs, held) {
		t.Fatalf("a held host's database joined the replication set: %v", dbs)
	}
	if !slices.Contains(skipped, held) {
		t.Fatalf("a held host's database was dropped silently: skipped = %v", skipped)
	}
	if !slices.Contains(dbs, kept) {
		t.Fatalf("a host with no journal was skipped: %v", dbs)
	}

	// No hook is the old behaviour: every route's database is claimed.
	dbs, _, err = FleetDBsWithSkipped(reg, dir, nil)
	if err != nil {
		t.Fatal(err)
	}
	if !slices.Contains(dbs, held) {
		t.Fatalf("with no hook the held host was skipped: %v", dbs)
	}
}

// LayoutJournalPresent answers true for anything at the journal path that
// is not definitely absent, a dangling symlink included: a journal that
// cannot be read is not proof there is none.
func TestLayoutJournalPresentIsLstatNotAbsent(t *testing.T) {
	dir := t.TempDir()
	present := LayoutJournalPresent(dir)
	if present("a.example.com") {
		t.Fatal("no journal directory, but a journal was reported")
	}
	link := provision.LayoutJournalPath(dir, "a.example.com")
	if err := os.MkdirAll(filepath.Dir(link), 0o700); err != nil {
		t.Fatal(err)
	}
	if err := os.Symlink(filepath.Join(dir, "nowhere"), link); err != nil {
		t.Fatal(err)
	}
	if !present("a.example.com") {
		t.Fatal("a dangling symlink at the journal path was read as no journal")
	}
	if present("b.example.com") {
		t.Fatal("another host was reported held")
	}
}