package pgstore

import (
	"context"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"

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

// readmodel.go: the reading surface's store half — the chapter tree with its pairs, the notes
// derived from what the stream resolved, and the bank as the client signs it.
//
// Two properties run through all of it and neither is decoration:
//
//   - a page and the revision it is stamped with are read in ONE transaction. The revision travels
//     as "what this page is a picture of" and a client MUST drop a read that is not above what it
//     applied, so a number read separately is newer than the rows it labels and everything
//     materialized in between is lost for good (PD-163).
//   - a walk across several pages is stamped with the LOWEST revision it saw, and that number is
//     also the watermark of the next delta read. Taking the newest would skip every row that changed
//     between the first page and the last (canon §Revision).
//
// The second is the CLIENT's arithmetic — the server answers one revision per page — but it decides
// the shape here: every collection answers a per-page revision rather than a per-collection one.

// ErrVersionTooOld is a delta watermark from before a collection was replaced wholesale. A delta
// cannot express a deletion, so the answer is a refusal that names the remedy (re-read in full)
// rather than a short list that looks complete.
var ErrVersionTooOld = errors.New("pgstore: that revision predates a wholesale replacement of this collection")

// ErrNoChapter is a chapter this book no longer has. Told apart from "no such book" because the
// remedies differ: the client re-reads the tree rather than checking the address (canon: 410).
var ErrNoChapter = errors.New("pgstore: this book has no such chapter")

// ErrTextUnknownForANewCut is a tree written from a manifest whose companion export could not be
// read, for a book that was cut AGAIN. Refused rather than written: see SaveStructure.
var ErrTextUnknownForANewCut = errors.New("pgstore: a book cut again may not be rewritten without its text")

// derivedID mints an opaque identifier that is a FUNCTION of what it names.
//
// Chapter and pair identities have to survive being re-materialized: the same chapter re-read from
// the same manifest must come back with the id a client already holds, or every refresh would
// invalidate every anchor. A random id would need a lookup table keyed by the engine's identity,
// which is the same mapping with a table to keep in step.
//
// The BOOK is hashed in, and that is not namespacing for its own sake: the engine derives a
// chapter's identity from its TEXT, so two accounts uploading the same book would otherwise collide
// on a primary key and one would read the other's rows.
func derivedID(prefix, bookID, engineID string) string {
	sum := sha256.Sum256([]byte(prefix + "\x00" + bookID + "\x00" + engineID))
	return prefix + "_" + hex.EncodeToString(sum[:])[:24]
}

// Structure is a book's tree as the engine last cut it, with the text of every pair.
type Structure struct {
	// ManifestKey is the engine's validity key. When it differs from the stored one the book was cut
	// again: the structure version moves, cursors and pair anchors die, and the client is told to
	// re-read.
	ManifestKey string
	// TextRead says whether the PAIRS channel ANSWERED — a different question from any pair's
	// TextKnown. An export carrying nothing is a book with no text yet; an export that failed is not
	// knowing, and only one of those may be written over existing text. See SaveStructure.
	TextRead bool
	Chapters []StructureChapter
	// Price is the engine's a-priori projection of the book, or nil when this build did not read a
	// whole one. A book with no projection is one the platform REFUSES to sell rather than one it
	// prices with a constant (pricing's own package comment), so nil is written through: a projection
	// that goes stale is worse than one that is honestly absent.
	Price *Projection
	// Structure is where the chapter boundaries came from, verbatim from the engine. Empty when the
	// engine did not say — a manifest older than the landing — and stored as such: the decision it
	// feeds (ingest.ChapterOrdersOffered) treats what it does not know as untrusted.
	Structure string
	// SourceChars is the engine's count of the book's ingested text in runes.
	//
	// ⛔ CARRIED APART FROM Price, AND THAT IS THE POINT. It is a property of the CUT, not of the
	// money: a manifest can name the text honestly and still fail the price witness (a renamed money
	// key, a chapter that lost its bill). Read off `Price`, such a book lost its honest character
	// count too and fell back to the intake's approximation — a number that for an EPUB counts the
	// runes of a ZIP archive — for a reason that has nothing to do with counting. Zero means the
	// engine did not say.
	SourceChars int64
}

// Projection is the MONEY half of the engine's a-priori figure, as the store holds it.
//
// ⛔ IT CARRIES NO CHARACTER COUNT, and the absence is the whole reason this type exists rather than
// the wire's own `ingest.BookPrice`. The engine publishes `source_chars` INSIDE its price object,
// because for the engine both are outputs of one projection; the store keeps them apart
// (Structure.SourceChars) because for the store they answer to different witnesses — a manifest can
// name the text honestly and fail on a MONEY key, and a book that lost its bill has no business
// losing its character count too.
//
// ⚠ Reusing the wire struct here put the same figure in TWO fields of one value, and a caller then
// filled one and not the other — which is not a hypothetical: it is how
// TestTheEnginesCharacterCountReachesTheLibraryRowBesideItsAccuracy went red. One fact, one carrier.
type Projection struct {
	// Expected is the whole book's expected bill, book-level passes INCLUDED.
	Expected money.MicroUSD
	// BookOnce is the part of Expected that belongs to the BOOK and not to any chapter — a flat bond
	// on the shipped arm, which is why an ORDER is priced from the chapters' own bills instead (see
	// pricing.Hold and ingest.BookPrice.BookOnceUSD for the measurement behind it).
	BookOnce money.MicroUSD
	// StepMax is the largest single reservation any one call of this book can ask for: the floor of
	// every hold, because a ceiling below it admits nothing.
	StepMax money.MicroUSD
}

// StructureChapter is one chapter and its pairs, in reading order.
type StructureChapter struct {
	// EngineID is the engine's content-derived chapter identity; Number its dense display ordinal.
	EngineID string
	Number   int
	Units    []StructureUnit
}

// StructureUnit is one pair.
type StructureUnit struct {
	// EngineID carries the cut, so it changes when the book is cut differently — which is exactly
	// what the contract promises about a pair id.
	EngineID string
	// Ordinal is the leader chunk index: reading order inside the chapter and the join key of every
	// resolution and every export record.
	Ordinal int
	Source  string
	Target  string
	State   string
	// TextKnown says whether the PAIRS channel answered about THIS pair. False leaves the stored text
	// alone: the tree and the text are two engine calls, and one that failed — or that answered about
	// a different cut — has nothing to say about a pair it did not carry.
	TextKnown bool
	// Expected and SourceChars are this unit's half of the projection. Kept per UNIT and not only per
	// chapter because a book whose structure was not recognised is ONE chapter, and the only partial
	// order it can carry is a number of characters — which resolves to a prefix of units.
	Expected    money.MicroUSD
	SourceChars int64
}

// SaveStructure replaces a book's tree with the one just read from the engine.
//
// A REPLACEMENT and not a merge, because that is what the engine's artifact is: a manifest is the
// whole cut, and a chapter absent from it does not exist any more. The structure version moves only
// when the engine's own validity key moved, so an ordinary refresh — same cut, new text — leaves
// every cursor and every anchor a client holds valid.
//
// ⚠ IT REFUSES A RE-CUT WHOSE TEXT COULD NOT BE READ. `TextKnown` protects a pair only where its row
// survives: a re-cut changes every identity, so the old rows are deleted and the new ones inserted
// with what the caller carried — nothing, whenever the export failed and the manifest did not. The
// debt is then written off and an empty surface freezes over text the account paid for. Refusing
// keeps the previous tree and leaves the debt for the next drain.
func (s *Store) SaveStructure(ctx context.Context, bookID string, in Structure) error {
	return s.inTx(ctx, func(tx pgx.Tx) error {
		var storedKey string
		var revision int64
		var structure int
		err := tx.QueryRow(ctx,
			`select manifest_key, revision, structure_version from books where id = $1 for update`,
			bookID).Scan(&storedKey, &revision, &structure)
		if errors.Is(err, pgx.ErrNoRows) {
			return ErrNoBook
		}
		if err != nil {
			return fmt.Errorf("pgstore: read book structure: %w", err)
		}
		recut := storedKey != in.ManifestKey
		// ⚠ A book whose cut has never been seen is not a book that was cut AGAIN. The distinction
		// only matters below, where a re-cut drops the resolutions of the previous cut: treating the
		// first materialization as one threw away the work of a run that finished before the tree
		// landed — which is exactly what happens when the intake's own refresh failed (PD-276).
		changed := storedKey != "" && recut
		if changed && !in.TextRead {
			return ErrTextUnknownForANewCut
		}
		revision++
		if recut {
			structure++
		}
		// The PROJECTION travels with the cut it was derived from, in one statement, and that is not
		// tidiness: the engine computes both from the same walk of the same text (priceprojection.go,
		// projectBook), so a price written a moment apart from the tree is a price for a cut that may
		// already be gone. Writing them together makes «this book is priced» mean «priced as it is now
		// cut» and nothing else.
		//
		// Nulls are written through when the book is not priced. A stale projection kept beside a fresh
		// tree would sell yesterday's book; an absent one refuses the sale, which is the honest half.
		var expected, bookOnce, stepMax *int64
		if in.Price != nil {
			e, b, sm := int64(in.Price.Expected), int64(in.Price.BookOnce), int64(in.Price.StepMax)
			expected, bookOnce, stepMax = &e, &b, &sm
		}
		// ⚠ INDEPENDENT OF THE PRICE — see Structure.SourceChars. A book the engine could not price is
		// still a book whose text the engine counted, and the screen's «Знаков» has no business
		// falling back to the intake's approximation because a MONEY key was renamed.
		var sourceChars *int64
		if in.SourceChars > 0 {
			sc := in.SourceChars
			sourceChars = &sc
		}
		var structureWord *string
		if in.Structure != "" {
			w := in.Structure
			structureWord = &w
		}
		if _, err := tx.Exec(ctx, `
			update books set manifest_key = $2, structure_version = $3, revision = $4,
			                 chapter_count = $5,
			                 structure_reset_revision = case when $6 then $4 else structure_reset_revision end,
			                 expected_micro_usd = $7, book_once_micro_usd = $8, step_max_micro_usd = $9,
			                 source_chars = $10, structure = $11
			 where id = $1`,
			bookID, in.ManifestKey, structure, revision, len(in.Chapters), recut,
			expected, bookOnce, stepMax, sourceChars, structureWord); err != nil {
			return fmt.Errorf("pgstore: record structure: %w", err)
		}
		if err := writeChapters(ctx, tx, bookID, in, revision, changed); err != nil {
			return err
		}
		// One `chapter` frame per chapter would be a frame per chapter of a 2283-chapter book for a
		// refresh that changed nothing a client is looking at. What a re-cut needs is the
		// structure_version the next frame carries — the stream turns that into `resync_required` by
		// itself — and what an ordinary refresh needs is one poke saying the text moved.
		return emitStatus(ctx, tx, bookID)
	})
}

func writeChapters(ctx context.Context, tx pgx.Tx, bookID string, in Structure, revision int64, recut bool) error {
	if recut {
		// ⚠ Resolutions are addressed by the engine's (chapter number, unit ordinal), which a re-cut
		// re-points: kept, a chapter reported the finished units of whoever held its number before.
		// Dropped rather than translated because no mapping between the two cuts survives; the
		// counters restart for the new cut, which is the truth about a book just cut differently.
		if _, err := tx.Exec(ctx,
			`delete from unit_resolutions where book_id = $1`, bookID); err != nil {
			return fmt.Errorf("pgstore: drop resolutions of the previous cut: %w", err)
		}
	}
	keep := make([]string, 0, len(in.Chapters))
	for _, c := range in.Chapters {
		id := derivedID("ch", bookID, c.EngineID)
		keep = append(keep, id)
		// ⚠ NO PRICE ON THE CHAPTER ROW. What an order costs is summed from its UNDELIVERED units, so a
		// chapter-level copy would be a second carrier of one fact — written here, read nowhere, and
		// free to drift. The engine's own chapter roll-up is cross-checked where a witness belongs,
		// inside the document (ingest.Priced).
		if _, err := tx.Exec(ctx, `
			insert into chapters (id, book_id, number, units_total, revision)
			values ($1, $2, $3, $4, $5)
			on conflict (id) do update set number = excluded.number,
			                               units_total = excluded.units_total,
			                               revision = excluded.revision`,
			id, bookID, c.Number, len(c.Units), revision); err != nil {
			return fmt.Errorf("pgstore: write chapter: %w", err)
		}
		if err := writeUnits(ctx, tx, bookID, id, c.Units, revision); err != nil {
			return err
		}
	}
	// A chapter the current cut does not have is GONE, and so are its pairs (the cascade). Deleting
	// is what makes `410` answerable at all: a chapter left behind would keep answering 200 with the
	// text of a book that no longer exists.
	//
	// ⚠ Runs after the upserts, which needs `chapters_book_id_number_key` DEFERRED (00018): a chapter
	// id is its text's hash and survives a re-cut while its number shifts, so mid-loop two rows hold
	// one number. Delete-first does not help — the colliding chapters are the surviving ones.
	if _, err := tx.Exec(ctx,
		`delete from chapters where book_id = $1 and not (id = any($2))`, bookID, keep); err != nil {
		return fmt.Errorf("pgstore: drop chapters outside the cut: %w", err)
	}
	// The chapters' own counters are derived from what the stream resolved, and a re-cut re-numbers
	// chapters — so they are recomputed here rather than carried across, or a chapter would inherit
	// the progress of whatever chapter used to hold its number.
	if _, err := tx.Exec(ctx, `
		update chapters c set
			units_draft_done = (select count(*) from unit_resolutions
			                     where book_id = c.book_id and chapter = c.number and wave = 'draft'),
			units_edit_done  = (select count(*) from unit_resolutions
			                     where book_id = c.book_id and chapter = c.number and wave = 'edit'),
			note_count       = (select count(*) from unit_resolutions
			                     where book_id = c.book_id and chapter = c.number and flagged)
		 where c.book_id = $1`, bookID); err != nil {
		return fmt.Errorf("pgstore: recompute chapter counters: %w", err)
	}
	return nil
}

func writeUnits(ctx context.Context, tx pgx.Tx, bookID, chapterID string, units []StructureUnit, revision int64) error {
	keep := make([]string, 0, len(units))
	for _, u := range units {
		keep = append(keep, derivedID("un", bookID, u.EngineID))
	}
	// ⚠ The removed rows go FIRST here, and that is the opposite of the bank's order for a reason
	// that is not symmetry: a pair id carries the cut, so re-cutting a chapter mints a new id for
	// every pair in it while the ORDINALS stay 0, 1, 2 — and (chapter, ordinal) is unique. Inserting
	// first, every re-cut collided on the row it was about to replace. Nothing references a pair, so
	// deleting first costs nothing; a bank decision references its term, which is why that one is the
	// other way round.
	if _, err := tx.Exec(ctx,
		`delete from units where chapter_id = $1 and not (id = any($2))`, chapterID, keep); err != nil {
		return fmt.Errorf("pgstore: drop units outside the cut: %w", err)
	}
	for i, u := range units {
		// ⚠ Text is written only where this materialization HAS text: a failed export arrives empty and
		// an unconditional upsert erased a translated book with empty strings. Emptiness cannot be the
		// signal — an empty target is legitimate (`pending`, `withheld`) — so the caller says, per pair.
		if _, err := tx.Exec(ctx, `
			insert into units (id, chapter_id, ordinal, source, target, state, revision,
			                   expected_micro_usd, source_chars)
			values ($1, $2, $3, $4, $5, $6, $7, $9, $10)
			on conflict (id) do update set
				ordinal  = excluded.ordinal,
				source   = case when $8 then excluded.source else units.source end,
				target   = case when $8 then excluded.target else units.target end,
				state    = case when $8 then excluded.state  else units.state  end,
				revision = excluded.revision,
				-- The projection follows the CUT and not the text: it is derived from the source the
				-- manifest just described, so it lands whether or not the pairs channel answered.
				expected_micro_usd = excluded.expected_micro_usd,
				source_chars       = excluded.source_chars`,
			keep[i], chapterID, u.Ordinal, u.Source, u.Target, u.State, revision, u.TextKnown,
			int64(u.Expected), u.SourceChars); err != nil {
			return fmt.Errorf("pgstore: write unit: %w", err)
		}
	}
	return nil
}

// SaveBank replaces a book's bank with the engine's read-out of it.
//
// A replacement, and the pipeline's own rule is why: it rebuilds a book's whole glossary from its
// deterministic inputs on every run, so a merge here would keep rows the engine has decided no
// longer exist.
//
// `bank_decisions` is not touched. Nothing writes that table today — the per-term verb it served was
// abolished (D39.144) and removed — but a rebuild must still leave it alone: what replaces it, an
// EDIT of a term, has to survive the rebuild for the same reason, and the note left where the write
// path stood promises that the storage is ready for it.
//
// ⚠ It takes the WHOLE read-out rather than its rows, and one transaction writes all of it. The
// document is one projection of one moment — the bank, what a stop is asking about, and how complete
// the pass that produced it managed to be — and two saves would let half of it land: a screen showing
// this run's suggestions against the previous run's bank is a worse answer than either half alone.
func (s *Store) SaveBank(ctx context.Context, bookID string, bank ingest.Bank) error {
	terms := bank.Terms
	return s.inTx(ctx, func(tx pgx.Tx) error {
		if err := lockBook(ctx, tx, bookID); err != nil {
			return err
		}
		var revision int64
		if err := tx.QueryRow(ctx,
			`update books set revision = revision + 1 where id = $1 returning revision`, bookID).
			Scan(&revision); err != nil {
			if errors.Is(err, pgx.ErrNoRows) {
				return ErrNoBook
			}
			return fmt.Errorf("pgstore: bump revision for the bank: %w", err)
		}
		// ⚠ Rows are written BEFORE the removed ones are deleted, and the order is load-bearing:
		// `bank_decisions` cascades on `bank_terms`, so clearing the table first would take that table
		// with it — today emptying nothing, and tomorrow the user's own corrections.
		keep := make([]string, 0, len(terms))
		for _, t := range terms {
			id := derivedID("tm", bookID, t.ID)
			keep = append(keep, id)
			if _, err := tx.Exec(ctx, `
				insert into bank_terms (id, book_id, src, dst, kind, status, origin, sense,
				                        since_chapter, until_chapter, revision)
				values ($1, $2, $3, $4, nullif($5, ''), $6, $7, $8, $9, $10, $11)
				on conflict (id) do update set src = excluded.src, dst = excluded.dst,
				                               kind = excluded.kind, status = excluded.status,
				                               origin = excluded.origin, sense = excluded.sense,
				                               since_chapter = excluded.since_chapter,
				                               until_chapter = excluded.until_chapter,
				                               revision = excluded.revision`,
				id, bookID, t.Src, t.Dst, t.Kind, t.Status, t.Origin, t.Sense,
				t.SinceChapter, t.UntilChapter, revision); err != nil {
				return fmt.Errorf("pgstore: write bank term: %w", err)
			}
		}
		tag, err := tx.Exec(ctx,
			`delete from bank_terms where book_id = $1 and not (id = any($2))`, bookID, keep)
		if err != nil {
			return fmt.Errorf("pgstore: drop bank terms outside the read-out: %w", err)
		}
		if tag.RowsAffected() > 0 {
			// A row that disappeared cannot be expressed as a delta, so every watermark from before
			// this moment is refused rather than answered with an incomplete list.
			if _, err := tx.Exec(ctx,
				`update books set bank_reset_revision = $2 where id = $1`, bookID, revision); err != nil {
				return fmt.Errorf("pgstore: record the bank reset: %w", err)
			}
		}
		if err := saveBankReadOut(ctx, tx, bookID, bank); err != nil {
			return err
		}
		counts, err := bankCountsTx(ctx, tx, bookID)
		if err != nil {
			return err
		}
		return emitFrame(ctx, tx, bookID, FrameBank, counts.payload())
	})
}

// saveBankReadOut stores the two sections a signing stop publishes beside the bank, and the boundary
// that published them (engine backlog rows 224 and 253).
//
// ⚠ The offered terms are deleted BEFORE they are written, the opposite of the order the bank rows
// above use, and each order is load-bearing for its own reason: a bank row goes first because
// `bank_decisions` cascades on it, while an offered term has no dependants and its key is a POSITION
// in a ranking that moves — writing first would leave the tail of a longer previous read-out standing.
func saveBankReadOut(ctx context.Context, tx pgx.Tx, bookID string, bank ingest.Bank) error {
	c := bank.Consolidation
	// Seven values or none: the schema refuses a half-written section, because a half-written one is
	// exactly the zeroed section the engine refuses to publish — it would answer "consolidated 0,
	// unanswered 0", which reads as "nothing is missing".
	var complete *bool
	var renderDropped, classifyDropped, consolidated, declined, unanswered, neverAsked *int
	if c != nil {
		complete = &c.Complete
		renderDropped, classifyDropped = &c.RenderBatchesDropped, &c.ClassifyBatchesDropped
		consolidated, declined = &c.Consolidated, &c.Declined
		unanswered, neverAsked = &c.Unanswered, &c.NeverAsked
	}
	if _, err := tx.Exec(ctx, `
		insert into bank_read_out (book_id, boundary, engine_run_id, offered_present,
		                           consolidation_measured, complete, render_batches_dropped,
		                           classify_batches_dropped, consolidated, declined, unanswered,
		                           never_asked)
		values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
		on conflict (book_id) do update set boundary = excluded.boundary,
		                                    engine_run_id = excluded.engine_run_id,
		                                    offered_present = excluded.offered_present,
		                                    consolidation_measured = excluded.consolidation_measured,
		                                    complete = excluded.complete,
		                                    render_batches_dropped = excluded.render_batches_dropped,
		                                    classify_batches_dropped = excluded.classify_batches_dropped,
		                                    consolidated = excluded.consolidated,
		                                    declined = excluded.declined,
		                                    unanswered = excluded.unanswered,
		                                    never_asked = excluded.never_asked`,
		bookID, ingest.BoundaryOrUnknown(bank.Boundary), bank.RunID, bank.Offered != nil, c != nil,
		complete, renderDropped, classifyDropped, consolidated, declined, unanswered, neverAsked,
	); err != nil {
		return fmt.Errorf("pgstore: write the bank read-out: %w", err)
	}
	if _, err := tx.Exec(ctx, `delete from bank_offered_terms where book_id = $1`, bookID); err != nil {
		return fmt.Errorf("pgstore: drop the previously offered terms: %w", err)
	}
	if bank.Offered == nil {
		return nil
	}
	for i, t := range *bank.Offered {
		if _, err := tx.Exec(ctx, `
			insert into bank_offered_terms (book_id, ordinal, src, dst, kind, channel, freq, spread,
			                                conventions, confidence, invented, contradicts, bank_holds,
			                                variants)
			values ($1, $2, $3, $4, nullif($5, ''), nullif($6, ''), $7, $8, $9, $10, $11, $12, $13, $14)`,
			bookID, i, t.Src, t.Dst, ingest.KindOrNone(t.Kind), ingest.ChannelOrNone(t.Channel),
			t.Freq, t.Spread,
			t.Conventions, t.Confidence, t.Invented,
			stringList(t.Contradicts), stringList(t.BankHolds), stringList(t.Variants),
		); err != nil {
			return fmt.Errorf("pgstore: write an offered term: %w", err)
		}
	}
	return nil
}

// stringList is the empty ARRAY for an absent list. The columns are `not null default '{}'` and nil
// would be a null; "the read-out named nothing here" and "the read-out does not have this field" are
// the same fact for these three, because the engine omits all of them when empty.
func stringList(v []string) []string {
	if v == nil {
		return []string{}
	}
	return v
}

// BankCounts are the whole-bank aggregates: the header of a signing screen, and INFORMATIONAL —
// never a gate. Signing is one act over the whole bank and `resume` lifts the stop with the
// decisions as they stand (D39.144), so nothing here decides whether continuing may be offered.
//
// ⚠ `pending_decisions` and `complete` are GONE (canon 0.5.0, PD-399), not trimmed for taste: they
// were fed by the `bank_decisions` table whose write path went with the per-term model, which left
// them a live count of `proposed` rows — a frozen-plausible number teaching the abolished model.
// The honest count of undecidedness is the ENGINE's (`SignatureState`, served on the correction
// receipt); this read model cannot compute it without re-implementing the engine's law, which the
// seam forbids (17-seam-inbound-law п.6).
type BankCounts struct {
	Total  int
	Signed int
}

func (c BankCounts) payload() map[string]any {
	return map[string]any{"total": c.Total, "signed": c.Signed}
}

func bankCountsTx(ctx context.Context, tx pgx.Tx, bookID string) (BankCounts, error) {
	const q = `
		select count(*), count(*) filter (where t.status = 'approved')
		  from bank_terms t where t.book_id = $1`
	var c BankCounts
	if err := tx.QueryRow(ctx, q, bookID).Scan(&c.Total, &c.Signed); err != nil {
		return BankCounts{}, fmt.Errorf("pgstore: count the bank: %w", err)
	}
	return c, nil
}

// bookStatus is the product status of a book with the three machine reasons that accompany it.
type bookStatus struct {
	Status        string
	PausedReason  string
	RejectReason  string
	FailureReason string
}

// derivedStatus is the contract's rule of precedence, written once as SQL: the book's status is
// that of its current or last run, EXCEPT the states a run cannot be in — which belong to the book
// before any run exists or instead of one (canon §BookStatus).
//
// Without it the two disagreed observably: a signing stop is written on the RUN row and the book's
// own column stayed `translating`, so the card said the work was going while the run was waiting for
// a signature.
const derivedStatus = `case when b.status in ('uploading', 'parsing', 'not_started', 'rejected')
                            then b.status else coalesce(r.status, b.status) end`

// bookColumns is the library row, in the ONE shape every path that hands out a book uses. Written
// once because a second copy is how a projection drifts: the intake's 201, the library page and the
// card all answer the same fields or the client sees a book change shape by the route it came from.
const bookColumns = `b.id, b.title, b.source_lang, b.target_lang, ` + derivedStatus + `, b.reject_reason,
	b.structure_version, b.shape_epoch, b.chapter_count, ` + chaptersDone + `, b.character_count,
	b.source_chars, coalesce(b.structure, ''), ` + noteCount + `,
	b.added_at, b.revision`

// epochWave is whether the pipeline this book's work goes through has an EDITOR — the shape of its
// CURRENT epoch, as the engine last announced it, over a `books b`. Three fallbacks in order: the
// epoch's own answer; the historical monotone flag, for a book whose last announcement predates the
// epoch columns; and finally "it has one", because a chapter only half-done must not count as
// finished.
//
// ⚠ It reads the EPOCH and not the monotone `edit_wave`, and the difference is two register rows.
// Read through the flag the count is wrong in BOTH directions, because the flag cannot come down:
// with the editor REMOVED the count freezes against an edit column no run will ever fill — the book
// stays half-done, no run's bar can reach one, and the purchase scale goes on offering chapters
// already translated, which is the user paying twice (PD-403); with the editor ADDED the flag flips
// on the run's first progress event and a book showing 7/10 shows 0/10, backwards, inside one
// `structure_version`, which the canon forbade in as many words (PD-404).
//
// The owner ruled the cure on 28.08 (D39.165 §2): a change of pipeline shape is an EVENT of the book,
// like cutting it again, and the count is legitimately recomputed at the boundary. `books.shape_epoch`
// is what makes that recomputation legible rather than indistinguishable from a server walking a
// counter down (canon 0.9.0).
//
// ⚠ `books.edit_wave` is NOT repealed and is NOT dead: it is still written, still monotone, still
// pinned (D39.153 §4б), and it is the second fallback above — the answer for every book whose shape
// was recorded before this epoch existed. What changed is that it stopped being the AUTHORITY.
//
// ⚠ Safe for the RUN's bar as well as the book's count, and that is worth stating because the bar is
// where PD-401 lived. The epoch moves only when a progress event announces a shape, and progress
// events come from the book's own live run — one per book by construction — so the epoch a bar reads
// is always that run's own announcement. PD-401 was never about WHICH flag chose the wave: it was
// about pairing each numerator with the baseline captured on ITS OWN column, and that pairing is
// untouched here.
const epochWave = `coalesce(b.epoch_editor, b.edit_wave, true)`

// editWave is the HISTORICAL, monotone answer to the same question, and it is what the RUN's bar
// reads. Unknown reads as "it has one", for the same reason.
//
// ⚠ The two are separated HERE and the separation is load-bearing, so it is worth being exact about
// which fact belongs to which reader. `epoch_editor` is ASSIGNED on every announcement, so it moves
// in BOTH directions; `edit_wave` only ever grows. That is right for the BOOK, whose lifetime count
// the owner ruled may be recomputed at a shape boundary (D39.165 §2) with `shape_epoch` announcing
// it — and WRONG for the run's bar, which the canon holds to one monotonic fraction over the run's
// whole work (row 200). An adversarial pass of this same pack landed the mistake and measured it:
// with the bar on the epoch, one run whose second attempt announced a draft-only shape read 4/4 and
// then 2/2 — the same row, the same `structure_version`, the fraction walking backwards. The bar
// stays on the flag.
//
// ⚠ Consequence, named rather than hidden: on a deployment where the editor was REMOVED, a run's bar
// still tariffs an edit wave that will not happen and cannot reach one. That is the second half of
// register row PD-403, and it stays OPEN — closing it needs a per-RUN record of the shape that run is
// actually working under, which is a decision of its own size (the shape is not known at StartRun —
// the engine announces it with the run's first progress event, which is exactly what PD-401 found).
const editWave = `coalesce(b.edit_wave, true)`

// finishedUnits is how many of a chapter's units are FINISHED: resolved by the LAST pass the book
// actually gets on this deployment.
//
// ⚠ Not always the edit wave, and naming that one outright was money. A pipeline with no editor gives
// the edit counter a denominator of zero, so counted as edit no chapter was ever finished: the book's
// bar stayed at zero and the purchase scale was never clamped, so the service went on offering
// chapters it had already translated.
//
// ⚠ The answer is the BOOK's (migration 00024) and not the latest run's. Read off the latest run it
// moved BACKWARDS: the latest run is the newest, and a run just admitted has announced nothing, so
// every book on a deployment with no editor fell back to zero chapters done until that run's first
// progress event.
const finishedUnits = `(case when ` + epochWave + ` then c.units_edit_done else c.units_draft_done end)`

// chapterUnitsDone is how many pairs of ONE chapter the CURRENT PASS has finished — the number the
// wire calls `units_done` — written once here and bound by each reader to the facts it has at hand.
//
// It is a rule about a chapter and it is evaluated PER CHAPTER, which is the property that matters
// most and the one two earlier editions of it lost. `signing` is the run-level fact (this run asked
// the engine to stop for the bank); everything else is columns of `c`. Nothing in it is book-wide, so
// a unit event on one chapter cannot change what a re-read says about another — and every chapter
// frame the sink pushes is announced by an event of that same chapter.
//
// The three arms, in order:
//   - A run that stops for signing, over a chapter the editor has not touched: the DRAFT pass is the
//     current one for that chapter, and its progress is the only honest number there is. This arm
//     covers the whole draft wave AND the stop itself, which is the point — a user watching a signing
//     run is watching the draft pass for most of it.
//   - Otherwise the book's epoch decides which pass is the last one, exactly as `finishedUnits` does
//     for the book's lifetime count.
//
// ⚠ The stored `bank_released` bit this replaces went out with the `--verify-bank` workaround
// (D39.158, ping #21), and the replacement is deliberately NOT a like-for-like port — the bit was
// wrong in a way that outlived every run: it moved only on a USER's resume, so a run that raised the
// flag and was never stopped by the engine (an empty delta, or a map the engine's presented memory
// had already shown — after D39.158 the ORDINARY case) kept it false for its whole life and counted
// DRAFTS as finished through the entire edit wave.
//
// ⚠ Two editions of this rule were landed and withdrawn inside one pack, and the reasons are worth
// keeping because they are the two ways to get it wrong. The first asked whether the run still owed
// draft passes — BOOK-wide, so it flipped on a draft unit of some OTHER chapter and silently changed
// chapters no event had touched. The second asked only whether the run was standing AT the stop,
// which is announced and per-run but far too narrow: it read the edit column — zeroes — through the
// whole draft wave, so a signing run showed nothing done for most of its life. Neither survived a
// measurement; this one is per-chapter and covers the wave.
// ⚠ ONE constant and not a function of its inputs, and that is the battery's rule rather than taste:
// every SQL of this package must fold to a compile-time constant so
// `TestEverySQLStatementParsesAgainstTheMigratedSchema` can plan it against the migrated schema. A
// helper taking its facts as arguments produced a non-constant expression at both call sites and the
// gate refused it — correctly. So the two readers share the TEXT: both join the book and its current
// run, which is what the sink's chapter frame already did and what the chapter listing now does too.
const chapterUnitsDone = `(case when r.verify_bank and c.units_edit_done = 0 then c.units_draft_done
	              when ` + epochWave + ` then c.units_edit_done
	              else c.units_draft_done end)`

// segmentUnits is the name the sink reads it by.
const segmentUnits = chapterUnitsDone

// chaptersDone is the BOOK's own progress: chapters fully translated, against chapter_count. It is a
// different question from the bar of a run — which measures what that run bought.
const chaptersDone = `(select count(*) from chapters c
	where c.book_id = b.id and c.units_total > 0 and ` + finishedUnits + ` >= c.units_total)`

// noteCount sums the per-chapter counters, which is what makes it describe the same set as the
// notes LIST: a note carries a chapter_id, so a resolution whose chapter has no row has no address
// on the wire — and no counter to be counted in either.
//
// Summed rather than counted from `unit_resolutions` with a join to `chapters`: that form answered
// the same number and cost an index scan of every note the book has, per book, on every library page
// (migration 00022 carries the measurement).
const noteCount = `(select coalesce(sum(c.note_count), 0) from chapters c where c.book_id = b.id)`

func scanBook(row pgx.Row) (Book, error) {
	var b Book
	err := row.Scan(&b.ID, &b.Title, &b.SourceLang, &b.TargetLang, &b.Status, &b.RejectReason,
		&b.StructureVersion, &b.ShapeEpoch, &b.ChapterCount, &b.ChaptersDone, &b.CharacterCount,
		&b.SourceChars, &b.Structure, &b.NoteCount, &b.AddedAt, &b.Revision)
	if errors.Is(err, pgx.ErrNoRows) {
		return Book{}, ErrNoBook
	}
	if err != nil {
		return Book{}, fmt.Errorf("pgstore: read book: %w", err)
	}
	return b, nil
}

// ReadBook answers one book in the read model's shape, outside any transaction.
//
// Exported for the intake, which cuts the book inside the upload request and must answer with the row
// as it stands after the cut (books.Accept).
func (s *Store) ReadBook(ctx context.Context, bookID string) (Book, error) {
	return scanBook(s.pool.QueryRow(ctx,
		`select `+bookColumns+` from books b `+lastRun+` where b.id = $1`, bookID))
}

// readBookTx answers one book in the shape above.
func readBookTx(ctx context.Context, tx pgx.Tx, bookID string) (Book, error) {
	return scanBook(tx.QueryRow(ctx,
		`select `+bookColumns+` from books b `+lastRun+` where b.id = $1`, bookID))
}

// draftChapters / editChapters are how many of the book's chapters each pass has fully resolved —
// each on its OWN column, never through the flag. The bar's baselines are captured on these same
// fixed predicates (StartRun), so a capture can never disagree with the numerator it will later be
// subtracted from, however `edit_wave` moves in between.
const draftChapters = `(select count(*) from chapters c
	where c.book_id = b.id and c.units_total > 0 and c.units_draft_done >= c.units_total)`

const editChapters = `(select count(*) from chapters c
	where c.book_id = b.id and c.units_total > 0 and c.units_edit_done >= c.units_total)`

// The run's bar measures the run's WHOLE work through both waves as ONE monotonic fraction (owner's
// word of 20.08, row 200): the draft pass and the last pass each contribute the chapters they
// finished, against what the run bought — so the bar no longer restarts from zero when the signing
// stop is lifted and the counting switches waves, which is exactly what it used to do.
//
// ⚠ Each half is measured from the run's own BASELINE and clamped into its bought range, and that
// part is unchanged deliberately. Resolutions persist across runs — a resumed run re-walks finished
// chapters at $0 and must not move them — so counting the book's totals would open a continuation
// run's bar at everything the previous runs did, against a denominator of what THIS one bought: a
// fraction starting above zero and able to exceed one. `least(…, ceiling_chapters)` closes the
// other end, where a run that bought 5 chapters watches a neighbouring run finish more of the same
// book. Two baselines because there are two halves: `chapters_before` for the last pass,
// `draft_before` for the draft one, both captured when the run row is created (StartRun) and never
// re-based after — monotonicity is exactly "the counters only grow and nothing under them moves".
const (
	// draftWork is how many DRAFT passes this run actually has to do: the bought range runs from the
	// edit baseline to `chapters_before + ceiling`, and whatever of it was already drafted when the
	// run started (`draft_before`) is not this run's work. Without this term the denominator
	// tariffed a draft wave the run never performs, and a continuation run over a drafted backlog
	// finished a clean `ready` at 50% with the caption stuck on drafting (refuter finding, P9).
	draftWork = `least(greatest(r.chapters_before + r.ceiling_chapters - r.draft_before, 0), r.ceiling_chapters)`
	draftBar  = `least(greatest(` + draftChapters + ` - r.draft_before, 0), ` + draftWork + `)`
	// ⚠ EACH numerator is paired with the baseline captured on ITS OWN column, and the pair is chosen
	// by the LIVE flag at read time — never by the flag as it stood at capture. The alternative was
	// the reviewer's blocker (P9): `edit_wave` legitimately flips false→true AFTER a run starts (the
	// engine announces its wave shape with its first progress event — sink.recordWaveShape, STACK §35),
	// and a baseline captured through the flag sat on the draft column while the numerator moved to
	// the edit one, so the bar read 0/N after all the bought work was done and never caught up.
	editBar      = `least(greatest(` + editChapters + ` - r.chapters_before, 0), r.ceiling_chapters)`
	draftOnlyBar = `least(greatest(` + draftChapters + ` - r.draft_before, 0), r.ceiling_chapters)`
	// A pipeline with no editor has ONE wave, and its draft column IS the last pass: counting both
	// halves there would count every chapter twice against a doubled total that one wave can never
	// reach.
	// A RE-PASS run (P10, D39.165 §3 + errata 28.08-к) buys no chapters — ceiling_chapters is 0,
	// the one shape that writes it — and its bar is ONE UNIT OF WORK: 0 until the run finishes
	// clean, 1 then. Declared, not smuggled (the orchestrator's canon note): no finer honest
	// granularity exists — the engine announces a unit once for the life of the book and
	// re-announces nothing on a re-pass (announce-once ledger; the first edition counted chapters
	// by resolution times and read 0/N forever — adversarial K3). The 0/0 frame stays unreachable:
	// the denominator is the literal 1.
	// ⚠ THE UNIT-SHAPED BRANCH IS FIRST, and it is entered only by a run whose order does not close
	// whole chapters (`ordered_units is not null`). Every other run — every one that existed before
	// this pack — falls through to exactly the arithmetic it always had.
	runDone = `(case when r.ordered_units is not null then ` + unitDone + `
	              when r.ceiling_chapters = 0 then
	              (case when r.finished_at is not null and r.status = 'ready' then 1 else 0 end)
	              when ` + editWave + ` then ` + draftBar + ` + ` + editBar + ` else ` + draftOnlyBar + ` end)`
	runTotal = `(case when r.ordered_units is not null then ` + unitTotal + `
	              when r.ceiling_chapters = 0 then 1
	              when ` + editWave + ` then ` + draftWork + ` + r.ceiling_chapters else r.ceiling_chapters end)`
	// runStage is the caption's machine value — what the run is doing NOW, for the client to phrase
	// (open vocabulary, like the correction receipt's `depth`; D39.163 — new values move no
	// version). Derived from the same counters as the bar so the two cannot disagree: the run is
	// `editing` once the draft passes IT owed are done — at once, when it owed none — always
	// `drafting` where drafting is the only pass there is, and `re_pass` for the whole of a re-pass
	// run, which is neither.
	// runDelivered is how many chapters of what this run BOUGHT it has actually handed over — the
	// LAST pass over them, which is what «delivered» means to a reader. Bounded by the order at the
	// top for the same reason the bar is: free and carried units ride outside the grant (volume.go),
	// so a run can finish more than it was sold, and a figure above what was bought would read as an
	// overdelivery rather than as the rounding it is. Zero for a re-pass, which buys no chapters.
	// ⚠ NULL — NOT ZERO — FOR A UNIT-SHAPED ORDER, and there is no second figure beside it. A run
	// that bought a prefix of a chapter has delivered no CHAPTER, and «0» would read as «nothing
	// happened» when work was done and paid for. What such a run has delivered is carried by the
	// BAR instead, whose `done`/`total` are counted in units for exactly this run (see runDone) —
	// so the two figures a client gets are a null chapter count and a unit-shaped fraction, and
	// nothing on the wire is called `delivered_units`.
	runDelivered = `(case when r.ordered_units is not null then null
	              when r.ceiling_chapters = 0 then 0
	              else least(greatest((case when ` + editWave + ` then ` + editChapters + ` - r.chapters_before
	                                        else ` + draftChapters + ` - r.draft_before end), 0),
	                         r.ceiling_chapters) end)`
	runStage = `(case when r.ordered_units is not null then ` + unitStage + `
	              when r.ceiling_chapters = 0 then 're_pass'
	              when ` + editWave + ` and ` + draftBar + ` >= ` + draftWork + `
	                  then 'editing' else 'drafting' end)`
)

// runOrderedChapters is what the contract calls `ordered_chapters`: how much BOOK this run was sold,
// in chapters — and NULL for a run that was not sold in chapters at all.
//
// ⛔ THE NULL IS THE POINT, and the figure it replaces was a lie of a particular kind: not a wrong
// number, but a right number answering a question nobody asked. The column underneath
// (`ceiling_chapters`) holds, for an order phrased in CHARACTERS, the chapter SPAN — how many
// chapters the order reaches INTO — and that overstates what was bought by as much as a whole
// chapter: an order of one unit out of four in the first chapter stored `1`, which is the cheapest
// order there is and the one a person tries the service with. Measured, not reasoned: see
// `runs.TestACharacterOrdersVolumeIsPublishedInTheUnitItWasSoldIn`.
//
// ⚠ IT NULLS THE VIEW AND NOT THE COLUMN, deliberately. `ceiling_chapters = 0` is the RE-PASS's own
// mark, and three places read it as exactly that — `runs.maxUnitsFor`, the reconciler's resume gate,
// and this package's own admission check. Writing a zero or a null into the row would make a
// character order indistinguishable from a re-pass to all three, and in the first of them the cost is
// money: zero is the engine's word for «no volume bound at all». So the row keeps the span, every
// internal reader keeps its meaning, and only what goes OUT changes.
const runOrderedChapters = `(case when r.ordered_units is not null then null else r.ceiling_chapters end)`

// newestRun is the order in which a book RESOLVES to one of its runs: the LIVE one first, however the
// clocks fell, and only then the newest-started.
//
// ⚠ The live term is not a tie-break, it is the whole point (PD-402). `RestartRun` re-opens a run
// WITHOUT re-stamping `started_at` — deliberately, because the run's baselines and its bar are the
// ones it started with — so a resumed run is older than the finished run that legitimately ran
// between its stop and its resume. Ordered by `started_at` alone the book then resolves to the
// FINISHED one: the card says `ready`, the bar stands frozen at somebody else's work, and the live
// run spends money invisibly behind them. The index `runs_one_live_per_book` (00002:66, unique on
// `book_id` where `finished_at is null`) is what makes this a TOTAL order and not a new ambiguity —
// at most one row can win the first term.
//
// It lives in one constant because the read model is not its only user: the resume gate reads the
// same question through `LatestRun`, and the two answering differently would INVERT this defect
// rather than fix it — the screen would follow the live run while the resume refused it as "not the
// latest one".
const newestRun = ` order by (finished_at is null) desc, started_at desc, id desc limit 1`

// lastRun is the join every book-scoped read uses: the current or last run of the book.
const lastRun = `left join lateral (
	select id, status, paused_reason, failure_reason, verify_bank, ceiling_chapters, bond_funded,
	       ordered_units, units_before, draft_units_before,
	       chapters_before, draft_before, eta_seconds, started_at, finished_at, stop_requested_at,
	       accept_rebill_micro
	  from runs r where book_id = b.id` + newestRun + `) r on true`

func readBookStatusTx(ctx context.Context, tx pgx.Tx, bookID string) (bookStatus, error) {
	var st bookStatus
	err := tx.QueryRow(ctx, `
		select `+derivedStatus+`, coalesce(r.paused_reason, ''), b.reject_reason,
		       coalesce(r.failure_reason, '')
		  from books b `+lastRun+` where b.id = $1`, bookID).
		Scan(&st.Status, &st.PausedReason, &st.RejectReason, &st.FailureReason)
	if errors.Is(err, pgx.ErrNoRows) {
		return bookStatus{}, ErrNoBook
	}
	if err != nil {
		return bookStatus{}, fmt.Errorf("pgstore: read book status: %w", err)
	}
	return st, nil
}

// emitStatus announces the book's product status, with the three machine reasons that travel with
// it — each translated into the CONTRACT's vocabulary here and not by whoever reads the frame.
//
// ⚠ The translation is the point. The columns hold this platform's own words (`daily_ceiling`,
// `parser_unavailable`), the wire has no name for some of them, and the frame used to carry them
// raw while the JSON path next door mapped them properly: the same fact then read `null` on the card
// and `daily_ceiling` on the stream.
//
// A read failure emits NOTHING rather than a frame with a null status: `EventStatus` requires the
// field, and a frame outside its own schema is worse than a frame that did not arrive — the client
// re-reads on the next one either way.
func emitStatus(ctx context.Context, tx pgx.Tx, bookID string) error {
	st, err := readBookStatusTx(ctx, tx, bookID)
	if err != nil {
		if errors.Is(err, ErrNoBook) {
			return nil
		}
		return err
	}
	return emitFrame(ctx, tx, bookID, FrameStatus, map[string]any{
		"status":         st.Status,
		"paused_reason":  nullable(ingest.ContractPausedReason(st.PausedReason)),
		"reject_reason":  nullable(ingest.ContractRejectReason(st.RejectReason)),
		"failure_reason": nullable(st.FailureReason),
	})
}

// Page is the envelope every collection answers in.
type Page struct {
	Revision         int64
	StructureVersion int
	NextCursor       string
}

// Chapter is one row of the chapter tree.
type Chapter struct {
	ID string
	// Number is the display ordinal, or nil for a book with no numbering.
	Number     *int
	UnitsTotal int
	// UnitsDone is the pairs finished IN THE CURRENT PASS — the same accounting as the run's own bar,
	// one level down. It legally returns to zero when a new pass re-walks a chapter.
	UnitsDone int
	NoteCount int
}

// ChapterPage is one page of the tree.
type ChapterPage struct {
	Page
	Chapters []Chapter
}

// ListChapters returns one page of the chapter tree, in reading order.
func (s *Store) ListChapters(ctx context.Context, userID, bookID string, limit int, cursor string) (ChapterPage, error) {
	limit = clampPage(limit)
	var out ChapterPage
	err := s.inReadTx(ctx, func(tx pgx.Tx) error {
		scope, err := bookScope(ctx, tx, userID, bookID)
		if err != nil {
			return err
		}
		after, err := decodeNumberCursor(scope.tag("chapters"), cursor)
		if err != nil {
			return err
		}
		out.Page = Page{Revision: scope.revision, StructureVersion: scope.structure}
		// The SAME TEXT the sink's chapter frame is computed from, which is why this query joins the
		// book and its current run at all: a page that answered a different number from the frame it
		// was pushed a moment ago is a screen disagreeing with itself, and two copies of one rule is
		// how that happens.
		rows, err := tx.Query(ctx, `
			select c.id, c.number, c.units_total,
			       `+segmentUnits+`,
			       c.note_count
			  from chapters c join books b on b.id = c.book_id `+lastRun+`
			 where c.book_id = $1 and ($2::integer is null or c.number > $2::integer)
			 order by c.number limit $3`, bookID, after, limit+1)
		if err != nil {
			return fmt.Errorf("pgstore: list chapters: %w", err)
		}
		defer rows.Close()
		for rows.Next() {
			var c Chapter
			var number int
			if err := rows.Scan(&c.ID, &number, &c.UnitsTotal, &c.UnitsDone, &c.NoteCount); err != nil {
				return fmt.Errorf("pgstore: scan chapter: %w", err)
			}
			c.Number = &number
			out.Chapters = append(out.Chapters, c)
		}
		if err := rows.Err(); err != nil {
			return fmt.Errorf("pgstore: list chapters: %w", err)
		}
		if len(out.Chapters) > limit {
			last := out.Chapters[limit-1]
			out.Chapters = out.Chapters[:limit]
			out.NextCursor = encodeNumberCursor(scope.tag("chapters"), *last.Number)
		}
		return nil
	})
	if err != nil {
		return ChapterPage{}, err
	}
	return out, nil
}

// Unit is one source/translation pair with the notes attached to it.
type Unit struct {
	ID     string
	Source string
	Target string
	State  string
	Notes  []Note
}

// UnitPage is one page of a chapter's pairs.
type UnitPage struct {
	Page
	Units []Unit
}

// ListUnits returns one page of a chapter's pairs.
//
// PER CHAPTER and only per chapter: a book-wide pairs endpoint is never introduced and the client's
// whole memory model stands on that — a corpus book's pairs are tens of megabytes (canon §listUnits;
// the figures are in PLATFORM_DIRECTION §5б).
func (s *Store) ListUnits(ctx context.Context, userID, bookID, chapterID string, limit int, cursor string) (UnitPage, error) {
	limit = clampPage(limit)
	var out UnitPage
	err := s.inReadTx(ctx, func(tx pgx.Tx) error {
		scope, err := bookScope(ctx, tx, userID, bookID)
		if err != nil {
			return err
		}
		var number int
		err = tx.QueryRow(ctx,
			`select number from chapters where id = $1 and book_id = $2`, chapterID, bookID).Scan(&number)
		if errors.Is(err, pgx.ErrNoRows) {
			return ErrNoChapter
		}
		if err != nil {
			return fmt.Errorf("pgstore: read chapter: %w", err)
		}
		after, err := decodeNumberCursor(scope.tag("units:"+chapterID), cursor)
		if err != nil {
			return err
		}
		out.Page = Page{Revision: scope.revision, StructureVersion: scope.structure}
		rows, err := tx.Query(ctx, `
			select id, ordinal, source, target, state from units
			 where chapter_id = $1 and ($2::integer is null or ordinal > $2::integer)
			 order by ordinal limit $3`, chapterID, after, limit+1)
		if err != nil {
			return fmt.Errorf("pgstore: list units: %w", err)
		}
		defer rows.Close()
		var ordinals []int // parallel to out.Units: the leader index each pair is addressed by
		for rows.Next() {
			var u Unit
			var ordinal int
			if err := rows.Scan(&u.ID, &ordinal, &u.Source, &u.Target, &u.State); err != nil {
				return fmt.Errorf("pgstore: scan unit: %w", err)
			}
			ordinals = append(ordinals, ordinal)
			out.Units = append(out.Units, u)
		}
		if err := rows.Err(); err != nil {
			return fmt.Errorf("pgstore: list units: %w", err)
		}
		if len(out.Units) > limit {
			out.Units, ordinals = out.Units[:limit], ordinals[:limit]
			out.NextCursor = encodeNumberCursor(scope.tag("units:"+chapterID), ordinals[limit-1])
		}
		byOrdinal := make(map[int]int, len(ordinals))
		for i, ordinal := range ordinals {
			byOrdinal[ordinal] = i
		}
		// The notes of these pairs travel WITH them, so a reader screen need not join two collections
		// (canon §Unit.notes). Read for the chapter and attached by the leader index the resolution
		// and the pair share.
		notes, err := chapterNotes(ctx, tx, bookID, chapterID, number)
		if err != nil {
			return err
		}
		for _, n := range notes {
			if i, ok := byOrdinal[n.unit]; ok {
				out.Units[i].Notes = append(out.Units[i].Notes, n.Note)
			}
		}
		return nil
	})
	if err != nil {
		return UnitPage{}, err
	}
	return out, nil
}

// Note is one remark about a piece of the translation.
//
// It is PROJECTED from `unit_resolutions` rather than stored: the resolution is the durable record
// the stream writes, and a note is a view of the flagged ones. Its identity is derived from the
// resolution's own key, so a re-delivered line re-derives the same note instead of minting a second.
type Note struct {
	ID        string
	CreatedAt time.Time
	// Reason is the ENGINE's flag reason and does NOT go on the wire: the contract's code and the
	// step are mapped from it at projection time, so a reason this build has never heard of gets a
	// neutral phrase instead of a hole.
	Reason    string
	ChapterID string
	UnitID    string
	Revision  int64
}

type chapterNote struct {
	Note
	unit int
}

func chapterNotes(ctx context.Context, tx pgx.Tx, bookID, chapterID string, number int) ([]chapterNote, error) {
	rows, err := tx.Query(ctx, `
		select ur.unit, ur.wave, ur.reason, ur.at, ur.revision, u.id
		  from unit_resolutions ur
		  left join units u on u.chapter_id = $2 and u.ordinal = ur.unit
		 where ur.book_id = $1 and ur.chapter = $3 and ur.flagged
		 order by ur.at, ur.unit, ur.wave`, bookID, chapterID, number)
	if err != nil {
		return nil, fmt.Errorf("pgstore: read chapter notes: %w", err)
	}
	defer rows.Close()
	var out []chapterNote
	for rows.Next() {
		var n chapterNote
		var wave string
		var unitID *string
		if err := rows.Scan(&n.unit, &wave, &n.Reason, &n.CreatedAt, &n.Revision, &unitID); err != nil {
			return nil, fmt.Errorf("pgstore: scan note: %w", err)
		}
		n.ID = noteID(bookID, number, n.unit, wave)
		n.ChapterID = chapterID
		if unitID != nil {
			n.UnitID = *unitID
		}
		out = append(out, n)
	}
	return out, rows.Err()
}

// noteID is a note's identity: derived from the resolution it is a view of, so it survives the same
// line being delivered twice and the projection being rebuilt.
func noteID(bookID string, chapter, unit int, wave string) string {
	return derivedID("nt", bookID, fmt.Sprintf("%d:%d:%s", chapter, unit, wave))
}

// NotePage is one page of a book's notes.
type NotePage struct {
	Page
	Notes []Note
}

// ListNotes returns one page of a book's notes, oldest first.
//
// Ordered by time rather than by position in the book so that a note arriving on the stream can be
// placed into a list the client already holds. Ties are broken by the server in a way the client
// does NOT reproduce — here by the chapter, pair and pass the resolution belongs to, which is a
// total order over rows that share a timestamp.
func (s *Store) ListNotes(ctx context.Context, userID, bookID string, limit int, cursor string, after *int64) (NotePage, error) {
	limit = clampPage(limit)
	var out NotePage
	err := s.inReadTx(ctx, func(tx pgx.Tx) error {
		scope, err := bookScope(ctx, tx, userID, bookID)
		if err != nil {
			return err
		}
		if after != nil && *after < scope.structureReset {
			// The tree was re-cut, which re-numbers chapters and re-points every note's address.
			return ErrVersionTooOld
		}
		pos, err := decodeNoteCursor(scope.tag("notes"), cursor)
		if err != nil {
			return err
		}
		out.Page = Page{Revision: scope.revision, StructureVersion: scope.structure}
		rows, err := tx.Query(ctx, `
			select ur.chapter, ur.unit, ur.wave, ur.reason, ur.at, ur.revision, c.id, u.id
			  from unit_resolutions ur
			  join chapters c on c.book_id = ur.book_id and c.number = ur.chapter
			  left join units u on u.chapter_id = c.id and u.ordinal = ur.unit
			 where ur.book_id = $1 and ur.flagged
			   and ($2::bigint is null or ur.revision >= $2::bigint)
			   and ($3::timestamptz is null
			        or (ur.at, ur.chapter, ur.unit, ur.wave) > ($3::timestamptz, $4::integer, $5::integer, $6::text))
			 order by ur.at, ur.chapter, ur.unit, ur.wave limit $7`,
			bookID, after, pos.at, pos.chapter, pos.unit, pos.wave, limit+1)
		if err != nil {
			return fmt.Errorf("pgstore: list notes: %w", err)
		}
		defer rows.Close()
		var tail notePos
		for rows.Next() {
			var n Note
			var chapter, unit int
			var wave string
			var unitID *string
			if err := rows.Scan(&chapter, &unit, &wave, &n.Reason, &n.CreatedAt, &n.Revision,
				&n.ChapterID, &unitID); err != nil {
				return fmt.Errorf("pgstore: scan note: %w", err)
			}
			n.ID = noteID(bookID, chapter, unit, wave)
			if unitID != nil {
				n.UnitID = *unitID
			}
			if len(out.Notes) < limit {
				tail = notePos{at: &n.CreatedAt, chapter: &chapter, unit: &unit, wave: &wave}
			}
			out.Notes = append(out.Notes, n)
		}
		if err := rows.Err(); err != nil {
			return fmt.Errorf("pgstore: list notes: %w", err)
		}
		if len(out.Notes) > limit {
			out.Notes = out.Notes[:limit]
			out.NextCursor = encodeNoteCursor(scope.tag("notes"), tail)
		}
		return nil
	})
	if err != nil {
		return NotePage{}, err
	}
	return out, nil
}

// BankTerm is one row of the bank as the client reads it.
type BankTerm struct {
	ID     string
	Src    string
	Dst    string
	Kind   string // "" means the kind could not be decided: a legal state
	Status string
	Origin string
	Sense  string
	Since  *int
	Until  *int
}

// BankOfferedTerm is one row of a signing stop's table as the client reads it — a surface the service
// mined and consolidated that is NOT in the bank yet.
//
// It has no id, here or anywhere: the engine publishes none and minting one on this side would be
// re-implementing engine law. What it has is a POSITION in the stop's ranking, which the list's order
// carries — so no delta is expressible over this list and it is always read whole.
type BankOfferedTerm struct {
	Src string
	Dst string // "" when the read-out carries no rendering — it does not say why
	// "" for what this build cannot name: a kind the engine did not decide, a detector outside the
	// closed set.
	Kind, Channel string
	// Nil is NOT zero: `freq = 0` is the engine's "only the draft side saw it", and `conventions` is
	// absent from every row of a read-out bought before that field existed.
	Freq, Spread, Conventions *int
	// Nil "not carried", negative for the engine's "the reply stated none", >= 0 for a stated one.
	Confidence             *int
	Invented               bool
	Contradicts, BankHolds []string
	Variants               []string
}

// BankConsolidation is how complete the bank being signed is, taken from the engine ready-made.
type BankConsolidation struct {
	Complete                                     bool
	RenderBatchesDropped, ClassifyBatchesDropped int
	Consolidated, Declined                       int
	Unanswered, NeverAsked                       int
}

// BankSigningStop is what the last signing stop asked about.
//
// ⛔ Unreadable is the whole reason this is a struct and not a slice: without it "this deployment
// cannot tell you what the stop is asking" is byte-identical to "the stop asked nothing" — the one
// thing this type must never say by accident. It is true for THREE different causes, and they need no
// telling apart because the remedy is one: no read-out has been materialized yet, the read-out carries
// no such section (an engine build older than it), or the read-out was taken at a boundary that cannot
// carry one, where rows would be a stop this build invented.
//
// It does NOT answer whether a stop is standing — that is the run's status. This is the LAST stop's
// question, in the past tense the correction receipt already uses for the same population.
type BankSigningStop struct {
	Unreadable bool
	Offered    []BankOfferedTerm
}

// BankPage is one page of the bank. The aggregates ride on the FIRST page only — any response to a
// request with no cursor — which puts them at the same moment as the oldest rows of the walk.
type BankPage struct {
	Page
	Terms  []BankTerm
	First  bool
	Counts BankCounts
	// Consolidation is another whole-bank aggregate and rides where they do — the first page only.
	// Nil means nothing has measured this bank's completeness, which is NOT "it is complete".
	//
	// ⚠ It belongs with the bank rather than with the stop's questions, and the engine's own code is
	// why: its terminology result is set BEFORE the stop/continue fork and outlives it, so this
	// section rides three of the five boundaries while the questions ride one. Two different axes of
	// freshness, and one read cannot be honest about both.
	Consolidation *BankConsolidation
}

// ListBank returns one page of the bank, ordered by source surface then by the term's window so
// that the several rows of one surface stand together.
func (s *Store) ListBank(ctx context.Context, userID, bookID string, limit int, cursor string, after *int64) (BankPage, error) {
	limit = clampPage(limit)
	var out BankPage
	err := s.inReadTx(ctx, func(tx pgx.Tx) error {
		scope, err := bookScope(ctx, tx, userID, bookID)
		if err != nil {
			return err
		}
		if after != nil && *after < scope.bankReset {
			return ErrVersionTooOld
		}
		pos, err := decodeBankCursor(scope.tag("bank"), cursor)
		if err != nil {
			return err
		}
		out.Page = Page{Revision: scope.revision, StructureVersion: scope.structure}
		out.First = cursor == ""
		// Ordered by the SOURCE SURFACE and then by the term's WINDOW, which is what the contract
		// declares and what puts the several rows of one surface together in the order a person reads
		// them. The keyset follows the same tuple — ordering by the derived id instead put one
		// surface's rows in hash order while the doc comment claimed otherwise. `coalesce(…, 0)` is
		// the boundary's own sentinel: "from the beginning" sorts first.
		rows, err := tx.Query(ctx, `
			select id, src, dst, coalesce(kind, ''), status, origin, sense, since_chapter, until_chapter
			  from bank_terms
			 where book_id = $1
			   and ($2::bigint is null or revision >= $2::bigint)
			   and ($3::text is null or (src, coalesce(since_chapter, 0), coalesce(until_chapter, 0), id)
			                          > ($3::text, $4::integer, $5::integer, $6::text))
			 order by src, coalesce(since_chapter, 0), coalesce(until_chapter, 0), id
			 limit $7`, bookID, after, pos.src, pos.since, pos.until, pos.id, limit+1)
		if err != nil {
			return fmt.Errorf("pgstore: list bank: %w", err)
		}
		defer rows.Close()
		for rows.Next() {
			var t BankTerm
			if err := rows.Scan(&t.ID, &t.Src, &t.Dst, &t.Kind, &t.Status, &t.Origin, &t.Sense,
				&t.Since, &t.Until); err != nil {
				return fmt.Errorf("pgstore: scan bank term: %w", err)
			}
			out.Terms = append(out.Terms, t)
		}
		if err := rows.Err(); err != nil {
			return fmt.Errorf("pgstore: list bank: %w", err)
		}
		if len(out.Terms) > limit {
			last := out.Terms[limit-1]
			out.Terms = out.Terms[:limit]
			out.NextCursor = encodeBankCursor(scope.tag("bank"), last)
		}
		if out.First {
			if out.Counts, err = bankCountsTx(ctx, tx, bookID); err != nil {
				return err
			}
			if out.Consolidation, err = bankConsolidationTx(ctx, tx, bookID); err != nil {
				return err
			}
		}
		return nil
	})
	if err != nil {
		return BankPage{}, err
	}
	return out, nil
}

// bankConsolidationTx reads how complete the bank being signed is. No row, or a row from a boundary
// that never measured it, both answer nil — "nothing measured this", never "it is complete".
func bankConsolidationTx(ctx context.Context, tx pgx.Tx, bookID string) (*BankConsolidation, error) {
	var measured bool
	var c BankConsolidation
	err := tx.QueryRow(ctx, `
		select consolidation_measured, coalesce(complete, false),
		       coalesce(render_batches_dropped, 0), coalesce(classify_batches_dropped, 0),
		       coalesce(consolidated, 0), coalesce(declined, 0), coalesce(unanswered, 0),
		       coalesce(never_asked, 0)
		  from bank_read_out where book_id = $1`, bookID).
		Scan(&measured, &c.Complete, &c.RenderBatchesDropped, &c.ClassifyBatchesDropped,
			&c.Consolidated, &c.Declined, &c.Unanswered, &c.NeverAsked)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("pgstore: read the bank consolidation: %w", err)
	}
	// The coalesces keep the scan targets plain and are safe ONLY because the schema refuses a
	// partly-written section: the zeroes they produce are read exclusively when `measured` is false,
	// and nothing then sees them.
	if !measured {
		return nil, nil
	}
	return &c, nil
}

