1245 lines
56 KiB
Go
1245 lines
56 KiB
Go
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
|
||
}
|