package moves

import (
	"bytes"
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/benbjohnson/litestream"
	lsfile "github.com/benbjohnson/litestream/file"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/lease/leasetest"
	"github.com/superfly/ltx"
	_ "modernc.org/sqlite"
)

// The tests here run real litestream v0.5.14 and ltx v0.5.1 over real
// SQLite files (controller ruling: real data, never a fake provider as
// proof). A source replica is produced the way internal/sim/litestreamfacts
// produces one: litestream's file replica client over a database written
// through modernc sqlite, every sync explicit.

const (
	testBucket = "data"
	testHost   = "h"
	testOp     = "op1"
	testLane   = "moves/h/op1/replica/"
)

func testGenPrefix() string { return GenerationPrefix("instances/h", testOp) }

// openLS opens path under litestream against client with no background
// monitors, as litestreamfacts does.
func openLS(t *testing.T, path, meta string, client litestream.ReplicaClient) *litestream.DB {
	t.Helper()
	db := litestream.NewDB(path)
	db.Logger = slog.New(slog.NewTextHandler(io.Discard, nil))
	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
}

// writeRows inserts n rows, each its own transaction.
func writeRows(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)
		}
	}
}

func countRows(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 pageCountPragma(t *testing.T, path string) int64 {
	t.Helper()
	d, err := sql.Open("sqlite", "file:"+path+"?mode=ro")
	if err != nil {
		t.Fatal(err)
	}
	defer d.Close()
	var n int64
	if err := d.QueryRow(`PRAGMA page_count`).Scan(&n); err != nil {
		t.Fatal(err)
	}
	return n
}

func fileSHA256(t *testing.T, path string) string {
	t.Helper()
	b, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	s := sha256.Sum256(b)
	return hex.EncodeToString(s[:])
}

// sourceReplica writes a database with rows rows in syncs separate
// syncs, takes a level-9 snapshot when snapshot is set, and returns the
// file replica client plus the newest level-0 position.
func sourceReplica(t *testing.T, syncs int, snapshot bool) (litestream.ReplicaClient, uint64) {
	t.Helper()
	dir := t.TempDir()
	path := filepath.Join(dir, "src.db")
	writeRows(t, path, 1)
	client := lsfile.NewReplicaClient(filepath.Join(dir, "replica"))
	db := openLS(t, path, filepath.Join(dir, "meta"), client)
	ctx := context.Background()
	for i := 0; i < syncs; i++ {
		if err := db.SyncAndWait(ctx); err != nil {
			t.Fatalf("sync: %v", err)
		}
		writeRows(t, path, 2)
	}
	if err := db.SyncAndWait(ctx); err != nil {
		t.Fatalf("sync: %v", err)
	}
	if snapshot {
		if _, err := db.Snapshot(ctx); err != nil {
			t.Fatal(err)
		}
	}
	if err := db.Close(ctx); err != nil {
		t.Fatal(err)
	}
	var n uint64
	for _, f := range levelFiles(t, client, 0) {
		if uint64(f.MaxTXID) > n {
			n = uint64(f.MaxTXID)
		}
	}
	if n == 0 {
		t.Fatal("source replica has no level-0 files")
	}
	return client, n
}

func levelFiles(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 readLTX(t *testing.T, client litestream.ReplicaClient, info *ltx.FileInfo) []byte {
	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)
	}
	return b
}

// sealReplica copies every LTX file of client into the bucket under lane,
// as the staging lane holds them, and returns the pinned version set.
func sealReplica(t *testing.T, b *leasetest.VersionedBucket, client litestream.ReplicaClient, lane string) []SealedObject {
	t.Helper()
	var out []SealedObject
	for level := 0; level <= litestream.SnapshotLevel; level++ {
		for _, info := range levelFiles(t, client, level) {
			body := readLTX(t, client, info)
			key := lane + fmt.Sprintf("%04x/", level) + ltx.FormatFilename(info.MinTXID, info.MaxTXID)
			v := b.Put(testBucket, key, body)
			out = append(out, SealedObject{Key: key, VersionID: v, Size: int64(len(body))})
		}
	}
	return out
}

func newBucket() *leasetest.VersionedBucket {
	b := leasetest.NewVersionedBucket()
	b.AddBucket(testBucket)
	return b
}

func versionsOf(t *testing.T, b *leasetest.VersionedBucket, key string) []lease.ObjectVersion {
	t.Helper()
	all, err := b.ListVersions(context.Background(), testBucket, key)
	if err != nil {
		t.Fatal(err)
	}
	var out []lease.ObjectVersion
	for _, v := range all {
		if v.Key == key {
			out = append(out, v)
		}
	}
	return out
}

