1677 lines
84 KiB
Go
1677 lines
84 KiB
Go
package pgstore
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"encoding/base64"
|
||
"errors"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
"unicode/utf8"
|
||
|
||
"github.com/jackc/pgx/v5"
|
||
"github.com/jackc/pgx/v5/pgconn"
|
||
|
||
"textmachine/platform/internal/ingest"
|
||
"textmachine/platform/internal/money"
|
||
)
|
||
|
||
// ErrNoBook is a book that does not exist, or does not belong to the caller. ONE error for both, on
|
||
// purpose: telling them apart would let anyone enumerate other people's libraries.
|
||
var ErrNoBook = errors.New("pgstore: no such book")
|
||
|
||
// ErrBadCursor is a pagination cursor this collection cannot use. Rejecting it is the SERVER's duty
|
||
// (contract, NextCursor): the client holds an opaque string and cannot judge it.
|
||
var ErrBadCursor = errors.New("pgstore: cursor does not apply to this collection")
|
||
|
||
// Book is the library row, in contract terms.
|
||
type Book struct {
|
||
ID string
|
||
Title string
|
||
SourceLang string
|
||
TargetLang string
|
||
// Status is the PRODUCT status, already resolved against the book's current or last run
|
||
// (derivedStatus): the book's status is the run's, except the states a run cannot be in. Read
|
||
// off the column alone it disagreed with the run observably — a signing stop is written on the
|
||
// run row, and the card went on saying `translating`.
|
||
Status string
|
||
// RejectReason is the PLATFORM's internal word for why an intake ended; the wire vocabulary is a
|
||
// projection of it (httpapi.contractRejectReason).
|
||
RejectReason string
|
||
StructureVersion int
|
||
// ShapeEpoch is the generation of the book's COUNT: it moves when the pipeline's shape changes
|
||
// under the book (an editor added or removed), which is the one other event besides a re-cut that
|
||
// legitimately recomputes `ChaptersDone`. A coordinate, like StructureVersion — the shape itself
|
||
// is deliberately not published (canon 0.9.0).
|
||
ShapeEpoch int
|
||
ChapterCount int
|
||
// ChaptersDone is the book's own lifetime progress — chapters fully translated — as opposed to
|
||
// the bar of a run, which measures what that run bought.
|
||
ChaptersDone int
|
||
// CharacterCount is the INTAKE's count: the non-continuation bytes of the write stream. For a
|
||
// UTF-8 text source it IS the character count; for an EPUB it is a property of a ZIP archive, and
|
||
// for GB18030 or UTF-16 an approximation (books.counter says all of this at length). It is the
|
||
// FALLBACK now, kept because it is the only figure a book still in intake has.
|
||
CharacterCount int64
|
||
// SourceChars is the ENGINE's count of the ingested text in runes, spaces included — the honest
|
||
// answer, and the one the wire prefers. Nil until a manifest has been read for this book, which
|
||
// is what the wire's accuracy flag reports (unified backlog row 282, form ratified D39.201 §5б:
|
||
// a flag BESIDE the number, never a changed meaning for `null` — `null` already means «still
|
||
// arriving»).
|
||
SourceChars *int64
|
||
// Structure is where the chapter boundaries came from, verbatim from the engine — `declared`,
|
||
// `detected`, `none`, or empty when no manifest has said. It travels to the client because the
|
||
// order form has to tell a buyer what the chapter numbers are worth (D39.196 §1).
|
||
Structure string
|
||
NoteCount int
|
||
AddedAt time.Time
|
||
// Revision is the book's own scope. Read on the card, where it orders the client's reads against
|
||
// the stream's frames; the LIBRARY has a revision of its own and the two are never compared.
|
||
Revision int64
|
||
}
|
||
|
||
// NewBook is one book being registered in the read model.
|
||
type NewBook struct {
|
||
OwnerID string
|
||
Title string
|
||
SourceLang string
|
||
TargetLang string
|
||
ChapterCount int
|
||
CharacterCount int64
|
||
// Workdir is the engine's project directory. The platform reads the book's journal there and
|
||
// spawns the engine with it as the working directory; it never writes into it (D39.110).
|
||
Workdir string
|
||
Now time.Time
|
||
}
|
||
|
||
// AddBook registers a book. `not_started` and not `uploading`: the file is already on disk, and a
|
||
// status describing an upload this route never performed would be a lie the library then shows.
|
||
func (s *Store) AddBook(ctx context.Context, in NewBook) (string, error) {
|
||
id := NewBookID()
|
||
const q = `
|
||
insert into books (id, owner_id, title, source_lang, target_lang, status,
|
||
chapter_count, character_count, added_at, workdir, engine_book_id, revision)
|
||
values ($1, $2, $3, $4, $5, 'not_started', $6, $7, $8, $9, $10, ` + nextLibraryRevision + `)`
|
||
// engine_book_id is the platform's id until the engine's handshake reports its own. It is not
|
||
// left empty: the column is the join back to whatever the engine calls this book, and an empty
|
||
// one would read as "the engine has no name for it" rather than "we have not been told yet".
|
||
_, err := s.pool.Exec(ctx, q, id, in.OwnerID, in.Title, in.SourceLang, in.TargetLang,
|
||
in.ChapterCount, in.CharacterCount, in.Now, in.Workdir, id)
|
||
if err != nil {
|
||
var pg *pgconn.PgError
|
||
if errors.As(err, &pg) && pg.ConstraintName == "books_owner_id_fkey" {
|
||
return "", ErrNoAccount
|
||
}
|
||
return "", fmt.Errorf("pgstore: add book: %w", err)
|
||
}
|
||
return id, nil
|
||
}
|
||
|
||
// NewBookID mints a book identifier. Exported because the intake needs one BEFORE the row exists:
|
||
// the book's directory is named after it, and the row records that directory.
|
||
func NewBookID() string { return newID("bk") }
|
||
|
||
// nextLibraryRevision is the revision a book JOINS the library with, as a SQL fragment over $2 (the
|
||
// owner).
|
||
//
|
||
// The library's revision is derived as the maximum over the account's books, and the contract asks
|
||
// a client to DROP a read whose revision is not above what it already applied. A book inserted at
|
||
// revision 0 therefore left the library's number unchanged, and the screen that had just uploaded
|
||
// it dropped the very read that carried it (register row PD-122). Starting above the account's
|
||
// current maximum is what makes an addition visible.
|
||
//
|
||
// It is the greatest of BOTH halves the library's own revision is read from — the maximum over the
|
||
// books and the account's floor — because taking only the maximum reproduced the very defect the
|
||
// floor exists to prevent, one step later: after a cancelled upload raised the floor above the
|
||
// maximum, the NEXT book joined below it, and the library then answered the same number for three
|
||
// different states of itself.
|
||
//
|
||
// Two concurrent additions can compute the same number, and that is harmless: the value still rises,
|
||
// so the client still refetches and still sees both. What stays open in PD-122 is the account-scope
|
||
// counter itself — one writer for membership and statuses — which is what would let two SIMULTANEOUS
|
||
// writers get distinct numbers instead of merely rising ones.
|
||
const nextLibraryRevision = `(select greatest(
|
||
coalesce((select max(revision) from books where owner_id = $2), 0),
|
||
coalesce((select library_revision from users where id = $2), 0)) + 1)`
|
||
|
||
// nextRevisionOfThisBooksLibrary is the same number for a statement that has the book but not its
|
||
// owner: the owner is read from the row being updated.
|
||
//
|
||
// ⚠ A status change may NOT simply do `revision + 1`, and that is the same defect as the one above
|
||
// wearing different clothes. A book whose own counter sits below the account's floor — it was
|
||
// uploaded before another upload was cancelled — would walk `uploading → parsing → not_started`
|
||
// three increments deep and still be under the floor, so `greatest(max, floor)` would not move at
|
||
// all. A client obeying the contract drops every read that is not ABOVE what it applied, and the
|
||
// screen that uploaded the book would keep showing it as arriving until some unrelated event pushed
|
||
// the number. Taking the library's own next number instead makes every status visible exactly once.
|
||
const nextRevisionOfThisBooksLibrary = `(select greatest(
|
||
coalesce((select max(b.revision) from books b where b.owner_id = books.owner_id), 0),
|
||
coalesce((select u.library_revision from users u where u.id = books.owner_id), 0)) + 1)`
|
||
|
||
// NewUpload is one book arriving through the contract's intake, before its file has been received.
|
||
type NewUpload struct {
|
||
OwnerID string
|
||
Title string
|
||
SourceLang string
|
||
TargetLang string
|
||
// Workdir is the engine project directory this book will own. It is named after the book id and
|
||
// created before the row, so a row always points at a directory that exists.
|
||
Workdir string
|
||
Now time.Time
|
||
}
|
||
|
||
// CreateUpload registers a book whose file is still arriving.
|
||
//
|
||
// `uploading` is a real, observable state and not a formality: the row is written BEFORE the body is
|
||
// read, so a 60 MB upload is visible in the library — with its languages, without its size — while
|
||
// it is still on the wire. The alternative, inserting once the bytes have landed, gives the status
|
||
// no writer at all and makes a half-finished upload invisible to everything, including the sweep
|
||
// that has to clean it up.
|
||
func (s *Store) CreateUpload(ctx context.Context, id string, in NewUpload) (Book, error) {
|
||
const q = `
|
||
insert into books (id, owner_id, title, source_lang, target_lang, status,
|
||
added_at, workdir, engine_book_id, revision)
|
||
values ($1, $2, $3, $4, $5, 'uploading', $6, $7, $1, ` + nextLibraryRevision + `)`
|
||
var b Book
|
||
err := s.inTx(ctx, func(tx pgx.Tx) error {
|
||
if _, err := tx.Exec(ctx, q, id, in.OwnerID, in.Title, in.SourceLang, in.TargetLang,
|
||
in.Now, in.Workdir); err != nil {
|
||
var pg *pgconn.PgError
|
||
if errors.As(err, &pg) && pg.ConstraintName == "books_owner_id_fkey" {
|
||
return ErrNoAccount
|
||
}
|
||
return fmt.Errorf("pgstore: create upload: %w", err)
|
||
}
|
||
var err error
|
||
if b, err = readBookTx(ctx, tx, id); err != nil {
|
||
return err
|
||
}
|
||
// The book EXISTS from this moment and a watcher of the library is entitled to hear so: the
|
||
// row is written before the bytes, which is the whole reason `uploading` is observable.
|
||
return emitStatus(ctx, tx, id)
|
||
})
|
||
if err != nil {
|
||
return Book{}, err
|
||
}
|
||
return b, nil
|
||
}
|
||
|
||
// StartParsing hands a received file to the parser: `uploading → parsing`, with the size the intake
|
||
// counted while the bytes went past.
|
||
//
|
||
// Conditional on the status, so an upload the sweep has already given up on cannot be resurrected by
|
||
// a request that finally finished.
|
||
//
|
||
// enqueue joins the caller's transaction for the same reason the run admission's does: a status that
|
||
// says "being parsed" with no job to do it waits for the backstop sweep's grace, and a job for a
|
||
// book that never became `parsing` is a worker with nothing to claim.
|
||
//
|
||
// ⚠ It is NIL on the path a deployment actually takes, and that is deliberate rather than dead code
|
||
// left behind. An intake that cuts the book inside the request must not have a job racing that cut
|
||
// for the parse claim, so it enqueues at the END of the cut instead (books.cutsItsOwnUploads, and
|
||
// ReleaseParseClaim below). What reaches this parameter is the other deployment — a queue and no
|
||
// engine — which has nothing to wait for and hands the book over at once.
|
||
func (s *Store) StartParsing(ctx context.Context, id string, characters int64,
|
||
enqueue func(context.Context, Tx, string) error) (Book, error) {
|
||
var b Book
|
||
err := s.inTx(ctx, func(tx pgx.Tx) error {
|
||
const q = `
|
||
update books set status = 'parsing', character_count = $2,
|
||
revision = ` + nextRevisionOfThisBooksLibrary + `
|
||
where id = $1 and status = 'uploading'`
|
||
tag, err := tx.Exec(ctx, q, id, characters)
|
||
if err != nil {
|
||
return fmt.Errorf("pgstore: start parsing: %w", err)
|
||
}
|
||
if tag.RowsAffected() == 0 {
|
||
return ErrNoBook
|
||
}
|
||
if b, err = readBookTx(ctx, tx, id); err != nil {
|
||
return err
|
||
}
|
||
if err := emitStatus(ctx, tx, id); err != nil {
|
||
return err
|
||
}
|
||
if enqueue != nil {
|
||
return enqueue(ctx, tx, id)
|
||
}
|
||
return nil
|
||
})
|
||
if err != nil {
|
||
return Book{}, err
|
||
}
|
||
return b, nil
|
||
}
|
||
|
||
// ParseClaim is one book this process may parse.
|
||
type ParseClaim struct {
|
||
BookID string
|
||
Workdir string
|
||
Attempts int
|
||
// What the USER declared at intake. It travels with the claim because the parse is also where a
|
||
// book's starting engine configuration is rendered (books/render.go), and reading it in a second
|
||
// query would be a second query for a row this one already has open.
|
||
Title string
|
||
SourceLang string
|
||
TargetLang string
|
||
// At is the stamp this claim was taken with. Every write that ENDS the parse carries it back, so
|
||
// a pass whose claim was taken over by another cannot record a verdict about work it is no longer
|
||
// doing — the same re-check-under-the-write the money paths use.
|
||
At time.Time
|
||
}
|
||
|
||
// ErrParseClaimed is a parse someone else is already doing, or a book that is no longer waiting for
|
||
// one. The ordinary case, not a failure: the queue job and the backstop sweep both aim at the same
|
||
// book on purpose.
|
||
var ErrParseClaimed = errors.New("pgstore: the parse of this book is already claimed")
|
||
|
||
// ClaimParse takes the right to parse a book, once.
|
||
//
|
||
// A compare-and-set rather than a plain read, for the same reason the spawn claim is one: the queue
|
||
// worker and the backstop sweep legitimately arrive together, and two `tmctl manifest` processes on
|
||
// one project directory would be two writers of one SQLite file — which is exactly what the engine's
|
||
// exclusive lock exists to refuse, noisily and after the work.
|
||
//
|
||
// staleAfter is what makes the claim recoverable: a process that died holding one leaves a stamp
|
||
// nothing clears, and without a grace the book would wait for parsing forever.
|
||
func (s *Store) ClaimParse(ctx context.Context, id string, now, staleBefore time.Time) (ParseClaim, error) {
|
||
const q = `
|
||
update books set parse_started_at = $2, parse_attempts = parse_attempts + 1
|
||
where id = $1 and status = 'parsing'
|
||
and (parse_started_at is null or parse_started_at < $3)
|
||
returning id, workdir, parse_attempts, title, source_lang, target_lang`
|
||
var c ParseClaim
|
||
err := s.pool.QueryRow(ctx, q, id, now, staleBefore).Scan(&c.BookID, &c.Workdir, &c.Attempts,
|
||
&c.Title, &c.SourceLang, &c.TargetLang)
|
||
c.At = now
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return ParseClaim{}, ErrParseClaimed
|
||
}
|
||
if err != nil {
|
||
return ParseClaim{}, fmt.Errorf("pgstore: claim parse: %w", err)
|
||
}
|
||
return c, nil
|
||
}
|
||
|
||
// RefundParseAttempt gives back an attempt that was claimed but never spent on the engine.
|
||
//
|
||
// The budget exists to bound how many times a broken HOST is asked to parse a book, and a book that
|
||
// is only waiting for its configuration asks nobody: counting those claims meant the budget ran out
|
||
// while nothing had been tried, and the first real answer from the engine was terminal the moment it
|
||
// arrived. Conditional on the claim, like every other write that ends a pass.
|
||
func (s *Store) RefundParseAttempt(ctx context.Context, id string, claimedAt time.Time) error {
|
||
_, err := s.pool.Exec(ctx, `
|
||
update books set parse_attempts = greatest(parse_attempts - 1, 0)
|
||
where id = $1 and status = 'parsing' and parse_started_at = $2`, id, claimedAt)
|
||
if err != nil {
|
||
return fmt.Errorf("pgstore: refund parse attempt: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ReleaseParseClaim hands a claim back untouched — stamp cleared, attempt uncounted — and enqueues
|
||
// the job that will finish the book, in ONE transaction.
|
||
//
|
||
// For the intake's synchronous cut only. A sweep that could not finish keeps its claim on purpose:
|
||
// the stamp is what spaces the retries (books.defer_). The intake is not a retry — it is one look
|
||
// taken while the uploader waits, and the queue has nothing to do until that look reaches no verdict.
|
||
// Hence the job is enqueued HERE and not when the book entered `parsing`: a job that exists while the
|
||
// intake is cutting races it for the claim, and whichever side loses, the book pays (books.Accept).
|
||
//
|
||
// Both halves in one transaction because either alone is a state nobody finishes: a released claim
|
||
// with no job waits out the sweep's grace, and a job with the claim still held does nothing.
|
||
//
|
||
// Conditional on the claim still being this caller's, like every other write that ends a pass.
|
||
func (s *Store) ReleaseParseClaim(ctx context.Context, id string, claimedAt time.Time,
|
||
enqueue func(context.Context, Tx, string) error) error {
|
||
return s.inTx(ctx, func(tx pgx.Tx) error {
|
||
tag, err := tx.Exec(ctx, `
|
||
update books set parse_started_at = null, parse_attempts = greatest(parse_attempts - 1, 0)
|
||
where id = $1 and status = 'parsing' and parse_started_at = $2`, id, claimedAt)
|
||
if err != nil {
|
||
return fmt.Errorf("pgstore: release parse claim: %w", err)
|
||
}
|
||
if tag.RowsAffected() == 0 || enqueue == nil {
|
||
return nil // somebody else holds the claim now, or this deployment has no queue
|
||
}
|
||
return enqueue(ctx, tx, id)
|
||
})
|
||
}
|
||
|
||
// ParsedBook is what the engine's manifest told the platform about a book it has now cut.
|
||
type ParsedBook struct {
|
||
Chapters int
|
||
SourceSHA256 []byte
|
||
ChunkerVersion string
|
||
}
|
||
|
||
// FinishParse records a parsed book — `parsing → not_started` — and the materialization it now owes.
|
||
// It returns the moment that debt was stamped, which is what discharges it (ClearReadModelDebt).
|
||
//
|
||
// Conditional on the CLAIM still being this caller's: a pass whose claim was taken over is not the
|
||
// pass whose answer counts, and two answers about one book are one answer too many.
|
||
//
|
||
// The debt is written HERE rather than after the engine call that pays it off, because everything
|
||
// between the two can fail: this is the transaction that makes the book parsed, and a book that is
|
||
// parsed owes a tree.
|
||
func (s *Store) FinishParse(ctx context.Context, id string, claimedAt time.Time, in ParsedBook) (time.Time, error) {
|
||
const q = `
|
||
update books set status = 'not_started', chapter_count = $2, source_sha256 = $3,
|
||
chunker_version = $4, parse_started_at = null,
|
||
` + owesAReadingSurface + `
|
||
revision = ` + nextRevisionOfThisBooksLibrary + `
|
||
where id = $1 and status = 'parsing' and parse_started_at = $5
|
||
returning read_model_owed_at`
|
||
var owed time.Time
|
||
err := s.inTx(ctx, func(tx pgx.Tx) error {
|
||
err := tx.QueryRow(ctx, q, id, in.Chapters, in.SourceSHA256, in.ChunkerVersion, claimedAt).Scan(&owed)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return ErrNoBook
|
||
}
|
||
if err != nil {
|
||
return fmt.Errorf("pgstore: finish parse: %w", err)
|
||
}
|
||
// The end of the parse is an ORDINARY status change and reaches the client as one: there is
|
||
// no "parsing finished" frame, deliberately, because the fact is the status (canon
|
||
// §streamBookEvents).
|
||
return emitStatus(ctx, tx, id)
|
||
})
|
||
return owed, err
|
||
}
|
||
|
||
// RejectBook is the terminal end of an intake that failed: `parsing → rejected`.
|
||
//
|
||
// The reason is the platform's own word for the class of failure and is not projected — contract v0
|
||
// has no field for it (migration 00013).
|
||
//
|
||
// Conditional on the CLAIM, like FinishParse and for a sharper reason: rejecting is destructive —
|
||
// the book's source goes with it — and a pass whose claim was stolen may be holding an exit code
|
||
// that says nothing about the book. The engine maps every failure onto exit 1, including a lock held
|
||
// by the parse that took the claim over.
|
||
func (s *Store) RejectBook(ctx context.Context, id string, claimedAt time.Time, reason string) error {
|
||
const q = `
|
||
update books set status = 'rejected', reject_reason = $2, parse_started_at = null,
|
||
revision = ` + nextRevisionOfThisBooksLibrary + `
|
||
where id = $1 and status = 'parsing' and parse_started_at = $3`
|
||
return s.inTx(ctx, func(tx pgx.Tx) error {
|
||
tag, err := tx.Exec(ctx, q, id, reason, claimedAt)
|
||
if err != nil {
|
||
return fmt.Errorf("pgstore: reject book: %w", err)
|
||
}
|
||
if tag.RowsAffected() == 0 {
|
||
return ErrNoBook
|
||
}
|
||
return emitStatus(ctx, tx, id)
|
||
})
|
||
}
|
||
|
||
// DeleteUpload removes a book whose file never arrived.
|
||
//
|
||
// Not `rejected`: that status means "the file could not be parsed" (contract §BookStatus), and a
|
||
// request the client abandoned produced no file to judge. The user's library must not accumulate a
|
||
// permanent row for every cancelled upload — there is no delete handle in the contract to clear one
|
||
// with.
|
||
//
|
||
// Narrow on purpose: only a book still in `uploading`, and only one no run was ever started for, so
|
||
// this can never become a way to delete a book that has cost money.
|
||
//
|
||
// ⚠ It also carries the library's revision ACROSS the deletion, and that is not bookkeeping: the
|
||
// library's number is derived as the maximum over the account's books, so removing the newest one
|
||
// would lower it — and the contract asks a client to DROP a read whose revision is below what it
|
||
// applied (§Revision). The screen would then keep rendering a book that no longer exists until some
|
||
// other book's counter grew past it. The floor lives in `users.library_revision`, the account-scope
|
||
// counter that has existed unused since 00001 (register row PD-122).
|
||
func (s *Store) DeleteUpload(ctx context.Context, id string) error {
|
||
return s.inTx(ctx, func(tx pgx.Tx) error {
|
||
return deleteIntakeRow(ctx, tx, id, func() pgx.Row {
|
||
return tx.QueryRow(ctx, `
|
||
delete from books where id = $1 and status = 'uploading'
|
||
and not exists (select 1 from runs where book_id = $1)
|
||
returning owner_id, revision`, id)
|
||
})
|
||
})
|
||
}
|
||
|
||
// DeleteRefusedIntake removes a book the intake refused during its own synchronous cut: the row is
|
||
// still `parsing` and the claim is the caller's.
|
||
//
|
||
// Conditional on the claim, like every other write that ends a parse pass.
|
||
func (s *Store) DeleteRefusedIntake(ctx context.Context, id string, claimedAt time.Time) error {
|
||
return s.inTx(ctx, func(tx pgx.Tx) error {
|
||
return deleteIntakeRow(ctx, tx, id, func() pgx.Row {
|
||
return tx.QueryRow(ctx, `
|
||
delete from books where id = $1 and status = 'parsing' and parse_started_at = $2
|
||
and not exists (select 1 from runs where book_id = $1)
|
||
returning owner_id, revision`, id, claimedAt)
|
||
})
|
||
})
|
||
}
|
||
|
||
// deleteIntakeRow runs the caller's delete under the book lock and carries the owner's library
|
||
// revision. `del` is a function so the statement is issued AFTER the lock: the package's global order
|
||
// is book first, then users (lockBook).
|
||
func deleteIntakeRow(ctx context.Context, tx pgx.Tx, id string, del func() pgx.Row) error {
|
||
if err := lockBook(ctx, tx, id); err != nil {
|
||
return err
|
||
}
|
||
var owner string
|
||
var revision int64
|
||
switch err := del().Scan(&owner, &revision); {
|
||
case errors.Is(err, pgx.ErrNoRows):
|
||
return ErrNoBook
|
||
case err != nil:
|
||
return fmt.Errorf("pgstore: delete intake: %w", err)
|
||
}
|
||
if _, err := tx.Exec(ctx, `
|
||
update users set library_revision = greatest(library_revision, $2) where id = $1`,
|
||
owner, revision+1); err != nil {
|
||
return fmt.Errorf("pgstore: carry the library revision: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// BookForMigration is one book and whether an engine SCHEMA migration of its project file is safe
|
||
// right now.
|
||
//
|
||
// It exists because an engine upgrade has an order and the order is money (unified backlog row 174).
|
||
// The engine's read path refuses a project file older than the binary reading it, so a deployment
|
||
// that migrates a book somebody is still using breaks that use — and the three ways a book is still
|
||
// in use are different facts that a single "is it running" question misses:
|
||
//
|
||
// - Live: a unit is going right now. Migrating under it is migrating under a writer.
|
||
// - Resumable: a run ENDED in a state its owner can continue from, and a resume runs on the build
|
||
// the run was PINNED to (row 139). Migrating moves the file past that build, so the resume either
|
||
// fails or is forced onto a newer one — and re-pinning a paid run is how already-bought calls get
|
||
// paid for twice.
|
||
// ⚠ `paused` LEFT this set in P7 and it is not an optimisation: since 0.3.0 a run stopped at a
|
||
// limit cannot be continued by `resume` at ALL — it answers 409 `ceiling_reached`, and the
|
||
// remedy is a NEW run, which is spawned with the CURRENT binary and has nothing pinned to
|
||
// migrate around (canon §resumeRun; register row PD-217, which is where a book stuck on the
|
||
// engine's daily ceiling blocked an upgrade for ever). The coupling is real and worth naming:
|
||
// the day `resume` continues a paused run again, this predicate has to come back with it.
|
||
// - Unsettled: an attempt's hold is still open, and the settlement reads the engine's committed
|
||
// figure with that attempt's pinned build. Migrate first and the figure can never be read, so the
|
||
// hold stays reserved with no sweep able to close it.
|
||
type BookForMigration struct {
|
||
ID string
|
||
Title string
|
||
Workdir string
|
||
// Live, Resumable and Unsettled are reported separately rather than folded into one boolean: an
|
||
// operator draining a stand needs to know WHICH of them to wait for, and they are cleared by
|
||
// different actions.
|
||
Live bool
|
||
Resumable bool
|
||
Unsettled bool
|
||
}
|
||
|
||
// Migratable reports whether this book's project file may be migrated now.
|
||
func (b BookForMigration) Migratable() bool { return !b.Live && !b.Resumable && !b.Unsettled }
|
||
|
||
// BooksForMigration lists every book with the three facts an engine upgrade turns on.
|
||
func (s *Store) BooksForMigration(ctx context.Context) ([]BookForMigration, error) {
|
||
const q = `
|
||
select b.id, b.title, b.workdir,
|
||
exists (select 1 from runs r
|
||
where r.book_id = b.id and r.finished_at is null),
|
||
exists (select 1 from runs r
|
||
where r.book_id = b.id and r.finished_at is not null
|
||
and r.status in ('stopped', 'awaiting_bank')),
|
||
exists (select 1 from runs r
|
||
join run_attempts a on a.run_id = r.id
|
||
join reservations res on res.engine_run_id = r.id || '#' || a.attempt_no
|
||
and res.state = 'open'
|
||
where r.book_id = b.id)
|
||
from books b
|
||
order by b.added_at, b.id`
|
||
rows, err := s.pool.Query(ctx, q)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("pgstore: list books for migration: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
var out []BookForMigration
|
||
for rows.Next() {
|
||
var b BookForMigration
|
||
if err := rows.Scan(&b.ID, &b.Title, &b.Workdir, &b.Live, &b.Resumable, &b.Unsettled); err != nil {
|
||
return nil, fmt.Errorf("pgstore: scan book for migration: %w", err)
|
||
}
|
||
out = append(out, b)
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// IntakeBook is a book the backstop sweep has to make a decision about.
|
||
type IntakeBook struct {
|
||
ID string
|
||
Status string
|
||
Workdir string
|
||
Attempts int
|
||
}
|
||
|
||
// owesAReadingSurface is what a transaction that ENDS a run adds to its `update books`. It is a
|
||
// fragment rather than three copies of one assignment because there are three such transactions —
|
||
// FinishRun, PauseRun and FinishUnspawnedStop — and the debt is only a mechanism while EVERY one of
|
||
// them writes it: a run whose ending forgets it is a run the user paid for whose text nothing will
|
||
// ever materialize, and nothing looks at that book again.
|
||
//
|
||
// ⚠ It also RESETS the attempt budget, and that is part of the same fragment on purpose: a new
|
||
// boundary is new evidence. A book whose debt was written off as unpayable (read_model_abandoned_at)
|
||
// has just finished a run, so the engine may well answer about it now — and a fresh boundary that
|
||
// inherited a spent budget would be written off without a single try.
|
||
const owesAReadingSurface = `read_model_owed_at = now(), read_model_attempts = 0,
|
||
read_model_abandoned_at = null, read_model_error = null, `
|
||
|
||
// OwedBook is a book whose reading surface is DUE a materialization. OwedAt is the stamp the caller
|
||
// read, and every write below is conditional on it still standing: a debt a later boundary stamped,
|
||
// or another worker claimed, is not the one this caller is holding.
|
||
type OwedBook struct {
|
||
ID string
|
||
Workdir string
|
||
OwedAt time.Time
|
||
// Attempts is how many passes have already failed to pay this debt. It travels with the book
|
||
// because the decision it feeds — try again later, or write it off — belongs to the worker that
|
||
// just failed, and a counter kept in the process would restart with the daemon and never reach
|
||
// any threshold. Same shape and same reason as `books.parse_attempts` at the intake.
|
||
Attempts int
|
||
}
|
||
|
||
// BooksOwedReadModel lists books whose reading surface is DUE a materialization and that are quiet
|
||
// enough to give it — the same "nothing is running" the stream ends on.
|
||
//
|
||
// Quiet, because the contract puts the freshness of a pair's text at the BOUNDARIES of the work: a
|
||
// book with a live run is not at one. (It is not a lock: the engine's read commands open the project
|
||
// WITHOUT its exclusive flock — backend/internal/store/store.go, NewReadOnlyRunner — precisely so an
|
||
// operator is not shut out of `status` for the length of a run.)
|
||
//
|
||
// DUE and not merely owed: a worker claims a book by pushing its stamp out (ClaimReadModelDebt), so
|
||
// the book the intake is materializing right now is not offered to the sweep as well. Without that
|
||
// every upload slower than one sweep interval was materialized twice at once — the cost this pack
|
||
// removed elsewhere, paid back with interest.
|
||
//
|
||
// Bounded, because a host that cannot run the engine would otherwise put every book of the
|
||
// deployment into one pass.
|
||
func (s *Store) BooksOwedReadModel(ctx context.Context, limit int) ([]OwedBook, error) {
|
||
const q = `
|
||
select b.id, b.workdir, b.read_model_owed_at, b.read_model_attempts
|
||
from books b
|
||
where b.read_model_owed_at is not null and b.read_model_owed_at <= now()
|
||
and ` + nothingIsRunning + `
|
||
order by b.read_model_owed_at limit $1`
|
||
rows, err := s.pool.Query(ctx, q, limit)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("pgstore: list books owed a reading surface: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
var out []OwedBook
|
||
for rows.Next() {
|
||
var b OwedBook
|
||
if err := rows.Scan(&b.ID, &b.Workdir, &b.OwedAt, &b.Attempts); err != nil {
|
||
return nil, fmt.Errorf("pgstore: scan book owed a reading surface: %w", err)
|
||
}
|
||
out = append(out, b)
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// ClaimReadModelDebt takes a debt for one worker, by pushing its DUE time `window` into the future.
|
||
// It answers the new stamp — which is what that worker must present to discharge or defer it — or
|
||
// the zero time when the debt is no longer the one the caller read: another worker claimed it, or a
|
||
// later boundary stamped a new one.
|
||
//
|
||
// The window is a LEASE and not a promise: a worker that dies holding one simply leaves the book
|
||
// invisible until it lapses, which is the same shape the intake's parse claim has.
|
||
func (s *Store) ClaimReadModelDebt(ctx context.Context, bookID string, owedAt time.Time, window time.Duration) (time.Time, error) {
|
||
var claimed time.Time
|
||
err := s.pool.QueryRow(ctx, `
|
||
update books set read_model_owed_at = now() + make_interval(secs => $3)
|
||
where id = $1 and read_model_owed_at = $2
|
||
returning read_model_owed_at`, bookID, owedAt, window.Seconds()).Scan(&claimed)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return time.Time{}, nil
|
||
}
|
||
if err != nil {
|
||
return time.Time{}, fmt.Errorf("pgstore: claim the reading-surface debt: %w", err)
|
||
}
|
||
return claimed, nil
|
||
}
|
||
|
||
// AttemptCost says whether a deferral counts against the book's attempt budget.
|
||
type AttemptCost bool
|
||
|
||
const (
|
||
// SpendsAnAttempt — the engine answered ABOUT THIS BOOK and the answer was a failure.
|
||
SpendsAnAttempt AttemptCost = true
|
||
// CostsNoAttempt — the failure belongs to the deployment (ingest.DeploymentFault) and applies to
|
||
// every book on the host, so counting it would write them all off in a few passes.
|
||
CostsNoAttempt AttemptCost = false
|
||
)
|
||
|
||
// DeferReadModelDebt moves a debt this pass could not pay to the BACK of the queue, counts the
|
||
// failure when it was the book's own, and answers how many there have now been.
|
||
//
|
||
// The queue is oldest-first, so a book the engine can never answer about — a workdir an operator
|
||
// moved, a project database this build cannot read — would be retried first forever, and enough of
|
||
// them would starve every book behind them out of text that was already paid for. Conditional on the
|
||
// stamp for the same reason the discharge is: a boundary recorded since this pass began is a newer
|
||
// debt and must not be overwritten.
|
||
//
|
||
// ⚠ `now()` was the whole of it, and "the back of the queue" was therefore a queue of one: the
|
||
// pass runs every 15 seconds, `<= now()` is true again immediately, and the same book was re-read
|
||
// by the engine four times a minute for as long as it existed. The caller passes a real deadline
|
||
// now, and the count it gets back is what ends the loop (AbandonReadModelDebt).
|
||
func (s *Store) DeferReadModelDebt(ctx context.Context, bookID string, owedAt, next time.Time, reason string, cost AttemptCost) (int, error) {
|
||
var attempts int
|
||
err := s.pool.QueryRow(ctx, `
|
||
update books set read_model_owed_at = $3, read_model_error = $4,
|
||
read_model_attempts = read_model_attempts + case when $5 then 1 else 0 end
|
||
where id = $1 and read_model_owed_at = $2
|
||
returning read_model_attempts`, bookID, owedAt, next, truncateReason(reason), bool(cost)).Scan(&attempts)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return 0, nil // a newer boundary or another worker owns it now: not this caller's debt to defer
|
||
}
|
||
if err != nil {
|
||
return 0, fmt.Errorf("pgstore: defer the reading-surface debt: %w", err)
|
||
}
|
||
return attempts, nil
|
||
}
|
||
|
||
// AbandonReadModelDebt writes a debt off as unpayable: the book keeps whatever surface it has and
|
||
// stops being asked about.
|
||
//
|
||
// Giving up is SAFE here and that asymmetry is why this exists at all while a stalled RUN is only
|
||
// ever handed to an operator. The money of the boundary that stamped this debt has already settled,
|
||
// so what is lost is the freshness of text — while what was lost by never giving up is three things,
|
||
// all of them ongoing and all of them named by the register: up to five minutes of engine processes
|
||
// per pass forever; a book whose event stream can NEVER end, because `AtRest` requires this column
|
||
// to be null and a browser therefore reconnects to it for good; and, where the manifest reads and
|
||
// the export does not, a committed `SaveStructure` on every pass — a revision bump and a frame every
|
||
// fifteen seconds, to say nothing changed.
|
||
//
|
||
// It is not final in the sense that matters to a user: the next boundary of real work stamps a fresh
|
||
// debt and clears this (owesAReadingSurface), because a new boundary is new evidence.
|
||
func (s *Store) AbandonReadModelDebt(ctx context.Context, bookID string, owedAt, now time.Time, reason string) error {
|
||
// The attempt that failed is COUNTED here as everywhere else. Without it the column said one less
|
||
// than had actually been tried, so `books --abandoned` reported four tries where the log line
|
||
// beside it said five — two numbers about one thing, which is how an operator learns to trust
|
||
// neither.
|
||
if _, err := s.pool.Exec(ctx, `
|
||
update books set read_model_owed_at = null, read_model_attempts = read_model_attempts + 1,
|
||
read_model_abandoned_at = $3, read_model_error = $4
|
||
where id = $1 and read_model_owed_at = $2`,
|
||
bookID, owedAt, now, truncateReason(reason)); err != nil {
|
||
return fmt.Errorf("pgstore: abandon the reading-surface debt: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// truncateReason bounds what an engine's or a driver's error string may put into a column. These
|
||
// fields are an OPERATOR's, never a client's: they carry paths, unit names and driver text, and the
|
||
// wire's vocabulary of failure is a closed enum decided elsewhere.
|
||
//
|
||
// ⚠ IT SANITISES BEFORE IT CUTS, AND BOTH HALVES ARE LOAD-BEARING — this is not defensive habit.
|
||
// What arrives here is the first line of the ENGINE's stderr, verbatim and unbounded
|
||
// (runner.firstLine), for a product whose sources are Chinese and Japanese: a message quoting the
|
||
// book's own text is both long and multi-byte. Postgres refuses a `text` value that is not valid
|
||
// UTF-8 (SQLSTATE 22021), so a byte-slice through the middle of a rune makes the write FAIL — and
|
||
// the write that fails is precisely the one that records the failure. The book or the run then
|
||
// keeps its old deadline, its counter never rises, and it stands at the head of the queue for good:
|
||
// the exact starvation this whole pack exists to end, re-created by the bookkeeping of it. Arbitrary
|
||
// stderr is not guaranteed to be valid UTF-8 to begin with either, which is why the repair comes
|
||
// first and the cut second.
|
||
//
|
||
// ⚠ NUL DEFEATS THE REPAIR ABOVE: U+0000 is valid UTF-8 and Postgres refuses it anyway, same
|
||
// SQLSTATE, same self-defeating failure. No live path carrying one was found — the engine gates NUL
|
||
// on every decode branch — so this closes a hole in the guard rather than a reproduced defect.
|
||
func truncateReason(reason string) string {
|
||
const max = 1000
|
||
reason = strings.ToValidUTF8(reason, "")
|
||
reason = strings.ReplaceAll(reason, "\x00", "")
|
||
if len(reason) <= max {
|
||
return reason
|
||
}
|
||
cut := max
|
||
for cut > 0 && !utf8.RuneStart(reason[cut]) {
|
||
cut--
|
||
}
|
||
return reason[:cut]
|
||
}
|
||
|
||
// RearmReadModelDebt is the operator's handle on a written-off debt: ask for the materialization
|
||
// again, with a clean budget, without waiting for the book's next run.
|
||
//
|
||
// It answers false when there was nothing to re-arm — the book already owes one, or has never owed
|
||
// one — so the CLI can say which of the two happened instead of reporting a success that did nothing.
|
||
func (s *Store) RearmReadModelDebt(ctx context.Context, bookID string, now time.Time) (bool, error) {
|
||
tag, err := s.pool.Exec(ctx, `
|
||
update books set read_model_owed_at = $2, read_model_attempts = 0,
|
||
read_model_abandoned_at = null, read_model_error = null
|
||
where id = $1 and read_model_owed_at is null`, bookID, now)
|
||
if err != nil {
|
||
return false, fmt.Errorf("pgstore: re-arm the reading-surface debt: %w", err)
|
||
}
|
||
return tag.RowsAffected() > 0, nil
|
||
}
|
||
|
||
// AbandonedSurface is a book whose reading surface the platform gave up on, as an operator sees it.
|
||
type AbandonedSurface struct {
|
||
ID string
|
||
Title string
|
||
Workdir string
|
||
Attempts int
|
||
AbandonedAt time.Time
|
||
LastError string
|
||
}
|
||
|
||
// AbandonedSurfaces lists them. The gauge says how many; this says which, and what they said.
|
||
func (s *Store) AbandonedSurfaces(ctx context.Context) ([]AbandonedSurface, error) {
|
||
rows, err := s.pool.Query(ctx, `
|
||
select id, title, workdir, read_model_attempts, read_model_abandoned_at,
|
||
coalesce(read_model_error, '')
|
||
from books where read_model_abandoned_at is not null
|
||
order by read_model_abandoned_at`)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("pgstore: list abandoned reading surfaces: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
var out []AbandonedSurface
|
||
for rows.Next() {
|
||
var b AbandonedSurface
|
||
if err := rows.Scan(&b.ID, &b.Title, &b.Workdir, &b.Attempts, &b.AbandonedAt, &b.LastError); err != nil {
|
||
return nil, fmt.Errorf("pgstore: scan abandoned reading surface: %w", err)
|
||
}
|
||
out = append(out, b)
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// ClearReadModelDebt discharges the debt stamped at one boundary, and only that one.
|
||
//
|
||
// The equality is the whole method: materializing a large book is minutes of engine time, and a run
|
||
// can finish inside that window. Clearing unconditionally would answer the NEW boundary with a read
|
||
// taken before it, and the text of that run would never reach its reader.
|
||
func (s *Store) ClearReadModelDebt(ctx context.Context, bookID string, owedAt time.Time) error {
|
||
if _, err := s.pool.Exec(ctx, `
|
||
update books set read_model_owed_at = null, read_model_attempts = 0, read_model_error = null
|
||
where id = $1 and read_model_owed_at = $2`,
|
||
bookID, owedAt); err != nil {
|
||
return fmt.Errorf("pgstore: clear the reading-surface debt: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// StuckIntake lists books that stopped moving through intake.
|
||
//
|
||
// Both halves have the same cause and different cures: an `uploading` row whose request is gone is
|
||
// litter to remove, a `parsing` row whose claim went stale is work to retry. Neither is reachable
|
||
// from the request that created it — the process holding it is the one that died — so the sweep is
|
||
// the only thing that can finish the walk.
|
||
func (s *Store) StuckIntake(ctx context.Context, uploadingBefore, claimedBefore time.Time) ([]IntakeBook, error) {
|
||
// coalesce(claimed, added): an unclaimed parse is stuck when the BOOK has been waiting, a claimed
|
||
// one when the CLAIM has. Written as one expression because they are the same question — how long
|
||
// has this book had nobody working on it — and a NULL claim is simply the first answer to it.
|
||
const q = `
|
||
select id, status, workdir, parse_attempts
|
||
from books
|
||
where (status = 'uploading' and added_at < $1)
|
||
or (status = 'parsing' and coalesce(parse_started_at, added_at) < $2)
|
||
order by added_at`
|
||
rows, err := s.pool.Query(ctx, q, uploadingBefore, claimedBefore)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("pgstore: list stuck intake: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
var out []IntakeBook
|
||
for rows.Next() {
|
||
var b IntakeBook
|
||
if err := rows.Scan(&b.ID, &b.Status, &b.Workdir, &b.Attempts); err != nil {
|
||
return nil, fmt.Errorf("pgstore: scan stuck intake: %w", err)
|
||
}
|
||
out = append(out, b)
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// Progress is how far a run's WHOLE work has got, as one monotonic fraction (owner's word of 20.08,
|
||
// row 200): `done` counts chapter-passes through both waves and `total` is what the run bought,
|
||
// doubled where the pipeline has an editor. The counting itself is pgstore.runDone/runTotal.
|
||
//
|
||
// ONE counter and not two per wave, and in chapter-passes and not units, and both halves are the
|
||
// same decision: the numerator and the denominator have to be in one unit, the denominator derives
|
||
// from what the run BOUGHT, and what a run buys is declared in chapters (`ceiling_chapters`). The
|
||
// per-phase split stays INSIDE the platform — the wire carried it until 0.3.0, which pinned the
|
||
// engine's architecture to a React component through six files. What crosses the wire beside the
|
||
// numbers is `stage`, the caption's machine value — the client draws its own phrase from it.
|
||
type Progress struct {
|
||
Done int
|
||
Total int
|
||
// Stage is what the run is doing now: `drafting`, `editing` or `re_pass` today, an open vocabulary
|
||
// by design (readmodel.go, the `runStage` expression).
|
||
Stage string
|
||
// ETASeconds is absent when there is nothing to estimate from.
|
||
ETASeconds *int
|
||
}
|
||
|
||
// Run is the run row, in contract terms.
|
||
type Run struct {
|
||
ID string
|
||
BookID string
|
||
Revision int64
|
||
Status string
|
||
// VerifyBank is the engine flag's name for what the product calls "stop for signing". The column
|
||
// keeps the flag's name because that is what it reaches the engine as; the wire carries the
|
||
// product's.
|
||
VerifyBank bool
|
||
// OrderedChapters and DeliveredChapters are what this run BOUGHT and what it has actually handed
|
||
// over, in chapters. The pair is what makes «continue» explicable to a buyer: the remedy for a run
|
||
// that stopped short is to order what is LEFT, and the two numbers are what «left» is made of
|
||
// (unified backlog row 279).
|
||
//
|
||
// ⚠ The column behind OrderedChapters is still called `ceiling_chapters`, and the word is the one
|
||
// thing about it that was wrong: it never was a ceiling in the money sense — the money ceiling is
|
||
// the hold — it was always how much book the run was sold.
|
||
//
|
||
// ⛔ NIL FOR A RUN SOLD IN CHARACTERS — see runOrderedChapters for why the row still carries a
|
||
// number there and only the view is null. OrderedUnits is the figure such a run was sold in.
|
||
OrderedChapters *int
|
||
// OrderedUnits is how much this run bought in the engine's own output units, and nil for a run
|
||
// bought in chapters. It is the other half of OrderedChapters: exactly one of the two is set, and
|
||
// which one says what unit this run — its bar, its allowance, its delivery — is measured in.
|
||
//
|
||
// ⚠ It is NOT the number the buyer typed. A character order is resolved into units at admission
|
||
// (`runs.resolveOrder` through `pricing.UnitsFor`) and the characters themselves are not stored
|
||
// anywhere; a reloaded screen therefore answers in units even though the question was asked in
|
||
// characters. Whether that echo is owed to the buyer is a product question, named in the pack's
|
||
// report rather than decided here.
|
||
OrderedUnits *int
|
||
// DeliveredChapters is NIL for a run whose order does not close whole chapters: such a run has
|
||
// delivered no CHAPTER, and reporting `0` would read as «nothing happened» rather than as «this
|
||
// is not the unit this run is measured in». Progress carries what it HAS done instead, and for
|
||
// such a run its `done`/`total` are counted in units rather than chapters.
|
||
DeliveredChapters *int
|
||
// BondFunded is whether this run's hold has room for the book-wide consistency pass on top of
|
||
// what it bought. It is the order form's promise, kept where it can still be read after the
|
||
// click — see migration 00033 for why the form's answer alone is not enough.
|
||
BondFunded bool
|
||
Progress Progress
|
||
PausedReason string
|
||
// FailureReason is the contract's RunFailureReason, and it is the one state that IS an error
|
||
// carrying why. Written when the run is closed, from what the engine's exit said.
|
||
FailureReason string
|
||
// StopRequested is the user's own click, and it answers what no status can: stopping is not
|
||
// instant, so a stop asked for mid-translation can meet the run reaching a stop of its OWN — the
|
||
// bank signature — and the run then ends in `awaiting_bank`, a status that offers to continue on
|
||
// a click that meant "stop" (canon §Run.stop_requested). Cleared by a resume, with the intent it
|
||
// records.
|
||
StopRequested bool
|
||
StartedAt time.Time
|
||
FinishedAt *time.Time
|
||
// ProgressLagging says the figures in Progress are BEHIND this run while the run itself goes on:
|
||
// the live attempt's projection is not following it. Two states produce exactly that and the
|
||
// buyer cannot tell them apart from outside — a PARK (the journal continued under another stream
|
||
// id and the tailer stopped there) and a QUARANTINE (the stream cannot be materialised at all) —
|
||
// so one fact answers for both. In both the run is spending money and the bar is standing.
|
||
//
|
||
// ⚠ IT EXISTS BECAUSE THE SCREEN DOES NOT SIMPLY FREEZE, IT MOVES WRONG. The repair channel keeps
|
||
// refreshing the estimate and the freshness stamp of a parked or quarantined attempt (ApplyStatus)
|
||
// while the bar is derived from `chapters`, which that channel does not write — so a client
|
||
// reading `eta_seconds` beside `done/total` shows a moving estimate over a standing bar, which
|
||
// reads as progress. Four operator-facing carriers already exist for the park (a column, a writer,
|
||
// this read, a gauge) and the paying reader had none (unified backlog row 399).
|
||
ProgressLagging bool
|
||
// RebillConsent is what this run may spend AGAIN on work already paid for — the consent a user's
|
||
// own resume of a corrected book grants (P10 §3.1) — and zero where none stands.
|
||
//
|
||
// ⚠ IT IS VISIBILITY AND PROVENANCE, NOT A CEILING, and the difference is the whole of what this
|
||
// zone can honestly do here. The figure is the run's OWN hold, because a projection of what
|
||
// re-making the corrected text would cost does not exist at the point of decision by construction
|
||
// (reconcile.reopen, errata 28.08-к): only the engine can compute it, and only after the next
|
||
// translation folds the correction in. Measured once at 78× the work actually re-bought (unified
|
||
// backlog row 416) — from over-purchase the run was saved by the engine's checkpoints, not by this
|
||
// number — so what the platform owes the buyer is to SAY which figure they are signing and where
|
||
// it came from.
|
||
RebillConsent money.MicroUSD
|
||
}
|
||
|
||
// runRow is the contract's Run, in the ONE shape every path that hands one out reads it — `r` the
|
||
// run and `b` its book. Written once because the three that used it wrote the same fourteen columns
|
||
// three times, in three orders that had to stay in step by hand.
|
||
const runRow = `r.id, b.id, b.revision, r.status, r.verify_bank, r.stop_requested_at is not null,
|
||
` + runOrderedChapters + `, r.ordered_units, ` + runDelivered + `, r.bond_funded, ` + runDone + `, ` + runTotal + `, ` + runStage + `, r.eta_seconds,
|
||
coalesce(r.paused_reason, ''), coalesce(r.failure_reason, ''), r.started_at, r.finished_at,
|
||
exists (select 1 from run_attempts a where a.run_id = r.id and a.ended_at is null
|
||
and (a.parked_at is not null or a.quarantine_reason is not null)),
|
||
r.accept_rebill_micro`
|
||
|
||
// scanRun reads runRow into a Run, plus whatever the caller selected after it.
|
||
func scanRun(row pgx.Row, out *Run, extra ...any) error {
|
||
return row.Scan(append([]any{&out.ID, &out.BookID, &out.Revision, &out.Status, &out.VerifyBank,
|
||
&out.StopRequested, &out.OrderedChapters, &out.OrderedUnits, &out.DeliveredChapters, &out.BondFunded,
|
||
&out.Progress.Done, &out.Progress.Total,
|
||
&out.Progress.Stage, &out.Progress.ETASeconds, &out.PausedReason, &out.FailureReason,
|
||
&out.StartedAt, &out.FinishedAt, &out.ProgressLagging, &out.RebillConsent}, extra...)...)
|
||
}
|
||
|
||
// Library is one page of a user's books plus the revision of the library scope itself.
|
||
type Library struct {
|
||
Revision int64
|
||
Books []Book
|
||
NextCursor string
|
||
}
|
||
|
||
// ListBooks returns one page of the library, newest first.
|
||
//
|
||
// Keyset, not offset: the library is written to while it is read (a run bumps its book's revision on
|
||
// every materialized transaction), and an offset page silently skips or repeats rows under that.
|
||
//
|
||
// ⚠ The page and the SCOPE's revision are read in ONE transaction, and that is the contract rather
|
||
// than tidiness. The revision travels with the page as "what this page is a picture of", and a
|
||
// client MUST DROP a read whose revision is not above one it has applied (§Revision). Read as a
|
||
// second statement, the number was NEWER than the rows it labelled — anything materialized between
|
||
// the two got a revision the client would then treat as already seen, and every frame in that window
|
||
// was lost for good. Register row PD-163, closed before the stream that would have made it visible.
|
||
func (s *Store) ListBooks(ctx context.Context, userID string, limit int, cursor string) (Library, error) {
|
||
limit = clampPage(limit)
|
||
after, err := decodeCursor(userID, cursor)
|
||
if err != nil {
|
||
return Library{}, err
|
||
}
|
||
var out Library
|
||
err = s.inReadTx(ctx, func(tx pgx.Tx) error {
|
||
var err error
|
||
out, err = listBooksTx(ctx, tx, userID, limit, after)
|
||
return err
|
||
})
|
||
if err != nil {
|
||
return Library{}, err
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func listBooksTx(ctx context.Context, tx pgx.Tx, userID string, limit int, after cursorPos) (Library, error) {
|
||
// The library row carries no progress bar: the bar belongs to the RUN, and the book's own
|
||
// figure is `chapters_done` against `chapter_count` (canon §Book).
|
||
const q = `
|
||
select ` + bookColumns + `
|
||
from books b ` + lastRun + `
|
||
where b.owner_id = $1
|
||
and ($2::timestamptz is null or (b.added_at, b.id) < ($2::timestamptz, $3::text))
|
||
order by b.added_at desc, b.id desc
|
||
limit $4`
|
||
rows, err := tx.Query(ctx, q, userID, after.time, after.id, limit+1)
|
||
if err != nil {
|
||
return Library{}, fmt.Errorf("pgstore: list books: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
var out Library
|
||
for rows.Next() {
|
||
b, err := scanBook(rows)
|
||
if err != nil {
|
||
return Library{}, err
|
||
}
|
||
out.Books = append(out.Books, b)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return Library{}, fmt.Errorf("pgstore: list books: %w", err)
|
||
}
|
||
// One row was asked for beyond the page so that "is there a next page" is answered by data
|
||
// rather than by "the page came back full", which is wrong exactly when the last page is full.
|
||
if len(out.Books) > limit {
|
||
last := out.Books[limit-1]
|
||
out.Books = out.Books[:limit]
|
||
out.NextCursor = encodeCursor(userID, last.AddedAt, last.ID)
|
||
}
|
||
// The revision of the LIBRARY scope — membership and statuses — and never compared with the
|
||
// revision of a book (contract §Revision).
|
||
//
|
||
// The maximum over the account's books, FLOORED by the account's own counter. The floor is what
|
||
// makes the number monotonic across a deletion: the maximum alone falls when the newest book goes
|
||
// (DeleteUpload raises the floor as it deletes), and a library revision that goes backwards is a
|
||
// library the client stops rendering.
|
||
if err := tx.QueryRow(ctx, `
|
||
select greatest(coalesce((select max(revision) from books where owner_id = $1), 0),
|
||
coalesce((select library_revision from users where id = $1), 0))`,
|
||
userID).Scan(&out.Revision); err != nil {
|
||
return Library{}, fmt.Errorf("pgstore: library revision: %w", err)
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// DefaultPage is how many rows a collection returns when `limit` is omitted, and maxPage is the most
|
||
// a client may ASK for. Exported because `GET /capabilities` answers the first one: a client's
|
||
// default and the server's have to be the same number, and the way to make them the same is for
|
||
// there to be only one.
|
||
const (
|
||
DefaultPage = 100
|
||
maxPage = 1000
|
||
)
|
||
|
||
// GetBook returns the book card: the book and its current or last run.
|
||
//
|
||
// ⚠ One TRANSACTION, for the reason ListBooks is one: the card carries a revision, a client drops
|
||
// anything not above what it applied, and the book's number was read before the run's rows. A run
|
||
// that moved in between produced a card labelled with a revision older than the figures on it, and
|
||
// the client obeying the contract discarded exactly that update (register row PD-163).
|
||
func (s *Store) GetBook(ctx context.Context, userID, bookID string) (Book, *Run, error) {
|
||
var b Book
|
||
var run *Run
|
||
err := s.inReadTx(ctx, func(tx pgx.Tx) error {
|
||
var err error
|
||
if b, err = scanBook(tx.QueryRow(ctx,
|
||
`select `+bookColumns+` from books b `+lastRun+` where b.id = $1 and b.owner_id = $2`,
|
||
bookID, userID)); err != nil {
|
||
return err
|
||
}
|
||
run, err = lastRunTx(ctx, tx, bookID)
|
||
return err
|
||
})
|
||
if err != nil {
|
||
return Book{}, nil, err
|
||
}
|
||
if run != nil {
|
||
// ONE counter per book, and this is the single place that says so for every run this store
|
||
// hands out (§Revision: "every book-scoped read … carry the same number"). Read off the run's
|
||
// own column the card lagged — a unit_done bumps the book and the chapter and not the run, so
|
||
// a client that had applied frame id=2 got revision 0 back and, obeying the contract, dropped
|
||
// the read. The stop and resume handles answer from ReadRun/RequestStop, which select the
|
||
// book's revision for the same reason.
|
||
run.Revision = b.Revision
|
||
}
|
||
return b, run, nil
|
||
}
|
||
|
||
// lastRunTx reads a book's current or last run, with the bar of its current segment.
|
||
func lastRunTx(ctx context.Context, tx pgx.Tx, bookID string) (*Run, error) {
|
||
const q = `select ` + runRow + ` from books b ` + lastRun + ` where b.id = $1 and r.id is not null`
|
||
var r Run
|
||
err := scanRun(tx.QueryRow(ctx, q, bookID), &r)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return nil, nil
|
||
}
|
||
if err != nil {
|
||
return nil, fmt.Errorf("pgstore: read run: %w", err)
|
||
}
|
||
return &r, nil
|
||
}
|
||
|
||
// BookRunContext is what starting a run needs to know about a book, read under the caller's
|
||
// ownership so a stranger's book is indistinguishable from a missing one.
|
||
type BookRunContext struct {
|
||
Workdir string
|
||
// Status is the book's own state. A run may only be started on a book whose intake finished:
|
||
// the directory of a book still `uploading` holds half a file and the directory of a `rejected`
|
||
// one may hold nothing at all.
|
||
Status string
|
||
// ⚠ ChaptersLeft is GONE, and its removal is the point rather than a tidy-up: it existed to clamp
|
||
// the ceiling SCALE, and there is no scale — what is left of a book is now read per chapter, with
|
||
// its price, by ReadBookForOrder. A field whose doc promises a mechanism that no longer exists is
|
||
// worse than no field: the next reader builds against the promise.
|
||
//
|
||
// ChapterCount is the book's whole length — what a re-pass hold prices, since a re-pass may in
|
||
// the worst case re-translate everything the correction touched, up to the book.
|
||
ChapterCount int
|
||
HasLiveRun bool
|
||
// LiveRunAwaitingBank narrows HasLiveRun for the correction door: a bank stop moves the run's
|
||
// status to `awaiting_bank` from the journal BEFORE the reconciler reads the exit marker and
|
||
// closes the row (sink TypeBankStop; reconcile.finish) — so for up to a sweep the row is live
|
||
// while the book already shows its signing screen. In that window the door must open, not
|
||
// answer run_in_flight; Start keeps refusing on HasLiveRun alone (the row IS still live, and a
|
||
// second one would break runs_one_live_per_book).
|
||
LiveRunAwaitingBank bool
|
||
// BankMoved is the re-pass door's fact (P10, errata 28.08-к): a correction landed that the
|
||
// already-translated text does not carry, and no --resnapshot run has finished `ready` since.
|
||
//
|
||
// ⚠ IT IS TRUE FOR TWO FACTS, NOT ONE, and the second is a write-ahead mark: a verb that was
|
||
// STARTED and whose outcome nobody recorded (migration 00037). A process that dies between the
|
||
// engine's files landing and the stamp would otherwise leave this false forever, and the next run
|
||
// dies on the engine's snapshot guard with its hold already taken (PD-425, unified backlog row
|
||
// 400). Reading the two together is what makes that death cost one harmless --resnapshot instead
|
||
// of one dead paid run — the asymmetry the clearing branch of reconcile.finish already argues.
|
||
// A plain standing flag, deliberately NOT a timestamp comparison: the door stamps it from its
|
||
// own receipt and reconcile.finish clears it explicitly on a ready resnapshot run, so there is
|
||
// no clock pairing to skew and no failed run to retire it early (the adversarial pass's K6 and
|
||
// its clock dispute both die with the predicate). The asymmetry is chosen: a STALE standing
|
||
// fact costs one harmless --resnapshot (the engine reads the flag only when a snapshot actually
|
||
// moved), a falsely-retired one kills the next run on the guard after its hold.
|
||
BankMoved bool
|
||
// HasPriorRun is whether this book has ever had a run at all. It is the second condition for
|
||
// `--resnapshot` beside the correction door's flag — see SpawnOrder.HasPriorRun for the whole of
|
||
// PD-422 and why the flag is harmless when it is not needed.
|
||
HasPriorRun bool
|
||
// HasTree is whether the book's chapter tree has been MATERIALISED — the `chapters` rows the whole
|
||
// read model counts over — as opposed to merely declared by `chapter_count`.
|
||
//
|
||
// The two are separate facts with separate writers, and the window between them is ordinary: the
|
||
// intake commits `not_started` in one transaction and materialises the tree afterwards, outside
|
||
// it. On the healthy path the window is milliseconds; when the materialisation FAILED or was
|
||
// deferred it is unbounded, and a run admitted inside it is a run whose bar can never move — every
|
||
// counter the screen shows is a count over `chapters`, and there are no rows to count (PD-405).
|
||
// Worse, the tree's debt is frozen for the run's whole life, because the sweep that would pay it
|
||
// skips a book with a live run. So the run is refused at admission instead, before its hold.
|
||
HasTree bool
|
||
}
|
||
|
||
// ReadBookForRun gathers the facts a run start is judged on.
|
||
func (s *Store) ReadBookForRun(ctx context.Context, userID, bookID string) (BookRunContext, error) {
|
||
const q = `
|
||
select b.workdir, b.status,
|
||
b.chapter_count,
|
||
exists (select 1 from runs lr where lr.book_id = b.id and lr.finished_at is null),
|
||
exists (select 1 from runs lr where lr.book_id = b.id and lr.finished_at is null
|
||
and lr.status = 'awaiting_bank'),
|
||
(b.bank_moved_at is not null or b.bank_move_pending_at is not null),
|
||
exists (select 1 from runs pr where pr.book_id = b.id),
|
||
exists (select 1 from chapters c where c.book_id = b.id)
|
||
from books b ` + lastRun + ` where b.id = $1 and b.owner_id = $2`
|
||
var out BookRunContext
|
||
err := s.pool.QueryRow(ctx, q, bookID, userID).Scan(&out.Workdir, &out.Status,
|
||
&out.ChapterCount, &out.HasLiveRun, &out.LiveRunAwaitingBank, &out.BankMoved,
|
||
&out.HasPriorRun, &out.HasTree)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return BookRunContext{}, ErrNoBook
|
||
}
|
||
if err != nil {
|
||
return BookRunContext{}, fmt.Errorf("pgstore: read book for run: %w", err)
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// RecordBankMove is the correction door's write: the bank moved. Stamped from the door's OWN
|
||
// receipt — never from the engine's status projection, which is blind to a correction until the
|
||
// next translate folds the bank into memory (P10 errata 28.08-к, the adversarial pass's K1). A
|
||
// plain flag: what re-passing would COST is unknowable at this moment by construction, so nothing
|
||
// pretends to price it.
|
||
// ⚠ IT CLEARS THE WRITE-AHEAD MARK IN THE SAME STATEMENT, and that is not tidiness: the mark exists
|
||
// only to stand in for this stamp when nobody got to write it, so a mark surviving its own stamp
|
||
// would be a fact that no ending retires — `ClearBankMove` would clear the stamp and leave the run
|
||
// carrying --resnapshot for ever.
|
||
func (s *Store) RecordBankMove(ctx context.Context, bookID string) error {
|
||
tag, err := s.pool.Exec(ctx,
|
||
`update books set bank_moved_at = now(), bank_move_pending_at = null where id = $1`, bookID)
|
||
if err != nil {
|
||
return fmt.Errorf("pgstore: record bank move: %w", err)
|
||
}
|
||
if tag.RowsAffected() == 0 {
|
||
return ErrNoBook
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ClearBankMove retires the fact — called by the reconciler when a run that CARRIED --resnapshot
|
||
// finished `ready`: the correction has demonstrably reached the translated text. Deliberately not
|
||
// keyed on time and not called for failed/stopped/paused endings: a partial re-pin leaves
|
||
// superseded units behind, and the next admission must still carry the flags.
|
||
// ⚠ BOTH COLUMNS, and leaving the second out is the mine this pack had to not step in: a
|
||
// write-ahead mark a dead process left behind is retired by nothing else, so a run that carried
|
||
// --resnapshot and finished `ready` would retire the stamp and meet the mark again on the next
|
||
// admission — the flag would become permanent and the re-pass door would never read a quiet book.
|
||
func (s *Store) ClearBankMove(ctx context.Context, bookID string) error {
|
||
if _, err := s.pool.Exec(ctx,
|
||
`update books set bank_moved_at = null, bank_move_pending_at = null where id = $1`, bookID); err != nil {
|
||
return fmt.Errorf("pgstore: clear bank move: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// MarkBankMovePending is the correction door's write-ahead: a verb is about to run on this book, and
|
||
// until its outcome is recorded the book is read as one whose bank MOVED (migration 00037).
|
||
//
|
||
// The mark is the only thing that survives a death with no channel — the stamp has one writer and no
|
||
// sweep — and its whole value is being committed BEFORE the verb starts. A caller that cannot write
|
||
// it has no backstop, which is why the door refuses rather than proceeds: the correction's remedy
|
||
// («re-send the same document») converges, while the state it would otherwise risk is a paid run
|
||
// that dies on the engine's snapshot guard after its hold was taken.
|
||
func (s *Store) MarkBankMovePending(ctx context.Context, bookID string, now time.Time) error {
|
||
tag, err := s.pool.Exec(ctx,
|
||
`update books set bank_move_pending_at = $2 where id = $1`, bookID, now)
|
||
if err != nil {
|
||
return fmt.Errorf("pgstore: mark a bank move pending: %w", err)
|
||
}
|
||
if tag.RowsAffected() == 0 {
|
||
return ErrNoBook
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ClearBankMovePending retires the write-ahead mark ALONE, leaving whatever the stamp says.
|
||
//
|
||
// Called only where the door holds a RECEIPT that proves nothing landed — the verb's own
|
||
// all-or-nothing refusal, or a clean report that neither changed anything nor re-found it. Every
|
||
// other way out of the door keeps the mark: a killed verb can die between its two renames, so
|
||
// «nothing landed» is not something an absent report may be read to say.
|
||
func (s *Store) ClearBankMovePending(ctx context.Context, bookID string) error {
|
||
if _, err := s.pool.Exec(ctx,
|
||
`update books set bank_move_pending_at = null where id = $1`, bookID); err != nil {
|
||
return fmt.Errorf("pgstore: clear a pending bank move: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// Usage is the ACCOUNT-WIDE credit state: a share and a halt word.
|
||
//
|
||
// ⚠ A SHARE HERE IS NO LONGER A PROHIBITION ON SUMS, it is the right shape for THIS question. The
|
||
// blanket ban on money crossing the boundary was revoked by the owner on 05.09 (D39.196 §2) and the
|
||
// balance now goes out in dollars — on the ORDER FORM, where a sum has an order beside it to be a
|
||
// price OF. The account screen asks a different question, «am I running low», and the owner's own
|
||
// note is that a percentage is honest exactly as an ALARM. Both are served, from the same fact.
|
||
type Usage struct {
|
||
// RemainingPercent is the balance as a share of everything ever granted to the account. Grants,
|
||
// not "a limit of a period": there are no periods (contract §/usage).
|
||
RemainingPercent int
|
||
// Spendable is whether there is anything left AT ALL. Separate from the percentage because the
|
||
// percentage is floored: $9 of a $1000 grant rounds to 0%, and "0%" is not the same fact as
|
||
// "nothing can be started".
|
||
Spendable bool
|
||
// PausedReason is set when the ACCOUNT is in a halted state.
|
||
PausedReason string
|
||
}
|
||
|
||
// ReadUsage computes the credit state.
|
||
func (s *Store) ReadUsage(ctx context.Context, userID string) (Usage, error) {
|
||
// ⚠ The halt is a state of the ACCOUNT and it is read off the ACCOUNT. It used to be read off the
|
||
// paused reason of each book's latest run, and the canon warns about exactly that at
|
||
// §AccountHaltReason: a run stops for reasons that say nothing about the account. Measured on the
|
||
// wire — $10 on the account, a ten-chapter run that spent the $0.30 IT bought — the answer was
|
||
// `{"state":"ok","remaining_percent":97,"halt_reason":"credit_exhausted"}`: a user with money
|
||
// told they have none, beside a percentage saying the opposite.
|
||
// ⚠ The denominator is everything ever ADDED to the account, not the `grant` rows alone. On the
|
||
// beta the signup grant is zero and an operator tops an account up with `adjust`, which writes a
|
||
// different kind — so counting grants only answered `remaining_percent: 0` to an account holding
|
||
// $20 that could start runs. A NEGATIVE adjustment is a correction and belongs on the other side,
|
||
// like a spend.
|
||
const q = `
|
||
select coalesce((select balance_micro_usd from account_balances where user_id = $1), 0),
|
||
coalesce((select sum(amount_micro_usd) from credit_ledger
|
||
where user_id = $1
|
||
and (kind = 'grant' or (kind = 'adjustment' and amount_micro_usd > 0))), 0)
|
||
from users where id = $1`
|
||
var balance, granted int64
|
||
err := s.pool.QueryRow(ctx, q, userID).Scan(&balance, &granted)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return Usage{}, ErrNoAccount
|
||
}
|
||
if err != nil {
|
||
return Usage{}, fmt.Errorf("pgstore: read usage: %w", err)
|
||
}
|
||
u := Usage{Spendable: balance > 0}
|
||
switch {
|
||
case granted <= 0:
|
||
// Nothing was ever granted. Zero out of zero is not 100% available; an account that cannot
|
||
// spend anything is exhausted, and saying so is what makes the screen actionable.
|
||
u.RemainingPercent = 0
|
||
case balance <= 0:
|
||
u.RemainingPercent = 0
|
||
case balance >= granted:
|
||
// A correction can push the balance above the sum of grants. The share is still capped: the
|
||
// contract's field is 0..100 and a client that got 137 would have to invent a meaning.
|
||
u.RemainingPercent = 100
|
||
default:
|
||
// Rounded DOWN so "1%" never means "already nothing": the figure is what is LEFT.
|
||
u.RemainingPercent = int(balance * 100 / granted)
|
||
}
|
||
if !u.Spendable {
|
||
// An account with nothing left to spend IS halted, and that is the one account-level reason
|
||
// this contract version has. It travels beside `state` rather than being derived from it by
|
||
// the client: the two vocabularies are separate, so a later reason lands here without touching
|
||
// what `state` means.
|
||
u.PausedReason = PausedCreditExhausted
|
||
}
|
||
return u, nil
|
||
}
|
||
|
||
// The reasons a run can be paused. The contract names TWO of them since the order form's minor —
|
||
// `run_limit_reached` (the run spent the order it was sold) and `credit_exhausted` (the ACCOUNT has
|
||
// nothing left) — and they are separate because their remedies are opposite: buy again, against top
|
||
// up. `daily_ceiling` and `ceiling_unknown` are INTERNAL and deliberately not projected — see
|
||
// CeilingPause and migration 00015.
|
||
const (
|
||
// PausedRunLimitReached is what a run's OWN ceiling now says — see ingest for why it is not
|
||
// `credit_exhausted` any more (PD-446).
|
||
PausedRunLimitReached = ingest.PausedRunLimitReached
|
||
PausedCreditExhausted = ingest.PausedCreditExhausted
|
||
PausedDailyCeiling = ingest.PausedDailyCeiling
|
||
// PausedCeilingUnknown is a ceiling halt with nothing to say WHOSE ceiling it was: an exit code
|
||
// with no event behind it — a journal that was never written, or a projection under quarantine,
|
||
// which never drains at all — or an event whose scope this build cannot place. It is resumable
|
||
// like the platform's own (it may BE the platform's own), and it deliberately is NOT
|
||
// `credit_exhausted`: that value lights the account-level halted flag, and an account whose run
|
||
// hit somebody else's limit has money on it.
|
||
PausedCeilingUnknown = ingest.PausedCeilingUnknown
|
||
)
|
||
|
||
// CeilingPause turns the stream's ceiling SCOPE into the reason a run is paused with.
|
||
//
|
||
// `day` is the engine's own daily limit out of a book.yaml the operator owns (`ceilings.day_usd`):
|
||
// the platform never chose it, cannot read it, and a run it stops has an account with money still
|
||
// on it. That is why it gets a reason of its own rather than `credit_exhausted` — which would tell
|
||
// the user their credit ran out and would light the ACCOUNT-level halted flag, which keys on
|
||
// exactly that value (ReadUsage).
|
||
//
|
||
// `book` is the platform's own, which it sets flush against the hold it took (PD-158). ANYTHING
|
||
// ELSE — an absent scope, a scope a later minor invents — is neither, and says so: this pack gave
|
||
// that state a name for the exit-code path, and the same reasoning applies to an event that names a
|
||
// ceiling this build cannot place. Defaulting it to `credit_exhausted` was the strictly worse of the
|
||
// three available answers: it is resumable either way, so nothing is bought
|
||
// by the guess, and being wrong tells an account with money on it that its credit ran out.
|
||
func CeilingPause(scope string) string {
|
||
switch scope {
|
||
case ingest.ScopeDay:
|
||
return PausedDailyCeiling
|
||
case ingest.ScopeBook:
|
||
// The platform's own ceiling, which it sets flush against the hold it took (PD-158). The run
|
||
// spent what was BOUGHT — which says nothing about the account, and used to claim it did.
|
||
return PausedRunLimitReached
|
||
default:
|
||
return PausedCeilingUnknown
|
||
}
|
||
}
|
||
|
||
// validPauseReason is the closed vocabulary the DDL also checks (migration 00015). Both, and not
|
||
// one: this value is produced by our own materializer, so an unknown one is a defect rather than
|
||
// data, and the constraint is what makes that true of a path this function never sees.
|
||
func validPauseReason(reason string) bool {
|
||
switch reason {
|
||
case PausedRunLimitReached, PausedCreditExhausted, PausedDailyCeiling, PausedCeilingUnknown:
|
||
return true
|
||
}
|
||
return false
|
||
}
|
||
|
||
// cursorPos is the keyset position a cursor encodes.
|
||
type cursorPos struct {
|
||
time *time.Time
|
||
id *string
|
||
}
|
||
|
||
// encodeCursor renders an opaque cursor. Opaque means the client MUST NOT parse it, not that it is
|
||
// secret: it carries nothing the same client cannot already see in the row it came from.
|
||
//
|
||
// It is BOUND to the collection it came from — the owner whose library produced it. Rejecting a
|
||
// cursor that does not apply is the SERVER's duty (contract §NextCursor) and the client cannot
|
||
// perform it, because the token is opaque to it by construction; without the binding, a page token
|
||
// handed to the wrong library was silently accepted and answered with someone else's window.
|
||
func encodeCursor(owner string, t time.Time, id string) string {
|
||
return base64.RawURLEncoding.EncodeToString(
|
||
[]byte(scopeTag(owner) + "\x00" + t.UTC().Format(time.RFC3339Nano) + "\x00" + id))
|
||
}
|
||
|
||
// scopeTag identifies the collection without carrying its identifier: the cursor travels to the
|
||
// client, and an account id in it would be one more place the id exists.
|
||
func scopeTag(owner string) string {
|
||
sum := sha256.Sum256([]byte("library\x00" + owner))
|
||
return base64.RawURLEncoding.EncodeToString(sum[:8])
|
||
}
|
||
|
||
func decodeCursor(owner, s string) (cursorPos, error) {
|
||
if s == "" {
|
||
return cursorPos{}, nil
|
||
}
|
||
raw, err := base64.RawURLEncoding.DecodeString(s)
|
||
if err != nil {
|
||
return cursorPos{}, ErrBadCursor
|
||
}
|
||
scope, rest, ok := strings.Cut(string(raw), "\x00")
|
||
if !ok || scope != scopeTag(owner) {
|
||
return cursorPos{}, ErrBadCursor
|
||
}
|
||
stamp, id, ok := strings.Cut(rest, "\x00")
|
||
if !ok || id == "" {
|
||
return cursorPos{}, ErrBadCursor
|
||
}
|
||
t, err := time.Parse(time.RFC3339Nano, stamp)
|
||
if err != nil {
|
||
return cursorPos{}, ErrBadCursor
|
||
}
|
||
return cursorPos{time: &t, id: &id}, nil
|
||
}
|
||
|
||
// BookDirectory is a book and where its engine project lives, which is all a backup needs to know
|
||
// about it.
|
||
type BookDirectory struct {
|
||
ID string
|
||
Workdir string
|
||
// Status is the book's intake status, and the backup needs it for ONE decision that cannot be
|
||
// made any other way: whether a book the engine refuses to copy is a book with nothing to copy
|
||
// yet, or a hole in the restore point. The engine's exit code cannot tell those apart and its
|
||
// prose must not be parsed for it.
|
||
Status string
|
||
}
|
||
|
||
// BooksForBackup lists every book this deployment holds, oldest first.
|
||
//
|
||
// EVERY book, with no filter on status, and that is the decision worth naming: a book still parsing
|
||
// has an uploaded source the user cannot re-supply from our side, and a book whose run failed holds
|
||
// exactly the work someone paid for. The one state that legitimately has nothing on disk — an upload
|
||
// that was rejected and took its directory with it — is recognised by the copier from the directory
|
||
// itself and recorded as a skip, rather than guessed at from a status here.
|
||
func (s *Store) BooksForBackup(ctx context.Context) ([]BookDirectory, error) {
|
||
const q = `select id, workdir, status from books order by added_at, id`
|
||
rows, err := s.pool.Query(ctx, q)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("pgstore: list books for backup: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
var out []BookDirectory
|
||
for rows.Next() {
|
||
var b BookDirectory
|
||
if err := rows.Scan(&b.ID, &b.Workdir, &b.Status); err != nil {
|
||
return nil, fmt.Errorf("pgstore: scan book for backup: %w", err)
|
||
}
|
||
out = append(out, b)
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// RenameBook changes a book's DISPLAY name and answers the card that results (canon §updateBook).
|
||
//
|
||
// ⚠ WHAT IT DOES NOT TOUCH, and this is the ratified boundary rather than a limitation of the
|
||
// implementation: «A title is DISPLAY and reaches nothing else — not the translation, whose
|
||
// configuration is written once at intake and never rewritten» (canon §updateBook). Nothing here
|
||
// goes near the book's directory, its `book.yaml` or the engine's project database. Two facts make
|
||
// that the only correct reading and not merely the cautious one:
|
||
//
|
||
// - The engine folds `title` into `BriefHash` (backend/internal/config/book.go, BriefHash: "Title
|
||
// is part of the brief"), which is the first field of the snapshot payload and therefore of
|
||
// every request hash. Editing the title in `book.yaml` moves the snapshot, so the next run of an
|
||
// unfinished book REFUSES until `--resnapshot`, and consenting re-buys the whole book. A rename
|
||
// is the cheapest thing a user does; it must not be the most expensive thing this system does.
|
||
// - It would also be a write into the engine's own file, which form Б (D39.130 п.2б,
|
||
// 17-seam-inbound-law §1а) permits exactly once, at the book's birth.
|
||
//
|
||
// Scoped by OWNER, like every other book read here: a stranger's book must be indistinguishable
|
||
// from one that is not there (API1/BOLA).
|
||
//
|
||
// The revision moves, and by the library's next number rather than by one — the same fragment every
|
||
// other change to a card uses, and for the same reason spelled out at nextRevisionOfThisBooksLibrary:
|
||
// a client obeying the contract drops a read whose revision is not ABOVE what it applied, so a
|
||
// rename that left the number alone would be a rename the screen never showed.
|
||
//
|
||
// No stream frame is emitted, and that is deliberate rather than forgotten: the only book-scoped
|
||
// frame this store has is FrameStatus, which carries the status vocabulary and no title, so emitting
|
||
// one here would announce a status change that did not happen. The client that renamed has the new
|
||
// card in the answer; another client learns it on its next read, which its revision tells it to take.
|
||
func (s *Store) RenameBook(ctx context.Context, userID, bookID, title string) (Book, *Run, error) {
|
||
const q = `
|
||
update books set title = $3,
|
||
revision = ` + nextRevisionOfThisBooksLibrary + `
|
||
where id = $1 and owner_id = $2`
|
||
var b Book
|
||
var run *Run
|
||
err := s.inTx(ctx, func(tx pgx.Tx) error {
|
||
tag, err := tx.Exec(ctx, q, bookID, userID, title)
|
||
if err != nil {
|
||
return fmt.Errorf("pgstore: rename book: %w", err)
|
||
}
|
||
if tag.RowsAffected() == 0 {
|
||
return ErrNoBook
|
||
}
|
||
// Read back INSIDE the same transaction, for the reason GetBook is one transaction: the card
|
||
// carries a revision, and a card assembled after the write had committed could be labelled
|
||
// with a number older than the figures on it.
|
||
if b, err = scanBook(tx.QueryRow(ctx,
|
||
`select `+bookColumns+` from books b `+lastRun+` where b.id = $1 and b.owner_id = $2`,
|
||
bookID, userID)); err != nil {
|
||
return err
|
||
}
|
||
run, err = lastRunTx(ctx, tx, bookID)
|
||
return err
|
||
})
|
||
if err != nil {
|
||
return Book{}, nil, err
|
||
}
|
||
if run != nil {
|
||
run.Revision = b.Revision
|
||
}
|
||
return b, run, nil
|
||
}
|
||
|
||
// RemainingChapter is one chapter still to be delivered, priced.
|
||
type RemainingChapter struct {
|
||
// ID is the chapter's IDENTITY, and it is what an order phrased in chapters is anchored on. See
|
||
// migration 00033: the ordinal beside it is a label, because an ordinal resolves after a re-cut
|
||
// to different text at the same number and nobody sees the difference.
|
||
ID string
|
||
// Number is the DISPLAY ordinal an order is phrased in — the label.
|
||
Number int
|
||
// Units is how many of the chapter's output units are still to be delivered — what
|
||
// `translate --max-units` counts, and what the price below is pro-rated by.
|
||
Units int
|
||
// Expected is what the engine expects the REMAINING units of this chapter to be billed, and
|
||
// SourceChars is how much of its text is still to be translated.
|
||
//
|
||
// ⛔ BOTH ARE SUMS OVER THE UNDELIVERED UNITS THEMSELVES, never the chapter's total pro-rated by a
|
||
// COUNT of them. The first edition of this query divided the chapter's bill by units_total and
|
||
// multiplied by what was left, and units are not equal: measured on a fixture whose chapter holds
|
||
// a 100-rune unit at $0.001 beside a 9900-rune one at $0.099, delivering the small one left the
|
||
// chapter quoted at $0.050 against a real remainder of $0.099 — a `covers_all` the engine then
|
||
// stops halfway through. Delivering the LARGE one quoted $0.050 against a real $0.001, refusing a
|
||
// purchase the buyer could afford. The engine publishes a price per UNIT and the platform stores
|
||
// it; using it is what makes the quote about this book rather than about an average of it.
|
||
Expected money.MicroUSD
|
||
SourceChars int64
|
||
}
|
||
|
||
// PricedBook is everything an ORDER is judged against: the book's projection, and what is left of it.
|
||
//
|
||
// It is a second read beside ReadBookForRun rather than more columns on it, because the two answer
|
||
// different questions of different shapes — one row of facts about the book against one row per
|
||
// chapter — and because only the order form needs the second.
|
||
type PricedBook struct {
|
||
// Priced is whether this book carries a whole projection. FALSE is a refusal to sell and not a
|
||
// free book: the per-chapter constant that used to answer here was measured 4.47× low and is what
|
||
// made the last chapters of every book unbuyable (PD-440), so its removal leaves exactly one
|
||
// source for the price and a loud silence when that source has not spoken.
|
||
Priced bool
|
||
Expected money.MicroUSD
|
||
BookOnce money.MicroUSD
|
||
StepMax money.MicroUSD
|
||
// Structure is where the chapter cut came from, verbatim (`declared`, `detected`, `none`, or
|
||
// empty). What it is worth is decided by ingest.ChapterOrdersOffered, not here.
|
||
Structure string
|
||
// Remaining are the chapters still to deliver, ascending by number.
|
||
//
|
||
// ⚠ There is deliberately NO unit total and no delivered count beside them. Both were here and
|
||
// both were read by nothing: `--max-units` is computed on the SPAWN path, from the run's own row
|
||
// (LiveRun), where the boundary is resolved against the tree as it stands at that moment. A
|
||
// second copy of that arithmetic, read from a different query at a different time, is the shape
|
||
// every drift in this package has had.
|
||
Remaining []RemainingChapter
|
||
}
|
||
|
||
// ReadBookForOrder reads what an order is priced from, under the caller's ownership.
|
||
func (s *Store) ReadBookForOrder(ctx context.Context, userID, bookID string) (PricedBook, error) {
|
||
var out PricedBook
|
||
var expected, bookOnce, stepMax *int64
|
||
err := s.pool.QueryRow(ctx, `
|
||
select b.expected_micro_usd, b.book_once_micro_usd, b.step_max_micro_usd,
|
||
coalesce(b.structure, '')
|
||
from books b where b.id = $1 and b.owner_id = $2`,
|
||
bookID, userID).Scan(&expected, &bookOnce, &stepMax, &out.Structure)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return PricedBook{}, ErrNoBook
|
||
}
|
||
if err != nil {
|
||
return PricedBook{}, fmt.Errorf("pgstore: read book for order: %w", err)
|
||
}
|
||
// ALL THREE OR NONE. The migration's own checks keep each positive, but «priced» is a statement
|
||
// about the three together: a book carrying two of them was read from a document this build did
|
||
// not read whole, and the one that would be missing is exactly the one whose absence is silent —
|
||
// a zero step_max takes the floor out of every hold and the run dies on its first reservation.
|
||
if expected == nil || bookOnce == nil || stepMax == nil {
|
||
return out, nil
|
||
}
|
||
out.Priced = true
|
||
out.Expected, out.BookOnce, out.StepMax =
|
||
money.MicroUSD(*expected), money.MicroUSD(*bookOnce), money.MicroUSD(*stepMax)
|
||
|
||
// ⚠ THE UNDELIVERED UNITS THEMSELVES, aggregated per chapter — not the chapter's own counters.
|
||
// The counters are recomputed FROM these same rows (writeChapters), so the two agree about HOW
|
||
// MANY; what only the rows can say is WHICH, and for money that is the whole question.
|
||
rows, err := s.pool.Query(ctx, `
|
||
select c.id, c.number, count(*),
|
||
coalesce(sum(u.expected_micro_usd), 0), coalesce(sum(u.source_chars), 0)
|
||
from chapters c
|
||
join units u on u.chapter_id = c.id
|
||
join books b on b.id = c.book_id
|
||
where c.book_id = $1 and `+unitUndelivered+`
|
||
group by c.id, c.number
|
||
order by c.number`, bookID)
|
||
if err != nil {
|
||
return PricedBook{}, fmt.Errorf("pgstore: read remaining chapters: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
for rows.Next() {
|
||
var c RemainingChapter
|
||
var expectedLeft int64
|
||
if err := rows.Scan(&c.ID, &c.Number, &c.Units, &expectedLeft, &c.SourceChars); err != nil {
|
||
return PricedBook{}, fmt.Errorf("pgstore: read remaining chapter: %w", err)
|
||
}
|
||
c.Expected = money.MicroUSD(expectedLeft)
|
||
out.Remaining = append(out.Remaining, c)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return PricedBook{}, fmt.Errorf("pgstore: read remaining chapters: %w", err)
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// RemainingUnit is one output unit still to be delivered: its identity, how much source text is in
|
||
// it, and what the engine expects it to be billed.
|
||
type RemainingUnit struct {
|
||
// ID is what a CHARACTER order is anchored on. A unit id carries the cut in its own bytes, so it
|
||
// dies WITH the cut — which is exactly the signal a partial order needs (migration 00033).
|
||
ID string
|
||
SourceChars int64
|
||
// Expected is this unit's own bill. It is what a CHARACTER order is PRICED from, and the first
|
||
// edition of that path did not read it: an order of one unit was quoted at the price of every
|
||
// chapter it touched, so on a book with no chapter structure — one chapter, many units — buying
|
||
// a thousand characters reserved the whole book. That is the one partial order such a book can
|
||
// carry (D39.196 §1), and priced that way it did not exist.
|
||
Expected money.MicroUSD
|
||
}
|
||
|
||
// RemainingUnits is every output unit still to be delivered, in reading order. It is read ONLY to
|
||
// resolve a CHARACTER order — the one form of partial order a book with no chapter structure can
|
||
// carry (D39.196 §1) — and never on the options path, where a book of several thousand units would
|
||
// be read on every poll to answer a question the chapter roll-up already answers.
|
||
func (s *Store) RemainingUnits(ctx context.Context, userID, bookID string) ([]RemainingUnit, error) {
|
||
rows, err := s.pool.Query(ctx, `
|
||
select u.id, u.source_chars, u.expected_micro_usd
|
||
from units u
|
||
join chapters c on c.id = u.chapter_id
|
||
join books b on b.id = c.book_id
|
||
where b.id = $1 and b.owner_id = $2 and `+unitUndelivered+`
|
||
order by c.number, u.ordinal`, bookID, userID)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("pgstore: read remaining unit sizes: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
out := []RemainingUnit{}
|
||
for rows.Next() {
|
||
var u RemainingUnit
|
||
if err := rows.Scan(&u.ID, &u.SourceChars, &u.Expected); err != nil {
|
||
return nil, fmt.Errorf("pgstore: read remaining unit size: %w", err)
|
||
}
|
||
out = append(out, u)
|
||
}
|
||
return out, rows.Err()
|
||
}
|