// Package litestreamfacts establishes, by observation of the pinned
// libraries (litestream v0.5.14, superfly/ltx v0.5.1), the facts the
// move-any-app spec's restore contract rests on (moves spec rev 7.7 §5.1;
// joint rounds 5 and 6, R5-1 to R5-3, R6-2, R6-5). It is test-only and
// drives litestream as internal/lease does, with the file replica client
// the lease tests use. Each test asserts the exact fact the spec relies on
// and fails on any alternative behaviour, so a library upgrade that
// changes a fact fails here first (R6-2): a passing run is the contract,
// not a record of one observation. TestGate_AnchoredLineageThroughStore
// takes one anchored lineage through the production Store configuration
// (internal/lease's Held.Replicate, intervals shortened) with repeated
// writes, compaction, snapshot rotation, time-based level-0 expiry, disk
// loss and repeated cold restores, asserting exact positions and contents.
//
// What the file client cannot show (R6 and R7 reviews): S3 lists LTX
// files in filename order and ages them for retention by LastModified,
// the file client by the encoded LTX timestamp kept as the file's mtime;
// S3 keeps versions and delete markers, the file client replaces and
// deletes outright; S3 requests can complete late, and an upload
// outstanding across a writer replacement is the R7-1 hazard the model
// (internal/sim/hostlife) encodes. The repository has no S3 fake that
// litestream's client can talk to (the lease tests' fakeMirror is a
// MirrorStore seam), so those remain the model's and §15 step 8's.
// TestFact_R7_1_UnconditionalPut reads the pinned S3 client's source for
// the two facts the hazard rests on: an unconditional PutObject, and a key
// of prefix, level and TXID range only.
//
// The observed facts are recorded in ~/.cache/nearest-region/litestream-facts.md.
package litestreamfacts

import (
	"bytes"
	"context"
	"database/sql"
	"fmt"
	"go/build"
	"io"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"github.com/benbjohnson/litestream"
	lsfile "github.com/benbjohnson/litestream/file"
	"github.com/superfly/ltx"
	_ "modernc.org/sqlite"
)

// openDB opens path under litestream against client, with its own metadata
// directory, no background monitors (every sync is explicit), and returns
// the DB. The caller closes it.
func openDB(t *testing.T, path, meta string, client litestream.ReplicaClient) *litestream.DB {
	t.Helper()
	db := litestream.NewDB(path)
	db.SetMetaPath(meta)
	db.MonitorInterval = time.Hour
	db.Replica = litestream.NewReplicaWithClient(db, client)
	db.Replica.SyncInterval = time.Hour
	if err := db.Open(); err != nil {
		t.Fatalf("open %s: %v", path, err)
	}
	return db
}

func closeDB(t *testing.T, db *litestream.DB) {
	t.Helper()
	if err := db.Close(context.Background()); err != nil {
		t.Fatalf("close: %v", err)
	}
}

// write inserts n rows, each in its own transaction, so each is a
// transaction litestream can see.
func write(t *testing.T, path string, n int) {
	t.Helper()
	d, err := sql.Open("sqlite", path)
	if err != nil {
		t.Fatal(err)
	}
	defer d.Close()
	if _, err := d.Exec(`CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY, body TEXT)`); err != nil {
		t.Fatal(err)
	}
	for i := 0; i < n; i++ {
		if _, err := d.Exec(`INSERT INTO t (body) VALUES (?)`, strings.Repeat("x", 100)); err != nil {
			t.Fatal(err)
		}
	}
}

// rows reads the table's contents, id -> body, so a restore can be
// compared row by row, updates and deletes included (R7-5).
func rows(t *testing.T, path string) map[int]string {
	t.Helper()
	d, err := sql.Open("sqlite", "file:"+path+"?mode=ro")
	if err != nil {
		t.Fatal(err)
	}
	defer d.Close()
	rs, err := d.Query(`SELECT id, body FROM t ORDER BY id`)
	if err != nil {
		t.Fatal(err)
	}
	defer rs.Close()
	out := map[int]string{}
	for rs.Next() {
		var id int
		var body string
		if err := rs.Scan(&id, &body); err != nil {
			t.Fatal(err)
		}
		out[id] = body
	}
	return out
}

// mustRows fails unless the database at path holds exactly want.
func mustRows(t *testing.T, path string, want map[int]string) {
	t.Helper()
	got := rows(t, path)
	if len(got) != len(want) {
		t.Fatalf("%s: %d rows, want %d", filepath.Base(path), len(got), len(want))
	}
	for id, body := range want {
		if got[id] != body {
			t.Fatalf("%s: row %d is %q, want %q", filepath.Base(path), id, got[id], body)
		}
	}
}

// exec runs one statement in its own transaction.
func exec(t *testing.T, path, stmt string, args ...any) {
	t.Helper()
	d, err := sql.Open("sqlite", path)
	if err != nil {
		t.Fatal(err)
	}
	defer d.Close()
	if _, err := d.Exec(stmt, args...); err != nil {
		t.Fatal(err)
	}
}

// waitFor polls cond until it holds or the deadline passes (R7-5: a
// completion condition, not a sleep).
func waitFor(t *testing.T, what string, d time.Duration, cond func() bool) {
	t.Helper()
	deadline := time.Now().Add(d)
	for !cond() {
		if time.Now().After(deadline) {
			t.Fatalf("timed out waiting for %s", what)
		}
		time.Sleep(20 * time.Millisecond)
	}
}

func count(t *testing.T, path string) int {
	t.Helper()
	d, err := sql.Open("sqlite", "file:"+path+"?mode=ro")
	if err != nil {
		t.Fatal(err)
	}
	defer d.Close()
	var n int
	if err := d.QueryRow(`SELECT count(*) FROM t`).Scan(&n); err != nil {
		t.Fatal(err)
	}
	return n
}

