package pgstore

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"

	"textmachine/platform/internal/ingest"
)

// RunSink materializes one attempt's event stream into the reporting database.
//
// It is bound to an ATTEMPT rather than to a run, because the ratified idempotency key is
// (engine_run_id, seq) and the engine mints a new run id — with seq restarting at 1 — on every
// invocation. Keying on the platform's run would drop the whole stream of attempt two.
type RunSink struct {
	store     *Store
	attemptID int64
	runID     string
	bookID    string
}

// NewRunSink builds the sink for one attempt.
func (s *Store) NewRunSink(attemptID int64, runID, bookID string) *RunSink {
	return &RunSink{store: s, attemptID: attemptID, runID: runID, bookID: bookID}
}

// Begin binds the engine's run id to this attempt.
//
// ⚠ It is a LEGACY path now: since the platform names an attempt's stream when the attempt row is
// created (pgstore.EngineStreamID), the tailer is always given the id it wants and never adopts a
// handshake, so this is reached only for attempts written by an older build. What it used to do
// besides the binding — recording the chunker version — moved into `effect`, where the handshake
// arrives as an ordinary event.
//
// The binding is refused if the attempt already carries a DIFFERENT id: that means two engine
// processes wrote into one journal under one attempt, and materializing either of them would mix
// two runs' counters into one projection.
//
// It is refused for an ENDED attempt too, and that half is newer than the rule above. A stop that
// landed before its engine was spawned leaves an attempt that is closed and carries no engine run id
// — a shape this zone could not produce before P5 — and an unbound, ended attempt would otherwise
// ADOPT the handshake of the attempt that replaced it: the same events would then be materialized
// twice, once per attempt, and every counter they increment would be counted twice.
func (r *RunSink) Begin(ctx context.Context, h ingest.Hello) error {
	tag, err := r.store.pool.Exec(ctx, `
		update run_attempts set engine_run_id = $2
		 where id = $1 and ended_at is null
		   and (engine_run_id is null or engine_run_id = $2)`, r.attemptID, h.EngineRunID)
	if err != nil {
		return fmt.Errorf("pgstore: bind engine run: %w", err)
	}
	if tag.RowsAffected() == 0 {
		return fmt.Errorf("pgstore: attempt %d is over, or already bound to another engine run", r.attemptID)
	}
	if h.ChunkerVersion != "" {
		if _, err := r.store.pool.Exec(ctx,
			`update books set chunker_version = $2 where id = $1`, r.bookID, h.ChunkerVersion); err != nil {
			return fmt.Errorf("pgstore: record chunker version: %w", err)
		}
	}
	return nil
}

// Apply materializes one event AND the cursor it moves, in ONE transaction.
//
// That is the whole reason this method exists rather than two: an implementation that applies the
// effect and then records the position has a window in which a crash re-applies the effect, and an
// implementation that records first has a window in which the effect is lost. The high-water mark is
// re-checked inside the transaction, so a duplicate line that raced a live writer is dropped here
// too and not only in the reader (PD-105: at-least-once delivery, a duplicate is not an error).
//
// The BOOK is locked first — see lockBook. This path used to take the attempt first and every other
// path that touches both takes the book first, which is a deadlock the moment a materializer and a
// reconciler work on one run: measured on the real API at 258 of 300 concurrent pairs.
func (r *RunSink) Apply(ctx context.Context, ev ingest.Envelope, c ingest.Cursor) error {
	return r.store.inTx(ctx, func(tx pgx.Tx) error {
		if err := lockBook(ctx, tx, r.bookID); err != nil {
			return err
		}
		var last int64
		if err := tx.QueryRow(ctx,
			`select last_seq from run_attempts where id = $1 for update`, r.attemptID).Scan(&last); err != nil {
			return fmt.Errorf("pgstore: lock attempt: %w", err)
		}
		if ev.Seq <= last {
			return nil // already applied; the cursor cannot move backwards either
		}
		if err := r.effect(ctx, tx, ev); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `
			update run_attempts
			   set last_seq = $2, last_offset = greatest(last_offset, $3), last_line_sha256 = $4
			 where id = $1`, r.attemptID, ev.Seq, c.Offset, c.SHA256); err != nil {
			return fmt.Errorf("pgstore: move cursor: %w", err)
		}
		return nil
	})
}

