textmachine/platform/internal/readmodel/readmodel.go

// Package readmodel materializes what the client READS: a book's chapter tree with the text of
// every pair, and its memory bank.
//
// It is the consumer half of three engine channels and it is deliberately ONE place, because the
// three have to agree with each other: the manifest carries the structure and the identities, the
// export carries the text and the state derived from it, and the bank sidecar carries the terms. A
// tree written from one of them and text written from another would drift apart at exactly the
// moment they disagree — a book cut again between two calls.
//
// WHEN it runs is the contract's own answer, not a schedule: "`target` is updated at the boundaries
// of the work and at stops, not continuously" (canon §Unit). Those boundaries are the end of an
// intake and the end of a run, and they are also the only moments the engine's project is not held
// exclusively by a translation.
package readmodel

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"time"

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

// Engine is what this needs from the engine's $0 read commands.
type Engine interface {
	Manifest(ctx context.Context, binary, workdir string) (ingest.Manifest, error)
	Export(ctx context.Context, binary, workdir string) (ingest.Export, error)
}

// Store is the write side of the read model, plus the ledger of what it still owes.
type Store interface {
	SaveStructure(ctx context.Context, bookID string, in pgstore.Structure) error
	SaveBank(ctx context.Context, bookID string, terms []ingest.BankTerm) error
	BooksOwedReadModel(ctx context.Context, limit int) ([]pgstore.OwedBook, error)
	ClaimReadModelDebt(ctx context.Context, bookID string, owedAt time.Time, window time.Duration) (time.Time, error)
	ClearReadModelDebt(ctx context.Context, bookID string, owedAt time.Time) error
	DeferReadModelDebt(ctx context.Context, bookID string, owedAt, next time.Time, reason string, cost pgstore.AttemptCost) (int, error)
	AbandonReadModelDebt(ctx context.Context, bookID string, owedAt, now time.Time, reason string) error
}

// MaterializeBudget is what ONE book's materialization may take. Generous because it is two full
// re-chunks of the source, and bounded because a host that cannot run the engine must not spend a
// whole pass finding that out on one book.
const MaterializeBudget = 5 * time.Minute

// booksPerDrain bounds one drain pass, for the same reason the budget above bounds one book.
const booksPerDrain = 4

// dischargeBudget is what writing DOWN a finished materialization gets. Short: it is one statement
// against a database this process is already connected to.
const dischargeBudget = 30 * time.Second

// maxAttempts is how many passes may fail to pay one debt before it is written off.
//
// Five, which is the intake's number (`books.parse_attempts`) and for the same reason: this is the
// same question about a different object — "how many times does a deployment try before it admits
// that trying is not the problem" — and two different answers to it would be two policies for an
// operator to learn. What was there before was no number at all: claim, fail, defer to `now()`,
// retried fifteen seconds later, forever.
const maxAttempts = 5

// retryIn is how long a failed materialization waits, doubling from a minute and capped.
//
// The cap matters as much as the growth: a book five minutes of engine time deep, retried every
// fifteen seconds, is a host that spends its whole pass on one book that cannot be read.
func retryIn(attempts int) time.Duration {
	const base, ceiling = time.Minute, 30 * time.Minute
	d := base
	for range min(attempts, 16) - 1 {
		if d >= ceiling {
			break
		}
		d *= 2
	}
	return min(d, ceiling)
}

// Service refreshes the reading surface of one book.
type Service struct {
	Store  Store
	Engine Engine
	// Binary is the versioned tmctl path. Empty means this deployment cannot read the engine at all,
	// and then a refresh is a no-op rather than a failure: an instance with no engine is a read
	// replica, and the surface it serves is whatever was materialized before.
	Binary string
	Log    *slog.Logger
}

func (s *Service) log() *slog.Logger {
	if s.Log == nil {
		return slog.New(slog.DiscardHandler)
	}
	return s.Log
}