// BankSigningStop answers what the book's last signing stop asked about.
//
// Its own read and its own resource, because it moves on a different axis from the bank: these rows
// appear and vanish with a RUN while the bank's revision does not move, and there is no identity to
// express a delta over. So it is served whole, uncursored, and stamped by nothing.
func (s *Store) BankSigningStop(ctx context.Context, userID, bookID string) (BankSigningStop, error) {
	var out BankSigningStop
	err := s.inReadTx(ctx, func(tx pgx.Tx) error {
		if _, err := bookScope(ctx, tx, userID, bookID); err != nil {
			return err
		}
		var boundary string
		var present bool
		err := tx.QueryRow(ctx, `select boundary, offered_present from bank_read_out where book_id = $1`,
			bookID).Scan(&boundary, &present)
		if errors.Is(err, pgx.ErrNoRows) {
			out.Unreadable = true
			return nil
		}
		if err != nil {
			return fmt.Errorf("pgstore: read the bank read-out: %w", err)
		}
		// ⛔ The boundary is a condition and not a label. The engine publishes this section at the
		// signing stop and nowhere else, so rows standing at any other boundary mean the document is
		// not what this build takes it for — and serving them would put a stop on the screen that no
		// run is holding.
		if !present || boundary != ingest.BoundarySignatureRequested {
			out.Unreadable = true
			return nil
		}
		rows, err := tx.Query(ctx, `
			select src, dst, coalesce(kind, ''), coalesce(channel, ''), freq, spread, conventions,
			       confidence, invented, contradicts, bank_holds, variants
			  from bank_offered_terms where book_id = $1 order by ordinal`, bookID)
		if err != nil {
			return fmt.Errorf("pgstore: list the offered terms: %w", err)
		}
		defer rows.Close()
		// Non-nil before the loop: a stop that asked nothing must read as an empty list, not as a
		// document that carried no section — `Unreadable` is the only carrier of the second fact.
		out.Offered = []BankOfferedTerm{}
		for rows.Next() {
			var t BankOfferedTerm
			if err := rows.Scan(&t.Src, &t.Dst, &t.Kind, &t.Channel, &t.Freq, &t.Spread,
				&t.Conventions, &t.Confidence, &t.Invented, &t.Contradicts, &t.BankHolds,
				&t.Variants); err != nil {
				return fmt.Errorf("pgstore: scan an offered term: %w", err)
			}
			out.Offered = append(out.Offered, t)
		}
		if err := rows.Err(); err != nil {
			return fmt.Errorf("pgstore: list the offered terms: %w", err)
		}
		return nil
	})
	if err != nil {
		return BankSigningStop{}, err
	}
	return out, nil
}

