textmachine/platform/internal/pgstore/sink.go

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.
//
// 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. Found by
// cross-family review of the acceptance dofix.
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.TypeProgress:
		var p ingest.Progress
		if err := decode(ev, &p); err != nil {
			return err
		}
		return 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))
	case ingest.TypeUnitDone:
		var u ingest.UnitDone
		if err := decode(ev, &u); err != nil {
			return err
		}
		return r.unitDone(ctx, tx, u)
	case ingest.TypeBankStop:
		return r.bump(ctx, tx, `update runs set status = 'awaiting_bank', revision = $2 where id = $1`, r.runID)
	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).
		return r.bump(ctx, tx, `
			update runs set status = 'paused', paused_reason = 'credit_exhausted', revision = $2
			 where id = $1`, r.runID)
	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 folds one resolved edit unit into its chapter.
//
// ⚠ Bounded by what exists: without the engine's persisted chapter manifest (unified backlog row
// 100) the platform has no mapping from the engine's chapter NUMBER to a stable chapter id, so a
// book whose chapters have never been materialized has nothing to update. The counters that DO have
// a home — the run's own progress — are carried by the progress event, so nothing is lost that the
// screen reads today.
func (r *RunSink) unitDone(ctx context.Context, tx pgx.Tx, u ingest.UnitDone) error {
	col := "units_draft_done"
	if u.Wave == "edit" {
		col = "units_edit_done"
	}
	tag, err := tx.Exec(ctx, `
		update chapters
		   set `+col+` = `+col+` + 1,
		       units_done = case when $3 then units_done + 1 else units_done end,
		       revision = (select revision + 1 from books where id = $1)
		 where book_id = $1 and number = $2`, r.bookID, u.Chapter, u.Wave == "edit")
	if err != nil {
		return fmt.Errorf("pgstore: fold unit: %w", err)
	}
	if tag.RowsAffected() == 0 {
		return nil // no manifest yet — see the note above
	}
	return r.bumpBook(ctx, tx)
}

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

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 := 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
	})
}

// 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.
//
// 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, runID string, attemptID int64, status, exitResult string, exitCode *int, now time.Time) (closed bool, err error) {
	if !validRunStatus(status) {
		return false, fmt.Errorf("pgstore: %q is not a run status", status)
	}
	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)
		}
		if err := tx.QueryRow(ctx, `
			update runs set status = $2, 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, status, now, attemptID).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, exitCode, exitResult); err != nil {
			return fmt.Errorf("pgstore: finish attempt: %w", err)
		}
		if _, err := tx.Exec(ctx, `
			update books set status = $2,
			                 revision = `+nextRevisionOfThisBooksLibrary+` where id = $1`, bookID, status); err != nil {
			return fmt.Errorf("pgstore: finish book: %w", err)
		}
		closed = true
		return nil
	})
	return closed, err
}

func validRunStatus(s string) bool {
	switch s {
	case "translating", "awaiting_bank", "finalizing", "ready", "stopped", "failed", "paused":
		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)
		}
		if _, err := tx.Exec(ctx, `
			update books set status = 'stopped',
			                 revision = `+nextRevisionOfThisBooksLibrary+` where id = $1`, bookID); err != nil {
			return fmt.Errorf("pgstore: finish book: %w", err)
		}
		finished = true
		return nil
	})
	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. 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.
func (s *Store) UnsettledRuns(ctx context.Context) ([]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'
		 order by a.ended_at`)
}

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