func decodeLTX(t *testing.T, data []byte) (ltx.Header, ltx.Trailer, int) {
	t.Helper()
	dec := ltx.NewDecoder(bytes.NewReader(data))
	if err := dec.DecodeHeader(); err != nil {
		t.Fatal(err)
	}
	page := make([]byte, dec.Header().PageSize)
	pages := 0
	for {
		var ph ltx.PageHeader
		if err := dec.DecodePage(&ph, page); err == io.EOF {
			break
		} else if err != nil {
			t.Fatal(err)
		}
		pages++
	}
	if err := dec.Close(); err != nil {
		t.Fatal(err)
	}
	if err := ltx.NewDecoder(bytes.NewReader(data)).Verify(); err != nil {
		t.Fatalf("verify: %v", err)
	}
	return dec.Header(), dec.Trailer(), pages
}

// plannedFiles is how many files litestream's planner reads to reach n
// over the pinned set.
func plannedFiles(t *testing.T, b *leasetest.VersionedBucket, src []SealedObject, n uint64) int {
	t.Helper()
	c, err := NewSealedClient(b, testBucket, src)
	if err != nil {
		t.Fatal(err)
	}
	plan, err := litestream.CalcRestorePlan(context.Background(), c, ltx.TXID(n), time.Time{}, slog.New(slog.NewTextHandler(io.Discard, nil)))
	if err != nil {
		t.Fatal(err)
	}
	return len(plan)
}

func sameManifest(a, b InstallManifest) bool {
	a.BuiltAt, b.BuiltAt = time.Time{}, time.Time{}
	return a == b
}

var testClock = time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC)

func inputs(b *leasetest.VersionedBucket, src []SealedObject, n uint64, work string) BuildInputs {
	return inputsAt(b, src, n, work, testClock)
}

// inputsAt is inputs with the build's clock at now: a replay runs later
// than the attempt it replays.
func inputsAt(store lease.VersionedStore, src []SealedObject, n uint64, work string, now time.Time) BuildInputs {
	return BuildInputs{
		Store: store, Bucket: testBucket, Source: src, N: n,
		GenerationPrefix: testGenPrefix(), Op: testOp, Host: testHost,
		WorkDir: work,
		Now:     func() time.Time { return now },
	}
}

// versionRewriter is a VersionedStore whose answers name a version id the
// builder must refuse: what a bucket whose versioning was suspended after
// the seal returns for a new write ("null"), or a provider that returns
// none (""). putAs rewrites the id a fresh PutIfAbsent returns for a key;
// listAs rewrites the current version a replay finds for a key.
type versionRewriter struct {
	lease.VersionedStore
	putAs  map[string]string
	listAs map[string]string
}

func (r *versionRewriter) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) {
	v, err := r.VersionedStore.PutIfAbsent(ctx, bucket, key, body)
	if as, ok := r.putAs[key]; ok && err == nil {
		return as, nil
	}
	return v, err
}

func (r *versionRewriter) ListVersions(ctx context.Context, bucket, prefix string) ([]lease.ObjectVersion, error) {
	vs, err := r.VersionedStore.ListVersions(ctx, bucket, prefix)
	for i := range vs {
		if as, ok := r.listAs[vs[i].Key]; ok && vs[i].IsLatest {
			vs[i].VersionID = as
		}
	}
	return vs, err
}

// restoreGeneration restores the built generation through litestream's own
// Restore, reading exactly the objects the manifest names.
func restoreGeneration(t *testing.T, b *leasetest.VersionedBucket, m InstallManifest, out string) {
	t.Helper()
	snapBody, err := b.GetVersion(context.Background(), testBucket, m.SnapshotKey, m.SnapshotVersion)
	if err != nil {
		t.Fatal(err)
	}
	anchorBody, err := b.GetVersion(context.Background(), testBucket, m.AnchorKey, m.AnchorVersion)
	if err != nil {
		t.Fatal(err)
	}
	gen, err := NewSealedClient(b, testBucket, []SealedObject{
		{Key: m.SnapshotKey, VersionID: m.SnapshotVersion, Size: int64(len(snapBody))},
		{Key: m.AnchorKey, VersionID: m.AnchorVersion, Size: int64(len(anchorBody))},
	})
	if err != nil {
		t.Fatal(err)
	}
	r := litestream.NewReplicaWithClient(nil, gen)
	opt := litestream.NewRestoreOptions()
	opt.OutputPath = out
	if err := r.Restore(context.Background(), opt); err != nil {
		t.Fatalf("restore the generation: %v", err)
	}
}

func TestGenerationPrefixAndKeys(t *testing.T) {
	if got := GenerationPrefix("instances/h", "op1"); got != "instances/h/g-op1/" {
		t.Fatalf("GenerationPrefix = %q", got)
	}
	if got := GenerationPrefix("instances/h/", "op1"); got != "instances/h/g-op1/" {
		t.Fatalf("GenerationPrefix with a slash = %q", got)
	}
	if got := GenerationPrefix("instances/h/inc2", "op1"); got != "instances/h/inc2/g-op1/" {
		t.Fatalf("GenerationPrefix with an incarnation = %q", got)
	}
	// The files twin (spec §7.6, merge round 6): every reader and writer of
	// a generation's files names them by this rule.
	for _, loc := range []string{"instances/h/inc2", "instances/h/inc2/"} {
		if got := GenerationFilesPrefix(loc, "op1"); got != "instances/h/inc2/files/g-op1/" || !strings.HasPrefix(got, FilesGenerationBase(loc)) {
			t.Fatalf("GenerationFilesPrefix(%q) = %q", loc, got)
		}
	}
	p := "instances/h/g-op1/"
	if got := SnapshotKey(p, 7); got != p+"0009/0000000000000001-0000000000000007.ltx" {
		t.Fatalf("SnapshotKey = %q", got)
	}
	if got := AnchorKey(p, 7); got != p+"0000/0000000000000007-0000000000000007.ltx" {
		t.Fatalf("AnchorKey = %q", got)
	}
	if got := BuildIntentKey(p); got != p+"build-intent.json" {
		t.Fatalf("BuildIntentKey = %q", got)
	}
	if got := InstallManifestKey(p); got != p+"install.json" {
		t.Fatalf("InstallManifestKey = %q", got)
	}
}