// scope is what every book-scoped read needs before it reads anything: that the caller may see the
// book at all, and the two numbers the page is stamped with.
type scope struct {
	bookID         string
	revision       int64
	structure      int
	structureReset int64
	bankReset      int64
	// signing is whether the book's current run asked the engine to stop for the bank — a FACT, never
	// a pre-decided wave. Which pass a chapter counts by is `chapterUnitsDone`, written once and
	// evaluated per chapter; a scope that answered "draft" or "edit" by itself was a SECOND copy of
	// that rule — book-wide, applied to every chapter of a page, and free to disagree with the frame
	// the sink had just pushed for one of them.
	signing bool
}

// The BOOK is in the tag: without it a cursor minted for one book was accepted on another at the
// same structure version, and answered that book's rows from this one's offset.
func (s scope) tag(collection string) string {
	return s.bookID + "\x00" + collection + "\x00" + strconv.Itoa(s.structure)
}

func bookScope(ctx context.Context, tx pgx.Tx, userID, bookID string) (scope, error) {
	out := scope{bookID: bookID}
	var signing *bool
	err := tx.QueryRow(ctx, `
		select b.revision, b.structure_version, b.structure_reset_revision, b.bank_reset_revision,
		       r.verify_bank
		  from books b `+lastRun+`
		 where b.id = $1 and b.owner_id = $2`, bookID, userID).
		Scan(&out.revision, &out.structure, &out.structureReset, &out.bankReset, &signing)
	if errors.Is(err, pgx.ErrNoRows) {
		return scope{}, ErrNoBook
	}
	if err != nil {
		return scope{}, fmt.Errorf("pgstore: read book scope: %w", err)
	}
	// A book with no run at all has no signing fact; that is not "signing".
	out.signing = signing != nil && *signing
	return out, nil
}

