package readmodel

import (
	"context"
	"errors"
	"fmt"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"testing"
	"time"

	"textmachine/platform/internal/ingest"
	"textmachine/platform/internal/money"
	"textmachine/platform/internal/pgstore"
)

type fakeEngine struct {
	manifest    ingest.Manifest
	export      ingest.Export
	manifestErr error
	exportErr   error
	// noEnvelope models an engine build from before the artifacts envelope (D39.158): its manifest
	// names no bank read-out place at all.
	noEnvelope bool
}

func (f *fakeEngine) Manifest(_ context.Context, _, workdir string) (ingest.Manifest, error) {
	m := f.manifest
	if !f.noEnvelope {
		// Derived from the workdir the way the real engine derives it — beside the project database.
		m.Artifacts = ingest.StatusArtifacts{BankExport: filepath.Join(workdir, "bk_1.db.bank.json")}
	}
	return m, f.manifestErr
}

func (f *fakeEngine) Export(context.Context, string, string) (ingest.Export, error) {
	return f.export, f.exportErr
}

type fakeStore struct {
	structure pgstore.Structure
	saved     bool
	bank      []ingest.BankTerm
	bankSaved bool
	owed      []pgstore.OwedBook
	cleared   []pgstore.OwedBook
	deferred  []pgstore.OwedBook
	// deferredUntil is WHEN each deferral pushed the debt to. It is the whole of the fix the register
	// row is about: `now()` put the book back at the head of the queue fifteen seconds later.
	deferredUntil []time.Time
	deferCost     []pgstore.AttemptCost
	attempts      int
	abandoned     []pgstore.OwedBook
	abandonReason string
	claimed       []pgstore.OwedBook
	// unclaimable models a debt somebody else already holds.
	unclaimable bool
	// beforeClear runs just before the discharge, so a test can spend the caller's context exactly
	// where the reads would have spent it; clearCtxErr is what the discharge saw.
	beforeClear func()
	clearCtxErr error
}

func (f *fakeStore) SaveStructure(_ context.Context, _ string, in pgstore.Structure) error {
	f.structure, f.saved = in, true
	return nil
}

func (f *fakeStore) SaveBank(_ context.Context, _ string, terms []ingest.BankTerm) error {
	f.bank, f.bankSaved = terms, true
	return nil
}

func (f *fakeStore) BooksOwedReadModel(_ context.Context, limit int) ([]pgstore.OwedBook, error) {
	if len(f.owed) > limit {
		return f.owed[:limit], nil
	}
	return f.owed, nil
}

func (f *fakeStore) ClaimReadModelDebt(_ context.Context, bookID string, owedAt time.Time, window time.Duration) (time.Time, error) {
	if f.unclaimable {
		return time.Time{}, nil
	}
	f.claimed = append(f.claimed, pgstore.OwedBook{ID: bookID, OwedAt: owedAt})
	return owedAt.Add(window), nil
}

func (f *fakeStore) DeferReadModelDebt(_ context.Context, bookID string, owedAt, next time.Time, reason string, cost pgstore.AttemptCost) (int, error) {
	f.deferred = append(f.deferred, pgstore.OwedBook{ID: bookID, OwedAt: owedAt})
	f.deferredUntil = append(f.deferredUntil, next)
	f.deferCost = append(f.deferCost, cost)
	if cost == pgstore.SpendsAnAttempt {
		f.attempts++
	}
	return f.attempts, nil
}

func (f *fakeStore) AbandonReadModelDebt(_ context.Context, bookID string, owedAt, _ time.Time, reason string) error {
	f.abandoned = append(f.abandoned, pgstore.OwedBook{ID: bookID, OwedAt: owedAt})
	f.abandonReason = reason
	return nil
}

func (f *fakeStore) ClearReadModelDebt(ctx context.Context, bookID string, owedAt time.Time) error {
	if f.beforeClear != nil {
		f.beforeClear()
	}
	f.clearCtxErr = ctx.Err()
	f.cleared = append(f.cleared, pgstore.OwedBook{ID: bookID, OwedAt: owedAt})
	return nil
}

// owedBook is the book every test below refreshes: an id, a workdir and the boundary that owes it.
func owedBook(workdir string) pgstore.OwedBook {
	return pgstore.OwedBook{ID: "bk_1", Workdir: workdir, OwedAt: boundary}
}

