textmachine/backend/internal/store/outbox.go

package store

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
)

// outbox.go: the durable half of the run-event seam (row 103). The ratified form is an OUTBOX, not a
// second write (D39.106 §2, research/25): a line of the event journal is a projection of a row this
// database has already committed, so the file can never claim something the ledger does not have.
//
// What lives here is the SEQUENCING and the STORAGE, and nothing about what an event means: the store
// takes an opaque line, exactly as it takes an opaque `disposition` string on chunk_status. Two
// properties are the whole point of putting it in SQLite rather than in a variable:
//
//   - a number is assigned INSIDE the transaction, so a transaction that rolls back consumes no
//     sequence number and the stream has no hole — and the reader treats a hole as fatal (ErrStreamGap);
//   - the exact BYTES are kept, so a write to the journal that failed can be retried later and produce
//     the identical line. A re-render with a fresher timestamp would read to the platform as the same
//     seq with a different payload, which quarantines the projection (ErrPayloadConflict).

// onceKeyLookup asks whether a unit has already been announced. The trailing `once_key <> ”` is not
// redundant with the equality: the uniqueness index is PARTIAL on exactly that predicate, and SQLite
// uses a partial index only when the query implies its WHERE clause SYNTACTICALLY — with a bind
// parameter it cannot prove ?1 <> ” while planning, so the bare equality plans as a full SCAN. That is
// O(rows) per announced unit, i.e. quadratic over a book, on the wave's own path. It is a constant so
// the plan can be asserted against the SHIPPING query rather than a copy of it
// (TestTheAnnounceLedgerLookupUsesItsIndex).
const onceKeyLookup = `SELECT 1 FROM events_outbox WHERE once_key = ? AND once_key <> ''`

// AnnouncedOnceKeys returns every announce-once key this database holds — the set of things a reader
// has already been TOLD about, as opposed to the set of things the store happens to have rows for.
//
// It is the READ half of EnqueueOnce, and it exists because the two facts are different and only this
// one is monotone. Whether a unit still has a row for every position the CURRENT pipeline runs changes
// the moment the pipeline's shape changes; whether a reader was told the unit was done does not, ever.
// A projection that asks the first question calls an already-delivered book "never delivered" the day a
// stage is added to the config, and invites its buyer to purchase it a second time (unified backlog row
// 232).
//
// The whole set rather than a per-key probe, and no filter on the key's SHAPE: the outbox is deliberately
// opaque to what an event MEANS (see the file comment), so the key format stays the emitter's business
// and the caller does its own matching. The size is bounded by the BOOK — one row per unit per wave, the
// bound ForgetEvents already relies on — not by how many times it has been run.
//
// The trailing `once_key <> ”` is the same predicate as onceKeyLookup and for the same reason: the
// uniqueness index is PARTIAL on it, and spelling it out is what lets SQLite use the index instead of
// scanning the buffered rows of the current run alongside the ledger.
func (s *Store) AnnouncedOnceKeys() (map[string]bool, error) {
	keys, err := queryAll(s.r, `SELECT once_key FROM events_outbox WHERE once_key <> ''`,
		func(rows *sql.Rows) (string, error) {
			var k string
			return k, rows.Scan(&k)
		})
	if err != nil {
		return nil, fmt.Errorf("store: read the announce-once ledger: %w", err)
	}
	out := make(map[string]bool, len(keys))
	for _, k := range keys {
		out[k] = true
	}
	return out, nil
}

// EventLine is one stored journal line: its sequence number and the exact bytes, without a newline.
type EventLine struct {
	Seq  int64
	Line []byte
}

// EventBuilder renders the line for the sequence number the outbox has just assigned. It runs INSIDE the
// transaction, so a render failure rolls the number back with everything else.
type EventBuilder func(seq int64) ([]byte, error)

// EnqueueEvent assigns runID's next sequence number, renders the line and stores it — one transaction.
func (s *Store) EnqueueEvent(runID string, build EventBuilder) error {
	ctx, cancel := opContext()
	defer cancel()
	tx, err := s.w.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	if _, err := enqueueEvent(ctx, tx, runID, build); err != nil {
		return err
	}
	return tx.Commit()
}

