package backup

import (
	"context"
	"encoding/json"
	"errors"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"textmachine/platform/internal/pgstore"
	"textmachine/platform/internal/runner"
)

// fakeEngine stands in for `tmctl backup`: it writes a file where the engine would and reports it,
// or refuses the way the engine refuses a book with no database.
type fakeEngine struct {
	refuse  bool
	fail    error
	calls   int
	content string
}

func (f *fakeEngine) Backup(_ context.Context, _, workdir string) (runner.BackupOutcome, error) {
	f.calls++
	if f.fail != nil {
		return runner.BackupOutcome{}, f.fail
	}
	if f.refuse {
		return runner.BackupOutcome{Exited: true, ExitCode: 1,
			Stderr: "tmctl backup: project database does not exist yet"}, nil
	}
	dir := filepath.Join(workdir, "backups")
	if err := os.MkdirAll(dir, 0o750); err != nil {
		return runner.BackupOutcome{}, err
	}
	path := filepath.Join(dir, "snapshot.db")
	body := f.content
	if body == "" {
		body = "CONSISTENT-COPY"
	}
	if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
		return runner.BackupOutcome{}, err
	}
	return runner.BackupOutcome{Exited: true, ExitCode: 0, Path: path}, nil
}

type fakeStore struct {
	books []pgstore.BookDirectory
	err   error
}

func (f *fakeStore) BooksForBackup(context.Context) ([]pgstore.BookDirectory, error) {
	return f.books, f.err
}

// fakePg writes a file where pg_dump would and exits 0, or fails. Written as a shell script because
// the code under test SPAWNS the tool: a Go double would prove the wiring this platform does not have.
func fakePg(t *testing.T, dir, name string, fail bool) string {
	t.Helper()
	sh, err := exec.LookPath("sh")
	if err != nil {
		t.Skip("sh not on PATH: the pg_dump double cannot be built")
	}
	path := filepath.Join(dir, name)
	script := "#!" + sh + "\n"
	if fail {
		script += "echo 'pg_dump: error: connection failed' >&2\nexit 1\n"
	} else {
		// --file=<path> is the argument the real tool takes and the one this code passes.
		script += "for a in \"$@\"; do case \"$a\" in --file=*) printf 'PGDUMP' > \"${a#--file=}\";; esac; done\nexit 0\n"
	}
	if err := os.WriteFile(path, []byte(script), 0o700); err != nil {
		t.Fatal(err)
	}
	return path
}

// fakeRefusing stands in for a `pg_restore --list` that cannot read the archive — a truncated dump,
// a version mismatch that wrote a header and stopped.
func fakeRefusing(t *testing.T, dir, name string) string {
	t.Helper()
	sh, err := exec.LookPath("sh")
	if err != nil {
		t.Skip("sh not on PATH")
	}
	path := filepath.Join(dir, name)
	body := "#!" + sh + "\necho 'pg_restore: error: did not find magic string in file header' >&2\nexit 1\n"
	if err := os.WriteFile(path, []byte(body), 0o700); err != nil {
		t.Fatal(err)
	}
	return path
}

func fakeOK(t *testing.T, dir, name string) string {
	t.Helper()
	sh, err := exec.LookPath("sh")
	if err != nil {
		t.Skip("sh not on PATH")
	}
	path := filepath.Join(dir, name)
	if err := os.WriteFile(path, []byte("#!"+sh+"\nexit 0\n"), 0o700); err != nil {
		t.Fatal(err)
	}
	return path
}

// aBook makes a book directory shaped like a real one: a configuration, a source, a LIVE database
// with its write-ahead sidecar, a sidecar artefact, and the engine's own `backups/` subdirectory.
func aBook(t *testing.T, root, id string) pgstore.BookDirectory {
	t.Helper()
	dir := filepath.Join(root, id)
	if err := os.MkdirAll(filepath.Join(dir, "backups"), 0o750); err != nil {
		t.Fatal(err)
	}
	for name, body := range map[string]string{
		"book.yaml":                     "book_id: " + id + "\ntitle: t\n",
		"source.txt":                    "第1章\n",
		id + ".db":                      "LIVE-AND-POSSIBLY-TORN",
		id + ".db-wal":                  "WAL",
		id + ".db.manifest.json":        `{"manifest_version":"tm-manifest-v2"}`,
		"events.jsonl":                  "{}\n",
		"backups/older-engine-point.db": "AN OLDER ENGINE RESTORE POINT",
	} {
		if err := os.WriteFile(filepath.Join(dir, name), []byte(body), 0o600); err != nil {
			t.Fatal(err)
		}
	}
	return pgstore.BookDirectory{ID: id, Workdir: dir, Status: "not_started"}
}

func service(t *testing.T, books []pgstore.BookDirectory, eng *fakeEngine, dumpFails bool) (*Service, string) {
	t.Helper()
	tools := t.TempDir()
	dest := t.TempDir()
	return &Service{
		Cfg: Config{
			Dir: dest, Every: time.Hour, Keep: 3,
			PgDumpBin:    fakePg(t, tools, "pg_dump", dumpFails),
			PgRestoreBin: fakeOK(t, tools, "pg_restore"),
			DSN:          "postgres://u:pw@127.0.0.1:5432/db?sslmode=disable",
			EngineBinary: "/opt/engine/tmctl",
		},
		Store: &fakeStore{books: books}, Engine: eng,
	}, dest
}