// effect is what one event changes. An unknown type changes nothing and is not an error: tolerating
// it is the minor-version rule of the stream (D39.85).
func (r *RunSink) effect(ctx context.Context, tx pgx.Tx, ev ingest.Envelope) error {
	switch ev.Type {
	case ingest.TypeHello:
		// The handshake reaches here as an ordinary event — it is seq 1 and it moves the cursor — and
		// what it changes is one fact: WHICH cut of the source the engine is working from. A reader
		// that persisted a chapter manifest cut by a different chunker would be joining on ordinals
		// that were silently re-numbered.
		//
		// ⚠ It used to be recorded in Begin, and Begin is now unreachable for any attempt this build
		// created: the platform names the stream when the attempt row is written, so the tailer never
		// has to adopt a handshake.
		var h ingest.Hello
		if err := decode(ev, &h); err != nil {
			return err
		}
		if h.ChunkerVersion == "" {
			return nil
		}
		if _, err := tx.Exec(ctx,
			`update books set chunker_version = $2 where id = $1`, r.bookID, h.ChunkerVersion); err != nil {
			return fmt.Errorf("pgstore: record chunker version: %w", err)
		}
		return nil
	case ingest.TypeProgress:
		var p ingest.Progress
		if err := decode(ev, &p); err != nil {
			return err
		}
		if err := r.bump(ctx, tx, `
			update runs set draft_done = $2, draft_total = $3, edit_done = $4, edit_total = $5,
			                eta_seconds = $6, revision = $7 where id = $1`,
			r.runID, p.Draft.Done, p.Draft.Total, p.Edit.Done, p.Edit.Total, etaOrNil(p.ETASeconds)); err != nil {
			return err
		}
		if err := recordWaveShape(ctx, tx, r.bookID, p.Draft.Total, p.Edit.Total); err != nil {
			return err
		}
		// The wire's counter is in CHAPTERS and this event is in units, so the frame carries the bar
		// as the read model computes it rather than the numbers that arrived.
		return r.emitProgress(ctx, tx)
	case ingest.TypeUnitDone:
		var u ingest.UnitDone
		if err := decode(ev, &u); err != nil {
			return err
		}
		return r.unitDone(ctx, tx, ev, u)
	case ingest.TypeBankStop:
		// The book's status follows its run's, so this IS a status change on the wire even though
		// only the run row moves (derivedStatus).
		if err := r.bump(ctx, tx, `update runs set status = 'awaiting_bank', revision = $2 where id = $1`, r.runID); err != nil {
			return err
		}
		return emitStatus(ctx, tx, r.bookID)
	case ingest.TypeCeiling:
		var c ingest.Ceiling
		if err := decode(ev, &c); err != nil {
			return err
		}
		if !c.Halted {
			return nil
		}
		// A ceiling stop is `paused`, never `failed`: it is resumable, and mapping it to a failure
		// would lie about that (contract §BookStatus). WHICH ceiling decides the reason — and, one
		// step later, whether a resume can do anything about it (see CeilingPause).
		if err := r.bump(ctx, tx, `
			update runs set status = 'paused', paused_reason = $2, revision = $3
			 where id = $1`, r.runID, CeilingPause(c.Scope)); err != nil {
			return err
		}
		return emitStatus(ctx, tx, r.bookID)
	case ingest.TypeSpend:
		var s ingest.Spend
		if err := decode(ev, &s); err != nil {
			return err
		}
		// The MAXIMUM seen, not a sum: the counter is cumulative, so a redelivered line is harmless
		// only as long as nothing adds it up. Freshness only — no balance moves here.
		_, err := tx.Exec(ctx, `
			update run_attempts set spend_micro_usd = greatest(spend_micro_usd, $2) where id = $1`,
			r.attemptID, s.CommittedMicroUSD)
		if err != nil {
			return fmt.Errorf("pgstore: record spend: %w", err)
		}
		return nil
	case ingest.TypeFinished:
		// The stream says the engine believes it is done. The RUN is not closed here: closing it
		// settles money, and money is settled from what the process actually did — which is known
		// once the unit is gone, not once a line was written.
		return nil
	default:
		return nil
	}
}