// EnqueueOnce is EnqueueEvent for an event that must be announced ONCE for the life of the book,
// whatever crashes and resumes happen in between. It reports whether THIS call is the one announcing it
// (false = a ledger row already says a reader has been told) and the sequence number it took, which the
// caller passes back to MarkAnnounced once the line is actually ON THE FILE.
//
// It exists for the vocabulary's only COUNTING event. A reader folds `unit_done` by increment, so the
// engine has to decide once per unit — and "has this unit been announced" cannot be answered from the
// dispositions alone: a process that resolved a unit and died before the line reached the file leaves a
// book whose SQLite says done and whose reader was never told, and the next process, seeing the unit
// already resolved, would stay silent forever.
//
// The ledger is written AFTER the file, never with the enqueue, and the order is the whole design: what
// it records is not "this line exists" but "a reader has seen it". An unprojected row of a dead run
// carries no key, ForgetEvents drops it, and its unit is announced again by the next process — the
// at-least-once side of a boundary that has no transaction across it (D39.119 п.3), because the other
// side loses counts permanently.
func (s *Store) EnqueueOnce(runID, onceKey string, build EventBuilder) (announcing bool, seq int64, err error) {
	if onceKey == "" {
		return false, 0, fmt.Errorf("store: an announce-once event needs a key")
	}
	ctx, cancel := opContext()
	defer cancel()
	tx, err := s.w.BeginTx(ctx, nil)
	if err != nil {
		return false, 0, err
	}
	defer tx.Rollback()
	var exists int
	// See onceKeyLookup: `AND once_key <> ''` is not redundant: the uniqueness index is PARTIAL (same predicate), and SQLite
	// will only use a partial index when the query implies its WHERE clause SYNTACTICALLY. With a bind
	// parameter it cannot prove ?1 <> '' at planning time, so the bare equality plans as SCAN — O(rows)
	// per announced unit, i.e. quadratic over a book, on the wave's own path. Spelling the predicate out
	// restores SEARCH ... USING COVERING INDEX; pinned by TestTheAnnounceLedgerLookupUsesItsIndex.
	switch err := tx.QueryRowContext(ctx, onceKeyLookup, onceKey).Scan(&exists); {
	case errors.Is(err, sql.ErrNoRows):
	case err != nil:
		return false, 0, fmt.Errorf("store: read the announce-once ledger: %w", err)
	default:
		return false, 0, nil // already announced, by this process or by one that ran before it
	}
	seq, err = enqueueEvent(ctx, tx, runID, build)
	if err != nil {
		return false, 0, err
	}
	if err := tx.Commit(); err != nil {
		return false, 0, err
	}
	return seq > 0, seq, nil
}

// MarkAnnounced records that these lines have reached the journal, keyed so a later process knows not
// to announce them again. One transaction for the batch a projection just wrote.
func (s *Store) MarkAnnounced(runID string, keyBySeq map[int64]string) error {
	if len(keyBySeq) == 0 {
		return nil
	}
	ctx, cancel := opContext()
	defer cancel()
	tx, err := s.w.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	for seq, key := range keyBySeq {
		if _, err := tx.ExecContext(ctx,
			`UPDATE events_outbox SET once_key = ? WHERE engine_run_id = ? AND seq = ?`, key, runID, seq); err != nil {
			return fmt.Errorf("store: mark event %d announced: %w", seq, err)
		}
	}
	return tx.Commit()
}