var boundary = time.Date(2026, 8, 20, 12, 0, 0, 0, time.UTC)

func tree() ingest.Manifest {
	return ingest.Manifest{
		Version: "tm-manifest-v2", Key: "k1", ChaptersTotal: 2, UnitsTotal: 3,
		Chapters: []ingest.ManifestChapter{
			{ID: "c1", Number: 1, UnitsTotal: 2, Units: []ingest.ManifestUnit{
				{ID: "c1:cut:0", FirstChunkIdx: 0}, {ID: "c1:cut:4", FirstChunkIdx: 4},
			}},
			{ID: "c2", Number: 2, UnitsTotal: 1, Units: []ingest.ManifestUnit{{ID: "c2:cut:0", FirstChunkIdx: 0}}},
		},
	}
}

// The two documents are joined by the key the engine publishes for exactly this — the chapter's
// ordinal and the unit's LEADER chunk index — and a pair the export says nothing about stays
// `pending` with no text rather than inheriting a neighbour's.
//
// Mutation caught: joining by position in the array instead of by (chapter, first_chunk_idx).
func TestTheTextIsJoinedOntoTheTreeByTheEnginesOwnKey(t *testing.T) {
	store := &fakeStore{}
	svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{
		manifest:   tree(),
		noEnvelope: true,
		export: ingest.Export{TotalUnits: 3, Units: []ingest.UnitText{
			{Chapter: 1, Unit: 4, Source: "第二节", Target: "Второй", State: ingest.StateTranslated},
			{Chapter: 2, Unit: 0, Source: "第三节", State: ingest.StateWithheld},
		}},
	}}
	// ⚠ The BANK half fails here — this engine publishes no artifacts envelope — and the tree lands
	// anyway: each channel is a different question about the same book, and answering two of three
	// is strictly better than answering none.
	if err := svc.Refresh(t.Context(), owedBook(t.TempDir())); err == nil {
		t.Error("a workdir with no bank read-out reported no failure at all")
	}
	if !store.saved || len(store.structure.Chapters) != 2 {
		t.Fatalf("structure: %+v", store.structure)
	}
	first := store.structure.Chapters[0]
	if first.Units[0].Ordinal != 0 || first.Units[0].State != ingest.StatePending || first.Units[0].Target != "" {
		t.Errorf("a pair the export did not mention: %+v", first.Units[0])
	}
	if first.Units[1].Target != "Второй" || first.Units[1].State != ingest.StateTranslated {
		t.Errorf("the pair at leader index 4: %+v", first.Units[1])
	}
	second := store.structure.Chapters[1]
	if second.Units[0].Source != "第三节" || second.Units[0].State != ingest.StateWithheld {
		t.Errorf("the pair of the second chapter: %+v", second.Units[0])
	}
	if store.structure.ManifestKey != "k1" {
		t.Errorf("the validity key did not travel: %q", store.structure.ManifestKey)
	}
}

// A book that has never been translated has a tree and no translations, and so does a book whose
// export this build could not read. The TREE lands either way: a reader screen that stayed empty
// because the text channel failed would hide a book the user can already navigate.
//
// Mutation caught: returning early when Export fails; claiming to KNOW the text that failed to read.
func TestAFailedTextReadStillLandsTheTree(t *testing.T) {
	store := &fakeStore{}
	svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{
		manifest: tree(), exportErr: errors.New("the engine refused"),
	}}
	_ = svc.Refresh(t.Context(), owedBook(t.TempDir()))
	if !store.saved {
		t.Fatal("a failed text read cost the whole tree")
	}
	for _, c := range store.structure.Chapters {
		for _, u := range c.Units {
			if u.State != ingest.StatePending || u.Target != "" {
				t.Errorf("a pair claims a translation nothing produced: %+v", u)
			}
			// …and it says so, which is what stops the store overwriting the stored text.
			if u.TextKnown {
				t.Errorf("a pair claims its text is known after the export failed: %+v", u)
			}
		}
	}
}