// Claim takes a book's debt for this worker before it starts reading the engine, and answers the
// book to work on — or `ok` false when the debt is no longer the caller's to pay.
//
// It is what stops two materializations of one book running at once: the intake pays its own debt
// inline, the sweep drains the rest every few seconds, and without a claim every upload slower than
// one sweep interval was read twice over.
func (s *Service) Claim(ctx context.Context, b pgstore.OwedBook) (pgstore.OwedBook, bool, error) {
	claimed, err := s.Store.ClaimReadModelDebt(ctx, b.ID, b.OwedAt, MaterializeBudget)
	if err != nil || claimed.IsZero() {
		return pgstore.OwedBook{}, false, err
	}
	b.OwedAt = claimed
	return b, true, nil
}

// Refresh re-reads a book from the engine and writes what a client reads.
//
// The three channels are read in one order and for one reason: the STRUCTURE first, because the text
// is addressed by it and a pair whose chapter this build has not seen has nowhere to go; the bank
// last, because it is independent of both and its failure must not cost the tree.
//
// A failure of one channel does not abandon the others. Each is a different question about the same
// book, and answering two of the three is strictly better than answering none — the alternative is a
// reader screen that stays empty because a bank read-out that has never existed could not be found.
// But a partial answer does not DISCHARGE the debt: everything that failed is returned, and the book
// stays in the queue until one pass answers all three.
func (s *Service) Refresh(ctx context.Context, b pgstore.OwedBook) error {
	return s.refresh(ctx, b, nil)
}

// RefreshCut is Refresh for a caller that has ALREADY read the manifest — the intake, which decoded
// the same document a moment earlier to learn the book's chapter count. Re-reading it there is a
// third full re-chunk of the source per upload (PD-248), paid for nothing.
func (s *Service) RefreshCut(ctx context.Context, b pgstore.OwedBook, cut ingest.Manifest) error {
	return s.refresh(ctx, b, &cut)
}

func (s *Service) refresh(ctx context.Context, b pgstore.OwedBook, cut *ingest.Manifest) error {
	if s.Binary == "" || s.Engine == nil {
		// A replica with no engine serves what was materialized before, and — the debt being untouched
		// — leaves the work to an instance that has one.
		return nil
	}
	// Bounded HERE and not by each caller: the budget is a property of this work — two full re-chunks
	// of the source — rather than of whoever asks for it, and a caller with less time than that still
	// wins, since the shorter deadline is the one that fires.
	ctx, cancel := context.WithTimeout(ctx, MaterializeBudget)
	defer cancel()
	var failed error
	if err := s.refreshStructure(ctx, b.ID, b.Workdir, cut); err != nil {
		failed = err
		s.log().ErrorContext(ctx, "the chapter tree could not be refreshed", "err", err)
	}
	if err := s.refreshBank(ctx, b.ID, b.Workdir); err != nil {
		failed = errors.Join(failed, err)
		s.log().ErrorContext(ctx, "the bank could not be refreshed", "err", err)
	}
	if failed != nil {
		return failed
	}
	// The discharge is a RECORD, and it gets a context of its own — detached and short, the same rule
	// the intake's terminal writes follow. The reads above can legitimately consume the whole budget,
	// and a write that then runs on an expired context is a materialization that HAPPENED and was not
	// written down: the next pass would do two full re-chunks of the same source again.
	c, cancel := context.WithTimeout(context.WithoutCancel(ctx), dischargeBudget)
	defer cancel()
	return s.Store.ClearReadModelDebt(c, b.ID, b.OwedAt)
}

// Drain materializes the books that owe a reading surface, oldest debt first.
//
// This is the ONLY retry there is, and every way a materialization can be lost ends here: an engine
// call that failed, a process that died between the boundary and the read, an instance that has no
// engine to read with. The debt is a column, so none of those forget it.
func (s *Service) Drain(ctx context.Context) error {
	if s.Binary == "" || s.Engine == nil {
		return nil
	}
	owed, err := s.Store.BooksOwedReadModel(ctx, booksPerDrain)
	if err != nil {
		return err
	}
	for _, b := range owed {
		if deadline, ok := ctx.Deadline(); ok && time.Until(deadline) < MaterializeBudget {
			// What is left of the pass is shorter than one book. Starting it anyway spends the engine's
			// CPU on a read that will be killed; the debt keeps the book for the next pass.
			return nil
		}
		b, ok, err := s.Claim(ctx, b)
		if err != nil {
			return err
		}
		if !ok {
			continue // somebody else is paying it, or a later boundary has replaced it
		}
		if err := s.Refresh(ctx, b); err != nil {
			// Loud: the boundary is past and its text is what the reader screen shows, so a book that
			// keeps failing here is a book that finished and looks empty.
			s.log().ErrorContext(ctx, "a book still owes its reading surface", "err", err)
			s.giveUpOrRetry(ctx, b, err)
		}
	}
	return nil
}