// The whole shape of a restore point in one pass: the money registry, every book's files, the
// engine's consistent copy standing in for the live database, and a manifest that can find each one.
//
// Mutation caught: skipping the engine call (the live, possibly torn file becomes the backup);
// copying the live database anyway (a restore has two candidates and no way to choose); leaving the
// engine's own `backups/` in (every pass copies every previous pass); publishing without the manifest
// (nothing can be verified afterwards).
func TestARestorePointCarriesTheLedgerAndEachBooksConsistentDatabase(t *testing.T) {
	root := t.TempDir()
	books := []pgstore.BookDirectory{aBook(t, root, "bk_one"), aBook(t, root, "bk_two")}
	eng := &fakeEngine{}
	svc, dest := service(t, books, eng, false)

	res, err := svc.Take(t.Context())
	if err != nil {
		t.Fatalf("Take: %v", err)
	}
	if res.Books != 2 || res.Skipped != 0 || !res.Complete {
		t.Fatalf("result does not describe two copied books: %+v", res)
	}
	if eng.calls != 2 {
		t.Errorf("the engine was asked for %d consistent copies, want one per book", eng.calls)
	}
	m, err := ReadManifest(res.Path)
	if err != nil {
		t.Fatal(err)
	}
	if m.Version != ManifestVersion || !m.Complete || len(m.Books) != 2 {
		t.Fatalf("manifest: %+v", m)
	}
	if m.Postgres.Name != DumpFile || m.Postgres.Bytes == 0 {
		t.Errorf("the money registry is not in the point: %+v", m.Postgres)
	}
	for _, b := range m.Books {
		names := map[string]bool{}
		for _, f := range b.Files {
			names[f.Name] = true
		}
		db := filepath.ToSlash(filepath.Join(BooksDir, b.BookID, ProjectDBName))
		if !names[db] {
			t.Errorf("%s: the consistent database is missing: %v", b.BookID, names)
		}
		body, err := os.ReadFile(filepath.Join(res.Path, filepath.FromSlash(db)))
		if err != nil || string(body) != "CONSISTENT-COPY" {
			// The point of the whole mechanism: what is stored is the engine's snapshot, never the
			// file a run may be writing.
			t.Errorf("%s: the stored database is not the engine's copy: %q, %v", b.BookID, body, err)
		}
		for _, want := range []string{"book.yaml", "source.txt", b.BookID + ".db.manifest.json", "events.jsonl"} {
			if !names[filepath.ToSlash(filepath.Join(BooksDir, b.BookID, want))] {
				t.Errorf("%s: %s did not make it into the point: %v", b.BookID, want, names)
			}
		}
		for _, unwanted := range []string{b.BookID + ".db", b.BookID + ".db-wal", "backups/older-engine-point.db"} {
			if names[filepath.ToSlash(filepath.Join(BooksDir, b.BookID, unwanted))] {
				t.Errorf("%s: %s should not be in a restore point: %v", b.BookID, unwanted, names)
			}
		}
	}
	// The engine's copy is taken AWAY, not left behind: otherwise every pass grows each book's own
	// directory by a whole database.
	for _, b := range books {
		if _, err := os.Stat(filepath.Join(b.Workdir, "backups", "snapshot.db")); !errors.Is(err, os.ErrNotExist) {
			t.Errorf("%s: the engine's copy was left in the book's directory (%v)", b.ID, err)
		}
	}
	if points, _ := List(dest); len(points) != 1 {
		t.Errorf("published points: %v", points)
	}
}

// The money registry has no partial outcome: a point without it answers nothing about who paid what,
// so a failed dump publishes NOTHING and leaves no staging directory behind either.
//
// Mutation caught: publishing anyway (a restore point that looks complete and carries no ledger);
// returning early without the cleanup (a full copy of the deployment accumulates per failure).
func TestAFailedLedgerDumpPublishesNothingAndLeavesNothing(t *testing.T) {
	root := t.TempDir()
	svc, dest := service(t, []pgstore.BookDirectory{aBook(t, root, "bk_one")}, &fakeEngine{}, true)
	if _, err := svc.Take(t.Context()); err == nil {
		t.Fatal("a failed pg_dump was reported as a restore point")
	}
	points, err := List(dest)
	if err != nil || len(points) != 0 {
		t.Errorf("a point was published without the ledger: %v (%v)", points, err)
	}
	entries, err := os.ReadDir(dest)
	if err != nil {
		t.Fatal(err)
	}
	if len(entries) != 0 {
		t.Errorf("the staging directory was left behind: %v", entries)
	}
}

