textmachine/platform/internal/readmodel/readmodel_test.go

package readmodel

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

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

type fakeEngine struct {
	manifest    ingest.Manifest
	export      ingest.Export
	manifestErr error
	exportErr   error
}

func (f *fakeEngine) Manifest(context.Context, string, string) (ingest.Manifest, error) {
	return f.manifest, 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
	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 time.Time) error {
	f.deferred = append(f.deferred, pgstore.OwedBook{ID: bookID, OwedAt: owedAt})
	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(),
		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 workdir has no engine artifacts — 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},
		}},
	}}
	// The bank half fails on an empty workdir, which is not what this pins.
	_ = 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 book IS configured and simply has no bank sidecar yet — which is the branch this pins.
	// Without the configuration the read failed one step earlier, on the missing book.yaml, and the
	// declared property was never reached: "a read-out that has not been made must not erase one that
	// was" was asserted by a test that could not tell the two failures apart.
	dir := t.TempDir()
	if err := os.WriteFile(filepath.Join(dir, "book.yaml"),
		[]byte("book_id: bk_1\nproject_db: bk_1.db\n"), 0o644); err != nil {
		t.Fatal(err)
	}
	store := &fakeStore{}
	svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifest: tree()}}
	if err := svc.Refresh(t.Context(), owedBook(dir)); 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")
	}
}

// A workdir with no configuration at all is a different fact, and it IS a failure: nothing about
// this book can be located, so the refresh says so instead of reporting an empty book.
func TestAWorkdirWithNoConfigurationIsAFailureRatherThanAnEmptyBank(t *testing.T) {
	store := &fakeStore{}
	svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifest: tree()}}
	if err := svc.Refresh(t.Context(), owedBook(t.TempDir())); err == nil {
		t.Fatal("a workdir with no configuration was read as a book with no bank")
	}
	if store.bankSaved {
		t.Error("an unreadable configuration saved a bank anyway")
	}
}

// configured is a workdir the bank channel can answer about: the book exists and has simply never
// produced terms, so the read-out is absent rather than unreadable.
func configured(t *testing.T) string {
	t.Helper()
	dir := t.TempDir()
	if err := os.WriteFile(filepath.Join(dir, "book.yaml"),
		[]byte("book_id: bk_1\nproject_db: bk_1.db\n"), 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(configured(t))); 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(configured(t))); 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(configured(t))); 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: t.TempDir(), OwedAt: boundary}, // no configuration: the bank read fails
		{ID: "bk_2", Workdir: configured(t), 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: configured(t), 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)
	}
}