// giveUpOrRetry decides what a failed materialization means: later, or never.
//
// The debt goes to the back of the queue with a real delay — the list is oldest-first, so a book the
// engine can never answer about held the FRONT of it forever and starved the books behind it out of
// text they were paid for, and deferring to `now()` (which is what this used to do) put it straight
// back at that front fifteen seconds later.
//
// After `maxAttempts` it is written off, and that is the half that did not exist. Giving up is safe
// here in a way it is never safe for a run: the money of this boundary has already settled, so what
// is lost is freshness of text — against three things that were lost by never giving up, all of them
// ongoing (see AbandonReadModelDebt). The next boundary of real work stamps a fresh debt, so "never"
// lasts exactly until the book does something again.
//
// The writes are DETACHED and short: the reads above are allowed to consume the whole budget, and a
// verdict recorded on an expired context is a pass that failed silently and starts over.
func (s *Service) giveUpOrRetry(ctx context.Context, b pgstore.OwedBook, cause error) {
	c, cancel := context.WithTimeout(context.WithoutCancel(ctx), dischargeBudget)
	defer cancel()
	// ⚠ A BROKEN HOST DOES NOT SPEND A BOOK'S BUDGET. The budget answers "this book cannot be read";
	// an engine binary that is missing, locked out or unmigrated says nothing about any book and
	// applies to all of them at once, so counting it would write off every book on the host within a
	// few passes. The intake holds the same rule for the same reason (books.defer_).
	if ingest.DeploymentFault(cause) {
		s.defer_(c, b, cause, pgstore.CostsNoAttempt)
		return
	}
	if b.Attempts+1 >= maxAttempts {
		if err := s.Store.AbandonReadModelDebt(c, b.ID, b.OwedAt, time.Now(), cause.Error()); err != nil {
			s.log().ErrorContext(c, "the unpayable debt could not be written off", "err", err)
			return
		}
		// The one line an operator is meant to act on, said once. The book keeps the surface it had;
		// `tmplatformctl books --abandoned` names it and `book refresh` asks for it again.
		s.log().ErrorContext(c, "a book's reading surface has failed too many times and is given up on; its text is whatever was materialized before",
			"attempts", b.Attempts+1, "err", cause)
		return
	}
	s.defer_(c, b, cause, pgstore.SpendsAnAttempt)
}

// defer_ puts a debt back with a real delay. The backoff grows either way — a host that cannot run
// the engine must not be asked every fifteen seconds — only the counting differs.
func (s *Service) defer_(ctx context.Context, b pgstore.OwedBook, cause error, cost pgstore.AttemptCost) {
	if _, err := s.Store.DeferReadModelDebt(ctx, b.ID, b.OwedAt,
		time.Now().Add(retryIn(b.Attempts+1)), cause.Error(), cost); err != nil {
		s.log().ErrorContext(ctx, "the unpaid debt could not be moved to the back of the queue", "err", err)
	}
}