// A book that cannot be copied does NOT sink the pass — nine books saved beat none — but the point
// says so about itself, and Verify repeats it. The alternative is the dangerous one: a point that
// silently holds less than it claims.
func TestABookThatCannotBeCopiedMarksThePointIncompleteRatherThanFailingIt(t *testing.T) {
	root := t.TempDir()
	ok := aBook(t, root, "bk_ok")

	// ⚠ A MISSING DIRECTORY IS TWO FACTS, AND THIS TEST ASSERTED ONLY THE HARMLESS ONE UNTIL 05.09.
	// Its earlier edition treated ANY absent directory as an ordinary skip that leaves the point
	// complete — reasoning that a rejected upload takes its directory with it. True for a rejected
	// book; false for every other status, and the false half is the dangerous one: an unmounted books
	// volume, or a books tree moved while the absolute workdirs in Postgres stayed put, makes EVERY
	// book an "ordinary skip" and publishes a point holding only the ledger, marked complete, with the
	// age gauge green. Retention then evicts the last point that held any paid work on schedule.
	// The assertion below is the corrected one and is strictly stronger.
	gone := pgstore.BookDirectory{ID: "bk_gone", Workdir: filepath.Join(root, "not_here"), Status: "not_started"}
	svc, _ := service(t, []pgstore.BookDirectory{ok, gone}, &fakeEngine{}, false)
	res, err := svc.Take(t.Context())
	if err != nil {
		t.Fatalf("one absent book ended the whole pass: %v", err)
	}
	if res.Books != 1 || res.Skipped != 1 {
		t.Fatalf("counts: %+v", res)
	}
	m, err := ReadManifest(res.Path)
	if err != nil {
		t.Fatal(err)
	}
	if m.Complete {
		t.Errorf("a book whose status says its directory should be there went missing, and the point still calls itself complete: %+v", m)
	}
	var skipped string
	for _, b := range m.Books {
		if b.BookID == "bk_gone" {
			skipped = b.Skipped
		}
	}
	if skipped == "" || !strings.Contains(skipped, "not_started") {
		t.Errorf("the hole does not name the status that makes it one: %q", skipped)
	}

	// The other half: a REJECTED book legitimately has no directory — its reject path removed it with
	// the source — and that must NOT make every point of every deployment read incomplete, which is
	// how the flag stops meaning anything.
	rejected := pgstore.BookDirectory{ID: "bk_rejected", Workdir: filepath.Join(root, "also_not_here"), Status: "rejected"}
	svcR, _ := service(t, []pgstore.BookDirectory{aBook(t, root, "bk_fine"), rejected}, &fakeEngine{}, false)
	resR, err := svcR.Take(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	if !resR.Complete || resR.Skipped != 1 {
		t.Errorf("a rejected book with no directory was treated as a hole: %+v", resR)
	}

	// An engine that FAILS is different: something is wrong and the point is not what it claims.
	svc2, _ := service(t, []pgstore.BookDirectory{aBook(t, root, "bk_two")}, &fakeEngine{fail: errors.New("no such binary")}, false)
	res2, err := svc2.Take(t.Context())
	if err != nil {
		t.Fatalf("a failing engine ended the whole pass: %v", err)
	}
	if res2.Complete {
		t.Error("a book the engine could not copy left the point marked complete")
	}
	problems, err := Verify(res2.Path)
	if err != nil {
		t.Fatal(err)
	}
	if len(problems) == 0 || !strings.Contains(strings.Join(problems, " "), "INCOMPLETE") {
		t.Errorf("Verify does not repeat the point's own verdict: %v", problems)
	}
}

// ⛔ THE WORST OBJECT THIS PACKAGE CAN PRODUCE: a point that says `complete: true` over a hole. It was
// REAL — measured 05.09 by running two `tmplatformctl backup` at once against a live stand — and it
// came from treating every non-zero engine exit as one ordinary "skip".
//
// The engine refuses to overwrite a restore point it already made under the same timestamp
// (backend/internal/store/backup.go: "refusing to overwrite a restore point"), so the loser of a
// same-second race made the winner's engine call fail — and the winner published `books=0 skipped=1
// complete=true`. The discriminator is the book's own intake status and nothing else.
//
// Mutation caught: collapsing the two branches back into one skip; reading the discriminator off the
// exit code or the engine's message; making a not-yet-cut book fail the whole pass.
func TestAnEngineRefusalOnACutBookIsAHoleAndNotASkip(t *testing.T) {
	root := t.TempDir()
	cut := aBook(t, root, "bk_cut") // status not_started: intake finished, a database must exist
	svc, _ := service(t, []pgstore.BookDirectory{cut}, &fakeEngine{refuse: true}, false)
	res, err := svc.Take(t.Context())
	if err != nil {
		t.Fatalf("one refused book ended the whole pass: %v", err)
	}
	if res.Complete {
		t.Error("a point holding NO copy of a cut book was published as complete")
	}
	m, err := ReadManifest(res.Path)
	if err != nil {
		t.Fatal(err)
	}
	if m.Complete || len(m.Books) != 1 || m.Books[0].Skipped == "" {
		t.Errorf("the manifest does not name the hole: %+v", m)
	}

	// The other half: a book whose intake has NOT finished has nothing to copy yet, and that must NOT
	// make every point of every deployment read "incomplete" — which is how the flag stops meaning
	// anything.
	for _, status := range []string{"uploading", "parsing"} {
		early := aBook(t, root, "bk_"+status)
		early.Status = status
		svc2, _ := service(t, []pgstore.BookDirectory{early}, &fakeEngine{refuse: true}, false)
		res2, err := svc2.Take(t.Context())
		if err != nil {
			t.Fatalf("%s: %v", status, err)
		}
		if !res2.Complete || res2.Skipped != 1 {
			t.Errorf("%s: a book with no database yet was treated as a hole: %+v", status, res2)
		}
	}
}

// Two passes in the same second must not share a staging directory: the winner renames it away under
// the loser's feet, and the loser's half-written files are what the winner then publishes.
//
// Mutation caught: MkdirAll instead of Mkdir — the exact one-word difference that produced the
// measured failure above.
func TestTwoPassesInOneSecondDoNotShareAStagingDirectory(t *testing.T) {
	root := t.TempDir()
	svc, dest := service(t, []pgstore.BookDirectory{aBook(t, root, "bk_one")}, &fakeEngine{}, false)
	fixed := time.Date(2026, 9, 5, 12, 0, 0, 0, time.UTC)
	svc.Now = func() time.Time { return fixed }
	if _, err := svc.Take(t.Context()); err != nil {
		t.Fatal(err)
	}
	// A second pass in the same second finds the published point and refuses BEFORE doing any work.
	if _, err := svc.Take(t.Context()); err == nil {
		t.Fatal("a second pass in the same second was allowed to run")
	}
	// And with the first pass still in flight — its staging directory present, nothing published —
	// the second must refuse on the CLAIM rather than start writing into it.
	staging := filepath.Join(dest, partialPrefix+"20260905T130000Z")
	if err := os.Mkdir(staging, 0o750); err != nil {
		t.Fatal(err)
	}
	svc.Now = func() time.Time { return fixed.Add(time.Hour) }
	_, err := svc.Take(t.Context())
	if err == nil || !strings.Contains(err.Error(), "already being written") {
		t.Fatalf("a pass wrote into another pass's staging directory: %v", err)
	}
	if points, _ := List(dest); len(points) != 1 {
		t.Errorf("published points: %v", points)
	}
}

// ⛔ THE ENGINE'S `.env` OF PROVIDER KEYS MUST NEVER REACH A RESTORE POINT. The runbook tells the
// operator to ship this directory OFF THE HOST; a backup that carried the deployment's API keys with
// it would be a credential leak on a schedule.
//
// Mutation caught: dropping the dotfile rule (the keys ride out with every generation).
func TestProviderKeysBesideABookAreNeverCopiedAndTheOmissionIsRecorded(t *testing.T) {
	root := t.TempDir()
	b := aBook(t, root, "bk_keys")
	if err := os.WriteFile(filepath.Join(b.Workdir, ".env"), []byte("DEEPSEEK_API_KEY=sk-live-do-not-copy"+"\n"), 0o600); err != nil {
		t.Fatal(err)
	}
	svc, _ := service(t, []pgstore.BookDirectory{b}, &fakeEngine{}, false)
	res, err := svc.Take(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	// Not by reading the manifest — by walking what actually landed on the disk.
	var leaked []string
	err = filepath.WalkDir(res.Path, func(p string, d os.DirEntry, err error) error {
		if err != nil || d.IsDir() {
			return err
		}
		body, rerr := os.ReadFile(p)
		if rerr == nil && strings.Contains(string(body), "sk-live-do-not-copy") {
			leaked = append(leaked, p)
		}
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if len(leaked) > 0 {
		t.Fatalf("provider keys are inside the restore point: %v", leaked)
	}
	// And the omission is NAMED, so nothing goes missing quietly.
	m, err := ReadManifest(res.Path)
	if err != nil {
		t.Fatal(err)
	}
	if !strings.Contains(strings.Join(m.Books[0].NotCopied, " "), ".env") {
		t.Errorf("the skipped dotfile is not recorded: %v", m.Books[0].NotCopied)
	}
}

// A book's own data can live in a SUBDIRECTORY — `langpack_extend` is a per-book overlay of the
// pair's canon, resolved against book.yaml. An earlier version of this package skipped every
// subdirectory and still called the point COMPLETE, which is a book's canon lost in silence.
//
// Mutation caught: going back to `if e.IsDir() { continue }`; or recursing into the engine's own
// `backups/`, which is what makes a backup directory grow quadratically.
func TestABooksSubdirectoriesAreCopiedButTheEnginesOwnBackupsAreNot(t *testing.T) {
	root := t.TempDir()
	b := aBook(t, root, "bk_deep")
	if err := os.MkdirAll(filepath.Join(b.Workdir, "langpack", "zh-ru"), 0o750); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(filepath.Join(b.Workdir, "langpack", "zh-ru", "heading.txt"), []byte("Глава {n}"+"\n"), 0o600); err != nil {
		t.Fatal(err)
	}
	svc, _ := service(t, []pgstore.BookDirectory{b}, &fakeEngine{}, false)
	res, err := svc.Take(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	if body, err := os.ReadFile(filepath.Join(res.Path, BooksDir, "bk_deep", "langpack", "zh-ru", "heading.txt")); err != nil ||
		string(body) != "Глава {n}"+"\n" {
		t.Errorf("a book's own data in a subdirectory did not survive: %q (%v)", body, err)
	}
	if _, err := os.Stat(filepath.Join(res.Path, BooksDir, "bk_deep", engineBackupsDir)); !errors.Is(err, os.ErrNotExist) {
		t.Errorf("the engine's own restore points were copied into ours (%v)", err)
	}
	m, err := ReadManifest(res.Path)
	if err != nil {
		t.Fatal(err)
	}
	if !strings.Contains(strings.Join(m.Books[0].NotCopied, " "), engineBackupsDir+"/") {
		t.Errorf("the skipped directory is not recorded: %v", m.Books[0].NotCopied)
	}
}

// A pass that runs out of its budget publishes NOTHING. Truncating instead would drop the NEWEST
// books — the listing is oldest-first — while resetting the age gauge, so every generation would omit
// exactly the work being paid for now and the one alertable number would stay green.
func TestAPassThatRunsOutOfItsBudgetPublishesNothing(t *testing.T) {
	root := t.TempDir()
	books := []pgstore.BookDirectory{aBook(t, root, "bk_a"), aBook(t, root, "bk_b")}
	svc, dest := service(t, books, &fakeEngine{}, false)
	ctx, cancel := context.WithCancel(t.Context())
	cancel() // the deadline has already passed when the books are reached
	if _, err := svc.Take(ctx); err == nil {
		t.Fatal("a pass with no budget left published a restore point")
	}
	if points, _ := List(dest); len(points) != 0 {
		t.Errorf("a truncated point was published: %v", points)
	}
	entries, _ := os.ReadDir(dest)
	if len(entries) != 0 {
		t.Errorf("staging was left behind: %v", entries)
	}
}

// The age gauge reports the newest COMPLETE point. A point that could not copy a book is worth
// keeping and is not a backup OF THAT BOOK, so letting its timestamp reset the age would hide exactly
// the state an operator needs paging for.
func TestTheAgeGaugeIgnoresAnIncompletePoint(t *testing.T) {
	root := t.TempDir()
	good := aBook(t, root, "bk_ok")
	svc, _ := service(t, []pgstore.BookDirectory{good}, &fakeEngine{}, false)
	at := time.Date(2026, 9, 5, 12, 0, 0, 0, time.UTC)
	svc.Now = func() time.Time { return at }
	if _, err := svc.Take(t.Context()); err != nil {
		t.Fatal(err)
	}
	age, has, err := svc.Age(at.Add(time.Hour))
	if err != nil || !has || age != time.Hour {
		t.Fatalf("a complete point: age=%v has=%v err=%v", age, has, err)
	}
	// A LATER point that is incomplete must not make the deployment look freshly backed up.
	svc.Store = &fakeStore{books: []pgstore.BookDirectory{good}}
	svc.Engine = &fakeEngine{refuse: true}
	at = at.Add(6 * time.Hour)
	if _, err := svc.Take(t.Context()); err != nil {
		t.Fatal(err)
	}
	age, has, err = svc.Age(at.Add(time.Hour))
	if err != nil || !has || age != 7*time.Hour {
		t.Errorf("an incomplete point reset the age: age=%v has=%v err=%v (want 7h, the last COMPLETE one)", age, has, err)
	}
}

// A book that fails PART WAY must leave NOTHING behind, not a half-copied database the manifest does
// not mention.
//
// ⚠ The runbook's restore step is `cp -a books/<id>/. <workdir>/`. A truncated `project.db` sitting in
// a point whose manifest says the book was skipped is therefore not an untidiness: it is an operator
// copying a torn database over a good one, having been told this book was not in the point at all.
//
// Mutation caught: dropping the RemoveAll — the partial files are published, Verify never hashes them
// because the manifest has no entry for them, and `--verify` reports the point sound.
func TestAPartlyCopiedBookLeavesNothingInThePoint(t *testing.T) {
	root := t.TempDir()
	b := aBook(t, root, "bk_halfway")
	// The engine's snapshot lands, and the rest of the copy then fails: a directory the copier cannot
	// read reproduces "failed after some files were already written".
	unreadable := filepath.Join(b.Workdir, "locked")
	if err := os.Mkdir(unreadable, 0o000); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = os.Chmod(unreadable, 0o750) })
	svc, _ := service(t, []pgstore.BookDirectory{b}, &fakeEngine{}, false)
	res, err := svc.Take(t.Context())
	if err != nil {
		t.Fatalf("one book's failure ended the whole pass: %v", err)
	}
	if res.Complete {
		t.Error("a point missing a book was published as complete")
	}
	if _, err := os.Stat(filepath.Join(res.Path, BooksDir, "bk_halfway")); !errors.Is(err, os.ErrNotExist) {
		left, _ := os.ReadDir(filepath.Join(res.Path, BooksDir, "bk_halfway"))
		names := []string{}
		for _, e := range left {
			names = append(names, e.Name())
		}
		t.Errorf("a skipped book left files in the point that the manifest does not mention: %v", names)
	}
	// And the manifest is an exact description of what is on the disk: everything it lists verifies,
	// and it lists everything.
	if problems, err := Verify(res.Path); err != nil {
		t.Fatal(err)
	} else if len(problems) != 1 || !strings.Contains(problems[0], "INCOMPLETE") {
		t.Errorf("Verify: %v", problems)
	}
}

// The dump is READ BACK before the point is published, and a dump that cannot be read back does not
// become a restore point. Nothing observed this: the double `pg_restore` exited 0 whatever it was
// handed, so removing the check entirely would have left every test green.
func TestADumpThatCannotBeReadBackIsNotPublished(t *testing.T) {
	root := t.TempDir()
	tools := t.TempDir()
	dest := t.TempDir()
	svc := &Service{
		Cfg: Config{Dir: dest, Every: time.Hour, Keep: 3,
			PgDumpBin:    fakePg(t, tools, "pg_dump", false),
			PgRestoreBin: fakeRefusing(t, tools, "pg_restore"),
			DSN:          "postgres://u:pw@127.0.0.1:5432/db?sslmode=disable",
			EngineBinary: "/opt/engine/tmctl"},
		Store: &fakeStore{books: []pgstore.BookDirectory{aBook(t, root, "bk_one")}}, Engine: &fakeEngine{},
	}
	_, err := svc.Take(t.Context())
	if err == nil {
		t.Fatal("a dump pg_restore cannot read was published as a restore point")
	}
	if !strings.Contains(err.Error(), "cannot be read back") {
		t.Errorf("the error does not say what is wrong: %v", err)
	}
	if points, _ := List(dest); len(points) != 0 {
		t.Errorf("published anyway: %v", points)
	}
}

// Two guards their comments argue for at length and nothing checked: a book directory holding
// something that is not a regular file (following a symlink would copy from outside the book), and
// the removal of a `.partial-` a killed pass left behind.
func TestTheSymlinkGuardAndTheStaleStagingPruneBothWork(t *testing.T) {
	root := t.TempDir()
	b := aBook(t, root, "bk_one")
	// A symlink pointing OUT of the book, which must not be followed into the point.
	outside := filepath.Join(t.TempDir(), "somebody-elses-secret")
	if err := os.WriteFile(outside, []byte("NOT THIS BOOK'S"), 0o600); err != nil {
		t.Fatal(err)
	}
	if err := os.Symlink(outside, filepath.Join(b.Workdir, "escape.txt")); err != nil {
		t.Skipf("this filesystem does not do symlinks: %v", err)
	}
	svc, dest := service(t, []pgstore.BookDirectory{b}, &fakeEngine{}, false)
	now := time.Date(2026, 9, 5, 12, 0, 0, 0, time.UTC)
	svc.Now = func() time.Time { return now }

	// A leftover staging directory from a pass killed two days ago.
	stale := filepath.Join(dest, partialPrefix+"20260903T120000Z")
	if err := os.MkdirAll(stale, 0o750); err != nil {
		t.Fatal(err)
	}
	// ...and one from a pass that may still be running.
	fresh := filepath.Join(dest, partialPrefix+"20260905T113000Z")
	if err := os.MkdirAll(fresh, 0o750); err != nil {
		t.Fatal(err)
	}

	res, err := svc.Take(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := os.Stat(filepath.Join(res.Path, BooksDir, "bk_one", "escape.txt")); !errors.Is(err, os.ErrNotExist) {
		t.Errorf("a symlink out of the book was followed into the restore point (%v)", err)
	}
	m, err := ReadManifest(res.Path)
	if err != nil {
		t.Fatal(err)
	}
	if !strings.Contains(strings.Join(m.Books[0].NotCopied, " "), "escape.txt") {
		t.Errorf("the skipped symlink is not recorded: %v", m.Books[0].NotCopied)
	}
	if _, err := os.Stat(stale); !errors.Is(err, os.ErrNotExist) {
		t.Errorf("a day-old staging leftover was not pruned (%v)", err)
	}
	if _, err := os.Stat(fresh); err != nil {
		t.Errorf("a staging directory young enough to be a LIVE pass was removed under it: %v", err)
	}
}

// ⛔ A BOOK WHOSE SOURCE IS NAMED `source.db` MUST NOT LOSE ITS SOURCE. The intake keeps the user's
// own extension verbatim (`extensionOf` is not an allowlist), so a reader who uploads `novel.db` gets
// `source.db` on disk — and a rule that skips everything ending in `.db` drops the book's own text
// from every restore point. The source is the one file in that directory nobody can be asked to
// supply again.
//
// Mutation caught: dropping the `!isTheUploadedSource(name)` half of the guard.
func TestASourceNamedLikeADatabaseIsStillCopied(t *testing.T) {
	root := t.TempDir()
	b := aBook(t, root, "bk_dbsource")
	// The book's real source, under the name the intake would have written for an upload called
	// `novel.db` — and the live project database beside it, which must still be skipped.
	if err := os.WriteFile(filepath.Join(b.Workdir, "source.db"), []byte("第一章 текст книги"), 0o600); err != nil {
		t.Fatal(err)
	}
	svc, _ := service(t, []pgstore.BookDirectory{b}, &fakeEngine{}, false)
	res, err := svc.Take(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	body, err := os.ReadFile(filepath.Join(res.Path, BooksDir, "bk_dbsource", "source.db"))
	if err != nil || string(body) != "第一章 текст книги" {
		t.Fatalf("the book's source was dropped because it is named like a database: %q (%v)", body, err)
	}
	// ...and the live database beside it is STILL skipped, so the guard did not simply turn the rule off.
	if _, err := os.Stat(filepath.Join(res.Path, BooksDir, "bk_dbsource", "bk_dbsource.db")); !errors.Is(err, os.ErrNotExist) {
		t.Errorf("the live project database was copied after all (%v)", err)
	}
	m, err := ReadManifest(res.Path)
	if err != nil {
		t.Fatal(err)
	}
	if !strings.Contains(strings.Join(m.Books[0].NotCopied, " "), "bk_dbsource.db ") {
		t.Errorf("the live database is not recorded as skipped: %v", m.Books[0].NotCopied)
	}
}

// ⛔ RETENTION MUST NOT BE A COUNTDOWN TO DATA LOSS. Retention counts DIRECTORIES, and a deployment
// producing degraded points — an unmounted books volume, an engine that stopped answering — produces
// them on the same schedule as good ones. With the defaults that is `Keep × Every` before every point
// holding the paid work has been evicted, and the operator's only signal arrives after the last real
// backup is already gone.
//
// Mutation caught: dropping the newestComplete exception from prune.
func TestARunOfDegradedPointsCannotEvictTheLastCompleteOne(t *testing.T) {
	root := t.TempDir()
	good := aBook(t, root, "bk_ok")
	svc, dest := service(t, []pgstore.BookDirectory{good}, &fakeEngine{}, false)
	svc.Cfg.Keep = 2
	at := time.Date(2026, 9, 5, 0, 0, 0, 0, time.UTC)
	svc.Now = func() time.Time { return at }

	// One good point.
	res, err := svc.Take(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	theGoodOne := res.Stamp

	// Then the outage: the book's directory is gone while its status says it should be there, so every
	// later point is incomplete. Four of them — twice the retention count.
	if err := os.RemoveAll(good.Workdir); err != nil {
		t.Fatal(err)
	}
	for range 4 {
		at = at.Add(6 * time.Hour)
		if _, err := svc.Take(t.Context()); err != nil {
			t.Fatalf("a degraded pass failed outright: %v", err)
		}
	}

	points, err := List(dest)
	if err != nil {
		t.Fatal(err)
	}
	var found bool
	for _, p := range points {
		if p == theGoodOne {
			found = true
		}
	}
	if !found {
		t.Fatalf("the only point holding the paid work was evicted by retention; what is left: %v", points)
	}
	// And it is still the one the gauge reports, so the operator is not told the deployment is fresh.
	age, has, err := svc.Age(at)
	if err != nil || !has {
		t.Fatalf("age: %v %v", has, err)
	}
	if age != 24*time.Hour {
		t.Errorf("age %v: the gauge is not reporting the last COMPLETE point", age)
	}
}

// ⛔ PV2-02: A BOOK STILL IN INTAKE MUST GET ITS FILES INTO THE POINT. It has no project database
// yet, and an earlier version answered that by taking NOTHING — not the uploaded source, not
// `book.yaml`, not the journal — while the point still called itself complete.
//
// ⚠ `parsing` is not a five-minute window: the intake's `not_configured` and `storage_unavailable`
// never become terminal (internal/books/parse.go), so a host with a broken book template holds every
// upload in that state indefinitely. Those uploads are exactly the ones with no other copy anywhere,
// and this package's own comment calls the source irreplaceable.
//
// Mutation caught: returning from copyBook before copyBookFiles on the not-yet-cut branch.
func TestABookStillInIntakeStillGetsItsSourceIntoThePoint(t *testing.T) {
	root := t.TempDir()
	for _, status := range []string{"uploading", "parsing"} {
		b := aBook(t, root, "bk_"+status)
		b.Status = status
		svc, _ := service(t, []pgstore.BookDirectory{b}, &fakeEngine{refuse: true}, false)
		res, err := svc.Take(t.Context())
		if err != nil {
			t.Fatalf("%s: %v", status, err)
		}
		if !res.Complete {
			t.Errorf("%s: a book that legitimately has no database made the point incomplete", status)
		}
		// THE POINT: its own files are here, whatever the database situation.
		for _, want := range []string{"source.txt", "book.yaml", "events.jsonl"} {
			p := filepath.Join(res.Path, BooksDir, b.ID, want)
			if _, err := os.Stat(p); err != nil {
				t.Errorf("%s: %s is not in the point — the user's own upload was dropped: %v", status, want, err)
			}
		}
		m, err := ReadManifest(res.Path)
		if err != nil {
			t.Fatal(err)
		}
		if len(m.Books[0].Files) == 0 {
			t.Errorf("%s: the manifest lists no files for a book whose directory was full", status)
		}
		if m.Books[0].Skipped == "" {
			t.Errorf("%s: the absent database is not explained", status)
		}
	}
}

// ⛔ PV2-03: A REJECTED BOOK THAT KEPT ITS DIRECTORY IS NOT A HOLE. Only one of the two terminal
// reject paths removes the directory — `ReasonSourceUnreadable` — while the other (the attempt budget
// exhausted because the engine would not run) deliberately KEEPS the user's file
// (internal/books/parse.go, `if reason == ReasonSourceUnreadable { s.removeDir(...) }`).
//
// Such a book has a directory and no database, which fell between the branches and made every point
// incomplete FOREVER: the age gauge then reads +Inf permanently, the one alertable signal of real
// backup loss is stuck red and worthless, and prune's "never evict the newest COMPLETE point" becomes
// vacuous because no complete point exists. And the cause that produces such books — the engine not
// running — is the same deployment incident.
//
// Mutation caught: dropping `|| b.Status == statusRejected` from the engine-refusal branch.
func TestARejectedBookThatKeptItsDirectoryIsNotAHole(t *testing.T) {
	root := t.TempDir()
	b := aBook(t, root, "bk_rejected_kept")
	b.Status = "rejected" // its directory is right there: the reject path kept the user's file
	svc, dest := service(t, []pgstore.BookDirectory{b}, &fakeEngine{refuse: true}, false)
	at := time.Date(2026, 9, 5, 12, 0, 0, 0, time.UTC)
	svc.Now = func() time.Time { return at }
	res, err := svc.Take(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	if !res.Complete {
		t.Fatal("a rejected book that kept its directory made the point incomplete — the gauge would stick at +Inf forever")
	}
	// The gauge must actually report an age, which is the half that matters operationally.
	age, has, err := svc.Age(at.Add(time.Hour))
	if err != nil || !has || age != time.Hour {
		t.Errorf("the age gauge has no complete point to report: age=%v has=%v err=%v", age, has, err)
	}
	// And the user's file the deployment deliberately kept is in the point.
	if _, err := os.Stat(filepath.Join(res.Path, BooksDir, b.ID, "source.txt")); err != nil {
		t.Errorf("the source the reject path kept was not backed up: %v", err)
	}
	if points, _ := List(dest); len(points) != 1 {
		t.Errorf("points: %v", points)
	}
}

// Verify is what makes the manifest a guarantee rather than a description: a copy that rotted on the
// disk is indistinguishable from a good one by size alone, and the moment to find out is not the
// moment you need it.
func TestVerifyCatchesAFileThatChangedUnderTheManifest(t *testing.T) {
	root := t.TempDir()
	svc, _ := service(t, []pgstore.BookDirectory{aBook(t, root, "bk_one")}, &fakeEngine{}, false)
	res, err := svc.Take(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	if problems, err := Verify(res.Path); err != nil || len(problems) != 0 {
		t.Fatalf("a fresh point does not verify: %v (%v)", problems, err)
	}
	db := filepath.Join(res.Path, BooksDir, "bk_one", ProjectDBName)
	// The SAME LENGTH, one byte different — which is the case a size check cannot see and the one
	// silent corruption actually looks like.
	if err := os.WriteFile(db, []byte("CONSISTENT-COPX"), 0o600); err != nil {
		t.Fatal(err)
	}
	problems, err := Verify(res.Path)
	if err != nil {
		t.Fatal(err)
	}
	if len(problems) != 1 || !strings.Contains(problems[0], "sha256") {
		t.Errorf("a silently corrupted database was not caught: %v", problems)
	}
}

// A manifest of a shape this build does not know is REFUSED rather than half-read. That is the whole
// reason the document carries a version, and the failure it prevents is the worst kind: a restore
// that reads three fields of five and reports success.
func TestARestorePointOfAnUnknownShapeIsRefused(t *testing.T) {
	dir := t.TempDir()
	point := filepath.Join(dir, "20260905T120000Z")
	if err := os.MkdirAll(point, 0o750); err != nil {
		t.Fatal(err)
	}
	raw, _ := json.Marshal(Manifest{Version: "tm-platform-restore-point-v9", Stamp: "20260905T120000Z"})
	if err := os.WriteFile(filepath.Join(point, ManifestFile), raw, 0o600); err != nil {
		t.Fatal(err)
	}
	if _, err := ReadManifest(point); err == nil || !strings.Contains(err.Error(), "this build reads") {
		t.Errorf("a future restore point was read as if it were this one's: %v", err)
	}
}

// Keep is enforced AFTER a publish and never before: a prune that ran first would, on a host that
// then failed to write the new point, leave fewer copies than it started with.
func TestOldPointsGoOnlyOnceANewOneExists(t *testing.T) {
	root := t.TempDir()
	svc, dest := service(t, []pgstore.BookDirectory{aBook(t, root, "bk_one")}, &fakeEngine{}, false)
	svc.Cfg.Keep = 2
	now := time.Date(2026, 9, 5, 12, 0, 0, 0, time.UTC)
	svc.Now = func() time.Time { return now }
	for range 4 {
		if _, err := svc.Take(t.Context()); err != nil {
			t.Fatal(err)
		}
		now = now.Add(time.Hour)
	}
	points, err := List(dest)
	if err != nil {
		t.Fatal(err)
	}
	if len(points) != 2 {
		t.Fatalf("kept %v, want the newest 2", points)
	}
	if points[1] != "20260905T150000Z" || points[0] != "20260905T140000Z" {
		t.Errorf("the wrong two survived: %v", points)
	}
}

// The pass decides for itself whether a point is due, so the schedule survives a restart with no
// state of its own — and a deployment on a fifteen-second sweep tick does not take a full copy of
// itself every fifteen seconds.
func TestTheSweepTakesAPointOnlyWhenOneIsDue(t *testing.T) {
	root := t.TempDir()
	svc, dest := service(t, []pgstore.BookDirectory{aBook(t, root, "bk_one")}, &fakeEngine{}, false)
	svc.Cfg.Every = 6 * time.Hour
	now := time.Date(2026, 9, 5, 12, 0, 0, 0, time.UTC)
	svc.Now = func() time.Time { return now }
	if err := svc.Sweep(t.Context()); err != nil { // no point at all: one is due
		t.Fatal(err)
	}
	now = now.Add(5 * time.Hour)
	if err := svc.Sweep(t.Context()); err != nil {
		t.Fatal(err)
	}
	if points, _ := List(dest); len(points) != 1 {
		t.Fatalf("a point was taken before it was due: %v", points)
	}
	now = now.Add(2 * time.Hour)
	if err := svc.Sweep(t.Context()); err != nil {
		t.Fatal(err)
	}
	if points, _ := List(dest); len(points) != 2 {
		t.Fatalf("no point was taken when one was due: %v", points)
	}
	age, has, err := svc.Age(now)
	if err != nil || !has || age != 0 {
		t.Errorf("age of the newest point: %v %v %v", age, has, err)
	}
}

// The DSN carries a password and it must never reach an argv: /proc/<pid>/cmdline is readable by
// every process on the host.
//
// Mutation caught: passing the connection string through as `-d <dsn>`, which is the shortest way to
// write this function and leaks the password to every local process.
func TestThePasswordTravelsInTheEnvironmentAndNeverInTheArgv(t *testing.T) {
	args, env, err := dumpArgs("postgres://alice:s3cr3t@db.example:6432/tm?sslmode=disable", "/tmp/out.dump")
	if err != nil {
		t.Fatal(err)
	}
	if strings.Contains(strings.Join(args, " "), "s3cr3t") {
		t.Fatalf("the password is on the command line: %q", args)
	}
	var found bool
	for _, kv := range env {
		if kv == "PGPASSWORD=s3cr3t" {
			found = true
		}
	}
	if !found {
		t.Error("the password did not reach the child's environment, so the dump would prompt and hang")
	}
	joined := strings.Join(args, " ")
	for _, want := range []string{"db.example", "6432", "alice", "/tm", "--format=custom",
		"--file=/tmp/out.dump", "--no-password"} {
		if !strings.Contains(joined, want) {
			t.Errorf("missing %s in %q", want, joined)
		}
	}
	// ⛔ AND EVERY OTHER PARAMETER THE OPERATOR PUT IN THE DSN, `sslmode` above all. An earlier version
	// rebuilt the connection from host/port/user/dbname and dropped the rest, so a deployment whose
	// DSN says `verify-full` had its nightly dump go out under libpq's default `prefer` — TLS optional,
	// certificate unchecked — and nothing said so, because pg_dump succeeds either way.
	strict, _, err := dumpArgs("postgres://alice:s3cr3t@db.example:6432/tm?sslmode=verify-full&sslrootcert=/etc/ca.pem&connect_timeout=9", "/tmp/o")
	if err != nil {
		t.Fatal(err)
	}
	sj := strings.Join(strict, " ")
	for _, want := range []string{"sslmode=verify-full", "sslrootcert=/etc/ca.pem", "connect_timeout=9"} {
		if !strings.Contains(sj, want) {
			t.Errorf("the DSN's %s did not reach pg_dump — the dump would run under libpq defaults: %q", want, sj)
		}
	}
	if strings.Contains(sj, "s3cr3t") {
		t.Errorf("forwarding the DSN carried the password onto the argv: %q", sj)
	}
	// A DSN that cannot be parsed must not put itself into the error: that is the same leak by
	// another door.
	_, _, err = dumpArgs("://not a dsn:hunter2@", "/tmp/x")
	if err == nil || strings.Contains(err.Error(), "hunter2") {
		t.Errorf("the unparseable DSN leaked into the message: %v", err)
	}
}

// The restore instructions travel INSIDE the point, because a restore happens on the worst day this
// deployment has and possibly from a copy that travelled away from this repository.
func TestAPointCarriesItsOwnRestoreInstructions(t *testing.T) {
	root := t.TempDir()
	svc, _ := service(t, []pgstore.BookDirectory{aBook(t, root, "bk_one")}, &fakeEngine{}, false)
	res, err := svc.Take(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	m, err := ReadManifest(res.Path)
	if err != nil {
		t.Fatal(err)
	}
	joined := strings.Join(m.Notes, "\n")
	for _, want := range []string{"pg_restore", ProjectDBName, "project_db"} {
		if !strings.Contains(joined, want) {
			t.Errorf("the notes do not say how to restore (%s missing): %v", want, m.Notes)
		}
	}
}