// The two engine calls are two re-cuts of the same source, and they can disagree: a pair in the
// manifest that the export did not carry is a SKEW, not an empty pair, and must not be written as
// one over text already materialized.
//
// Mutation caught: setting TextKnown for every pair of a successful export.
func TestAPairTheExportDidNotCarryDoesNotSpeakAboutItsText(t *testing.T) {
	// The manifest has three pairs; the export carries ONE of them, as a cut that moved between the
	// two calls would leave it.
	store := &fakeStore{}
	svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{
		manifest: tree(),
		export: ingest.Export{TotalUnits: 3, Units: []ingest.UnitText{
			{Chapter: 1, Unit: 0, Source: "第一节", Target: "Первый", State: ingest.StateTranslated},
		}},
	}}
	_ = svc.Refresh(t.Context(), owedBook(t.TempDir()))
	var known, unknown int
	for _, c := range store.structure.Chapters {
		for _, u := range c.Units {
			if u.TextKnown {
				known++
			} else {
				unknown++
			}
		}
	}
	if known != 1 || unknown == 0 {
		t.Errorf("%d pairs claim known text and %d do not, want only the exported one", known, unknown)
	}
}

// A book with no bank read-out has never produced terms, and that is not an empty bank: saving one
// would erase a bank the engine simply has not written yet.
//
// Mutation caught: treating ErrNoBank as an empty read-out.
func TestAMissingBankReadOutSavesNothing(t *testing.T) {
	// ⚠ The engine PUBLISHED the read-out's place (the artifacts envelope, row 213) and no file is
	// there yet — which is the branch this pins. An engine that published no place at all is the
	// loud failure next door, and telling the two apart is what the envelope is for.
	store := &fakeStore{}
	svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifest: tree()}}
	if err := svc.Refresh(t.Context(), owedBook(t.TempDir())); err != nil {
		t.Fatalf("a book that has never produced terms is not a failure: %v", err)
	}
	if store.bankSaved {
		t.Error("an absent bank read-out was saved as an empty bank")
	}
	if !store.saved {
		t.Error("the tree was not materialized, though only the bank was missing")
	}
}

// An engine that publishes no artifacts envelope — a build from before D39.158 — is a different
// fact, and it IS a failure: the bank's place cannot be located, and reading that as "no bank"
// would silently stop every bank refresh on such a deployment.
func TestAnEngineWithoutTheEnvelopeIsAFailureRatherThanAnEmptyBank(t *testing.T) {
	store := &fakeStore{}
	svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifest: tree(), noEnvelope: true}}
	if err := svc.Refresh(t.Context(), owedBook(t.TempDir())); err == nil {
		t.Fatal("a manifest naming no bank read-out place was read as a book with no bank")
	}
	if store.bankSaved {
		t.Error("a missing envelope saved a bank anyway")
	}
}

// garbageBank is a workdir whose published bank read-out exists and is not a bank — the one way a
// book's bank channel fails while its neighbours' answer fine.
func garbageBank(t *testing.T) string {
	t.Helper()
	dir := t.TempDir()
	if err := os.WriteFile(filepath.Join(dir, "bk_1.db.bank.json"), []byte("not a bank"), 0o644); err != nil {
		t.Fatal(err)
	}
	return dir
}

// The debt is what brings the materializer back to a book, so only a materialization that answered
// EVERY channel may discharge it. A tree with no text is not a materialized book: before this the
// text channel's failure was logged and the caller was told the book was done.
//
// Mutation caught: returning nil from refreshStructure when Export fails.
func TestAPartialMaterializationDoesNotDischargeTheDebt(t *testing.T) {
	store := &fakeStore{}
	svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{
		manifest: tree(), exportErr: errors.New("the engine refused"),
	}}
	if err := svc.Refresh(t.Context(), owedBook(t.TempDir())); err == nil {
		t.Fatal("a materialization with no text reported success")
	}
	if !store.saved {
		t.Fatal("this fixture never reached the write, so it cannot observe the property")
	}
	if len(store.cleared) != 0 {
		t.Errorf("a partial materialization discharged the debt: %+v", store.cleared)
	}
}