// unitDone records one resolved output unit and re-derives the counters of its chapter.
//
// ⚠ The fold is an ASSIGNMENT and not an increment, and that is ratified rather than stylistic
// (D39.131 п.2г). Delivery on this seam is at-least-once, so absorbing a duplicate is the
// CONSUMER's duty: `+ 1` per line counted a re-read line twice, and re-reading is the ordinary
// consequence of a cursor that was written before its effect — or of the emitter's own named
// residual, a line committed to the outbox that never reached the file and is therefore announced
// again by the next process. Recording the (chapter, unit, wave) identity the event carries and
// counting the set makes the answer the same however many times the line arrives, and it self-heals
// the undrained tail rather than merely tolerating it.
//
// Unknown waves are dropped rather than stored: `wave` is a closed vocabulary on both sides of the
// seam, and a value from outside it would land in a column whose constraint refuses it and take the
// whole materialization down with it. Under the minor-version rule an unknown value is ignored.
//
// ⚠ Bounded by what exists, unchanged: without the engine's persisted chapter manifest (unified
// backlog row 100) there is no `chapters` row to carry the derived counters, so the fold records the
// unit and stops there. The counters the screen reads today come from the progress event.
func (r *RunSink) unitDone(ctx context.Context, tx pgx.Tx, ev ingest.Envelope, u ingest.UnitDone) error {
	if u.Wave != ingest.WaveDraft && u.Wave != ingest.WaveEdit {
		return nil
	}
	// The LATEST resolution wins, and "latest" is by the event's own timestamp rather than by arrival.
	// A unit can legitimately be resolved twice with different dispositions — a redrive re-attacks a
	// flagged one — and the read model wants the second.
	//
	// ⚠ The `where` guards against ARRIVAL ORDER, and it is worth being exact about what it does not
	// do, because the acceptance found this comment claiming the opposite: the engine does not emit
	// an inversion. Every announcement carries the clock of the process making it (pipeline/events.go,
	// `at := e.now()`), the re-announcement of a line a dead process never got onto the file included
	// — such a row carries no key, ForgetEvents drops it, and the next process announces it afresh
	// with its own stamp (store/outbox.go). So this is not the repair of a known regression.
	//
	// What it buys is that the fold's result stops depending on WHO WROTE LAST. The seam is
	// at-least-once (D39.119 п.3) and this consumer assigns rather than increments, so a disposition
	// is otherwise decided by delivery order — which is a property of the tailer, not of the data. One
	// predicate makes it a property of the data, and a redrive that moved a unit from flagged to
	// shipped cannot be walked backwards by a re-read.
	if tag, err := tx.Exec(ctx, `
		insert into unit_resolutions (book_id, chapter, unit, wave, shipped, flagged, reason, at, revision)
		values ($1, $2, $3, $4, $5, $6, $7, $8, (select revision + 1 from books where id = $1))
		on conflict (book_id, chapter, unit, wave)
		do update set shipped = excluded.shipped, flagged = excluded.flagged,
		              reason = excluded.reason, at = excluded.at, revision = excluded.revision
		 where excluded.at >= unit_resolutions.at`,
		r.bookID, u.Chapter, u.Unit, u.Wave, u.Shipped, u.Flagged, u.Reason, ev.Time); err != nil {
		return fmt.Errorf("pgstore: record unit: %w", err)
	} else if tag.RowsAffected() == 0 {
		// The predicate above rejected this delivery as older than what is stored. Nothing changed, so
		// nothing is announced: a frame built from it would carry the superseded verdict under the
		// note's own id, and a frame is stored wire-ready and replayed verbatim.
		return nil
	}
	// ⚠ INVARIANT, recorded rather than defended: a delivery that turned `flagged` from true to false
	// would take a note off the list, and the contract does not let a collection lose a row quietly —
	// it is replaced wholesale, with `resync_required` (canon §EventEnvelope). The transition is
	// unreachable today: the engine announces a unit's verdict once per wave and its ledger does not
	// re-announce it (verified by reading backend/internal/runevents), so the only re-delivery is the
	// SAME verdict. The day something can retract a flag, this fold owes that frame — the counter
	// below would otherwise drift away from the list in silence.
	//
	// One counter per wave and nothing derived from them here: which of the two says a chapter is
	// FINISHED is a property of the pipeline (pgstore.finishedUnits), read where the question is asked.
	tag, 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),
		       revision = (select revision + 1 from books where id = $1)
		 where c.book_id = $1 and c.number = $2`, r.bookID, u.Chapter)
	if err != nil {
		return fmt.Errorf("pgstore: fold unit: %w", err)
	}
	if tag.RowsAffected() == 0 {
		// No tree yet — see the note above. The unit itself IS recorded, so the counters come out
		// exact the moment a chapter row exists; the book's revision is deliberately NOT bumped,
		// because nothing a client can read has changed and every bump costs one refetch.
		return nil
	}
	if err := r.bumpBook(ctx, tx); err != nil {
		return err
	}
	return r.emitChapter(ctx, tx, u, ev.Time)
}

// emitProgress announces the run's bar as the WIRE counts it: one monotonic number in chapter-passes
// through both waves, against what the run bought, plus the stage caption. The event that triggers
// it is in units and per phase — that split stays inside the platform.
func (r *RunSink) emitProgress(ctx context.Context, tx pgx.Tx) error {
	var done, total int
	var stage string
	var eta *int
	err := tx.QueryRow(ctx, `
		select `+runDone+`, `+runTotal+`, `+runStage+`, r.eta_seconds
		  from books b `+lastRun+` where b.id = $1`, r.bookID).Scan(&done, &total, &stage, &eta)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil
	}
	if err != nil {
		return fmt.Errorf("pgstore: read the run bar: %w", err)
	}
	return emitFrame(ctx, tx, r.bookID, FrameProgress, map[string]any{
		"progress": map[string]any{"done": done, "total": total, "stage": stage, "eta_seconds": eta},
	})
}

// emitChapter announces one chapter's own progress, and a note when the unit that moved it was
// flagged. The two are separate frames because they are separate rules: a chapter frame is state and
// may be replaced by a later one, a note is an ADDITION and may never be dropped.
func (r *RunSink) emitChapter(ctx context.Context, tx pgx.Tx, u ingest.UnitDone, at time.Time) error {
	var chapterID string
	var unitsDone, noteCount int
	err := tx.QueryRow(ctx, `
		select c.id, `+segmentUnits+`, c.note_count
		  from chapters c join books b on b.id = c.book_id `+lastRun+`
		 where c.book_id = $1 and c.number = $2`, r.bookID, u.Chapter).
		Scan(&chapterID, &unitsDone, &noteCount)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil // no tree yet: the counters are exact the moment one exists
	}
	if err != nil {
		return fmt.Errorf("pgstore: read the chapter bar: %w", err)
	}
	if err := emitFrame(ctx, tx, r.bookID, FrameChapter, map[string]any{
		"id": chapterID, "units_done": unitsDone, "note_count": noteCount,
	}); err != nil {
		return err
	}
	if !u.Flagged {
		return nil
	}
	// The note travels WITH the frame — it has an identity of its own, so it is a delta the client
	// applies rather than a poke that costs it a re-read of the whole list.
	var unitID *string
	if err := tx.QueryRow(ctx,
		`select id from units where chapter_id = $1 and ordinal = $2`, chapterID, u.Unit).
		Scan(&unitID); err != nil && !errors.Is(err, pgx.ErrNoRows) {
		return fmt.Errorf("pgstore: read the flagged pair: %w", err)
	}
	// ⚠ The frame carries the CONTRACT's note and never the engine's own reason, through the same map
	// the read path uses: a frame is stored wire-ready and replayed verbatim, so a word that gets in
	// here stays in the buffer after the code is fixed.
	code, severity := ingest.NoteCode(u.Reason)
	note := map[string]any{
		"id":         noteID(r.bookID, u.Chapter, u.Unit, u.Wave),
		"created_at": at.UTC().Format(time.RFC3339Nano),
		"severity":   severity,
		"code":       code,
		"chapter_id": chapterID,
	}
	// OPTIONAL and absent rather than null: `unit_id` is a string or it is not there — a note about a
	// whole chapter simply has none (canon §Note).
	if unitID != nil {
		note["unit_id"] = *unitID
	}
	return emitFrame(ctx, tx, r.bookID, FrameNote, map[string]any{"note": note})
}

// bump runs a statement whose LAST argument is the book's next revision, and stamps the book with
// it. One transaction is one revision but possibly several rows, which is why catch-up reads use
// `>=` and not `>` (contract §Revision).
func (r *RunSink) bump(ctx context.Context, tx pgx.Tx, q string, args ...any) error {
	rev, err := r.nextRevision(ctx, tx)
	if err != nil {
		return err
	}
	if _, err := tx.Exec(ctx, q, append(args, rev)...); err != nil {
		return fmt.Errorf("pgstore: materialize event: %w", err)
	}
	return r.bumpBook(ctx, tx)
}

func (r *RunSink) nextRevision(ctx context.Context, tx pgx.Tx) (int64, error) {
	var rev int64
	if err := tx.QueryRow(ctx, `select revision + 1 from books where id = $1 for update`, r.bookID).Scan(&rev); err != nil {
		return 0, fmt.Errorf("pgstore: read revision: %w", err)
	}
	return rev, nil
}

func (r *RunSink) bumpBook(ctx context.Context, tx pgx.Tx) error {
	if _, err := tx.Exec(ctx, `update books set revision = revision + 1 where id = $1`, r.bookID); err != nil {
		return fmt.Errorf("pgstore: bump book revision: %w", err)
	}
	return nil
}

func decode(ev ingest.Envelope, into any) error {
	if len(ev.Data) == 0 {
		return fmt.Errorf("pgstore: event %s seq %d carries no data", ev.Type, ev.Seq)
	}
	if err := json.Unmarshal(ev.Data, into); err != nil {
		return fmt.Errorf("pgstore: event %s seq %d: %w", ev.Type, ev.Seq, err)
	}
	return nil
}

// recordWaveShape keeps the book's answer to "does this pipeline have an editor" — the fact
// finishedUnits reads. The engine announces the two denominators at the start of a run, so the FIRST
// progress event of a book's first run is what settles it.
//
// Monotone in one direction on purpose: once a book has been through an editing pipeline, a chapter
// of it is finished when it is EDITED, and a later run reporting no editor must not make half-done
// chapters count as done. The other direction — an operator removing the editor — shows up as more
// chapters finished, which is honest and never walks a counter backwards.
func recordWaveShape(ctx context.Context, tx pgx.Tx, bookID string, draftTotal, editTotal int) error {
	if draftTotal+editTotal == 0 {
		return nil // this event announced no shape at all
	}
	if _, err := tx.Exec(ctx,
		`update books set edit_wave = coalesce(edit_wave, false) or $2 where id = $1`,
		bookID, editTotal > 0); err != nil {
		return fmt.Errorf("pgstore: record the wave shape: %w", err)
	}
	return nil
}

func etaOrNil(s int) *int {
	if s <= 0 {
		return nil // absent, not zero: the screen renders without it rather than showing "0 s left"
	}
	return &s
}

// ApplyStatus folds a `tmctl status --json` report into the read model.
//
// This is the RESYNC channel: a snapshot of a run that keeps moving, taken every few minutes. Since
// D39.122 the report carries the per-wave split, so it materializes through the same four counters
// as the stream and no longer flattens them — what stays true of it is that it is STALE by up to
// the poll interval, which is why it stamps last_resync_at rather than leaving a reader to guess
// how old the figures on a quarantined run are.
func (s *Store) ApplyStatus(ctx context.Context, runID, bookID string, rep ingest.StatusReport, now time.Time) error {
	return s.inTx(ctx, func(tx pgx.Tx) error {
		var rev int64
		if err := tx.QueryRow(ctx, `select revision + 1 from books where id = $1 for update`, bookID).Scan(&rev); err != nil {
			return fmt.Errorf("pgstore: read revision: %w", err)
		}
		// greatest(), so a resync can never move a counter BACKWARDS. A report taken before the
		// engine's own figures caught up would otherwise walk a visible progress bar back down — the
		// one thing the contract asks a client never to do and which the server must not do either.
		// It is also what makes this safe to run over a projection the stream has already moved.
		if _, err := tx.Exec(ctx, `
			update runs set draft_done = greatest(draft_done, $2), draft_total = greatest(draft_total, $3),
			                edit_done = greatest(edit_done, $4), edit_total = greatest(edit_total, $5),
			                eta_seconds = $6, last_resync_at = $7, revision = $8
			 where id = $1`, runID, rep.Progress.Draft.Done, rep.Progress.Draft.Total,
			rep.Progress.Edit.Done, rep.Progress.Edit.Total,
			etaOrNil(int(rep.ETASeconds)), now, rev); err != nil {
			return fmt.Errorf("pgstore: apply status: %w", err)
		}
		if err := recordWaveShape(ctx, tx, bookID, rep.Progress.Draft.Total, rep.Progress.Edit.Total); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx,
			`update books set revision = revision + 1 where id = $1`, bookID); err != nil {
			return fmt.Errorf("pgstore: bump book revision: %w", err)
		}
		return nil
	})
}

// RunEnding is one attempt's end, as the reconciler decided it.
type RunEnding struct {
	RunID     string
	AttemptID int64
	// Status is the run's product status. PausedReason accompanies `paused` and must be empty for
	// every other status: a run that is not paused carrying a reason it is paused would be read by
	// the resume path, which asks that column what to do next.
	Status       string
	PausedReason string
	// FailureReason accompanies `failed` and must be empty for every other status, for the same
	// reason PausedReason must: a client reads it to decide whether to offer a retry.
	FailureReason string
	// ExitResult is systemd's $SERVICE_RESULT and ExitCode the engine's own code, when it exited.
	ExitResult string
	ExitCode   *int
	Now        time.Time
}

// FinishRun closes a run and its attempt: the read-model status, the end of the attempt and what
// systemd said about it. Money is NOT touched here — see Settlement.
//
// ⚠ The paused reason is written HERE and not only by PauseRun, and that is the ceiling half of
// PD-113. A ceiling halt reaches this platform twice — as a stream event and as exit code 4 — and
// the two fail independently: a journal that could not be written still leaves the code. On that
// path there is no event to have set the reason, and a `paused` run with a null reason gives the
// screen nothing to say and the resume path nothing to judge.
//
// closed is false when the write did not apply, and the caller must then do nothing else: the run was
// already finished, or — the case that made this a bool — the attempt it was asked to close is no
// longer the run's live one.
//
// ⚠ That second guard is money. A sweep decides from a snapshot and writes seconds later; between the
// two, the user can stop the run and RESUME it, and the resumed run is live again with a second
// attempt holding a second reservation. The old attempt's exit marker is still on disk — nothing
// deletes markers of attempts that ended — so the stale pass would close the run from it, leaving
// attempt 2's hold in NO worklist: `ListLiveRuns` selects runs with `finished_at is null` and
// `UnsettledRuns` attempts with `ended_at is not null`, and the resumed run matches neither once it
// has been re-finished. Reachable only since a run can come back to life at all, which is this pack.
func (s *Store) FinishRun(ctx context.Context, in RunEnding) (closed bool, err error) {
	if !validRunStatus(in.Status) {
		return false, fmt.Errorf("pgstore: %q is not a run status", in.Status)
	}
	switch {
	case in.PausedReason != "" && in.Status != "paused":
		return false, fmt.Errorf("pgstore: a %s run cannot carry a paused reason", in.Status)
	case in.PausedReason != "" && !validPauseReason(in.PausedReason):
		return false, fmt.Errorf("pgstore: %q is not a pause reason", in.PausedReason)
	case in.FailureReason != "" && in.Status != "failed":
		return false, fmt.Errorf("pgstore: a %s run cannot carry a failure reason", in.Status)
	case in.FailureReason != "" && !validFailureReason(in.FailureReason):
		return false, fmt.Errorf("pgstore: %q is not a failure reason", in.FailureReason)
	}
	runID, attemptID, now := in.RunID, in.AttemptID, in.Now
	err = s.inTx(ctx, func(tx pgx.Tx) error {
		// The BOOK's row is locked first, here and in the materializer. The two used to take them in
		// opposite orders — the materializer books-then-runs, this one runs-then-books — and two
		// reconcilers on one run (overlapping deploy generations) then deadlocked in both directions;
		// Postgres aborts one side, so it cost a failed sweep rather than corruption. Measured.
		var bookID string
		if err := tx.QueryRow(ctx,
			`select id from books where id = (select book_id from runs where id = $1) for update`,
			runID).Scan(&bookID); err != nil {
			if errors.Is(err, pgx.ErrNoRows) {
				return nil // the run is gone
			}
			return fmt.Errorf("pgstore: lock book: %w", err)
		}
		// nullif: a run that ends in any other status must not keep the reason a PREVIOUS pause of it
		// left behind — the resume path reads that column to decide what it may do.
		if err := tx.QueryRow(ctx, `
			update runs set status = $2, paused_reason = nullif($5, ''), failure_reason = nullif($6, ''),
			                finished_at = $3,
			                revision = (select revision + 1 from books where id = runs.book_id)
			 where id = $1 and finished_at is null
			   and exists (select 1 from run_attempts a
			                where a.id = $4 and a.run_id = runs.id and a.ended_at is null)
			returning book_id`, runID, in.Status, now, attemptID, in.PausedReason, in.FailureReason).
			Scan(&bookID); err != nil {
			if errors.Is(err, pgx.ErrNoRows) {
				// Already finished, or asked about an attempt that is no longer the live one. Both are
				// "someone else got here first"; finishing is idempotent by refusal, not by repetition.
				return nil
			}
			return fmt.Errorf("pgstore: finish run: %w", err)
		}
		if _, err := tx.Exec(ctx, `
			update run_attempts set ended_at = $2, exit_code = $3, exit_result = $4
			 where id = $1 and ended_at is null`, attemptID, now, in.ExitCode, in.ExitResult); err != nil {
			return fmt.Errorf("pgstore: finish attempt: %w", err)
		}
		// The end of a run is a boundary of the work, which is where the contract puts the freshness of
		// a pair's text — so the book leaves this transaction OWING a materialization. Stamped with the
		// status rather than queued by the caller: the caller's queue died with its process, and a run
		// that was paid for then never showed its text (see BooksOwedReadModel).
		if _, err := tx.Exec(ctx, `
			update books set status = $2, `+owesAReadingSurface+`
			                 revision = `+nextRevisionOfThisBooksLibrary+` where id = $1`, bookID, in.Status); err != nil {
			return fmt.Errorf("pgstore: finish book: %w", err)
		}
		closed = true
		return emitStatus(ctx, tx, bookID)
	})
	return closed, err
}

func validRunStatus(s string) bool {
	switch s {
	case "translating", "awaiting_bank", "ready", "stopped", "failed", "paused":
		return true
	}
	return false
}

// validFailureReason is the contract's RunFailureReason, checked here as well as in the DDL for the
// reason every other closed vocabulary of this store is: the value is produced by our own code, so
// one outside the set is a defect rather than data.
func validFailureReason(s string) bool {
	switch s {
	case "source_unreadable", "service_error", "interrupted":
		return true
	}
	return false
}

// FinishUnspawnedStop closes a run that was stopped before its unit ever existed, and only if that is
// still true when the write happens.
//
// The re-check is the whole method. The reconciler decides from a snapshot in which the attempt had
// no unit; between that read and this write the queue worker can claim the attempt and create one,
// and closing the run then leaves an engine spending against a book with no open reservation and no
// list that looks at it — the reconciler lists by `finished_at is null` and settlement by an open
// reservation, so it would be in neither. `ReleaseUnspawned` already guards the MONEY of exactly this
// window under a lock (PD-159); this is the same guard for the lifecycle the money follows.
//
// finished is false when the attempt was spawned after all: the caller does nothing, and the next
// sweep meets an ordinary live run — with a unit to signal and an intent that says to.
func (s *Store) FinishUnspawnedStop(ctx context.Context, runID string, attemptID int64, now time.Time) (finished bool, err error) {
	err = s.inTx(ctx, func(tx pgx.Tx) error {
		// Book first, then the attempt: the order every transaction in this package takes (lockBook).
		var bookID string
		if err := tx.QueryRow(ctx,
			`select id from books where id = (select book_id from runs where id = $1) for update`,
			runID).Scan(&bookID); err != nil {
			if errors.Is(err, pgx.ErrNoRows) {
				return nil // the run is gone
			}
			return fmt.Errorf("pgstore: lock book: %w", err)
		}
		var unit *string
		var ended *time.Time
		if err := tx.QueryRow(ctx,
			`select unit_name, ended_at from run_attempts where id = $1 and run_id = $2 for update`,
			attemptID, runID).Scan(&unit, &ended); err != nil {
			if errors.Is(err, pgx.ErrNoRows) {
				return nil
			}
			return fmt.Errorf("pgstore: lock attempt: %w", err)
		}
		if unit != nil && *unit != "" {
			return nil // spawned inside the window; not this path's business any more
		}
		if ended != nil {
			// The attempt this pass is holding is over, which means the run moved on without it — a stop
			// and a RESUME can both have happened since the snapshot was taken, and the run is live
			// again on a second attempt holding a second reservation. Closing it from here would put
			// that hold in no worklist at all. The same guard FinishRun carries (PD-181); acceptance
			// found this path missing it.
			return nil
		}
		tag, err := tx.Exec(ctx, `
			update runs set status = 'stopped', finished_at = $2,
			                revision = (select revision + 1 from books where id = runs.book_id)
			 where id = $1 and finished_at is null`, runID, now)
		if err != nil {
			return fmt.Errorf("pgstore: finish stopped run: %w", err)
		}
		if tag.RowsAffected() == 0 {
			return nil // already finished by an earlier pass
		}
		if _, err := tx.Exec(ctx, `
			update run_attempts set ended_at = $2, exit_result = $3
			 where id = $1 and ended_at is null`, attemptID, now, StopRequestedResult); err != nil {
			return fmt.Errorf("pgstore: finish attempt: %w", err)
		}
		// Owing a materialization like every other ending: this attempt may have had an engine behind
		// it — a spawn claim released while the process it asked for kept running is exactly the state
		// this path exists for — and that engine's text is paid for whether or not the platform ever
		// saw its unit.
		if _, err := tx.Exec(ctx, `
			update books set status = 'stopped', `+owesAReadingSurface+`
			                 revision = `+nextRevisionOfThisBooksLibrary+` where id = $1`, bookID); err != nil {
			return fmt.Errorf("pgstore: finish book: %w", err)
		}
		finished = true
		// Same reason as every other terminal write: this one sets `finished_at`, so the book comes to
		// rest and a client that hears nothing is told 204 on its next reconnect.
		return emitStatus(ctx, tx, bookID)
	})
	return finished, err
}

// StopRequestedResult is the exit_result of an attempt that ended by this platform's request without
// systemd getting to write a marker. Deliberately a value systemd cannot produce: an operator reading
// the column must be able to tell what the machine saw from what the platform concluded.
const StopRequestedResult = "stop-requested"

// MarkSettled records that the money of a run has been resolved.
func (s *Store) MarkSettled(ctx context.Context, runID string, now time.Time) error {
	_, err := s.pool.Exec(ctx, `update runs set settled_at = $2 where id = $1 and settled_at is null`, runID, now)
	if err != nil {
		return fmt.Errorf("pgstore: mark settled: %w", err)
	}
	return nil
}

// UnsettledRuns lists finished runs whose money is still open, minus the attempts that are serving a
// deferral. A run can finish and fail to settle — the engine's committed figure is read from a
// process that has to be asked, and asking can fail — and without this list that hold would stay
// reserved forever.
//
// ⚠ THE SAME DEFERRAL the reconciliation list uses, and the second half of the starvation fix: this
// list is ordered oldest-first too, so one attempt whose engine cannot answer spent the settlement's
// whole share of every pass and every other account's hold waited behind it.
func (s *Store) UnsettledRuns(ctx context.Context, now time.Time) ([]LiveRun, error) {
	// Keyed on the ATTEMPT being over, not on the RUN being over. An attempt that was interrupted and
	// replaced leaves its reservation open while its run goes on, and a list that filtered on the run
	// never looked at it again — the hold stayed reserved for the life of the account.
	return s.queryRuns(ctx, `
		  join run_attempts a on a.run_id = r.id and a.ended_at is not null
		  join reservations res on res.engine_run_id = r.id || '#' || a.attempt_no and res.state = 'open'
		 where a.reconcile_after is null or a.reconcile_after <= $1
		 order by a.ended_at`, now)
}

// ReservationKey is the attempt's reservation id, exported so the reconciler can settle without
// re-deriving a format that lives in this package.
func ReservationKey(runID string, attempt int) string { return engineRunKey(runID, attempt) }