func TestSealedClientServesOnlySealedVersions(t *testing.T) {
	b := newBucket()
	ctx := context.Background()
	key := testLane + "0000/" + ltx.FormatFilename(1, 1)
	// At least an LTX header's worth of bytes: the client refuses a
	// smaller object, as litestream's restore would.
	sealed := []byte(strings.Repeat("sealed bytes of 0000000000000001-0000000000000001 ", 3))
	v1 := b.Put(testBucket, key, sealed)
	b.Put(testBucket, key, []byte("a newer version that landed after the seal"))
	// An unsealed sibling exists in the lane but is not in the pinned set.
	b.Put(testBucket, testLane+"0000/"+ltx.FormatFilename(2, 2), []byte("unsealed"))

	c, err := NewSealedClient(b, testBucket, []SealedObject{{Key: key, VersionID: v1, Size: int64(len(sealed))}})
	if err != nil {
		t.Fatal(err)
	}
	if c.Type() != "sealed" {
		t.Fatalf("Type = %q", c.Type())
	}
	if err := c.Init(ctx); err != nil {
		t.Fatal(err)
	}

	rc, err := c.OpenLTXFile(ctx, 0, 1, 1, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	got, _ := io.ReadAll(rc)
	rc.Close()
	if !bytes.Equal(got, sealed) {
		t.Fatalf("OpenLTXFile served %q, want the sealed version", got)
	}
	// A ranged open, as litestream's resumable reader asks for it.
	rc, err = c.OpenLTXFile(ctx, 0, 1, 1, 7, 5)
	if err != nil {
		t.Fatal(err)
	}
	got, _ = io.ReadAll(rc)
	rc.Close()
	if string(got) != string(sealed[7:12]) {
		t.Fatalf("ranged open = %q, want %q", got, sealed[7:12])
	}
	rc, err = c.OpenLTXFile(ctx, 0, 1, 1, 7, 0)
	if err != nil {
		t.Fatal(err)
	}
	got, _ = io.ReadAll(rc)
	rc.Close()
	if string(got) != string(sealed[7:]) {
		t.Fatalf("open to the end = %q, want %q", got, sealed[7:])
	}

	if _, err := c.OpenLTXFile(ctx, 0, 2, 2, 0, 0); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("an unsealed key opened: %v", err)
	}
	infos := levelFiles(t, c, 0)
	if len(infos) != 1 || infos[0].MinTXID != 1 || infos[0].MaxTXID != 1 || infos[0].Size != int64(len(sealed)) || infos[0].Level != 0 {
		t.Fatalf("LTXFiles(0) = %+v", infos)
	}
	if infos := levelFiles(t, c, 9); len(infos) != 0 {
		t.Fatalf("LTXFiles(9) = %+v, want none", infos)
	}
	// seek skips files below it, as the file client does.
	itr, err := c.LTXFiles(ctx, 0, 2, false)
	if err != nil {
		t.Fatal(err)
	}
	if itr.Next() {
		t.Fatal("seek past the only file still yielded it")
	}

	if _, err := c.WriteLTXFile(ctx, 0, 3, 3, bytes.NewReader(nil)); !errors.Is(err, ErrSealedReadOnly) {
		t.Fatalf("WriteLTXFile = %v", err)
	}
	if err := c.DeleteLTXFiles(ctx, infos); !errors.Is(err, ErrSealedReadOnly) {
		t.Fatalf("DeleteLTXFiles = %v", err)
	}
	if err := c.DeleteAll(ctx); !errors.Is(err, ErrSealedReadOnly) {
		t.Fatalf("DeleteAll = %v", err)
	}
	if n := len(versionsOf(t, b, key)); n != 2 {
		t.Fatalf("the sealed key has %d versions after the refused writes, want 2", n)
	}

	// Each refusal below isolates one property: every other field of the
	// fixture is valid (a usable version, an LTX key, a header-sized object).
	okSize := int64(len(sealed))
	// G27: a version id that pins nothing is refused before any read.
	for _, bad := range []string{"", "null"} {
		if _, err := NewSealedClient(b, testBucket, []SealedObject{{Key: key, VersionID: bad, Size: okSize}}); err == nil {
			t.Fatalf("version %q accepted", bad)
		}
	}
	// A key that is not <level>/<min>-<max>.ltx has no position.
	if _, err := NewSealedClient(b, testBucket, []SealedObject{{Key: testLane + "lock.json", VersionID: v1, Size: okSize}}); err == nil {
		t.Fatal("a non-LTX key accepted")
	}
	// Two objects at one position would make the plan ambiguous.
	if _, err := NewSealedClient(b, testBucket, []SealedObject{
		{Key: key, VersionID: v1, Size: okSize},
		{Key: "other/" + "0000/" + ltx.FormatFilename(1, 1), VersionID: v1, Size: okSize},
	}); err == nil {
		t.Fatal("two objects at one position accepted")
	}
	// A size below an LTX header is refused on its own.
	if _, err := NewSealedClient(b, testBucket, []SealedObject{{Key: key, VersionID: v1, Size: ltx.HeaderSize - 1}}); err == nil {
		t.Fatal("an object smaller than an LTX header accepted")
	}
	if _, err := NewSealedClient(b, testBucket, []SealedObject{{Key: key, VersionID: v1, Size: ltx.HeaderSize}}); err != nil {
		t.Fatalf("a header-sized object refused: %v", err)
	}
}