// And a complete one discharges THAT boundary, carried through unchanged. Whether the STORE then
// refuses a debt stamped since is pgstore's own property and is pinned there, on a live database —
// here what is checked is that the boundary reaches it at all.
//
// Mutation caught: discharging with the current time instead of the boundary the caller was given.
func TestACompleteMaterializationDischargesTheBoundaryItRead(t *testing.T) {
	store := &fakeStore{}
	svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{
		manifest: tree(),
		export: ingest.Export{TotalUnits: 3, Units: []ingest.UnitText{
			{Chapter: 1, Unit: 0, Source: "第一节", Target: "Первый", State: ingest.StateTranslated},
		}},
	}}
	if err := svc.Refresh(t.Context(), owedBook(t.TempDir())); err != nil {
		t.Fatal(err)
	}
	if len(store.cleared) != 1 || store.cleared[0].ID != "bk_1" || !store.cleared[0].OwedAt.Equal(boundary) {
		t.Errorf("the debt discharged: %+v, want the boundary this read answered", store.cleared)
	}
}

// A materialization that HAPPENED is written down even when the context that paid for the reads is
// already spent. The reads legitimately consume the whole budget on a large book, and a discharge
// that then fails leaves the debt standing — so the next pass re-does two full re-chunks of the same
// source, and every pass after it.
//
// Mutation caught: discharging on the caller's own context.
func TestTheDischargeSurvivesAContextTheReadsUsedUp(t *testing.T) {
	store := &fakeStore{}
	svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{
		manifest: tree(),
		export: ingest.Export{TotalUnits: 3, Units: []ingest.UnitText{
			{Chapter: 1, Unit: 0, Source: "第一节", Target: "Первый", State: ingest.StateTranslated},
		}},
	}}
	ctx, cancel := context.WithCancel(t.Context())
	store.beforeClear = cancel // the budget runs out exactly between the last read and the record
	if err := svc.Refresh(ctx, owedBook(t.TempDir())); err != nil {
		t.Fatal(err)
	}
	if len(store.cleared) != 1 {
		t.Fatalf("the debt was not discharged at all: %+v", store.cleared)
	}
	if store.clearCtxErr != nil {
		t.Errorf("the discharge ran on a context that was already over: %v", store.clearCtxErr)
	}
}

// The drain is the only retry there is, so it must reach every owed book and must not let one book's
// failure end the pass: these are independent books of independent accounts.
//
// Mutation caught: returning on the first error.
func TestTheDrainMaterializesEveryOwedBookAndKeepsWhatFailed(t *testing.T) {
	store := &fakeStore{owed: []pgstore.OwedBook{
		{ID: "bk_1", Workdir: garbageBank(t), OwedAt: boundary}, // its read-out is not a bank: the read fails
		{ID: "bk_2", Workdir: t.TempDir(), OwedAt: boundary},
	}}
	svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifest: tree()}}
	if err := svc.Drain(t.Context()); err != nil {
		t.Fatal(err)
	}
	if len(store.cleared) != 1 || store.cleared[0].ID != "bk_2" {
		t.Errorf("discharged %+v, want only the book that was materialized whole", store.cleared)
	}
	// …and the one that failed goes to the BACK of the queue. The list is oldest-first, so a book the
	// engine can never answer about would hold the front of it forever and the books behind it would
	// never be materialized at all.
	if len(store.deferred) != 1 || store.deferred[0].ID != "bk_1" {
		t.Errorf("deferred %+v, want the book whose materialization failed", store.deferred)
	}
}

// A deployment with no engine binary is a read replica, not a broken one: a refresh is a no-op and
// the surface it serves is whatever was materialized before.
func TestWithoutAnEngineARefreshIsANoOp(t *testing.T) {
	store := &fakeStore{}
	svc := &Service{Store: store, Engine: &fakeEngine{manifest: tree()}}
	if err := svc.Refresh(t.Context(), owedBook("/srv/books/bk_1")); err != nil {
		t.Fatal(err)
	}
	if store.saved || store.bankSaved {
		t.Error("a deployment with no engine binary materialized something")
	}
	// …and it does not discharge the debt either: the work belongs to an instance that has an engine.
	if len(store.cleared) != 0 {
		t.Errorf("a replica cleared a debt it did not pay: %+v", store.cleared)
	}
}

