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 time.Time) 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

// 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)
			// …and it 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 starve the books behind it out
			// of text they were paid for.
			c, cancel := context.WithTimeout(context.WithoutCancel(ctx), dischargeBudget)
			if err := s.Store.DeferReadModelDebt(c, b.ID, b.OwedAt); err != nil {
				s.log().ErrorContext(ctx, "the unpaid debt could not be moved to the back of the queue", "err", err)
			}
			cancel()
		}
	}
	return nil
}

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 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
	}
	in := pgstore.Structure{ManifestKey: manifest.Key,
		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)
}