// Every artifact identity the build records, and every current version a
// replay chooses, must pin an immutable object: "" and "null" fail the
// build before they are read or recorded, and no install.json is written.
func TestBuildRefusesUnusableArtifactVersions(t *testing.T) {
	src, n := sourceReplica(t, 1, true)
	prefix := testGenPrefix()
	ctx := context.Background()
	keys := map[string]string{
		"intent":   BuildIntentKey(prefix),
		"snapshot": SnapshotKey(prefix, n),
		"anchor":   AnchorKey(prefix, n),
		"manifest": InstallManifestKey(prefix),
	}
	for name, key := range keys {
		for _, bad := range []string{"", "null"} {
			t.Run("created "+name+" as "+fmt.Sprintf("%q", bad), func(t *testing.T) {
				b := newBucket()
				seal := sealReplica(t, b, src, testLane)
				store := &versionRewriter{VersionedStore: b, putAs: map[string]string{key: bad}}
				_, err := BuildGeneration(ctx, inputsAt(store, seal, n, filepath.Join(t.TempDir(), "build"), testClock))
				if !errors.Is(err, ErrBuildUnpinned) {
					t.Fatalf("err = %v, want ErrBuildUnpinned", err)
				}
				if _, ok := b.Current(testBucket, InstallManifestKey(prefix)); ok && name != "manifest" {
					t.Fatal("install.json was written")
				}
			})
		}
	}
	// On replay: the attempt crashed after the snapshot PUT; the replay's
	// listing names an unusable current version for the intent or the
	// snapshot, as a bucket whose versioning was suspended would.
	for _, name := range []string{"intent", "snapshot"} {
		for _, bad := range []string{"", "null"} {
			t.Run("replayed "+name+" as "+fmt.Sprintf("%q", bad), func(t *testing.T) {
				b := newBucket()
				seal := sealReplica(t, b, src, testLane)
				work := filepath.Join(t.TempDir(), "build")
				boom := errors.New("the box died")
				b.FailOn = func(op, key string) error {
					if op == "put" && key == keys["anchor"] {
						return boom
					}
					return nil
				}
				if _, err := BuildGeneration(ctx, inputs(b, seal, n, work)); !errors.Is(err, boom) {
					t.Fatalf("first run = %v, want the injected failure", err)
				}
				b.FailOn = nil
				store := &versionRewriter{VersionedStore: b, listAs: map[string]string{keys[name]: bad}}
				_, err := BuildGeneration(ctx, inputsAt(store, seal, n, work, testClock.Add(time.Hour)))
				if !errors.Is(err, ErrBuildUnpinned) {
					t.Fatalf("replay = %v, want ErrBuildUnpinned", err)
				}
				if _, ok := b.Current(testBucket, InstallManifestKey(prefix)); ok {
					t.Fatal("install.json was written on the replay")
				}
				if vs := versionsOf(t, b, keys["snapshot"]); len(vs) != 1 {
					t.Fatalf("the snapshot has %d versions, want the attempt's one", len(vs))
				}
			})
		}
	}
}