// The engine's manifest carries a `heading` and the contract forbids a deployment to project it —
// it is a rendered ordinal («Глава N»), not a label out of the book's data. The allowlist is
// enforced by the DECODER: there is nowhere for the field to land.
//
// Mutation caught: adding a Heading field to ingest.ManifestChapter.
func TestTheManifestsRenderedHeadingHasNowhereToLand(t *testing.T) {
	doc := `{"manifest_version":"tm-manifest-v2","key":"k","chapters":[
		{"id":"c1","number":1,"heading":"Глава 1","units_total":1,
		 "units":[{"id":"c1:cut:0","first_chunk_idx":0}]}]}`
	m, err := ingest.DecodeManifest([]byte(doc))
	if err != nil {
		t.Fatal(err)
	}
	if len(m.Chapters) != 1 || m.Chapters[0].ID != "c1" {
		t.Fatalf("manifest: %+v", m)
	}
	// Whatever the engine renders, this side has no field holding it: the check is over the decoded
	// STRUCT, so a field added later fails here rather than on a reader's screen.
	if rendered := fmt.Sprintf("%#v", m.Chapters[0]); strings.Contains(rendered, "Глава") {
		t.Errorf("the engine's rendered ordinal survived the decode: %s", rendered)
	}
}

// Two materializations of one book must not run at once. The intake pays its own debt inline while
// the sweep drains the rest every few seconds, so without a claim every upload slower than one sweep
// interval was read by both at the same time — two full re-chunks of the source, which is the cost
// this package removed elsewhere.
//
// Mutation caught: draining without claiming; ignoring a claim somebody else holds.
func TestTheDrainSkipsABookSomebodyElseIsAlreadyMaterializing(t *testing.T) {
	store := &fakeStore{owed: []pgstore.OwedBook{{ID: "bk_1", Workdir: t.TempDir(), OwedAt: boundary}}}
	svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifest: tree()}}
	store.unclaimable = true
	if err := svc.Drain(t.Context()); err != nil {
		t.Fatal(err)
	}
	if store.saved {
		t.Error("the drain read the engine for a book it does not hold the debt of")
	}
	// …and it DOES take a debt nobody holds — the claim is a lease, not a refusal.
	store.unclaimable = false
	if err := svc.Drain(t.Context()); err != nil {
		t.Fatal(err)
	}
	if len(store.claimed) != 1 || !store.saved {
		t.Errorf("claimed %+v, saved=%v — want the free debt taken and paid", store.claimed, store.saved)
	}
	// The discharge presents the stamp the CLAIM produced, not the one the queue was listed with: the
	// claim moved it, and a discharge holding the old value would clear nothing.
	if len(store.cleared) != 1 || !store.cleared[0].OwedAt.Equal(boundary.Add(MaterializeBudget)) {
		t.Errorf("discharged %+v, want the stamp the claim produced", store.cleared)
	}
}

// A manifest whose lists are shorter than its own counts NEVER reaches the store.
//
// The tree is written by replacement — `SaveStructure` deletes every chapter outside the list it is
// handed and the cascade takes the text with it — so an empty list is the deletion of the whole
// book. No engine produces one (`buildManifest` fills the chapters and the file is written
// atomically), which is exactly why the danger is silent: the case this refuses is not a broken
// engine but a document THIS build read wrongly, and the counts printed beside the lists are the
// only witness it has of that.
//
// The asymmetry that made it a defect: the intake refuses precisely this document and has since P6,
// because there "no chapters" is the verdict that deletes a user's upload. The materializer's write
// is as destructive and checked nothing.
//
// Mutation caught: dropping either floor from refreshStructure — the self-description check or the
// "no chapters at all" one.
func TestAManifestThatDoesNotDescribeItselfNeverReachesTheTree(t *testing.T) {
	for name, broken := range map[string]func(*ingest.Manifest){
		"a chapter list this build could not read": func(m *ingest.Manifest) { m.Chapters = nil },
		"one chapter short":                        func(m *ingest.Manifest) { m.Chapters = m.Chapters[:1] },
		"a unit list this build could not read":    func(m *ingest.Manifest) { m.Chapters[0].Units = nil },
		"counts no chapters at all": func(m *ingest.Manifest) {
			m.Chapters, m.ChaptersTotal, m.UnitsTotal = nil, 0, 0
		},
		// The BOOK-level pair count disagreeing with the sum over chapters. Its own case because the
		// per-chapter check cannot see it: every chapter can describe itself correctly while the
		// document as a whole says a different number of pairs — which is what a renamed field does.
		"counts pairs the chapters do not carry": func(m *ingest.Manifest) { m.UnitsTotal = 99 },
		// …and a chapter numbered from zero, which chapter numbering cannot produce.
		"a chapter numbered zero": func(m *ingest.Manifest) { m.Chapters[0].Number = 0 },
		// A per-chapter count that is wrong while the BOOK's total still adds up. Its own case because
		// nothing else reaches it: every other broken shape here also breaks the book-level sum, so the
		// per-chapter equality could be deleted and the battery would not notice.
		"chapter counts that swap while the book's total holds": func(m *ingest.Manifest) {
			m.Chapters[0].UnitsTotal, m.Chapters[1].UnitsTotal = 1, 2
		},
		// The identities, which no counter stands beside: a renamed key decodes to "" for every row and
		// the replacement write then keeps one of them.
		"a chapter with no id": func(m *ingest.Manifest) { m.Chapters[0].ID = "" },
		"a pair with no id":    func(m *ingest.Manifest) { m.Chapters[0].Units[1].ID = "" },
		"no identities at all, as a renamed key decodes": func(m *ingest.Manifest) {
			for i := range m.Chapters {
				m.Chapters[i].ID = ""
				for j := range m.Chapters[i].Units {
					m.Chapters[i].Units[j].ID = ""
				}
			}
		},
	} {
		t.Run(name, func(t *testing.T) {
			m := tree()
			broken(&m)
			store := &fakeStore{}
			svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifest: m}}
			err := svc.Refresh(t.Context(), owedBook(t.TempDir()))
			if err == nil {
				t.Fatal("the refresh reported success")
			}
			if store.saved {
				t.Errorf("the tree was rewritten from a document that does not describe itself: %+v", store.structure)
			}
			// …and the debt is NOT discharged: the reader keeps the tree it has and the book comes back.
			if len(store.cleared) != 0 {
				t.Errorf("the debt was discharged by a materialization that wrote nothing: %+v", store.cleared)
			}
		})
	}
}

