package pgstore

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

	"github.com/jackc/pgx/v5"

	"textmachine/platform/internal/ingest"
)

// 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
}

// 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
}

// 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++
		}
		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
			 where id = $1`,
			bookID, in.ManifestKey, structure, revision, len(in.Chapters), recut); 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)
		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)
			values ($1, $2, $3, $4, $5, $6, $7)
			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`,
			keep[i], chapterID, u.Ordinal, u.Source, u.Target, u.State, revision, u.TextKnown); 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.
func (s *Store) SaveBank(ctx context.Context, bookID string, terms []ingest.BankTerm) error {
	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)
			}
		}
		counts, err := bankCountsTx(ctx, tx, bookID)
		if err != nil {
			return err
		}
		return emitFrame(ctx, tx, bookID, FrameBank, counts.payload())
	})
}

// 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, ` + 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.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
}

// 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.
	runDone = `(case 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.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.
	runStage = `(case when r.ceiling_chapters = 0 then 're_pass'
	              when ` + editWave + ` and ` + draftBar + ` >= ` + draftWork + `
	                  then 'editing' else 'drafting' 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,
	       chapters_before, draft_before, eta_seconds, started_at, finished_at, stop_requested_at
	  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
}

// 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
}

// 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
			}
		}
		return nil
	})
	if err != nil {
		return BankPage{}, 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
}