func TestBuildGenerationReachesNAndWritesSnapshotAndAnchor(t *testing.T) {
	src, n := sourceReplica(t, 3, true)
	b := newBucket()
	seal := sealReplica(t, b, src, testLane)
	work := filepath.Join(t.TempDir(), "build")
	ctx := context.Background()

	m, err := BuildGeneration(ctx, inputs(b, seal, n, work))
	if err != nil {
		t.Fatalf("BuildGeneration: %v", err)
	}
	prefix := testGenPrefix()
	if m.V != 1 || m.Op != testOp || m.Host != testHost || m.N != n {
		t.Fatalf("manifest identity = %+v", m)
	}
	if m.SnapshotKey != SnapshotKey(prefix, n) || m.AnchorKey != AnchorKey(prefix, n) {
		t.Fatalf("manifest keys = %q %q", m.SnapshotKey, m.AnchorKey)
	}
	if !lease.UsableVersionID(m.SnapshotVersion) || !lease.UsableVersionID(m.AnchorVersion) {
		t.Fatalf("manifest versions = %q %q", m.SnapshotVersion, m.AnchorVersion)
	}
	if m.IntegrityCheck != "ok" {
		t.Fatalf("IntegrityCheck = %q", m.IntegrityCheck)
	}
	dbPath := filepath.Join(work, testHost+".db")
	if got := fileSHA256(t, dbPath); got != m.DBSHA256 {
		t.Fatalf("DBSHA256 = %s, file is %s", m.DBSHA256, got)
	}
	if !m.BuiltAt.Equal(time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC)) {
		t.Fatalf("BuiltAt = %v", m.BuiltAt)
	}

	snapBody, err := b.GetVersion(ctx, testBucket, m.SnapshotKey, m.SnapshotVersion)
	if err != nil {
		t.Fatal(err)
	}
	sh, st, _ := decodeLTX(t, snapBody)
	if sh.MinTXID != 1 || sh.MaxTXID != ltx.TXID(n) || sh.Flags != ltx.HeaderFlagNoChecksum || sh.WALOffset != 0 {
		t.Fatalf("snapshot header = %+v", sh)
	}
	if uint64(st.FileChecksum) != m.SnapshotCRC64 {
		t.Fatalf("SnapshotCRC64 = %x, trailer %x", m.SnapshotCRC64, uint64(st.FileChecksum))
	}
	anchorBody, err := b.GetVersion(ctx, testBucket, m.AnchorKey, m.AnchorVersion)
	if err != nil {
		t.Fatal(err)
	}
	ah, at, _ := decodeLTX(t, anchorBody)
	if ah.MinTXID != ltx.TXID(n) || ah.MaxTXID != ltx.TXID(n) || ah.Flags != ltx.HeaderFlagNoChecksum || ah.WALOffset != litestream.WALHeaderSize {
		t.Fatalf("anchor header = %+v", ah)
	}
	if ah.PageSize != sh.PageSize || ah.Commit != sh.Commit || ah.Timestamp != sh.Timestamp {
		t.Fatalf("anchor and snapshot disagree: %+v vs %+v", ah, sh)
	}
	if uint64(at.FileChecksum) != m.AnchorCRC64 {
		t.Fatalf("AnchorCRC64 = %x, trailer %x", m.AnchorCRC64, uint64(at.FileChecksum))
	}
	ps, pc, err := ReadDBHeader(dbPath)
	if err != nil {
		t.Fatal(err)
	}
	if sh.PageSize != ps || sh.Commit != pc {
		t.Fatalf("snapshot header %d/%d, database %d/%d", sh.PageSize, sh.Commit, ps, pc)
	}

	// The intent was written and names the encoding inputs.
	intentBody, ok := b.Current(testBucket, BuildIntentKey(prefix))
	if !ok {
		t.Fatal("no build intent")
	}
	var intent BuildIntent
	if err := json.Unmarshal(intentBody, &intent); err != nil {
		t.Fatal(err)
	}
	if intent.V != 1 || intent.Op != testOp || intent.Host != testHost || intent.N != n || intent.PageSize != ps || intent.Commit != pc || intent.Timestamp != sh.Timestamp {
		t.Fatalf("intent = %+v", intent)
	}

	// The manifest in the bucket is the one returned.
	manifestBody, ok := b.Current(testBucket, InstallManifestKey(prefix))
	if !ok {
		t.Fatal("no install manifest")
	}
	var stored InstallManifest
	if err := json.Unmarshal(manifestBody, &stored); err != nil {
		t.Fatal(err)
	}
	if !sameManifest(stored, m) || !stored.BuiltAt.Equal(m.BuiltAt) {
		t.Fatalf("stored manifest %+v, returned %+v", stored, m)
	}

	// Nothing but the four objects was written to the generation.
	gen, err := b.ListVersions(ctx, testBucket, prefix)
	if err != nil {
		t.Fatal(err)
	}
	var keys []string
	for _, v := range gen {
		keys = append(keys, v.Key)
	}
	sort.Strings(keys)
	want := []string{AnchorKey(prefix, n), SnapshotKey(prefix, n), BuildIntentKey(prefix), InstallManifestKey(prefix)}
	sort.Strings(want)
	if strings.Join(keys, ",") != strings.Join(want, ",") {
		t.Fatalf("generation holds %v, want %v", keys, want)
	}

	// Litestream's own restore of the generation yields the same database.
	out := filepath.Join(t.TempDir(), "restored.db")
	restoreGeneration(t, b, m, out)
	if got := fileSHA256(t, out); got != m.DBSHA256 {
		t.Fatalf("the generation restores to %s, the build's file is %s", got, m.DBSHA256)
	}
	if err := lease.CheckDB(out); err != nil {
		t.Fatalf("CheckDB on the restored generation: %v", err)
	}
	if got, want := countRows(t, out), countRows(t, dbPath); got != want {
		t.Fatalf("restored %d rows, built %d", got, want)
	}
}