// …and the same document reaching through the INTAKE's door is refused too. `RefreshCut` exists to
// save a third re-chunk per upload by passing the manifest the intake already decoded, and a floor
// on only one of the two entrances is a floor on neither.
func TestTheIntakesOwnCutIsHeldToTheSameFloor(t *testing.T) {
	m := tree()
	m.Chapters = nil
	store := &fakeStore{}
	svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifest: tree()}}
	if err := svc.RefreshCut(t.Context(), owedBook(t.TempDir()), m); err == nil {
		t.Fatal("the cut handed in by the intake was written without being checked")
	}
	if store.saved {
		t.Errorf("the tree was rewritten from the intake's own broken cut: %+v", store.structure)
	}
}

// A host that cannot run the engine does not write off every book on it.
//
// The attempt budget answers "this book cannot be read". An engine binary that is missing, locked
// out by another process or refusing an unmigrated schema says nothing about any book and applies to
// all of them at once — so counting it would abandon the whole library within a few passes, each
// book keeping whatever stale surface it had. The intake holds the same rule (books.defer_).
//
// Mutation caught: spending an attempt on a deployment fault; not deferring one at all.
func TestABrokenDeploymentDoesNotSpendABooksAttempts(t *testing.T) {
	for _, c := range []struct {
		name string
		err  error
	}{
		{"the binary cannot be run", fmt.Errorf("readmodel: read the manifest: %w", exec.ErrNotFound)},
		{"the project is locked", exitError(t, ingest.ExitProjectLocked)},
		{"the schema is not migrated", exitError(t, ingest.ExitSchemaMismatch)},
	} {
		t.Run(c.name, func(t *testing.T) {
			store := &fakeStore{owed: []pgstore.OwedBook{
				// One failure short of the write-off: if this pass counts, the book is given up on.
				{ID: "bk_1", Workdir: t.TempDir(), OwedAt: boundary, Attempts: maxAttempts - 1},
			}}
			svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifestErr: c.err}}
			if err := svc.Drain(t.Context()); err != nil {
				t.Fatal(err)
			}
			if len(store.abandoned) != 0 {
				t.Errorf("a book was given up on over a fault of the host: every book on it would be, within a few passes")
			}
			if len(store.deferred) != 1 {
				t.Fatalf("the debt was not put back: %+v", store.deferred)
			}
			if store.deferCost[0] != pgstore.CostsNoAttempt {
				t.Errorf("the deployment's fault was charged to the book's budget")
			}
		})
	}
}