// clampPage bounds a page size to what the contract lets a client ASK for, and CLAMPS rather than
// refuses: a server that answered the default instead made "ask for more, get fewer rows than a
// smaller request" discoverable only by experiment (canon §Limit).
func clampPage(limit int) int {
	switch {
	case limit <= 0:
		return DefaultPage
	case limit > maxPage:
		return maxPage
	}
	return limit
}

// ── Cursors ─────────────────────────────────────────────────────────────────────────────────────
//
// Opaque, and BOUND to the collection they came from and to the structure version they were minted
// under. Rejecting a cursor that no longer applies is the SERVER's duty and the client cannot
// perform it, because the token is opaque to it by construction (canon §NextCursor).

func cursorScope(tag string) string {
	sum := sha256.Sum256([]byte(tag))
	return base64.RawURLEncoding.EncodeToString(sum[:8])
}

func encodeCursorParts(tag string, parts ...string) string {
	return base64.RawURLEncoding.EncodeToString(
		[]byte(cursorScope(tag) + "\x00" + strings.Join(parts, "\x00")))
}

func decodeCursorParts(tag, s string, want int) ([]string, error) {
	raw, err := base64.RawURLEncoding.DecodeString(s)
	if err != nil {
		return nil, ErrBadCursor
	}
	fields := strings.Split(string(raw), "\x00")
	if len(fields) != want+1 || fields[0] != cursorScope(tag) {
		return nil, ErrBadCursor
	}
	return fields[1:], nil
}