func (s *Service) refreshStructure(ctx context.Context, bookID, workdir string, cut *ingest.Manifest) error {
	manifest := ingest.Manifest{}
	if cut != nil {
		manifest = *cut
	} else {
		read, err := s.Engine.Manifest(ctx, s.Binary, workdir)
		if err != nil {
			return fmt.Errorf("readmodel: read the manifest: %w", err)
		}
		manifest = read
	}
	// ⚠ THE FLOOR, and it guards the one write in this zone that can lose a whole book's text.
	// `SaveStructure` writes the tree by REPLACEMENT — every chapter outside the list it is given is
	// deleted and the cascade takes the pairs with it — so an empty or short list is not a small
	// answer, it is "this book has no chapters any more". No engine produces one today, which is why
	// this is a floor and not a bug fix: what it catches is a document THIS build failed to read (a
	// key the engine renamed decodes into an empty list through the allowlist above), and the counts
	// printed beside the list are the only independent witness that the read worked.
	//
	// The intake has had exactly this floor since P6, on exactly this asymmetry: there "no chapters"
	// is the verdict that DELETES the upload, so it is never believed from a document whose own
	// counters disagree with it (books.parse). The materializer's write is as destructive and had no
	// floor at all.
	//
	// Refused rather than trimmed: the debt is not discharged, the previous tree stands, and the book
	// goes back into the queue — where the attempt budget now ends it if the engine can never answer.
	if err := manifest.Whole(); err != nil {
		return fmt.Errorf("readmodel: refusing to rewrite the chapter tree from a manifest this build did not read whole: %w", err)
	}
	if manifest.ChaptersTotal < 1 {
		// Zero chapters is internally consistent and still not something to write. A book reaches this
		// materializer only after an intake that demanded at least one chapter, so a zero here is a
		// project that was replaced or emptied under the platform — and the answer to that is to keep
		// what the reader already has, not to delete it.
		return fmt.Errorf("readmodel: refusing to rewrite the chapter tree of a book whose manifest counts no chapters at all")
	}
	// The text is a SEPARATE call and a failing one does not cost the tree, which lands either way.
	//
	// ⚠ But this read must not SPEAK about text it does not have. A pair carries `TextKnown` and the
	// store then leaves the stored text alone — which covers the whole failure (nothing is known) and
	// the partial one alike: the two calls are two re-cuts of the same source, and a pair the export
	// did not carry is a skew between them, not an empty pair.
	//
	// ⚠ And it is RETURNED, not merely logged. A tree with no text is not a materialized book: the
	// caller counted this as done, the debt was discharged, and a first materialization that lost its
	// text left the reader unable to see even their own source.
	text, textErr := s.Engine.Export(ctx, s.Binary, workdir)
	if textErr != nil {
		textErr = fmt.Errorf("readmodel: read the pairs: %w", textErr)
	}
	byKey := make(map[[2]int]ingest.UnitText, len(text.Units))
	for _, u := range text.Units {
		byKey[[2]int{u.Chapter, u.Unit}] = u
	}
	// Whether the pairs channel ANSWERED, which is a different fact from what it carried: an export
	// that returned nothing is a book with no translated text, an export that failed is this build not
	// knowing. The store refuses to rewrite a re-cut book on the second (SaveStructure), because there
	// the old rows are deleted rather than left alone and the text is gone for good.
	in := pgstore.Structure{ManifestKey: manifest.Key, TextRead: textErr == nil,
		Chapters: make([]pgstore.StructureChapter, 0, len(manifest.Chapters))}
	for _, c := range manifest.Chapters {
		chapter := pgstore.StructureChapter{
			EngineID: c.ID, Number: c.Number,
			Units: make([]pgstore.StructureUnit, 0, len(c.Units)),
		}
		for _, u := range c.Units {
			pair := pgstore.StructureUnit{
				EngineID: u.ID, Ordinal: u.FirstChunkIdx, State: ingest.StatePending,
			}
			if t, ok := byKey[[2]int{c.Number, u.FirstChunkIdx}]; ok {
				pair.Source, pair.Target, pair.State, pair.TextKnown = t.Source, t.Target, t.State, true
			}
			chapter.Units = append(chapter.Units, pair)
		}
		in.Chapters = append(in.Chapters, chapter)
	}
	if err := s.Store.SaveStructure(ctx, bookID, in); err != nil {
		return errors.Join(err, textErr)
	}
	// Counts, never a book id: the same rule as everywhere else on this side of the log.
	s.log().InfoContext(ctx, "the reading surface was refreshed",
		"chapters", len(in.Chapters), "pairs", manifest.UnitsTotal, "text", len(text.Units))
	return textErr
}

func (s *Service) refreshBank(ctx context.Context, bookID, workdir string) error {
	bank, err := runner.ReadBank(workdir)
	if errors.Is(err, runner.ErrNoBank) {
		// A book that has never produced terms. Not an empty bank written over a full one: nothing is
		// saved at all, so a read-out that has simply not been made yet cannot erase one that was.
		return nil
	}
	if err != nil {
		return err
	}
	return s.Store.SaveBank(ctx, bookID, bank.Terms)
}