// exitError is what an engine that ANSWERED with a refusal looks like from the runner: the platform
// tells that from "could not be run" by the type, so a fixture has to produce a real one.
func exitError(t *testing.T, code int) error {
	t.Helper()
	err := exec.CommandContext(t.Context(), "sh", "-c", "exit "+strconv.Itoa(code)).Run()
	var exit *exec.ExitError
	if !errors.As(err, &exit) || exit.ExitCode() != code {
		t.Fatalf("the fixture could not produce exit %d: %v", code, err)
	}
	return fmt.Errorf("readmodel: read the manifest: %w", err)
}

// A materialization that failed waits a REAL interval, and after enough of them the debt is given up
// on rather than retried for the life of the deployment.
//
// What was there before is the whole defect: claim, fail, `DeferReadModelDebt` writing `now()`, and
// the sweep running again fifteen seconds later — so "the back of the queue" was a queue of one and
// the book was re-read by the engine four times a minute, forever. Three things followed and every
// one of them was ongoing: up to five minutes of engine processes per pass; a book whose event
// stream can NEVER end, because `AtRest` requires the debt to be null and a browser reconnects to it
// for good; and, where the manifest reads and the export does not, a committed write on every pass —
// a revision bump and a frame every fifteen seconds to say nothing changed.
//
// Giving up is safe HERE in a way it never is for a run: the money of the boundary that stamped this
// debt has already settled, so what is lost is freshness of text. It is also not final — the next
// boundary of real work stamps a fresh debt (owesAReadingSurface resets the budget).
//
// Mutation caught: deferring to `now()`; not counting the attempt; never abandoning; abandoning on
// the first failure.
func TestAnUnpayableDebtBacksOffAndIsEventuallyGivenUpOn(t *testing.T) {
	// A book whose published bank read-out exists and is not a bank: the read fails on every pass, so
	// the materialization never completes — the shape of a project directory an operator broke.
	broken := garbageBank(t)
	for attempts := range maxAttempts {
		store := &fakeStore{owed: []pgstore.OwedBook{
			{ID: "bk_1", Workdir: broken, OwedAt: boundary, Attempts: attempts},
		}}
		svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifest: tree()}}
		before := time.Now()
		if err := svc.Drain(t.Context()); err != nil {
			t.Fatal(err)
		}
		if len(store.cleared) != 0 {
			t.Fatalf("attempt %d: a failed materialization discharged the debt: %+v", attempts, store.cleared)
		}
		last := attempts+1 >= maxAttempts
		switch {
		case last && len(store.abandoned) != 1:
			t.Errorf("after %d failures the debt is still being retried; nothing ever ends this loop", attempts+1)
		case last && store.abandonReason == "":
			t.Error("the debt was written off with no reason recorded: an operator cannot tell why")
		case !last && len(store.abandoned) != 0:
			t.Errorf("the debt was given up on after %d failures, before its budget was spent", attempts+1)
		case !last && len(store.deferredUntil) != 1:
			t.Fatalf("attempt %d: deferred %d times, want once", attempts, len(store.deferredUntil))
		case !last && !store.deferredUntil[0].After(before.Add(30*time.Second)):
			// The number that matters: a deferral shorter than the sweep interval is not a deferral.
			t.Errorf("attempt %d: the debt was pushed to %v, barely past %v — the next pass picks it straight back up",
				attempts, store.deferredUntil[0], before)
		case !last && !store.deferredUntil[0].After(before.Add(retryIn(attempts+1)-time.Second)):
			// …and it GROWS with the count. Asserting the function alone leaves the CALL free to hand it
			// a constant, which is a flat one-minute retry wearing the shape of a backoff.
			t.Errorf("attempt %d: deferred to %v, which is not the %v this many failures have earned",
				attempts, store.deferredUntil[0], retryIn(attempts+1))
		}
	}
}

// …and the delay grows and stops growing. Without a ceiling a book that healed would wait days;
// without growth the first retry is the only one that costs little.
func TestTheMaterializationBackoffGrowsAndIsCapped(t *testing.T) {
	if retryIn(1) != time.Minute {
		t.Errorf("the first retry is %v, want a minute", retryIn(1))
	}
	if retryIn(3) <= retryIn(1) {
		t.Errorf("the retry does not grow: %v then %v", retryIn(1), retryIn(3))
	}
	if got := retryIn(1000); got != 30*time.Minute {
		t.Errorf("the retry of a long-dead book is %v, want the cap", got)
	}
}