func encodeNumberCursor(tag string, n int) string {
	return encodeCursorParts(tag, strconv.Itoa(n))
}

func decodeNumberCursor(tag, s string) (*int, error) {
	if s == "" {
		return nil, nil
	}
	fields, err := decodeCursorParts(tag, s, 1)
	if err != nil {
		return nil, err
	}
	n, err := strconv.Atoi(fields[0])
	if err != nil {
		return nil, ErrBadCursor
	}
	return &n, nil
}

// notePos is the keyset position of the note list: the timestamp the contract orders by, plus the
// server's own tiebreak.
type notePos struct {
	at      *time.Time
	chapter *int
	unit    *int
	wave    *string
}

// encodeNoteCursor packs the server's own tiebreak. ⚠ The pass is carried as a NUMBER and not as its
// name: a cursor is opaque by contract but trivially decodable in practice, and `draft`/`edit` are
// exactly the words this seam exists to keep off the wire.
func encodeNoteCursor(tag string, p notePos) string {
	if p.at == nil {
		return ""
	}
	return encodeCursorParts(tag, p.at.UTC().Format(time.RFC3339Nano),
		strconv.Itoa(*p.chapter), strconv.Itoa(*p.unit), strconv.Itoa(waveOrdinal(*p.wave)))
}

// waveOrdinal and waveOf are the two halves of that packing: the ordinal is the platform's own and
// the order it imposes is the order the query sorts by.
func waveOrdinal(wave string) int {
	if wave == ingest.WaveDraft {
		return 0
	}
	return 1
}