// enqueueEvent is the shared body, inside a caller's transaction — which is also how the spend event
// joins the settle that produced it (SettleWithCheckpoint): the formula "the event line is written in
// the SAME transaction as the checkpoint", at the one place where a divergence would be about money.
// It returns the number it assigned, or 0 when the builder declined to produce a line.
func enqueueEvent(ctx context.Context, tx *sql.Tx, runID string, build EventBuilder) (int64, error) {
	var next int64
	if err := tx.QueryRowContext(ctx,
		`SELECT COALESCE(MAX(seq), 0) + 1 FROM events_outbox WHERE engine_run_id = ?`, runID).Scan(&next); err != nil {
		return 0, fmt.Errorf("store: next event seq: %w", err)
	}
	line, err := build(next)
	if err != nil {
		return 0, err
	}
	if line == nil {
		return 0, nil
	}
	if _, err := tx.ExecContext(ctx,
		`INSERT INTO events_outbox (engine_run_id, seq, line) VALUES (?, ?, ?)`,
		runID, next, string(line)); err != nil {
		return 0, fmt.Errorf("store: enqueue event: %w", err)
	}
	return next, nil
}

// PendingEvents returns at most `limit` of runID's stored lines with seq > after, in sequence order —
// the next batch the journal has not been told about.
//
// The order is dense and complete by construction, not by hope: every number is assigned by the query
// above under the single-connection write pool, so a transaction can only see MAX(seq)=N once the
// transaction that took N has committed. A caller that projects strictly in this order therefore never
// writes seq N+1 above seq N, which is the one interleaving the reader cannot survive.
//
// The LIMIT is what keeps a DEGRADED journal from becoming quadratic. While the file refuses writes the
// caller's cursor cannot advance, so the unprojected prefix grows with every event and an unbounded read
// would materialize all of it again on each one — under the emitter's mutex, on the failure path, where
// the run is least able to afford it. Bounded, each attempt costs a fixed batch and the caller drains
// the rest by looping.
func (s *Store) PendingEvents(runID string, after int64, limit int) ([]EventLine, error) {
	return queryAll(s.r, `SELECT seq, line FROM events_outbox WHERE engine_run_id = ? AND seq > ? ORDER BY seq LIMIT ?`,
		func(rows *sql.Rows) (EventLine, error) {
			var e EventLine
			var line string
			err := rows.Scan(&e.Seq, &line)
			e.Line = []byte(line)
			return e, err
		}, runID, after, limit)
}

// EventsUsed reports whether runID has ever written a row for this book — buffered or ledgered.
//
// It answers one question the seam cannot get wrong: is this process's stream identity FRESH. The
// identity is per PROCESS by contract (the reader's key is (engine_run_id, seq), and its handshake must
// carry seq 1), but since row 102 the identity can be supplied by the caller — and a caller that reuses
// one, e.g. by exporting TM_TRACE_ID in a wrapper, would otherwise have this run continue the previous
// one's numbering and re-project its whole stream.
func (s *Store) EventsUsed(runID string) (bool, error) {
	ctx, cancel := opContext()
	defer cancel()
	var one int
	switch err := s.r.QueryRowContext(ctx,
		`SELECT 1 FROM events_outbox WHERE engine_run_id = ? LIMIT 1`, runID).Scan(&one); {
	case errors.Is(err, sql.ErrNoRows):
		return false, nil
	case err != nil:
		return false, fmt.Errorf("store: check the run's event identity: %w", err)
	}
	return true, nil
}

// ForgetEvents drops the BUFFERED rows of every run except runID. For those the outbox is a projection
// buffer, not an archive — the authoritative state is chunk_status/checkpoints/spend — so it is bounded
// to one run instead of growing with the number of times a book has been resumed. Called once, at open,
// before the process writes its own handshake.
//
// Announce-once rows are kept: they are the ledger that says what a reader has already been told, and
// deleting them would make every resume re-announce the whole book. They are bounded by the BOOK's size
// (one per unit per wave), not by the number of runs.
func (s *Store) ForgetEvents(runID string) error {
	ctx, cancel := opContext()
	defer cancel()
	if _, err := s.w.ExecContext(ctx,
		`DELETE FROM events_outbox WHERE engine_run_id <> ? AND once_key = ''`, runID); err != nil {
		return fmt.Errorf("store: forget the previous runs' events: %w", err)
	}
	return nil
}