// ⛔ THE SEAM BETWEEN THE ENGINE'S DOCUMENT AND THE STORE'S ROW, and it went untested until it broke.
//
// Both halves had tests: `ingest` decides whether a manifest is priced, `pgstore` writes what it is
// handed. What nobody asked was whether this function hands over the RIGHT THING — and the answer was
// no. `pgstore.Structure` carried the book's character count in two places at once (its own field and
// the wire struct's), this mapping filled one of them, and the row went in null. The type no longer
// admits that, so what is pinned here is the mapping's SHAPE: every money figure arrives, and the
// character count arrives on the CUT, where it survives a price this build cannot read.
//
// Mutation caught: dropping a member of the Projection literal; reading SourceChars off the price
// object again; gating the character count on `Priced`.
func TestTheProjectionReachesTheStoreWholeAndTheCharacterCountTravelsBesideIt(t *testing.T) {
	m := tree()
	m.Structure = ingest.StructureDetected
	m.Price = &ingest.BookPrice{ExpectedUSD: 2_090_000, BookOnceUSD: 2_000_000, StepMaxUSD: 69_828, SourceChars: 3000}
	pricedTree(&m)
	store := &fakeStore{}
	svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifest: m, noEnvelope: true}}
	_ = svc.Refresh(t.Context(), owedBook(t.TempDir())) // the bank half fails; the tree lands anyway
	if !store.saved {
		t.Fatal("the tree did not land at all")
	}
	got := store.structure
	if got.Price == nil {
		t.Fatal("a manifest this build read whole reached the store with no projection")
	}
	if got.Price.Expected != 2_090_000 || got.Price.BookOnce != 2_000_000 || got.Price.StepMax != 69_828 {
		t.Errorf("a money figure was lost in the mapping: %+v", *got.Price)
	}
	// ⛔ ON THE CUT, and this is the assertion the defect would have failed: the count is a property of
	// the TEXT and the store keeps it apart from the money.
	if got.SourceChars != 3000 {
		t.Errorf("the engine's rune count reached the store as %d", got.SourceChars)
	}
	if got.Structure != ingest.StructureDetected {
		t.Errorf("the cut's provenance reached the store as %q", got.Structure)
	}
}

// …and the other direction of the same independence: a manifest whose MONEY this build cannot read
// still delivers the count of its text. The refusal is `ingest`'s (a chapter with no price), so what
// is pinned here is that this mapping does not widen it into the character count.
func TestAPriceThisBuildCannotReadDoesNotCostTheStoreItsCharacterCount(t *testing.T) {
	m := tree()
	m.Structure = ingest.StructureDetected
	m.Price = &ingest.BookPrice{ExpectedUSD: 2_090_000, BookOnceUSD: 2_000_000, StepMaxUSD: 69_828, SourceChars: 3000}
	pricedTree(&m)
	// One unit loses its bill, and `ingest` then refuses the projection as a whole — the half-read
	// shape is the dangerous one. The book's own `source_chars` is untouched and must still travel.
	m.Chapters[0].Units[0].Price = nil
	store := &fakeStore{}
	svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifest: m, noEnvelope: true}}
	_ = svc.Refresh(t.Context(), owedBook(t.TempDir()))
	if store.structure.Price != nil {
		t.Errorf("a projection this build could not read whole was written anyway: %+v", *store.structure.Price)
	}
	if store.structure.SourceChars != 3000 {
		t.Errorf("the character count went down with the price: %d", store.structure.SourceChars)
	}
}

// pricedTree gives `tree()` a projection its own witness accepts: 30_000 micro-USD and 1000 runes a
// unit, the chapters' roll-up summing to the book's figure less the flat book-level bond.
func pricedTree(m *ingest.Manifest) {
	for i := range m.Chapters {
		total := money.MicroUSD(0)
		for j := range m.Chapters[i].Units {
			m.Chapters[i].Units[j].Price = &ingest.UnitPrice{ExpectedUSD: 30_000, SourceChars: 1000}
			total += 30_000
		}
		m.Chapters[i].Price = &ingest.UnitPrice{ExpectedUSD: total, SourceChars: int64(len(m.Chapters[i].Units)) * 1000}
	}
}