func waveOf(ordinal int) string {
	if ordinal == 0 {
		return ingest.WaveDraft
	}
	return ingest.WaveEdit
}

func decodeNoteCursor(tag, s string) (notePos, error) {
	if s == "" {
		return notePos{}, nil
	}
	fields, err := decodeCursorParts(tag, s, 4)
	if err != nil {
		return notePos{}, err
	}
	at, err := time.Parse(time.RFC3339Nano, fields[0])
	if err != nil {
		return notePos{}, ErrBadCursor
	}
	chapter, err := strconv.Atoi(fields[1])
	if err != nil {
		return notePos{}, ErrBadCursor
	}
	unit, err := strconv.Atoi(fields[2])
	if err != nil {
		return notePos{}, ErrBadCursor
	}
	ordinal, err := strconv.Atoi(fields[3])
	if err != nil {
		return notePos{}, ErrBadCursor
	}
	wave := waveOf(ordinal)
	return notePos{at: &at, chapter: &chapter, unit: &unit, wave: &wave}, nil
}

// bankPos is the keyset position of the bank: the declared order, tuple for tuple.
type bankPos struct {
	src   *string
	since *int
	until *int
	id    *string
}

func encodeBankCursor(tag string, t BankTerm) string {
	return encodeCursorParts(tag, t.Src, strconv.Itoa(bound(t.Since)), strconv.Itoa(bound(t.Until)), t.ID)
}