func sync(t *testing.T, db *litestream.DB) {
	t.Helper()
	if err := db.SyncAndWait(context.Background()); err != nil {
		t.Fatalf("sync: %v", err)
	}
}

// files lists one level of a replica, in litestream's order.
func files(t *testing.T, client litestream.ReplicaClient, level int) []*ltx.FileInfo {
	t.Helper()
	itr, err := client.LTXFiles(context.Background(), level, 0, false)
	if err != nil {
		t.Fatal(err)
	}
	infos, err := ltx.SliceFileIterator(itr)
	if err != nil {
		t.Fatal(err)
	}
	return infos
}

func ranges(infos []*ltx.FileInfo) string {
	var out []string
	for _, f := range infos {
		out = append(out, fmt.Sprintf("[%d,%d]", f.MinTXID, f.MaxTXID))
	}
	if len(out) == 0 {
		return "none"
	}
	return strings.Join(out, " ")
}

// copyFile copies one LTX file between replica clients, as the install's
// CopyObject would.
func copyFile(t *testing.T, from, to litestream.ReplicaClient, info *ltx.FileInfo) {
	t.Helper()
	ctx := context.Background()
	rc, err := from.OpenLTXFile(ctx, info.Level, info.MinTXID, info.MaxTXID, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	defer rc.Close()
	if _, err := to.WriteLTXFile(ctx, info.Level, info.MinTXID, info.MaxTXID, rc); err != nil {
		t.Fatal(err)
	}
}

// restore restores client's newest state to out, litestream's own planner.
func restore(t *testing.T, client litestream.ReplicaClient, out string) error {
	t.Helper()
	r := litestream.NewReplicaWithClient(nil, client)
	opt := litestream.NewRestoreOptions()
	opt.OutputPath = out
	return r.Restore(context.Background(), opt)
}

// decode reads an LTX file's header and trailer and verifies it.
func decode(t *testing.T, client litestream.ReplicaClient, info *ltx.FileInfo) (ltx.Header, ltx.Trailer) {
	t.Helper()
	rc, err := client.OpenLTXFile(context.Background(), info.Level, info.MinTXID, info.MaxTXID, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	defer rc.Close()
	b, err := io.ReadAll(rc)
	if err != nil {
		t.Fatal(err)
	}
	dec := ltx.NewDecoder(bytes.NewReader(b))
	if err := dec.Verify(); err != nil {
		t.Fatalf("verify %s: %v", ltx.FormatFilename(info.MinTXID, info.MaxTXID), err)
	}
	return dec.Header(), dec.Trailer()
}

// encodeAnchor re-encodes a snapshot [1,N] as a level-0 file [N,N] with
// every page: the file litestream's position initialization fetches, so
// that replication continues from N (R5-1, the remedy under test). The
// header timestamp is an input (R6-5): with the same inputs the bytes are
// identical, which is what lets a replayed PUT repeat the same object.
func encodeAnchor(t *testing.T, from litestream.ReplicaClient, snap *ltx.FileInfo, ts int64) []byte {
	t.Helper()
	ctx := context.Background()
	rc, err := from.OpenLTXFile(ctx, snap.Level, snap.MinTXID, snap.MaxTXID, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	defer rc.Close()
	dec := ltx.NewDecoder(rc)
	if err := dec.DecodeHeader(); err != nil {
		t.Fatal(err)
	}
	hdr := dec.Header()
	var buf bytes.Buffer
	enc, err := ltx.NewEncoder(&buf)
	if err != nil {
		t.Fatal(err)
	}
	// WALOffset at the WAL header: litestream's verify reads the position
	// from the newest local level-0 file and, finding the WAL does not
	// match it, takes a fresh snapshot at the next TXID; an offset below
	// the header size is refused outright ("prev WAL offset is less than
	// the header size", observed)
	if err := enc.EncodeHeader(ltx.Header{
		Version:   ltx.Version,
		Flags:     ltx.HeaderFlagNoChecksum,
		PageSize:  hdr.PageSize,
		Commit:    hdr.Commit,
		MinTXID:   snap.MaxTXID,
		MaxTXID:   snap.MaxTXID,
		Timestamp: ts,
		WALOffset: litestream.WALHeaderSize,
	}); err != nil {
		t.Fatal(err)
	}
	page := make([]byte, hdr.PageSize)
	for {
		var ph ltx.PageHeader
		if err := dec.DecodePage(&ph, page); err == io.EOF {
			break
		} else if err != nil {
			t.Fatal(err)
		}
		if err := enc.EncodePage(ph, page); err != nil {
			t.Fatal(err)
		}
	}
	if err := dec.Close(); err != nil {
		t.Fatal(err)
	}
	if err := enc.Close(); err != nil {
		t.Fatal(err)
	}
	return buf.Bytes()
}

// anchorL0 writes the anchor for snap into to.
func anchorL0(t *testing.T, from litestream.ReplicaClient, snap *ltx.FileInfo, to litestream.ReplicaClient) {
	t.Helper()
	b := encodeAnchor(t, from, snap, time.Now().UnixMilli())
	if _, err := to.WriteLTXFile(context.Background(), 0, snap.MaxTXID, snap.MaxTXID, bytes.NewReader(b)); err != nil {
		t.Fatal(err)
	}
}

// mustRanges fails unless the level's current files are exactly want.
func mustRanges(t *testing.T, client litestream.ReplicaClient, level int, want string) {
	t.Helper()
	if got := ranges(files(t, client, level)); got != want {
		t.Fatalf("level %d: have %s, want exactly %s", level, got, want)
	}
}

// mustCount fails unless the database at path holds exactly want rows.
func mustCount(t *testing.T, path string, want int) {
	t.Helper()
	if n := count(t, path); n != want {
		t.Fatalf("%s: %d rows, want exactly %d", filepath.Base(path), n, want)
	}
}

// source builds a replicated source database with n rows and a level-9
// snapshot of it, and returns the client and the snapshot's info.
func source(t *testing.T, n int) (litestream.ReplicaClient, *ltx.FileInfo) {
	t.Helper()
	dir := t.TempDir()
	path := filepath.Join(dir, "src.db")
	write(t, path, 1)
	client := lsfile.NewReplicaClient(filepath.Join(dir, "replica"))
	db := openDB(t, path, filepath.Join(dir, "meta"), client)
	sync(t, db)
	write(t, path, n-1)
	sync(t, db)
	snap, err := db.Snapshot(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	closeDB(t, db)
	t.Logf("source: L0 %s; snapshot [%d,%d]", ranges(files(t, client, 0)), snap.MinTXID, snap.MaxTXID)
	return client, snap
}

// wipe removes the local database, its WAL and SHM, and litestream's
// metadata: a lost disk.
func wipe(t *testing.T, path, meta string) {
	t.Helper()
	for _, p := range []string{path, path + "-wal", path + "-shm", path + "-journal"} {
		os.Remove(p)
	}
	if err := os.RemoveAll(meta); err != nil {
		t.Fatal(err)
	}
}

// TestFact_R5_1_SnapshotOnlyInstallRestartsAtOne: install only the level-9
// snapshot [1,N], restore it, start replication with fresh metadata and no
// remote level 0. The next upload carries TXID 1, not N+1, and a cold
// restore afterwards loses the new writes.
func TestFact_R5_1_SnapshotOnlyInstallRestartsAtOne(t *testing.T) {
	src, snap := source(t, 5)
	dir := t.TempDir()
	gen := lsfile.NewReplicaClient(filepath.Join(dir, "gen"))
	copyFile(t, src, gen, snap)
	out := filepath.Join(dir, "out.db")
	if err := restore(t, gen, out); err != nil {
		t.Fatal(err)
	}
	if n := count(t, out); n != 5 {
		t.Fatalf("restored %d rows, want 5", n)
	}
	meta := filepath.Join(dir, "meta")
	db := openDB(t, out, meta, gen)
	write(t, out, 2)
	sync(t, db)
	closeDB(t, db)
	l0 := files(t, gen, 0)
	t.Logf("FACT R5-1a: after a snapshot-only install (snapshot [1,%d]) the first upload is level 0 %s", snap.MaxTXID, ranges(l0))
	if len(l0) == 0 || l0[0].MinTXID != 1 {
		t.Fatalf("expected replication to restart at TXID 1, got %s", ranges(l0))
	}
	wipe(t, out, meta)
	out2 := filepath.Join(dir, "out2.db")
	if err := restore(t, gen, out2); err != nil {
		t.Fatal(err)
	}
	n := count(t, out2)
	t.Logf("FACT R5-1b: a cold restore after that holds %d rows of 7: the planner roots at the snapshot and skips level-0 files below its TXID", n)
	if n != 5 {
		t.Fatalf("expected the two new rows to be lost (5 rows), got %d", n)
	}
}

// TestFact_R5_1_AnchorL0ContinuesFromN: the remedy. The install also writes
// a level-0 file [N,N] with every page. Litestream's open fetches the
// newest remote level-0 file as its local position, so the next upload
// carries N+1 and a cold restore has the new writes.
func TestFact_R5_1_AnchorL0ContinuesFromN(t *testing.T) {
	src, snap := source(t, 5)
	dir := t.TempDir()
	gen := lsfile.NewReplicaClient(filepath.Join(dir, "gen"))
	copyFile(t, src, gen, snap)
	anchorL0(t, src, snap, gen)
	out := filepath.Join(dir, "out.db")
	if err := restore(t, gen, out); err != nil {
		t.Fatal(err)
	}
	if n := count(t, out); n != 5 {
		t.Fatalf("restored %d rows, want 5", n)
	}
	meta := filepath.Join(dir, "meta")
	db := openDB(t, out, meta, gen)
	write(t, out, 2)
	sync(t, db)
	closeDB(t, db)
	l0 := files(t, gen, 0)
	t.Logf("FACT R5-1c: with a level-0 anchor [%d,%d] beside the snapshot, level 0 after two writes and one sync is %s", snap.MaxTXID, snap.MaxTXID, ranges(l0))
	// exactly: the anchor, then the next TXID N+1 and nothing else (R6-2)
	n := snap.MaxTXID
	mustRanges(t, gen, 0, fmt.Sprintf("[%d,%d] [%d,%d]", n, n, n+1, n+1))
	wipe(t, out, meta)
	out2 := filepath.Join(dir, "out2.db")
	if err := restore(t, gen, out2); err != nil {
		t.Fatal(err)
	}
	t.Logf("FACT R5-1d: a cold restore after that holds %d rows of 7", count(t, out2))
	mustCount(t, out2, 7)
}

// TestFact_R5_1_TXIDSidecarIsFollowOnly: a -txid sidecar beside the
// restored database does not set the replication position of an
// ordinary open.
func TestFact_R5_1_TXIDSidecarIsFollowOnly(t *testing.T) {
	src, snap := source(t, 5)
	dir := t.TempDir()
	gen := lsfile.NewReplicaClient(filepath.Join(dir, "gen"))
	copyFile(t, src, gen, snap)
	out := filepath.Join(dir, "out.db")
	if err := restore(t, gen, out); err != nil {
		t.Fatal(err)
	}
	if err := litestream.WriteTXIDFile(out, snap.MaxTXID); err != nil {
		t.Fatal(err)
	}
	db := openDB(t, out, filepath.Join(dir, "meta"), gen)
	write(t, out, 1)
	sync(t, db)
	closeDB(t, db)
	l0 := files(t, gen, 0)
	t.Logf("FACT R5-1e: with a -txid sidecar of %d, the first upload is level 0 %s", snap.MaxTXID, ranges(l0))
	if len(l0) == 0 || l0[0].MinTXID != 1 {
		t.Fatalf("the sidecar unexpectedly set the position: %s", ranges(l0))
	}
}

// TestFact_R5_1_L0ExpiryAndAFreshBox: a fresh box's position after
// litestream's own level-0 expiry. TXID-based retention keeps the newest
// level-0 file, and so does time-based retention (L0Retention, the store's
// monitor); a fresh box restored from the replica therefore continues at
// N+1 and a cold restore afterwards holds every row. Each of those is
// asserted exactly (R6-2): a library that removed the newest level-0 file,
// restarted numbering or lost the new writes fails here.
func TestFact_R5_1_L0ExpiryAndAFreshBox(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "src.db")
	write(t, path, 1)
	client := lsfile.NewReplicaClient(filepath.Join(dir, "replica"))
	db := openDB(t, path, filepath.Join(dir, "meta"), client)
	for i := 0; i < 3; i++ {
		write(t, path, 2)
		sync(t, db)
	}
	if _, err := db.Compact(context.Background(), 1); err != nil {
		t.Fatal(err)
	}
	snap, err := db.Snapshot(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	// TXID-based retention at the snapshot level
	if err := db.EnforceRetentionByTXID(context.Background(), 0, snap.MaxTXID+1); err != nil {
		t.Fatal(err)
	}
	n := snap.MaxTXID // the newest level-0 file's TXID, the position
	newest := fmt.Sprintf("[%d,%d]", n, n)
	byTXID := ranges(files(t, client, 0))
	if byTXID != newest {
		t.Fatalf("TXID-based retention at %d left level 0 %s, want exactly the newest file %s retained", n+1, byTXID, newest)
	}
	// time-based level-0 expiry, as the store's L0 retention monitor runs it
	db.L0Retention = time.Millisecond
	time.Sleep(20 * time.Millisecond)
	if err := db.EnforceL0RetentionByTime(context.Background()); err != nil {
		t.Fatal(err)
	}
	closeDB(t, db)
	t.Logf("FACT R5-1f: after TXID-based level-0 retention at %d the level-0 listing is %s; after time-based L0 expiry it is %s (L9 %s; L1 %s)", n+1, byTXID, ranges(files(t, client, 0)), ranges(files(t, client, 9)), ranges(files(t, client, 1)))
	mustRanges(t, client, 0, newest) // the newest level-0 file survives time-based expiry too, by identity
	out := filepath.Join(dir, "out.db")
	if err := restore(t, client, out); err != nil {
		t.Fatal(err)
	}
	mustCount(t, out, 7)
	meta := filepath.Join(dir, "meta2")
	db2 := openDB(t, out, meta, client)
	write(t, out, 2)
	sync(t, db2)
	closeDB(t, db2)
	t.Logf("FACT R5-1g: on a fresh box restored at TXID %d, the first upload is level 0 %s", n, ranges(files(t, client, 0)))
	mustRanges(t, client, 0, fmt.Sprintf("%s [%d,%d]", newest, n+1, n+1)) // exactly N+1: a restart at 1 fails here
	wipe(t, out, meta)
	out2 := filepath.Join(dir, "out2.db")
	if err := restore(t, client, out2); err != nil {
		t.Fatal(err)
	}
	t.Logf("FACT R5-1h: a cold restore after that holds %d rows of 9", count(t, out2))
	mustCount(t, out2, 9) // every row: a run that kept 7 of 9 fails here
}

// TestFact_R5_2_Checksums: what the pinned writers put in LTX headers and
// trailers: the level-0 writer, the snapshot writer and the compactor.
func TestFact_R5_2_Checksums(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "src.db")
	write(t, path, 1)
	client := lsfile.NewReplicaClient(filepath.Join(dir, "replica"))
	db := openDB(t, path, filepath.Join(dir, "meta"), client)
	for i := 0; i < 3; i++ {
		write(t, path, 2)
		sync(t, db)
	}
	snap, err := db.Snapshot(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	l1, err := db.Compact(context.Background(), 1)
	if err != nil {
		t.Fatal(err)
	}
	closeDB(t, db)
	check := func(name string, info *ltx.FileInfo) {
		hdr, tr := decode(t, client, info)
		t.Logf("FACT R5-2 %s [%d,%d]: HeaderFlagNoChecksum=%v PreApplyChecksum=%#x PostApplyChecksum=%#x FileChecksum!=0=%v (Verify passed: CRC-64 of the file)",
			name, info.MinTXID, info.MaxTXID, hdr.NoChecksum(), uint64(hdr.PreApplyChecksum), uint64(tr.PostApplyChecksum), tr.FileChecksum != 0)
		if !hdr.NoChecksum() || tr.PostApplyChecksum != 0 || tr.FileChecksum == 0 {
			t.Fatalf("%s: unexpected checksum fields", name)
		}
	}
	l0 := files(t, client, 0)
	check("level 0", l0[len(l0)-1])
	check("snapshot", snap)
	check("level 1 (compactor)", l1)
}

// TestFact_R5_3_SnapshotRetention: newer snapshots appear, and snapshot
// retention expires older snapshots and the lower-level ranges below the
// oldest remaining one. Afterwards no current file continues from the
// original snapshot's TXID, so a restore rooted there finds a gap, while
// litestream's own restore from the newest snapshot succeeds.
func TestFact_R5_3_SnapshotRetention(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "src.db")
	write(t, path, 1)
	client := lsfile.NewReplicaClient(filepath.Join(dir, "replica"))
	db := openDB(t, path, filepath.Join(dir, "meta"), client)
	sync(t, db)
	snap1, err := db.Snapshot(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	// keep the original snapshot's bytes: on S3 it would be a pinned,
	// noncurrent version; the file client deletes outright
	rc, err := client.OpenLTXFile(context.Background(), 9, snap1.MinTXID, snap1.MaxTXID, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	snap1Bytes, _ := io.ReadAll(rc)
	rc.Close()
	// two more rotations: writes, a level-1 compaction, a snapshot, as the
	// store's monitors would over days
	var snap2 *ltx.FileInfo
	for round := 0; round < 2; round++ {
		for i := 0; i < 3; i++ {
			write(t, path, 2)
			sync(t, db)
		}
		if _, err := db.Compact(context.Background(), 1); err != nil {
			t.Fatal(err)
		}
		time.Sleep(20 * time.Millisecond)
		if snap2, err = db.Snapshot(context.Background()); err != nil {
			t.Fatal(err)
		}
	}
	before := fmt.Sprintf("L9 %s; L1 %s; L0 %s", ranges(files(t, client, 9)), ranges(files(t, client, 1)), ranges(files(t, client, 0)))
	// snapshot retention keeps the newest snapshot; minSnapshotTXID is the
	// MaxTXID of the snapshot before the oldest kept one, and the store
	// then expires levels 1 and up below it (level 0 expires by time,
	// L0Retention, 5 minutes by default: here it is enforced by TXID too)
	minSnap, err := db.EnforceSnapshotRetention(context.Background(), time.Now().Add(time.Second))
	if err != nil {
		t.Fatal(err)
	}
	if err := db.EnforceRetentionByTXID(context.Background(), 1, minSnap); err != nil {
		t.Fatal(err)
	}
	if err := db.EnforceRetentionByTXID(context.Background(), 0, minSnap); err != nil {
		t.Fatal(err)
	}
	after := fmt.Sprintf("L9 %s; L1 %s; L0 %s", ranges(files(t, client, 9)), ranges(files(t, client, 1)), ranges(files(t, client, 0)))
	t.Logf("FACT R5-3a: before retention: %s", before)
	t.Logf("FACT R5-3b: after snapshot retention (minSnapshotTXID %d): %s", minSnap, after)
	if len(files(t, client, 9)) != 1 || files(t, client, 9)[0].MaxTXID != snap2.MaxTXID {
		t.Fatalf("expected only the newest snapshot to remain")
	}
	continuation := func() bool {
		// any current file with MinTXID <= snap1.MaxTXID+1 and MaxTXID > snap1.MaxTXID
		for _, lvl := range []int{0, 1, 2, 3} {
			for _, f := range files(t, client, lvl) {
				if f.MinTXID <= snap1.MaxTXID+1 && f.MaxTXID > snap1.MaxTXID {
					return true
				}
			}
		}
		return false
	}
	t.Logf("FACT R5-3c: after one retention pass a continuation from the original snapshot [1,%d] remains: %v (the level-1 file that starts at 1 ends at minSnapshotTXID and deletion is strictly below)", snap1.MaxTXID, continuation())
	if !continuation() {
		t.Fatalf("expected the level-1 file starting at 1 to survive the first pass (it ends at minSnapshotTXID %d)", minSnap)
	}
	// one more rotation: the level-1 file starting at 1 now ends below the
	// new minSnapshotTXID and is expired
	for i := 0; i < 3; i++ {
		write(t, path, 2)
		sync(t, db)
	}
	if _, err := db.Compact(context.Background(), 1); err != nil {
		t.Fatal(err)
	}
	time.Sleep(20 * time.Millisecond)
	if snap2, err = db.Snapshot(context.Background()); err != nil {
		t.Fatal(err)
	}
	if minSnap, err = db.EnforceSnapshotRetention(context.Background(), time.Now().Add(time.Second)); err != nil {
		t.Fatal(err)
	}
	for _, lvl := range []int{1, 0} {
		if err := db.EnforceRetentionByTXID(context.Background(), lvl, minSnap); err != nil {
			t.Fatal(err)
		}
	}
	closeDB(t, db)
	after2 := fmt.Sprintf("L9 %s; L1 %s; L0 %s", ranges(files(t, client, 9)), ranges(files(t, client, 1)), ranges(files(t, client, 0)))
	cont := continuation()
	t.Logf("FACT R5-3f: after a second rotation and retention (minSnapshotTXID %d): %s; a continuation from the original snapshot remains: %v", minSnap, after2, cont)
	if cont {
		t.Fatalf("expected every continuation from the original snapshot [1,%d] to be expired after the second rotation", snap1.MaxTXID)
	}
	mustRanges(t, client, 9, fmt.Sprintf("[1,%d]", snap2.MaxTXID)) // only the newest snapshot
	// restore rooted at the original snapshot: the original's bytes plus the
	// current lower levels, without the newer snapshot
	rooted := lsfile.NewReplicaClient(filepath.Join(dir, "rooted"))
	if _, err := rooted.WriteLTXFile(context.Background(), 9, snap1.MinTXID, snap1.MaxTXID, bytes.NewReader(snap1Bytes)); err != nil {
		t.Fatal(err)
	}
	for _, lvl := range []int{0, 1, 2, 3} {
		for _, f := range files(t, client, lvl) {
			copyFile(t, client, rooted, f)
		}
	}
	errRooted := restore(t, rooted, filepath.Join(dir, "rooted.db"))
	t.Logf("FACT R5-3d: restore rooted at the original snapshot after retention: %v", errRooted)
	errNewest := restore(t, client, filepath.Join(dir, "newest.db"))
	if errNewest != nil {
		t.Fatal(errNewest)
	}
	t.Logf("FACT R5-3e: restore from the newest snapshot: ok (rows %d)", count(t, filepath.Join(dir, "newest.db")))
	mustCount(t, filepath.Join(dir, "newest.db"), 19) // 1 row, then 3 rounds of 3 syncs of 2 rows
	// the rooted restore fails closed, with litestream's own error (R6-2)
	want := fmt.Sprintf("non-contiguous ltx files: have up to %016x but next file starts at %016x", uint64(snap1.MaxTXID), 5)
	if errRooted == nil || !strings.Contains(errRooted.Error(), want) {
		t.Fatalf("expected the rooted restore to fail closed with %q, got %v", want, errRooted)
	}
}

// TestFact_R6_5_AnchorReplayBytes: the anchor's bytes depend on the header
// timestamp. Encoded twice with the same inputs the bytes are identical;
// with another timestamp they differ, so a builder that replays the anchor
// PUT (a restarted install) must persist the timestamp it used, or reuse
// the object it already wrote (moves spec §5.1 step 2, §7.6 building).
func TestFact_R6_5_AnchorReplayBytes(t *testing.T) {
	src, snap := source(t, 5)
	ts := time.Now().UnixMilli()
	a := encodeAnchor(t, src, snap, ts)
	b := encodeAnchor(t, src, snap, ts)
	c := encodeAnchor(t, src, snap, ts+1)
	t.Logf("FACT R6-5: anchor [%d,%d], %d bytes; same timestamp: identical=%v; timestamp+1 ms: identical=%v", snap.MaxTXID, snap.MaxTXID, len(a), bytes.Equal(a, b), bytes.Equal(a, c))
	if !bytes.Equal(a, b) {
		t.Fatalf("the same inputs produced different anchor bytes")
	}
	if bytes.Equal(a, c) {
		t.Fatalf("a different timestamp produced identical bytes: the premise of persisting it is gone, review the contract")
	}
	// the replayed object is accepted by the decoder and equals the first
	dec := ltx.NewDecoder(bytes.NewReader(b))
	if err := dec.Verify(); err != nil {
		t.Fatal(err)
	}
}

// TestGate_AnchoredLineageThroughStore: one anchored lineage through the
// production Store configuration (internal/lease Held.Replicate: a DB with
// its own metadata directory, a replica client, litestream.NewStore with
// the compaction ladder, the L0 retention monitor), with every interval
// shortened from hours to fractions of a second so the monitors run inside
// the test: repeated writes with distinct bodies, updates and deletes,
// level-1 to level-3 compaction, snapshot rotation and retention,
// time-based level-0 expiry, then disk loss and a cold restore, three
// times over. Exact positions (every sync advances the position by one,
// and a reopened box continues at exactly the newest remote level-0 TXID
// plus one) and exact row contents are asserted; each maintenance event
// is awaited as a condition, not assumed after a sleep; the last round
// forces a restore that consumes compacted continuation files (R7-5).
func TestGate_AnchoredLineageThroughStore(t *testing.T) {
	src, snap := source(t, 5)
	dir := t.TempDir()
	gen := lsfile.NewReplicaClient(filepath.Join(dir, "gen"))
	copyFile(t, src, gen, snap)
	anchorL0(t, src, snap, gen)
	path := filepath.Join(dir, "db")
	if err := restore(t, gen, path); err != nil {
		t.Fatal(err)
	}
	want := rows(t, path) // the source's five rows, distinct bodies
	if len(want) != 5 {
		t.Fatalf("restored %d rows, want 5", len(want))
	}
	pos := int(snap.MaxTXID)
	next := 6 // the next row id to insert
	// internal/lease's ladder is L1 1h, L2 6h, L3 24h, the snapshot interval
	// 24h, L0 retention 5 min checked hourly, the DB monitor 10 s: here each
	// is shortened by the same shape
	levels := litestream.CompactionLevels{
		{Level: 0, Interval: 0},
		{Level: 1, Interval: 150 * time.Millisecond},
		{Level: 2, Interval: 400 * time.Millisecond},
		{Level: 3, Interval: 900 * time.Millisecond},
	}
	ctx := context.Background()
	maxOf := func(level int) int {
		fs := files(t, gen, level)
		if len(fs) == 0 {
			return 0
		}
		return int(fs[len(fs)-1].MaxTXID)
	}
	for round := 0; round < 3; round++ {
		meta := filepath.Join(dir, fmt.Sprintf("meta%d", round))
		db := litestream.NewDB(path)
		db.SetMetaPath(meta)
		db.MonitorInterval = 50 * time.Millisecond
		db.Replica = litestream.NewReplicaWithClient(db, gen)
		db.Replica.SyncInterval = 50 * time.Millisecond
		store := litestream.NewStore([]*litestream.DB{db}, levels)
		store.SnapshotInterval = 300 * time.Millisecond
		store.SnapshotRetention = 400 * time.Millisecond
		store.SetL0Retention(200 * time.Millisecond) // NewStore copied the default into the DB; SetL0Retention propagates
		store.L0RetentionCheckInterval = 50 * time.Millisecond
		if err := store.Open(ctx); err != nil {
			t.Fatalf("round %d: open store: %v", round, err)
		}
		snapsBefore := maxOf(9)
		l0Before := len(files(t, gen, 0))
		// eight transactions: inserts of distinct bodies, an update and a
		// delete among them, each advancing the position by exactly one
		for i := 0; i < 8; i++ {
			switch {
			case i == 3:
				id := next - 2
				exec(t, path, `UPDATE t SET body = ? WHERE id = ?`, fmt.Sprintf("r%d-updated-in-round-%d", id, round), id)
				want[id] = fmt.Sprintf("r%d-updated-in-round-%d", id, round)
			case i == 6:
				id := next - 4
				exec(t, path, `DELETE FROM t WHERE id = ?`, id)
				delete(want, id)
			default:
				exec(t, path, `INSERT INTO t (id, body) VALUES (?, ?)`, next, fmt.Sprintf("r%d-round-%d", next, round))
				want[next] = fmt.Sprintf("r%d-round-%d", next, round)
				next++
			}
			sync(t, db)
			pos++
			p, err := db.Pos()
			if err != nil {
				t.Fatal(err)
			}
			if int(p.TXID) != pos {
				t.Fatalf("round %d write %d: position %d, want exactly %d (a reopened box continues at the newest remote level-0 TXID plus one)", round, i, p.TXID, pos)
			}
			time.Sleep(120 * time.Millisecond) // the monitors run between writes
		}
		// the maintenance events, each as a condition: a snapshot rotated
		// past the previous newest, level 1, 2 and 3 files exist, the older
		// snapshot was removed by retention, and level-0 expiry removed
		// compacted files while keeping the newest
		waitFor(t, "a snapshot rotation", 5*time.Second, func() bool { return maxOf(9) > snapsBefore })
		for _, lvl := range []int{1, 2, 3} {
			waitFor(t, fmt.Sprintf("a level-%d compaction", lvl), 5*time.Second, func() bool { return len(files(t, gen, lvl)) > 0 })
		}
		// Retention is awaited at the state that lasts: one snapshot, at the
		// round's final position. One snapshot alone can be a moment in
		// passing. The monitor snapshots every interval and keeps the previous
		// snapshot while it is younger than the retention, so a tick that runs
		// 100 ms late removes the previous one and leaves a single snapshot
		// that is behind the position; the next tick adds the final snapshot
		// beside it, and a round that had stopped waiting closed the store
		// with two (#266). Once the only snapshot is at the final position a
		// later tick can only rewrite that same file.
		waitFor(t, "snapshot retention", 5*time.Second, func() bool {
			l9 := files(t, gen, 9)
			return len(l9) == 1 && int(l9[0].MaxTXID) == pos
		})
		waitFor(t, "level-0 expiry", 5*time.Second, func() bool { return len(files(t, gen, 0)) < l0Before+8 })
		if err := store.Close(ctx); err != nil {
			t.Fatal(err)
		}
		l0, l1, l9 := files(t, gen, 0), files(t, gen, 1), files(t, gen, 9)
		t.Logf("GATE round %d: position %d, rows %d; L0 %s (%d files, %d before); L1 %s; L2 %s; L3 %s; L9 %s", round, pos, len(want), ranges(l0), len(l0), l0Before, ranges(l1), ranges(files(t, gen, 2)), ranges(files(t, gen, 3)), ranges(l9))
		if len(l0) == 0 || int(l0[len(l0)-1].MaxTXID) != pos {
			t.Fatalf("round %d: newest remote level-0 file is %s, want MaxTXID exactly %d", round, ranges(l0), pos)
		}
		if len(l9) != 1 || int(l9[0].MaxTXID) <= snapsBefore {
			t.Fatalf("round %d: snapshot retention did not leave exactly the newest snapshot: L9 %s", round, ranges(l9))
		}
		// disk loss, cold restore: exact contents, row by row
		wipe(t, path, meta)
		if err := restore(t, gen, path); err != nil {
			t.Fatalf("round %d: cold restore: %v", round, err)
		}
		mustRows(t, path, want)
	}
	// a restore that must consume compacted continuation files: with the
	// monitors stopped, more writes are synced, compacted into level 1, and
	// every level-0 file but the newest is expired by time; the newest
	// snapshot is behind them, so the planner has to take the level-1 file
	meta := filepath.Join(dir, "meta-tail")
	db := openDB(t, path, meta, gen)
	for i := 0; i < 3; i++ {
		exec(t, path, `INSERT INTO t (id, body) VALUES (?, ?)`, next, fmt.Sprintf("r%d-tail", next))
		want[next] = fmt.Sprintf("r%d-tail", next)
		next++
		sync(t, db)
		pos++
	}
	exec(t, path, `DELETE FROM t WHERE id = ?`, next-1)
	delete(want, next-1)
	sync(t, db)
	pos++
	if _, err := db.Compact(ctx, 1); err != nil {
		t.Fatal(err)
	}
	db.L0Retention = time.Millisecond
	time.Sleep(20 * time.Millisecond)
	if err := db.EnforceL0RetentionByTime(ctx); err != nil {
		t.Fatal(err)
	}
	closeDB(t, db)
	l0, l1, l9 := files(t, gen, 0), files(t, gen, 1), files(t, gen, 9)
	t.Logf("GATE tail: position %d; L0 %s; L1 %s; L9 %s", pos, ranges(l0), ranges(l1), ranges(l9))
	if len(l0) != 1 || int(l0[0].MaxTXID) != pos {
		t.Fatalf("tail: level 0 should hold the newest file alone, have %s", ranges(l0))
	}
	if maxOf(9) >= pos || maxOf(1) != pos {
		t.Fatalf("tail: the newest snapshot (%d) must be behind the level-1 file ending at %d", maxOf(9), maxOf(1))
	}
	wipe(t, path, meta)
	if err := restore(t, gen, path); err != nil {
		t.Fatalf("tail: cold restore: %v", err)
	}
	mustRows(t, path, want) // the rows past the snapshot came through level 1
}

// TestFact_R7_1_UnconditionalPut: the pinned S3 client's PutObject carries
// no If-Match or If-None-Match, and its key is prefix, level and TXID range
// only, so two writers of one generation write the same key with
// unconditional PUTs: the R7-1 hazard (claim rev 6.6 §6.4a). Read from
// the pinned source so an upgrade that changes either fails here.
func TestFact_R7_1_UnconditionalPut(t *testing.T) {
	pkg, err := build.Default.Import("github.com/benbjohnson/litestream/s3", "", build.FindOnly)
	if err != nil {
		t.Skipf("pinned s3 client source not found: %v", err)
	}
	src, err := os.ReadFile(filepath.Join(pkg.Dir, "replica_client.go"))
	if err != nil {
		t.Fatal(err)
	}
	text := string(src)
	i := strings.Index(text, "s3.PutObjectInput{")
	if i < 0 {
		t.Fatalf("no PutObjectInput in the pinned client")
	}
	block := text[i:]
	block = block[:strings.Index(block, "}")]
	t.Logf("FACT R7-1a: the pinned PutObjectInput fields: %s", strings.Join(strings.Fields(strings.ReplaceAll(block, "\n", " ")), " "))
	for _, cond := range []string{"IfNoneMatch", "IfMatch"} {
		if strings.Contains(block, cond) {
			t.Fatalf("the pinned client's PutObject is conditional (%s): the writer-epoch premise changed, review the contract", cond)
		}
	}
	if !strings.Contains(text, `fmt.Sprintf("%04x/%s", level, filename)`) || !strings.Contains(text, "ltx.FormatFilename(minTXID, maxTXID)") {
		t.Fatalf("the pinned client's key is no longer prefix/level/TXID range")
	}
	t.Logf("FACT R7-1b: the key is c.Path + \"/\" + %%04x level + ltx.FormatFilename(min, max): two writers of one generation share it")
}

// TestFact_R8_3_EmptyBootstrap: a created incarnation's first wake has
// neither a database file nor a replica (R8-3). The pinned encoder refuses
// the files a position-0 build would carry, a snapshot [1,0] and an anchor
// [0,0], so an empty generation has no LTX file at all: litestream opened
// on an absent path against an empty replica syncs nothing, a cold restore
// of the empty generation has nothing to restore, and the first
// transaction is the first upload, which a cold restore then holds.
func TestFact_R8_3_EmptyBootstrap(t *testing.T) {
	for _, h := range []ltx.Header{
		{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: 4096, Commit: 1, MinTXID: 1, MaxTXID: 0, Timestamp: 1, WALOffset: litestream.WALHeaderSize},
		{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: 4096, Commit: 1, MinTXID: 0, MaxTXID: 0, Timestamp: 1},
	} {
		var buf bytes.Buffer
		enc, err := ltx.NewEncoder(&buf)
		if err != nil {
			t.Fatal(err)
		}
		err = enc.EncodeHeader(h)
		t.Logf("FACT R8-3a: encoding an LTX header [%d,%d]: %v", h.MinTXID, h.MaxTXID, err)
		if err == nil {
			t.Fatalf("the encoder accepted a zero TXID [%d,%d]; the spec's empty-bootstrap rule rests on it refusing", h.MinTXID, h.MaxTXID)
		}
	}
	dir := t.TempDir()
	gen := lsfile.NewReplicaClient(filepath.Join(dir, "gen"))
	path := filepath.Join(dir, "new.db")
	meta := filepath.Join(dir, "meta")
	if _, err := os.Stat(path); !os.IsNotExist(err) {
		t.Fatalf("the database path exists before the first wake: %v", err)
	}
	if err := restore(t, gen, filepath.Join(dir, "nothing.db")); err == nil {
		t.Fatalf("a restore from an empty replica succeeded; the empty generation must have nothing to restore")
	} else {
		t.Logf("FACT R8-3b: a cold restore of an empty generation: %v", err)
	}
	db := openDB(t, path, meta, gen)
	err := db.Sync(context.Background())
	t.Logf("FACT R8-3c: litestream opened on an absent database against an empty replica; first sync: %v; level 0 %s, level 9 %s", err, ranges(files(t, gen, 0)), ranges(files(t, gen, 9)))
	if err != nil {
		t.Fatalf("the first sync of an absent database failed: %v", err)
	}
	mustRanges(t, gen, 0, "none")
	mustRanges(t, gen, 9, "none")
	write(t, path, 1) // CREATE TABLE, then one row: two SQLite transactions, one sync
	sync(t, db)
	l0 := ranges(files(t, gen, 0))
	t.Logf("FACT R8-3d: after the first transactions and one sync the replica's level 0 is %s, level 9 %s (a TXID is one sync's WAL frames, not one SQLite transaction)", l0, ranges(files(t, gen, 9)))
	if l0 != "[1,1]" {
		t.Fatalf("expected the first upload to be level 0 [1,1], got %s", l0)
	}
	closeDB(t, db)
	out := filepath.Join(dir, "out.db")
	if err := restore(t, gen, out); err != nil {
		t.Fatalf("restore after the first transactions: %v", err)
	}
	mustCount(t, out, 1)
	t.Logf("FACT R8-3e: a cold restore after the first transactions holds the row; the first transaction was the first upload")
}