func TestInstallManifestKeys(t *testing.T) {
	src, n := sourceReplica(t, 1, true)
	b := newBucket()
	seal := sealReplica(t, b, src, testLane)
	if _, err := BuildGeneration(context.Background(), inputs(b, seal, n, filepath.Join(t.TempDir(), "build"))); err != nil {
		t.Fatal(err)
	}
	body, ok := b.Current(testBucket, InstallManifestKey(testGenPrefix()))
	if !ok {
		t.Fatal("no install manifest")
	}
	var got map[string]any
	if err := json.Unmarshal(body, &got); err != nil {
		t.Fatal(err)
	}
	var keys []string
	for k := range got {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	want := []string{"anchor_crc64", "anchor_key", "anchor_version", "built_at", "db_sha256", "host", "integrity_check", "op", "snapshot_crc64", "snapshot_key", "snapshot_txid", "snapshot_version", "v"}
	if strings.Join(keys, " ") != strings.Join(want, " ") {
		t.Fatalf("install.json keys = %v, want %v", keys, want)
	}
}

// Review Focus 3, witness wBuildCrashes: a build that dies after the
// snapshot PUT, or during the restore, is replayed in the same WorkDir
// and ends with one snapshot version and one manifest.
func TestBuildReplayReusesIdenticalSnapshot(t *testing.T) {
	src, n := sourceReplica(t, 2, true)
	prefix := testGenPrefix()
	ctx := context.Background()

	t.Run("after the snapshot PUT", func(t *testing.T) {
		b := newBucket()
		seal := sealReplica(t, b, src, testLane)
		work := filepath.Join(t.TempDir(), "build")
		boom := errors.New("the box died")
		b.FailOn = func(op, key string) error {
			if op == "put" && key == AnchorKey(prefix, n) {
				return boom
			}
			return nil
		}
		if _, err := BuildGeneration(ctx, inputs(b, seal, n, work)); !errors.Is(err, boom) {
			t.Fatalf("first run = %v, want the injected failure", err)
		}
		snaps := versionsOf(t, b, SnapshotKey(prefix, n))
		if len(snaps) != 1 {
			t.Fatalf("after the crash the snapshot has %d versions", len(snaps))
		}
		if _, ok := b.Current(testBucket, InstallManifestKey(prefix)); ok {
			t.Fatal("a manifest exists after a crash before the anchor")
		}
		if _, err := os.Stat(filepath.Join(work, testHost+".db")); err != nil {
			t.Fatalf("the restored file should still be on disk: %v", err)
		}
		first, _ := b.GetVersion(ctx, testBucket, SnapshotKey(prefix, n), snaps[0].VersionID)

		// The replay runs later: a build that encoded with its own clock
		// instead of the intent's would write other bytes.
		b.FailOn = nil
		m, err := BuildGeneration(ctx, inputsAt(b, seal, n, work, testClock.Add(time.Hour)))
		if err != nil {
			t.Fatalf("replay: %v", err)
		}
		if !m.BuiltAt.Equal(testClock.Add(time.Hour)) {
			t.Fatalf("BuiltAt = %v, want the replay's clock", m.BuiltAt)
		}
		snaps = versionsOf(t, b, SnapshotKey(prefix, n))
		if len(snaps) != 1 {
			t.Fatalf("after the replay the snapshot has %d versions, want 1", len(snaps))
		}
		if m.SnapshotVersion != snaps[0].VersionID {
			t.Fatalf("manifest names snapshot version %s, the bucket has %s", m.SnapshotVersion, snaps[0].VersionID)
		}
		again, _ := b.GetVersion(ctx, testBucket, SnapshotKey(prefix, n), m.SnapshotVersion)
		if !bytes.Equal(first, again) {
			t.Fatal("the replay's snapshot bytes differ from the first run's")
		}
		if _, ok := b.Current(testBucket, InstallManifestKey(prefix)); !ok {
			t.Fatal("no manifest after the replay")
		}
		if n := len(versionsOf(t, b, BuildIntentKey(prefix))); n != 1 {
			t.Fatalf("intent has %d versions, want 1", n)
		}
		// A third run, later again, is a pure replay: the stored manifest,
		// its BuiltAt included, and no new versions.
		m2, err := BuildGeneration(ctx, inputsAt(b, seal, n, work, testClock.Add(2*time.Hour)))
		if err != nil {
			t.Fatalf("second replay: %v", err)
		}
		if !sameManifest(m2, m) || !m2.BuiltAt.Equal(m.BuiltAt) {
			t.Fatalf("second replay returned %+v, first %+v", m2, m)
		}
		if n := len(versionsOf(t, b, AnchorKey(prefix, n))); n != 1 {
			t.Fatalf("anchor has %d versions, want 1", n)
		}
		if n := len(versionsOf(t, b, InstallManifestKey(prefix))); n != 1 {
			t.Fatalf("manifest has %d versions, want 1", n)
		}
	})

	t.Run("during the restore", func(t *testing.T) {
		b := newBucket()
		seal := sealReplica(t, b, src, testLane)
		work := filepath.Join(t.TempDir(), "build")
		boom := errors.New("the network died")
		// Every planned file is read once to verify it before the restore;
		// the restore's own reads come after. Fail from then on.
		planned := plannedFiles(t, b, seal, n)
		gets := 0
		b.FailOn = func(op, key string) error {
			if op != "get" {
				return nil
			}
			gets++
			if gets > planned {
				return boom
			}
			return nil
		}
		// Litestream's restore pipeline does not carry the read error's
		// identity (its compactor closes a pipe and the decoder reports the
		// EOF), so the assertion is that the restore failed and wrote
		// nothing, not that boom survives the wrapping.
		if _, err := BuildGeneration(ctx, inputs(b, seal, n, work)); err == nil {
			t.Fatal("the restore succeeded with its reads failing")
		} else if errors.Is(err, ErrBuildIncomplete) || errors.Is(err, ErrBuildDiverged) {
			t.Fatalf("first run = %v, want a plain restore failure", err)
		} else {
			t.Logf("first run failed as injected: %v", err)
		}
		if gets <= planned {
			t.Fatalf("the failure was not injected: %d gets, %d planned", gets, planned)
		}
		if vs, err := b.ListVersions(ctx, testBucket, prefix); err != nil || len(vs) != 0 {
			t.Fatalf("a failed restore left %d objects in the generation", len(vs))
		}
		// A crash leaves whatever the writer had on disk: a partial output
		// and litestream's temporary file.
		if err := os.MkdirAll(work, 0o700); err != nil {
			t.Fatal(err)
		}
		for _, p := range []string{testHost + ".db", testHost + ".db.tmp", testHost + ".db-wal", testHost + ".db-shm", testHost + ".db-txid"} {
			if err := os.WriteFile(filepath.Join(work, p), []byte("partial"), 0o600); err != nil {
				t.Fatal(err)
			}
		}
		b.FailOn = nil
		m, err := BuildGeneration(ctx, inputsAt(b, seal, n, work, testClock.Add(time.Hour)))
		if err != nil {
			t.Fatalf("replay: %v", err)
		}
		if got := fileSHA256(t, filepath.Join(work, testHost+".db")); got != m.DBSHA256 {
			t.Fatalf("the replay's file is %s, manifest %s", got, m.DBSHA256)
		}
		out := filepath.Join(t.TempDir(), "restored.db")
		restoreGeneration(t, b, m, out)
		if got := fileSHA256(t, out); got != m.DBSHA256 {
			t.Fatalf("the generation restores to %s, want %s", got, m.DBSHA256)
		}
	})
}

func TestBuildRefusesAGapOrAShortRestore(t *testing.T) {
	// Level 0 only: with no level-9 snapshot every level-0 file is on the
	// restore path, so a missing one is a gap.
	src, n := sourceReplica(t, 4, false)
	b := newBucket()
	seal := sealReplica(t, b, src, testLane)
	if len(seal) < 3 {
		t.Fatalf("need at least three level-0 files, have %d", len(seal))
	}
	prefix := testGenPrefix()
	ctx := context.Background()

	gapped := append(append([]SealedObject{}, seal[:1]...), seal[2:]...)
	_, err := BuildGeneration(ctx, inputs(b, gapped, n, filepath.Join(t.TempDir(), "build")))
	if !errors.Is(err, ErrBuildIncomplete) {
		t.Fatalf("gap: %v, want ErrBuildIncomplete", err)
	}
	if vs, _ := b.ListVersions(ctx, testBucket, prefix); len(vs) != 0 {
		t.Fatalf("a gap wrote %d objects into the generation", len(vs))
	}

	_, err = BuildGeneration(ctx, inputs(b, seal, n+5, filepath.Join(t.TempDir(), "build")))
	if !errors.Is(err, ErrBuildIncomplete) {
		t.Fatalf("short: %v, want ErrBuildIncomplete", err)
	}
	if vs, _ := b.ListVersions(ctx, testBucket, prefix); len(vs) != 0 {
		t.Fatalf("a short restore wrote %d objects into the generation", len(vs))
	}

	// The whole set reaches n.
	if _, err := BuildGeneration(ctx, inputs(b, seal, n, filepath.Join(t.TempDir(), "build"))); err != nil {
		t.Fatalf("the complete set: %v", err)
	}
}

func TestBuildRefusesADivergentIntent(t *testing.T) {
	src, n := sourceReplica(t, 1, true)
	b := newBucket()
	seal := sealReplica(t, b, src, testLane)
	prefix := testGenPrefix()
	ctx := context.Background()
	// An intent from another build of this generation with other inputs.
	other, _ := json.Marshal(BuildIntent{V: 1, Op: testOp, Host: testHost, N: n, PageSize: 1024, Commit: 1, Timestamp: 5})
	b.Put(testBucket, BuildIntentKey(prefix), other)
	_, err := BuildGeneration(ctx, inputs(b, seal, n, filepath.Join(t.TempDir(), "build")))
	if !errors.Is(err, ErrBuildDiverged) {
		t.Fatalf("divergent intent: %v, want ErrBuildDiverged", err)
	}
	if _, ok := b.Current(testBucket, SnapshotKey(prefix, n)); ok {
		t.Fatal("a snapshot was written under a divergent intent")
	}
}

func TestBuildRefusesADivergentSnapshot(t *testing.T) {
	src, n := sourceReplica(t, 1, true)
	b := newBucket()
	seal := sealReplica(t, b, src, testLane)
	prefix := testGenPrefix()
	ctx := context.Background()
	b.Put(testBucket, SnapshotKey(prefix, n), []byte("not this build's bytes"))
	_, err := BuildGeneration(ctx, inputs(b, seal, n, filepath.Join(t.TempDir(), "build")))
	if !errors.Is(err, ErrBuildDiverged) {
		t.Fatalf("divergent snapshot: %v, want ErrBuildDiverged", err)
	}
	if _, ok := b.Current(testBucket, AnchorKey(prefix, n)); ok {
		t.Fatal("an anchor was written beside a divergent snapshot")
	}
}

func TestReadDBHeader(t *testing.T) {
	path := filepath.Join(t.TempDir(), "x.db")
	writeRows(t, path, 3)
	ps, pc, err := ReadDBHeader(path)
	if err != nil {
		t.Fatal(err)
	}
	st, _ := os.Stat(path)
	if ps != 4096 || int64(pc)*int64(ps) != st.Size() {
		t.Fatalf("header %d x %d, file %d bytes", pc, ps, st.Size())
	}
	if err := os.WriteFile(path, []byte("not a database at all, just text"), 0o600); err != nil {
		t.Fatal(err)
	}
	if _, _, err := ReadDBHeader(path); err == nil {
		t.Fatal("garbage accepted")
	}
}

// TestEncodeFullSkipsTheLockPage: a database that spans SQLite's lock page
// encodes without it, verifies, and restores through litestream to a file
// whose page count is the original's.
func TestEncodeFullSkipsTheLockPage(t *testing.T) {
	if testing.Short() {
		t.Skip("writes a 1 GiB database")
	}
	dir := t.TempDir()
	path := filepath.Join(dir, "big.db")
	writeRows(t, path, 1)
	const pageSize = 4096
	lock := ltx.LockPgno(pageSize)
	pageCount := lock + 1
	f, err := os.OpenFile(path, os.O_RDWR, 0)
	if err != nil {
		t.Fatal(err)
	}
	if err := f.Truncate(int64(pageCount) * pageSize); err != nil {
		t.Fatal(err)
	}
	// The in-header size follows the file, as SQLite keeps it.
	var hdr [4]byte
	binary.BigEndian.PutUint32(hdr[:], pageCount)
	if _, err := f.WriteAt(hdr[:], 28); err != nil {
		t.Fatal(err)
	}
	ps, pc, err := ReadDBHeader(path)
	if err != nil {
		t.Fatal(err)
	}
	if ps != pageSize || pc != pageCount {
		t.Fatalf("header %d x %d", pc, ps)
	}
	origPages := pageCountPragma(t, path)
	if origPages != int64(pageCount) {
		t.Fatalf("PRAGMA page_count = %d, want %d", origPages, pageCount)
	}

	data, crc, err := EncodeFull(f, ps, pc, 1, 1, 1234, 0)
	f.Close()
	if err != nil {
		t.Fatalf("EncodeFull: %v", err)
	}
	h, tr, pages := decodeLTX(t, data)
	if pages != int(pageCount)-1 {
		t.Fatalf("encoded %d pages, want %d (every page but the lock page)", pages, pageCount-1)
	}
	if h.Commit != pageCount || uint64(tr.FileChecksum) != crc {
		t.Fatalf("header commit %d crc %x, returned crc %x", h.Commit, uint64(tr.FileChecksum), crc)
	}

	b := newBucket()
	key := SnapshotKey(testGenPrefix(), 1)
	v := b.Put(testBucket, key, data)
	gen, err := NewSealedClient(b, testBucket, []SealedObject{{Key: key, VersionID: v, Size: int64(len(data))}})
	if err != nil {
		t.Fatal(err)
	}
	out := filepath.Join(dir, "restored.db")
	r := litestream.NewReplicaWithClient(nil, gen)
	opt := litestream.NewRestoreOptions()
	opt.OutputPath = out
	if err := r.Restore(context.Background(), opt); err != nil {
		t.Fatalf("restore: %v", err)
	}
	if got := pageCountPragma(t, out); got != origPages {
		t.Fatalf("restored PRAGMA page_count = %d, original %d", got, origPages)
	}
	st, err := os.Stat(out)
	if err != nil {
		t.Fatal(err)
	}
	if st.Size() != int64(pageCount)*pageSize {
		t.Fatalf("restored file is %d bytes, want %d", st.Size(), int64(pageCount)*pageSize)
	}
}