// bound is the window's own sentinel: an open boundary sorts as 0, which chapter numbering cannot
// produce (it starts at 1).
func bound(n *int) int {
	if n == nil {
		return 0
	}
	return *n
}

func decodeBankCursor(tag, s string) (bankPos, error) {
	if s == "" {
		return bankPos{}, nil
	}
	fields, err := decodeCursorParts(tag, s, 4)
	if err != nil {
		return bankPos{}, err
	}
	since, err := strconv.Atoi(fields[1])
	if err != nil {
		return bankPos{}, ErrBadCursor
	}
	until, err := strconv.Atoi(fields[2])
	if err != nil {
		return bankPos{}, ErrBadCursor
	}
	return bankPos{src: &fields[0], since: &since, until: &until, id: &fields[3]}, nil
}

// orderBoundary is the (chapter number, unit ordinal) the book's order stops at, over a `books b`.
//
// ONE expression for both kinds of partial order, because they differ only in how fine the boundary
// is: a CHAPTER order stops at the end of a chapter, so its ordinal is «everything in it», while a
// CHARACTER order stops at a named unit inside one. Written once so the two cannot be resolved by
// two arithmetics that drift apart.
//
// It answers NOTHING — no row — when the boundary does not resolve, and that is the load-bearing
// case rather than an edge: a book cut again mints new unit ids and may drop a chapter altogether,
// and the reference then dangles ON PURPOSE (migration 00033). Its absence must never be read as
// «the whole book», which is what a foreign key with `on delete set null` would have made it.
//
// ⚠ TWO GUARDS THAT LOOK REDUNDANT AND ARE NOT, both found by an adversarial pass rather than by a
// failure. `c.book_id = b.id` — without it a chapter id belonging to ANOTHER book resolves by its
// NUMBER against this one, which no writer produces today (a chapter id hashes its book in) and
// which nothing else would catch if one ever did. And the two ids are MUTUALLY EXCLUSIVE by the
// same predicate: with both set, a unit id killed by a re-cut would resolve through the surviving
// chapter id to the END of that chapter — a dangling reference silently widening an order, which is
// the one thing the identity form exists to prevent. `resolveOrder` sets exactly one; this is the
// schema-side statement of that, in the place that would otherwise believe both.
const orderBoundary = `(select c.number as ch, coalesce(u.ordinal, 2147483647) as ord
	  from chapters c
	  left join units u on u.id = b.ordered_through_unit_id
	 where c.book_id = b.id
	   and c.id = coalesce(b.ordered_through_chapter_id, u.chapter_id)
	   and (b.ordered_through_chapter_id is null or b.ordered_through_unit_id is null))`

// orderedUnits is the order RESOLVED against the tree as it stands: how many output units the book
// is bought through, counted from its start. Zero for the whole-book order (there is no volume to
// bound) and zero for a boundary that does not resolve — the two are told apart by orderResolved,
// never by this number.
const orderedUnits = `coalesce((select count(*)
	  from units u2 join chapters c2 on c2.id = u2.chapter_id, ` + orderBoundary + ` bnd
	 where c2.book_id = b.id and (c2.number, u2.ordinal) <= (bnd.ch, bnd.ord)), 0)`

// orderResolved is whether the order's boundary can be found at all. TRUE for the whole-book order,
// which has no boundary to lose.
const orderResolved = `(b.ordered_through_chapter_id is null and b.ordered_through_unit_id is null
	or exists ` + orderBoundary + `)`

// unitUndelivered is one unit of a book still to be handed over, over `units u`, `chapters c` and
// `books b`. Written once because the order form asks it three ways — per chapter for a quote, per
// unit for a character order, and as a total — and three spellings of one predicate is how a price
// and the volume it is a price OF stop describing the same work.
//
// It reads the RESOLUTION ROWS rather than the chapters' counters. The counters are recomputed from
// exactly these rows (writeChapters), so the two agree about HOW MANY units are done; what only the
// rows can say is WHICH, and a chapter's units are not equal — for money that is the whole question.
const unitUndelivered = `not exists (select 1 from unit_resolutions ur
	 where ur.book_id = b.id and ur.chapter = c.number and ur.unit = u.ordinal
	   and ur.wave = case when ` + epochWave + ` then 'edit' else 'draft' end)`

// deliveredWithinOrder is how many of the units the book is bought THROUGH have been handed over —
// the subtrahend of `--max-units`, over a `books b`.
//
// ⛔ BOUNDED BY THE SAME BOUNDARY the order is, and the first edition of it was not. Counting the
// book's delivered units book-WIDE while counting the ordered ones only up to the boundary makes the
// two describe different sets, and work delivered BEYOND the boundary then cancels work owed inside
// it: measured on a fixture whose order ran through chapter 2 while chapters 1 and 3 came back
// delivered, the allowance fell to its floor of one and a two-unit chapter was bought and half
// delivered. It costs no money — settlement is by fact — and it breaks «you bought N», which is the
// promise this whole pack exists to make true.
const deliveredWithinOrder = `coalesce((select count(*)
	  from units u2 join chapters c2 on c2.id = u2.chapter_id, ` + orderBoundary + ` bnd
	 where c2.book_id = b.id and (c2.number, u2.ordinal) <= (bnd.ch, bnd.ord)
	   and exists (select 1 from unit_resolutions ur
	                where ur.book_id = b.id and ur.chapter = c2.number and ur.unit = u2.ordinal
	                  and ur.wave = case when ` + epochWave + ` then 'edit' else 'draft' end)), 0)`

// bookUnitsLeft is how many output units of the book are still to be handed over, book-wide — the
// allowance a WHOLE-BOOK order gets on a continuation. See LiveRun.UnitsLeft for why a whole-book
// order is bounded at all.
const bookUnitsLeft = `coalesce((select count(*)
	  from units u join chapters c on c.id = u.chapter_id
	 where c.book_id = b.id and ` + unitUndelivered + `), 0)`

// bookUnitsDraftDone / bookUnitsEditDone are how many of a book's output units each WAVE has
// resolved, over a `books b`. The unit-level twins of `draftChapters`/`editChapters`, and finer in
// the one way that matters: a unit counts the moment it resolves, where a chapter counts only once
// every unit in it has.
//
// ⚠ Written out twice rather than made from a helper, and the repetition is deliberate: every SQL
// fragment in this package is a compile-time CONSTANT, which is what lets the whole read model be
// assembled from them and still be extracted and planned by the SQL gate. A function returning a
// string turns every expression that touches it into a `var`, and the cascade reaches `runRow` and
// half the queries in the package.
const bookUnitsDraftDone = `(select count(*) from unit_resolutions ur
	 where ur.book_id = b.id and ur.wave = 'draft')`

const bookUnitsEditDone = `(select count(*) from unit_resolutions ur
	 where ur.book_id = b.id and ur.wave = 'edit')`

// The bar of a run whose order does NOT close whole chapters — a CHARACTER order, which buys a
// prefix of a chapter and is the only partial order a book with no sellable chapter cut can carry.
//
// ⛔ IT IS THE SAME ARITHMETIC AS THE CHAPTER BAR, ONE LEVEL FINER, and the shape is copied rather
// than invented so the two cannot drift: each numerator is paired with the baseline captured on ITS
// OWN wave (units_before on the edit column, draft_units_before on the draft one), the pair is chosen
// by the LIVE wave flag at read time, and each half is clamped by what the run bought. Counted in
// chapters instead, such a run reads `0/N` for its whole life — the state this platform refuses a
// book for at admission (PD-405), arriving through a different door.
const (
	unitDraftBar = `least(greatest(` + bookUnitsDraftDone + ` - r.draft_units_before, 0), r.ordered_units)`
	unitEditBar  = `least(greatest(` + bookUnitsEditDone + ` - r.units_before, 0), r.ordered_units)`
	// unitDone/unitTotal mirror runDone/runTotal: with an editor the run owes both passes over what it
	// bought, without one the draft pass IS the last pass and counting both would double a total one
	// wave can never reach.
	unitDone  = `(case when ` + editWave + ` then ` + unitDraftBar + ` + ` + unitEditBar + ` else ` + unitDraftBar + ` end)`
	unitTotal = `(case when ` + editWave + ` then r.ordered_units * 2 else r.ordered_units end)`
	// The caption follows the same counters, so the two cannot disagree: the run is `editing` once the
	// draft pass over what it bought is done.
	unitStage = `(case when ` + editWave + ` and ` + unitDraftBar + ` >= r.ordered_units
	                   then 'editing' else 'drafting' end)`
)