rindle_replica/

consumer.rs

//! The **cluster consumer**: the engine-side composition a normalized-protocol server
//! front builds on (CLUSTER-FOLD-IN-DESIGN.md WS6). Lifted from the `@rindle/replica`
//! napi addon's `cluster_core` so the Node daemon and the Rust daemon (`rindle-server`'s
//! network front) drive the SAME stack:
//!
//! - the coordinator ([`Cluster`]) lives on ONE thread (`!Send`): `register_table` /
//!   `query_normalized` / `commit*` / `destroy_query` all land here and return promptly;
//! - per-query normalized batches, per-connection progress frames, and faults are produced
//!   **asynchronously** by the drain thread and pushed into the caller's [`DrainSink`].
//!
//! The apply half — registration, `apply_muts`, the cursor discipline, replicated DDL —
//! is the engine-free [`ApplyConsumer`] (design 309), shared over the SAME store as the
//! cluster: every apply method here **delegates**, so the live follower and a headless
//! replay (`rindle-backup-sqlite`) run one implementation. This type adds what only a
//! derivation host has: the drain/query surface, the lmid mutation plane, and the room
//! ledger.
//!
//! The table helpers ([`ColType`], [`create_table_ddl`], [`build_mutation_sql`]) are shared
//! with the single-thread napi `Replica` so both schemas/SQL are identical.
//!
//! [`ColType`]: crate::ColType
//! [`create_table_ddl`]: crate::create_table_ddl
//! [`build_mutation_sql`]: crate::build_mutation_sql

use std::path::Path;

use rindle::value::OwnedValue;
use rindle::Ast;
use rindle_wire::family_key::{Binding, FamilyTemplate};
use rusqlite::OptionalExtension;

use crate::apply::{upsert_source_offset_hashed, ApplyConsumer};
use crate::cluster::{Cluster, ClusterWriteTxn, DdlMigrationError};
use crate::drain::{ConnId, Drain, DrainHandle, DrainSink};
use crate::normalize::{reject_unsupported_sync_aggregate, table_tree};
use crate::normalize_protocol::{
    build_query_parts, hello_from_parts, NormalizedHello, TableWireSchema,
};
use crate::table_shape::{BaseTableSchema, ColType, Mutation, TableMeta};
use crate::writeplane::{
    self, BookkeepingError, MigrationRecord, PublicOperationCommit, SqlOutcomeNamespace,
    StoredPublicOutcome,
};
use crate::ReplicaError;
use crate::{
    cas_cell_matches, RoomRowConflict, CLIENT_MUTATIONS_TABLE, OUTCOME_RETENTION_LMIDS,
    ROOM_CLIENT_MUTATIONS_TABLE, ROOM_MUTATION_OUTCOMES_TABLE, ROOM_PLACEMENT_TABLE,
    ROOM_WATERMARK_TABLE, SCOPE_SESSIONS_TABLE, SOURCE_OFFSET_WHOLE_RUN,
};
use crate::{QueryId, SqlStatementRequest, StatementResult, StatementRunError, TableNode};

pub use crate::apply::{SourceHead, APPLIED_DDL_TABLE};

/// The query-wide pieces a server needs to frame each subscriber's `hello` for a shared
/// query (see [`ClusterConsumer::register_shared_query`]): the deterministic table set and
/// its `normalized_fp`. The footprint fold itself lives on the drain thread.
#[derive(Clone, Debug)]
pub struct SharedQueryInfo {
    pub tables: Vec<TableWireSchema>,
    pub normalized_fp: u64,
}

/// The outcome of a CAS-guarded room flush ([`ClusterConsumer::commit_room_flush`]).
pub enum RoomFlushOutcome {
    /// Effects + cursor + identity committed in one transaction.
    Applied { cv: u64 },
    /// At least one precondition missed: NOTHING applied (rolled back); each miss
    /// carries the authoritative current image for the room's §5.4 convergence.
    Conflicts(Vec<RoomRowConflict>),
}

/// The cluster-backed consumer (coordinator side). `!Send` — lives on one thread; the
/// drain pushes batches/progress/faults into the sink from its own thread.
pub struct ClusterConsumer {
    cluster: Cluster,
    /// Kept alive for the process; its `Drop` stops the drain loop.
    _drain: Drain,
    handle: DrainHandle,
    /// The engine-free apply surface (design 309), sharing the cluster's store + pool
    /// fan-out — owner of the [`TableMeta`] registry and every `apply_*`/`commit_*`
    /// implementation this type delegates to.
    apply: ApplyConsumer,
}

impl ClusterConsumer {
    /// Open a cluster-backed consumer over `path` with `n_workers` IVM worker threads,
    /// wiring drain output into `sink`. The caller owns the file's lifecycle. A fresh
    /// file gets plain `wal` (design 306 D5); the daemon's roles pass their wal2 opt-in
    /// via [`open_with_journal`](Self::open_with_journal).
    pub fn open(
        path: &Path,
        n_workers: usize,
        sink: impl DrainSink + 'static,
    ) -> Result<ClusterConsumer, ReplicaError> {
        Self::open_with_journal(path, n_workers, crate::JournalMode::default(), sink)
    }

    /// [`open`](Self::open) with an explicit fresh-file [`crate::JournalMode`] — the
    /// daemon passes [`crate::JournalMode::Wal2`] (checkpoint headroom + the
    /// follower/backup fleet contract; an existing wal/wal2 file keeps its mode).
    pub fn open_with_journal(
        path: &Path,
        n_workers: usize,
        journal: crate::JournalMode,
        sink: impl DrainSink + 'static,
    ) -> Result<ClusterConsumer, ReplicaError> {
        Self::open_with(
            path,
            n_workers,
            crate::OpenOptions {
                journal,
                ..crate::OpenOptions::default()
            },
            sink,
        )
    }

    /// The full-combination opener: every [`crate::OpenOptions`] field is honored —
    /// the daemon's escape hatch for combining wal2 with a non-default derivation
    /// mode or operator storage (e.g. the design-306 soak fallback) without a new
    /// rung per combination.
    pub fn open_with(
        path: &Path,
        n_workers: usize,
        opts: crate::OpenOptions,
        sink: impl DrainSink + 'static,
    ) -> Result<ClusterConsumer, ReplicaError> {
        let (cluster, events) = Cluster::open_with(path, n_workers, opts)?;
        let apply = ApplyConsumer::from_parts(cluster.store_rc(), cluster.fanout_rc());
        let drain = Drain::spawn(events, cluster.committed_tx_id().0, sink);
        let handle = drain.handle();
        Ok(ClusterConsumer {
            cluster,
            _drain: drain,
            handle,
            apply,
        })
    }

    /// The underlying coordinator (controlled writes, lmid reads, raw event taps).
    pub fn cluster(&self) -> &Cluster {
        &self.cluster
    }

    /// Walk every declared foreign key and report violating rows — the opt-in audit a
    /// follower runs instead of enforcing per row. Delegates to
    /// [`Cluster::foreign_key_audit`](crate::Cluster::foreign_key_audit); see there for
    /// the cost, the cap, and why enforcement and verification are separable.
    pub fn foreign_key_audit(
        &self,
        max_rows: usize,
    ) -> Result<crate::ForeignKeyAudit, ReplicaError> {
        self.cluster.foreign_key_audit(max_rows)
    }

    /// Define + register a base table (`CREATE TABLE` + cluster source). Idempotent.
    pub fn register_table(
        &self,
        table: &str,
        columns: &[String],
        pk: &[usize],
        col_types: &[ColType],
    ) -> Result<(), ReplicaError> {
        self.apply
            .register_table_via(table, columns, pk, col_types, || {
                self.cluster.register_table(table)
            })
    }

    /// Register an EXISTING base table (already created by the host's own DDL): reads its
    /// column order + pk from SQLite and registers it for capture + derivation.
    pub fn register_existing_table(&self, table: &str) -> Result<(), ReplicaError> {
        self.apply
            .register_existing_table_via(table, || self.cluster.register_table(table))
    }

    /// One-time setup for client mutations: the `_rindle_client_mutations` table is created,
    /// captured, AND engine-hosted (lmid-as-data) — its meta joins the table map so the
    /// per-client system query's `hello` resolves like any table's.
    pub fn enable_client_mutations(&self) -> Result<(), ReplicaError> {
        self.apply
            .enable_client_mutations_via(|| self.cluster.enable_client_mutations())
    }

    /// One-time setup for the §4 realtime lifecycle ([`Cluster::enable_realtime_lifecycle`] —
    /// see [`Db::enable_realtime_lifecycle`](crate::Db::enable_realtime_lifecycle) for the
    /// table roster and why each is registered): the four lifecycle tables join the table
    /// map so doorbell / fence / outcome / ledger subscriptions resolve their `hello` like
    /// any table's — [`ROOM_CLIENT_MUTATIONS_TABLE`] had NO meta before this (Slice C kept
    /// it direct-SQL-only). All four are `_rindle_*`-prefixed and so excluded from
    /// [`base_table_schemas`](Self::base_table_schemas). Requires
    /// [`enable_client_mutations`](Self::enable_client_mutations) first. Enabling also
    /// switches on [`commit_room_flush`](Self::commit_room_flush)'s §4.2 watermark
    /// co-commit.
    pub fn enable_realtime_lifecycle(&self) -> Result<(), ReplicaError> {
        self.cluster.enable_realtime_lifecycle()?;
        let mut tables = self.apply.tables();
        tables.insert(
            SCOPE_SESSIONS_TABLE.to_string(),
            TableMeta {
                columns: vec![
                    "scope".to_string(),
                    "client_id".to_string(),
                    "expires_at".to_string(),
                ],
                pk: vec![0, 1],
                col_types: vec![ColType::String, ColType::String, ColType::Number],
                nullable: vec![false, false, false],
            },
        );
        tables.insert(
            ROOM_WATERMARK_TABLE.to_string(),
            TableMeta {
                columns: vec!["doc".to_string(), "flush_seq".to_string()],
                pk: vec![0],
                col_types: vec![ColType::String, ColType::Number],
                nullable: vec![false, false],
            },
        );
        tables.insert(
            ROOM_MUTATION_OUTCOMES_TABLE.to_string(),
            TableMeta {
                columns: vec![
                    "doc".to_string(),
                    "client_id".to_string(),
                    "mid".to_string(),
                    "kind".to_string(),
                    "reason".to_string(),
                    "name".to_string(),
                    "args".to_string(),
                ],
                pk: vec![0, 1, 2],
                col_types: vec![
                    ColType::String,
                    ColType::String,
                    ColType::Number,
                    ColType::String,
                    ColType::String,
                    ColType::String,
                    ColType::String,
                ],
                // `reason`/`name`/`args` mirror the frame's optional fields
                // (`mutationOutcome {mid, kind, reason?, name?, args?}`); the rest are row
                // identity + the verdict.
                nullable: vec![false, false, false, false, true, true, true],
            },
        );
        tables.insert(
            ROOM_CLIENT_MUTATIONS_TABLE.to_string(),
            TableMeta {
                columns: vec![
                    "doc".to_string(),
                    "client_id".to_string(),
                    "last_mutation_id".to_string(),
                ],
                pk: vec![0, 1],
                col_types: vec![ColType::String, ColType::String, ColType::Number],
                nullable: vec![false, false, false],
            },
        );
        Ok(())
    }

    /// The registered **base** tables' schemas (name + ordered columns/types + PK names), sorted by
    /// name, for client-schema codegen via `/schema` (DRIZZLE-MIGRATIONS-DESIGN.md §6.2). Excludes
    /// the daemon's own bookkeeping/internal tables — `_rindle_*` (e.g. `_rindle_client_mutations`),
    /// `__*` (e.g. the `__replica_meta` commit watermark), and `sqlite_*` — which must stay invisible
    /// to the client schema just as they are to CDC + planning (rindle-cdc skips `sqlite_*` + `__*`).
    /// Reads straight from the in-memory table map — the live introspected schema, no DB round-trip.
    pub fn base_table_schemas(&self) -> Vec<BaseTableSchema> {
        self.apply.base_table_schemas()
    }

    /// Re-read the visibility sidecars after a `ddl` entry (design 406 §10); `true` when the
    /// advertised schema changed and a bounce is due.
    pub fn refresh_visibility(&self) -> Result<bool, ReplicaError> {
        self.apply.refresh_visibility()
    }

    /// The durably stored high-water mutation id for `client_id` (0 if new).
    pub fn client_lmid(&self, client_id: &str) -> Result<u64, ReplicaError> {
        self.cluster.client_lmid(client_id)
    }

    /// The durable last sequence a foreign `producer` wrote (0 if new). This deliberately
    /// runs the shared connection-only helper on the observed writer while no mutation is open,
    /// keeping the read serialized with every write-plane decision.
    pub fn producer_seq(&self, producer: &str) -> Result<u64, BookkeepingError> {
        writeplane::read_producer_seq(self.cluster.writer_connection(), producer)
    }

    /// Read the distinct-producer count into `census` off the observed writer — the one scan the
    /// incrementally-maintained gauge takes, at open (design 306 §4). Same narrow-access rule as
    /// [`producer_seq`](Self::producer_seq): the shared helper runs here rather than exposing the
    /// writer connection across crates.
    pub fn seed_producer_census(
        &self,
        census: &writeplane::ProducerCensus,
    ) -> Result<(), BookkeepingError> {
        census.seed(self.cluster.writer_connection())
    }

    /// Read one durable deploy/public migration identity while the standalone engine owns the
    /// idle writer. Kept narrow instead of exposing the raw writer connection across crates.
    pub fn migration_record(&self, id: &str) -> Result<Option<MigrationRecord>, BookkeepingError> {
        writeplane::migration_record(self.cluster.writer_connection(), id)
    }

    /// Read the captured data-migration marker for `id` on the serialized writer.
    pub fn data_migration_checksum(&self, id: &str) -> Result<Option<String>, BookkeepingError> {
        writeplane::data_migration_checksum(self.cluster.writer_connection(), id)
    }

    /// Persist the standalone producer's DDL journal row after the marker transaction. A crash
    /// between the two is healed by replay: the durable marker skips DDL and this insert retries.
    pub fn journal_local_migration(
        &self,
        id: &str,
        checksum: Option<&str>,
        content_checksum: &str,
        run_id: &str,
        statements: &str,
        applied_at: i64,
    ) -> Result<(), BookkeepingError> {
        self.local_bookkeeping_transaction(|conn| {
            writeplane::journal_migration(
                conn,
                id,
                checksum,
                content_checksum,
                run_id,
                statements,
                applied_at,
            )
        })
    }

    /// Fill legacy migration identity columns without minting a captured data commit.
    pub fn adopt_migration_checksums(
        &self,
        id: &str,
        checksum: Option<&str>,
        content_checksum: &str,
    ) -> Result<(), BookkeepingError> {
        self.local_bookkeeping_transaction(|conn| {
            writeplane::adopt_migration_checksums(conn, id, checksum, content_checksum)
        })
    }

    /// Read one exact public-SQL outcome while the serialized standalone writer is idle.
    pub fn stored_public_outcome(
        &self,
        outcome_key: &str,
    ) -> Result<Option<StoredPublicOutcome>, BookkeepingError> {
        writeplane::read_sql_outcome(self.cluster.writer_connection(), outcome_key)
    }

    /// The durable one-shot outcome retention floor at `now_ms`.
    pub fn public_operation_floor_ms(&self, now_ms: i64) -> Result<i64, BookkeepingError> {
        writeplane::sql_outcome_floor_ms(
            self.cluster.writer_connection(),
            SqlOutcomeNamespace::Operation,
            now_ms,
        )
    }

    /// Resolve a retried public transaction commit from standalone's co-transactional outcome.
    pub fn public_commit_outcome(
        &self,
        transaction_id: &str,
        now_ms: i64,
    ) -> Result<Option<Option<String>>, crate::session::SessionError> {
        writeplane::public_commit_outcome_on_conn(
            self.cluster.writer_connection(),
            transaction_id,
            now_ms,
            |_, _, _| Ok(None),
        )
    }

    /// Sweep the bounded outcome cache in one host-local metadata transaction.
    pub fn sweep_sql_outcomes(&self, now_ms: i64) -> Result<(), BookkeepingError> {
        self.local_bookkeeping_transaction(|conn| writeplane::sweep_sql_outcomes(conn, now_ms))
    }

    /// Fresh standalone public DDL: schema + desired-index effects + exact replay outcome + TxId
    /// watermark commit as one SQLite atom.
    #[allow(clippy::too_many_arguments)]
    pub fn apply_public_ddl_operation<F>(
        &self,
        statement: &SqlStatementRequest,
        declared_tables: &[String],
        outcome_key: &str,
        request_identity: &str,
        result_byte_limit: usize,
        now_ms: i64,
        apply_step_effects: F,
    ) -> Result<(PublicOperationCommit, crate::DdlApplyReport), DdlMigrationError>
    where
        F: FnMut(&rusqlite::Connection, &crate::DdlStep) -> rusqlite::Result<()>,
    {
        self.apply.apply_public_ddl_operation(
            statement,
            declared_tables,
            outcome_key,
            request_identity,
            result_byte_limit,
            now_ms,
            apply_step_effects,
        )
    }

    /// Fresh standalone DDL migration, with its permanent identity row and exact TxId cursor
    /// committed in the checked DDL transaction rather than backfilled afterward. Both front
    /// doors — the public `/v1/sql/migrate` route and the private deploy route — apply through
    /// this one primitive so they mint identical journal rows and either can absorb the other's
    /// replay; only the opaque checksum is optional (the deploy surface accepts checksum-less
    /// DDL files, whose identity is the statement vector alone).
    #[allow(clippy::too_many_arguments)]
    pub fn apply_public_ddl_migration<F>(
        &self,
        id: &str,
        supplied_checksum: Option<&str>,
        content_checksum: &str,
        normalized: &[String],
        identity_json: &str,
        declared_tables: &[String],
        now_ms: i64,
        apply_step_effects: F,
    ) -> Result<(String, crate::DdlApplyReport), DdlMigrationError>
    where
        F: FnMut(&rusqlite::Connection, &crate::DdlStep) -> rusqlite::Result<()>,
    {
        self.apply.apply_public_ddl_migration(
            id,
            supplied_checksum,
            content_checksum,
            normalized,
            identity_json,
            declared_tables,
            now_ms,
            apply_step_effects,
        )
    }

    fn local_bookkeeping_transaction(
        &self,
        apply: impl FnOnce(&rusqlite::Connection) -> Result<(), BookkeepingError>,
    ) -> Result<(), BookkeepingError> {
        let conn = self.cluster.writer_connection();
        conn.execute_batch("BEGIN")?;
        let outcome = apply(conn).and_then(|()| {
            conn.execute_batch("COMMIT")?;
            Ok(())
        });
        if outcome.is_err() {
            let _ = conn.execute_batch("ROLLBACK");
        }
        outcome
    }

    /// Register a connection (drain-side progress bookkeeping).
    pub fn connect(&self, conn: ConnId) {
        self.handle.connect(conn);
    }

    /// Drop a connection's drain-side progress bookkeeping. The caller destroys the
    /// connection's queries separately (via [`destroy_query`](Self::destroy_query)).
    pub fn disconnect(&self, conn: ConnId) {
        self.handle.disconnect(conn);
    }

    /// Register a NORMALIZED live query for `conn` under `server_qid`. Returns the slim
    /// `hello` **synchronously** (schema-derived); the seq-0 snapshot and every later batch
    /// arrive **asynchronously** through the sink.
    ///
    /// The 1:1 convenience over the shared-query primitives: it registers an engine query
    /// **and** attaches a single subscriber whose sub id is the same `server_qid`. A server
    /// that dedups identical ASTs across subscribers uses [`register_shared_query`] +
    /// [`attach_subscriber`] directly instead.
    ///
    /// [`register_shared_query`]: Self::register_shared_query
    /// [`attach_subscriber`]: Self::attach_subscriber
    pub fn query_normalized(
        &self,
        conn: ConnId,
        server_qid: u64,
        ast: Ast,
        epoch: u64,
    ) -> Result<NormalizedHello, ReplicaError> {
        let info = self.register_shared_query(server_qid, ast)?;
        let hello = hello_from_parts(epoch, info.tables, info.normalized_fp);
        self.attach_subscriber(server_qid, server_qid, conn, epoch);
        Ok(hello)
    }

    /// Register a **shared** engine query under `eqid` (one IVM pipeline + one footprint
    /// fold), with NO subscriber yet. Identical ASTs registered under one `eqid` are
    /// computed once; subscribers attach via [`attach_subscriber`](Self::attach_subscriber),
    /// each with its own epoch / seq cursor / connection route (`RINDLE-SERVER-DESIGN.md`
    /// §6/§11). Returns the deterministic table set + fingerprint the caller turns into each
    /// subscriber's `hello`.
    pub fn register_shared_query(
        &self,
        eqid: u64,
        ast: Ast,
    ) -> Result<SharedQueryInfo, ReplicaError> {
        // The precomputed-sync path is count-only; a relationship sum/avg has no synthetic
        // table encoding, so reject it here before any of the normalize machinery runs.
        reject_unsupported_sync_aggregate(&ast).map_err(ReplicaError::Build)?;
        let schemas = self.normalized_table_schemas(&ast)?;
        let (fold, tables, normalized_fp, proj) = build_query_parts(&ast, schemas);
        let hydrated_cv = self.cluster.committed_tx_id().0;
        let worker = self.cluster.query(QueryId(eqid), ast)?;
        self.handle.register_query(
            QueryId(eqid),
            worker,
            fold,
            normalized_fp,
            proj,
            hydrated_cv,
        );
        Ok(SharedQueryInfo {
            tables,
            normalized_fp,
        })
    }

    /// Register a **parameterized query family** under `eqid` (design 310 §5.1/§5.2): one
    /// IVM pipeline over the template, plus one footprint fold **per binding**, each built
    /// from the concrete member AST (`FamilyTemplate::instantiate`, impl plan D8) so every
    /// frame a partition emits equals the standalone query's byte for byte. Returns each
    /// binding's [`SharedQueryInfo`] (its tables + fingerprint — the same for every member,
    /// modulo the literal, but computed per member so nothing is assumed).
    pub fn register_shared_family(
        &self,
        eqid: u64,
        template: &FamilyTemplate,
        bindings: &[Binding],
    ) -> Result<Vec<(Binding, SharedQueryInfo)>, ReplicaError> {
        let param_cols = self.family_param_cols(template)?;
        let mut parts = Vec::with_capacity(bindings.len());
        for b in bindings {
            parts.push((b.clone(), self.partition_parts(template, b)?));
        }
        let hydrated_cv = self.cluster.committed_tx_id().0;
        let worker = self.cluster.family(
            QueryId(eqid),
            template.stripped.clone(),
            template.params.clone(),
            bindings.to_vec(),
        )?;
        self.handle
            .register_family(QueryId(eqid), worker, param_cols, hydrated_cv);
        let mut infos = Vec::with_capacity(parts.len());
        for (b, (fold, tables, normalized_fp, proj)) in parts {
            self.handle.bind_partition(
                QueryId(eqid),
                b.clone(),
                fold,
                normalized_fp,
                proj,
                hydrated_cv,
            );
            infos.push((
                b,
                SharedQueryInfo {
                    tables,
                    normalized_fp,
                },
            ));
        }
        Ok(infos)
    }

    /// Bind one more partition of the family registered under `eqid` (design 310 §4.4):
    /// the engine hydrates only that partition; the drain gets its own footprint.
    pub fn bind_shared(
        &self,
        eqid: u64,
        template: &FamilyTemplate,
        binding: &Binding,
    ) -> Result<SharedQueryInfo, ReplicaError> {
        let (fold, tables, normalized_fp, proj) = self.partition_parts(template, binding)?;
        let hydrated_cv = self.cluster.committed_tx_id().0;
        self.cluster.bind(QueryId(eqid), binding.clone())?;
        self.handle.bind_partition(
            QueryId(eqid),
            binding.clone(),
            fold,
            normalized_fp,
            proj,
            hydrated_cv,
        );
        Ok(SharedQueryInfo {
            tables,
            normalized_fp,
        })
    }

    /// Unbind one partition of the family under `eqid`: the engine drains it silently, the
    /// drain drops its footprint (faulting any subscriber still on it).
    pub fn unbind_shared(&self, eqid: u64, binding: &Binding) -> Result<bool, ReplicaError> {
        let bound = self.cluster.unbind(QueryId(eqid), binding.clone())?;
        self.handle.unbind_partition(QueryId(eqid), binding.clone());
        Ok(bound)
    }

    /// Attach subscriber `sub` to one partition of a shared family (the family form of
    /// [`attach_subscriber`](Self::attach_subscriber)).
    pub fn attach_partition_subscriber(
        &self,
        eqid: u64,
        binding: &Binding,
        sub: u64,
        conn: ConnId,
        epoch: u64,
    ) {
        self.handle
            .attach_partition_subscriber(QueryId(eqid), binding.clone(), sub, conn, epoch);
    }

    /// The one-shot snapshot of ONE partition of a family (the family form of
    /// [`query_snapshot`](Self::query_snapshot)): the family's assembled view filtered to
    /// the rows whose partition key is `binding`, rendered under the concrete member's
    /// view schema.
    pub fn partition_snapshot(
        &self,
        eqid: u64,
        template: &FamilyTemplate,
        binding: &Binding,
    ) -> Result<serde_json::Value, ReplicaError> {
        let concrete = template.instantiate(binding);
        let schema = self.cluster.view_schema(&concrete)?;
        let param_cols = self.family_param_cols(template)?;
        let changes: Vec<crate::ChangeEvent> = self
            .cluster
            .read_snapshot(QueryId(eqid))?
            .into_iter()
            .filter(|c| rindle::canon::canonical_key(c.root_row(), &param_cols) == *binding)
            .collect();
        Ok(crate::wire_json::assembled_snapshot_to_json(
            &changes, &schema,
        ))
    }

    /// The drain-side parts of one family partition: the concrete member's fold, tables,
    /// fingerprint and projection.
    fn partition_parts(
        &self,
        template: &FamilyTemplate,
        binding: &Binding,
    ) -> Result<
        (
            crate::normalize::NormalizeFold,
            Vec<TableWireSchema>,
            u64,
            crate::normalize_protocol::ProjMap,
        ),
        ReplicaError,
    > {
        let concrete = template.instantiate(binding);
        reject_unsupported_sync_aggregate(&concrete).map_err(ReplicaError::Build)?;
        let schemas = self.normalized_table_schemas(&concrete)?;
        Ok(build_query_parts(&concrete, schemas))
    }

    /// The family's partition key as root-row column indices: the template's parameter
    /// names resolved against the root table's wire schema (whose column order is the
    /// engine's row order).
    fn family_param_cols(&self, template: &FamilyTemplate) -> Result<Vec<usize>, ReplicaError> {
        let schemas = self.normalized_table_schemas(&template.stripped)?;
        let root = schemas
            .iter()
            .find(|t| t.name == template.stripped.table)
            .ok_or_else(|| {
                ReplicaError::Schema(format!(
                    "unknown root table for query family: {}",
                    template.stripped.table
                ))
            })?;
        template
            .params
            .iter()
            .map(|p| {
                root.columns.iter().position(|c| c == p).ok_or_else(|| {
                    ReplicaError::Schema(format!(
                        "unknown parameter column {p} on {}",
                        template.stripped.table
                    ))
                })
            })
            .collect()
    }

    /// Read the **assembled** view of a registered shared query as a one-shot snapshot — the
    /// SSR REST path (`SSR-DESIGN.md` §3). No subscriber, no streaming, no lease: it re-reads
    /// the live view of `eqid` and renders it to the flat/nested wire JSON a stateless API
    /// server can hydrate directly (cells keyed by name, relationships nested inline). The
    /// query must already be registered (via
    /// [`register_shared_query`](Self::register_shared_query)); an absent query (or degraded
    /// shard) yields an empty array. `ast` is the same AST the query was registered under —
    /// it supplies the view schema (relationship names / projection / `.one()` shape).
    pub fn query_snapshot(&self, eqid: u64, ast: &Ast) -> Result<serde_json::Value, ReplicaError> {
        let schema = self.cluster.view_schema(ast)?;
        let changes = self.cluster.read_snapshot(QueryId(eqid))?;
        Ok(crate::wire_json::assembled_snapshot_to_json(
            &changes, &schema,
        ))
    }

    /// Attach subscriber `sub` (its own `conn` route + `epoch`) to a shared engine query.
    /// Its seq-0 snapshot (from the cached footprint if the query is already hydrated, else
    /// when it hydrates) and every later batch arrive asynchronously through the sink. Build
    /// the subscriber's `hello` from the [`SharedQueryInfo`] [`register_shared_query`]
    /// returned (see [`hello_for`](crate::normalize_protocol::hello_from_parts)).
    ///
    /// [`register_shared_query`]: Self::register_shared_query
    pub fn attach_subscriber(&self, eqid: u64, sub: u64, conn: ConnId, epoch: u64) {
        self.handle
            .attach_subscriber(QueryId(eqid), sub, conn, epoch);
    }

    /// Detach one subscriber; the shared engine query and its peers keep running.
    pub fn detach_subscriber(&self, sub: u64) {
        self.handle.detach_subscriber(sub);
    }

    /// Fault every subscriber on a shared engine query (each gets a terminal `faulted`) and
    /// detach them, leaving the query for a following [`destroy_query`](Self::destroy_query)
    /// — gives active subscribers a re-subscribe signal before an explicit dematerialize.
    pub fn fault_subscribers(&self, eqid: u64, reason: &str) {
        self.handle
            .fault_subscribers(QueryId(eqid), reason.to_string());
    }

    /// Tear down a registered query (cluster pipeline + drain bookkeeping, dropping every
    /// subscriber still on it). For the 1:1 [`query_normalized`](Self::query_normalized)
    /// path this removes the lone subscriber too.
    pub fn destroy_query(&self, server_qid: u64) {
        self.cluster.destroy_query(QueryId(server_qid));
        self.handle.remove_query(QueryId(server_qid));
    }

    /// Apply a batch of positional mutations as one **raw foreign write** (no `lmid`,
    /// confirms nothing), returning the commit version synchronously.
    pub fn commit_normalized(&self, muts: &[Mutation]) -> Result<u64, ReplicaError> {
        self.apply.commit_normalized(muts)
    }

    /// Apply a change-source batch AND advance the source's durable cursor in ONE write txn
    /// (CHANGE-SOURCE-DESIGN.md §4). The `_rindle_source_offsets` upsert rides the same
    /// transaction as the effects — exactly the `upsert_lmid` discipline — so a crash can
    /// never commit the data without the cursor (or vice-versa). The caller owns the
    /// monotonic-absorb dedup (`offset <= stored` ⇒ skip) BEFORE calling this; there is no
    /// gap rejection (the source owns contiguity, §4).
    pub fn commit_normalized_with_offset(
        &self,
        muts: &[Mutation],
        source: &str,
        offset: &str,
        chunk_seq: i64,
        run_id: Option<&str>,
    ) -> Result<u64, ReplicaError> {
        self.apply
            .commit_normalized_with_offset(muts, source, offset, chunk_seq, run_id)
    }

    /// Terminal step of the **streaming-follower apply** (`REPLICATOR-PRECOMMIT-STREAMING-DESIGN.md`
    /// §7): the caller has opened ONE [`ClusterWriteTxn`] via [`cluster`](Self::cluster)`.write()`
    /// and driven [`apply_muts`](Self::apply_muts) into it once per `chunk` frame; this upserts the
    /// source cursor in that SAME open txn (co-transactional with the chunk applies — a crash can
    /// never commit the data without the cursor) and commits, returning the commit version. It is
    /// exactly [`commit_normalized_with_offset`](Self::commit_normalized_with_offset)'s cursor
    /// discipline, but with the row-changes already applied incrementally as chunks arrived rather
    /// than handed over as one batch.
    pub fn commit_follower_txn(
        &self,
        txn: ClusterWriteTxn,
        source: &str,
        offset: &str,
        chunk_seq: i64,
        run_id: Option<&str>,
    ) -> Result<u64, ReplicaError> {
        self.apply
            .commit_follower_txn(txn.into_core(), source, offset, chunk_seq, run_id)
    }

    /// [`commit_follower_txn`](Self::commit_follower_txn) plus the frame's durable row-count/commit stamps.
    /// Effects, cursor, run fence, and head accounting land in one transaction.
    pub fn commit_follower_txn_with_head(
        &self,
        txn: ClusterWriteTxn,
        source: &str,
        offset: &str,
        chunk_seq: i64,
        run_id: Option<&str>,
        head: SourceHead,
    ) -> Result<u64, ReplicaError> {
        self.apply.commit_follower_txn_with_head(
            txn.into_core(),
            source,
            offset,
            chunk_seq,
            run_id,
            head,
        )
    }

    /// Advance only a CDC transport locator at an unchanged semantic cursor.
    ///
    /// The source-offset table is deliberately unregistered, so this empty
    /// application commit emits no IVM row delta. The compare-and-swap keeps a
    /// stale connection from replacing a newer locator, and the update remains
    /// a real SQLite transaction so a portable image observes either locator
    /// in full, never torn metadata.
    pub fn refresh_follower_run_id(
        &self,
        source: &str,
        offset: &str,
        previous_run_id: &str,
        next_run_id: &str,
    ) -> Result<u64, ReplicaError> {
        self.apply
            .refresh_follower_run_id(source, offset, previous_run_id, next_run_id)
    }

    /// Create the `_rindle_source_offsets` bookkeeping table (idempotent). Not registered for
    /// capture — daemon metadata, like `_rindle_sql_outcomes`.
    pub fn ensure_source_offsets_table(&self) -> Result<(), ReplicaError> {
        self.apply.ensure_source_offsets_table()
    }

    /// The durably-stored `(offset, chunk_seq, run_id)` checkpoint for `source` (`None` ⇒ never
    /// applied; the caller treats that as the genesis `""` and subscribes from the start).
    /// `chunk_seq` is [`SOURCE_OFFSET_WHOLE_RUN`] for a whole-run checkpoint (the common case) or a
    /// real within-run ordinal for a mid-run segment left by the commit-at-DDL-boundary follower
    /// (§6.6). The resume/dedup keyset is `(offset, chunk_seq)`; `run_id` is the checkpointed run's
    /// identity token, echoed on the subscribe as the fencing proof
    /// (RELAY-CURSOR-EPOCH-FENCING-DESIGN.md §2) — `None` for a pre-fence checkpoint.
    #[allow(clippy::type_complexity)]
    pub fn source_checkpoint(
        &self,
        source: &str,
    ) -> Result<Option<(String, i64, Option<String>)>, ReplicaError> {
        self.apply.source_checkpoint(source)
    }

    /// Persisted accounting carried beside the source checkpoint.
    pub fn source_head(&self, source: &str) -> Result<Option<SourceHead>, ReplicaError> {
        self.apply.source_head(source)
    }

    /// The stored §8.3 batch identity for `source`'s checkpoint (`None` = no row, or a
    /// hash-less source). Compared — never recomputed — against a resubmission's
    /// declared hash at the exact stored offset.
    pub fn source_checkpoint_hash(&self, source: &str) -> Result<Option<String>, ReplicaError> {
        self.apply.source_checkpoint_hash(source)
    }

    /// Create the `_rindle_producer_offsets` foreign-write watermark table (idempotent). The DDL
    /// is the shared one, so this cannot drift from the write-master's or the restore's copy.
    /// Callers register it for capture afterwards — it is replicated data, not host bookkeeping
    /// (design 306 §3.3).
    pub fn ensure_producer_offsets_table(&self) -> Result<(), ReplicaError> {
        self.apply.ensure_producer_offsets_table()
    }

    // ------------------- the room write-behind (RINDLE-REALTIME §5.3) -------------------

    /// Create the `_rindle_room_placement` fencing table (idempotent). Unregistered
    /// bookkeeping like the offsets table.
    pub fn ensure_room_placement_table(&self) -> Result<(), ReplicaError> {
        self.cluster.exec_ddl(&format!(
            "CREATE TABLE IF NOT EXISTS {ROOM_PLACEMENT_TABLE} \
                 (doc TEXT PRIMARY KEY, epoch INTEGER NOT NULL)"
        ))
    }

    /// Claim the next placement epoch for `doc` (§2.5): one write transaction, one
    /// monotone bump. The claim is what fences every prior epoch's flushes.
    pub fn claim_room_epoch(&self, doc: &str) -> Result<i64, ReplicaError> {
        let mut txn = self.cluster.write()?;
        txn.exec(
            &format!(
                "INSERT INTO {ROOM_PLACEMENT_TABLE} (doc, epoch) VALUES (?1, 1) \
                 ON CONFLICT(doc) DO UPDATE SET epoch = epoch + 1"
            ),
            &[OwnedValue::str(doc)],
        )?;
        let rows = txn.query(
            &format!("SELECT epoch FROM {ROOM_PLACEMENT_TABLE} WHERE doc = ?1"),
            &[OwnedValue::str(doc)],
        )?;
        let epoch = match rows.first().and_then(|r| r.first()) {
            Some(OwnedValue::Int(e)) => *e,
            other => {
                return Err(ReplicaError::Schema(format!(
                    "room placement epoch read back {other:?}"
                )))
            }
        };
        txn.commit_with_info()?;
        Ok(epoch)
    }

    /// The current placement epoch for `doc` (`None` = never claimed — a fence-bearing
    /// request against it is stale by definition).
    pub fn room_epoch(&self, doc: &str) -> Result<Option<i64>, ReplicaError> {
        self.cluster.read(|conn| {
            conn.query_row(
                &format!("SELECT epoch FROM {ROOM_PLACEMENT_TABLE} WHERE doc = ?1"),
                [doc],
                |r| r.get::<_, i64>(0),
            )
            .optional()
        })
    }

    /// The **domain-scoped** ledger's `last_mutation_id` per client under `doc` (absent =
    /// 0 to the caller) — the room's boot probe (§3.3): what lets a rebooted room absorb
    /// already-durable mutations as replay dedup instead of double-applying them. Reads
    /// [`ROOM_CLIENT_MUTATIONS_TABLE`] keyed by `(doc, client_id)` (§7.1) — NOT the
    /// slow-path [`CLIENT_MUTATIONS_TABLE`], so a client's room stream and its daemon
    /// stream never alias.
    pub fn room_lmids(
        &self,
        doc: &str,
        clients: &[String],
    ) -> Result<Vec<(String, i64)>, ReplicaError> {
        self.cluster.read(|conn| {
            let mut out = Vec::with_capacity(clients.len());
            for c in clients {
                let lmid = conn
                    .query_row(
                        &format!(
                            "SELECT last_mutation_id FROM {ROOM_CLIENT_MUTATIONS_TABLE} \
                             WHERE doc = ?1 AND client_id = ?2"
                        ),
                        [doc, c.as_str()],
                        |r| r.get::<_, i64>(0),
                    )
                    .optional()?;
                if let Some(lmid) = lmid {
                    out.push((c.clone(), lmid));
                }
            }
            Ok(out)
        })
    }

    /// Apply a room flush batch (§5.3 step 5): optional **CAS preconditions** — every
    /// change's old image (`Edit.old` / `Remove`'s old / `Add`'s asserted absence)
    /// must match the current row under the identity comparator (`null == null`, one
    /// number domain; NOT join semantics) — then effects + cursor + batch identity in
    /// ONE transaction. Any CAS miss rolls the whole batch back and returns the
    /// authoritative current images. The caller owns the fence and the dedup (both
    /// race-free on the single-threaded engine).
    ///
    /// When `doc` is `Some` this is a **room** flush: its ledger co-edits (§5.3 step 3 —
    /// the changes targeting [`CLIENT_MUTATIONS_TABLE`]) are domain-scoped and retarget to
    /// [`ROOM_CLIENT_MUTATIONS_TABLE`] keyed by `(doc, client_id)` (§7.1). They are pulled
    /// out of the CAS + effects pass and upserted monotonically in the same transaction,
    /// so the daemon's slow-path [`CLIENT_MUTATIONS_TABLE`] row is never perturbed by a
    /// room flush (§8.5's ledger-isolation invariant — the Rev 1 data-loss bug). When
    /// `doc` is `None` (a plain change source / replicator follower) the batch applies
    /// verbatim, so the slow-path lmid stream still rides its own `CLIENT_MUTATIONS_TABLE`
    /// rows unchanged.
    ///
    /// When the §4 lifecycle is enabled
    /// ([`enable_realtime_lifecycle`](Self::enable_realtime_lifecycle)) a room flush also
    /// co-commits the §4.2 downgrade-fence watermark —
    /// `ROOM_WATERMARK_TABLE(doc, flush_seq = offset)`, monotone — in the SAME
    /// transaction, so the fence rides the flush's echo through every authority shape
    /// (see the inline comment for the dedup/regression reasoning).
    ///
    /// A room flush's **outcome co-edits** (Slice I-ii — changes targeting
    /// [`ROOM_MUTATION_OUTCOMES_TABLE`], doc-less 6-wide rows) split off exactly like the
    /// ledger's: they NEVER enter the CAS pass or `apply_muts` — the split is
    /// unconditional for a `doc` flush (a lifecycle-disabled daemon would otherwise fail
    /// the whole batch on the unregistered table), while the upsert itself is gated on
    /// `realtime_lifecycle_enabled`, so **a lifecycle-disabled daemon accepts a batch
    /// carrying outcome rows and silently discards them** (it has no downgrade path to
    /// resolve through — graceful, not an error). With the lifecycle on, each row upserts
    /// keyed `(doc, client_id, mid)` (idempotent on replay by PK) in the SAME transaction
    /// as the lmid rows covering those mids — the I-ii soundness contract: an absent
    /// outcome row under a covering daemon-carried lmid means `applied`. Retention prunes
    /// in the same transaction by lmid DISTANCE ([`OUTCOME_RETENTION_LMIDS`]), for exactly
    /// the `(doc, client)` rows this flush advanced.
    pub fn commit_room_flush(
        &self,
        muts: &[Mutation],
        source: &str,
        offset: &str,
        doc: Option<&str>,
        batch_hash: Option<&str>,
        cas: bool,
    ) -> Result<RoomFlushOutcome, ReplicaError> {
        let mut txn = self.cluster.write()?;
        // A room flush (doc present) domain-scopes its ledger co-edits: split them off so
        // the CAS check + `apply_muts` below see only the data rows, and the ledger rows
        // land in `ROOM_CLIENT_MUTATIONS_TABLE(doc, …)` instead of the slow-path table.
        // The outcome co-edits (I-ii) split with them — unconditionally, lifecycle or not:
        // left in the data pass they would error in `apply_muts` (the table is
        // unregistered without the lifecycle) or CAS-trip (the wire row is doc-less,
        // width-mismatched against the registered 7-column table).
        let data: Vec<Mutation> = if doc.is_some() {
            muts.iter()
                .filter(|m| {
                    m.table() != CLIENT_MUTATIONS_TABLE && m.table() != ROOM_MUTATION_OUTCOMES_TABLE
                })
                .cloned()
                .collect()
        } else {
            Vec::new()
        };
        let data_muts: &[Mutation] = if doc.is_some() { &data } else { muts };
        if cas {
            let tables = self.apply.tables();
            let mut conflicts: Vec<RoomRowConflict> = Vec::new();
            for m in data_muts {
                let table = m.table();
                let meta = tables
                    .get(table)
                    .ok_or_else(|| ReplicaError::Schema(format!("unknown table: {table}")))?;
                let (asserted, keyed): (Option<&[OwnedValue]>, &[OwnedValue]) = match m {
                    Mutation::Add { row, .. } => (None, row),
                    Mutation::Remove { row, .. } => (Some(row), row),
                    Mutation::Edit { old, .. } => (Some(old), old),
                    Mutation::Truncate { .. } => (None, &[][..]),
                    Mutation::Upsert { row, .. } => (None, row),
                };
                // A width drift (a migration landed under the batch, §8.4/T9) can't
                // key the row — surface it as a conflict on a best-effort pk.
                let widthy = keyed.len() == meta.columns.len();
                let pk_cells: Vec<OwnedValue> = if widthy {
                    meta.pk.iter().map(|&c| keyed[c].clone()).collect()
                } else {
                    keyed.first().cloned().into_iter().collect()
                };
                let current = if widthy {
                    let cols = meta.columns.join(", ");
                    let pred = meta
                        .pk
                        .iter()
                        .enumerate()
                        .map(|(i, &c)| format!("{} = ?{}", meta.columns[c], i + 1))
                        .collect::<Vec<_>>()
                        .join(" AND ");
                    txn.query(
                        &format!("SELECT {cols} FROM {table} WHERE {pred}"),
                        &pk_cells,
                    )?
                    .into_iter()
                    .next()
                } else {
                    None
                };
                let holds = match (asserted, &current) {
                    (None, None) => widthy, // an add asserts absence
                    (Some(a), Some(c)) => {
                        a.len() == c.len()
                            && a.iter().zip(c.iter()).all(|(x, y)| cas_cell_matches(x, y))
                    }
                    _ => false,
                };
                if !holds {
                    conflicts.push(RoomRowConflict {
                        table: table.to_string(),
                        pk: pk_cells,
                        current: current.clone(),
                    });
                }
            }
            if !conflicts.is_empty() {
                // Dropping the txn rolls everything back: nothing applied (§5.3). The
                // domain-scoped ledger upsert below is never reached, so a data CAS miss
                // also rolls back the lmid advance — the batch stays atomic.
                return Ok(RoomFlushOutcome::Conflicts(conflicts));
            }
        }
        self.apply_muts(&mut txn, data_muts)?;
        // §5.3 step 3, domain-scoped (§7.1): the ledger co-edits retarget to
        // `ROOM_CLIENT_MUTATIONS_TABLE` keyed by `(doc, client_id)`. Monotonic upsert (the
        // room owns this stream and flushes are epoch-fenced, so last-writer-wins is safe);
        // a `Remove` would retire a row, though the room never emits one for lmids.
        if let Some(doc) = doc {
            // The `(client, lmid)` rows this flush advances — the outcome prune below keys
            // on exactly these (prune only when the covering watermark moves).
            let mut advanced: Vec<(OwnedValue, i64)> = Vec::new();
            for m in muts.iter().filter(|m| m.table() == CLIENT_MUTATIONS_TABLE) {
                match m {
                    Mutation::Truncate { .. } => {}
                    Mutation::Add { row, .. }
                    | Mutation::Edit { new: row, .. }
                    | Mutation::Upsert { row, .. } => {
                        txn.exec(
                            &format!(
                                "INSERT INTO {ROOM_CLIENT_MUTATIONS_TABLE} \
                                 (doc, client_id, last_mutation_id) VALUES(?1, ?2, ?3) \
                                 ON CONFLICT(doc, client_id) DO UPDATE \
                                 SET last_mutation_id = excluded.last_mutation_id"
                            ),
                            &[OwnedValue::str(doc), row[0].clone(), row[1].clone()],
                        )?;
                        if let Some(OwnedValue::Int(lmid)) = row.get(1) {
                            advanced.push((row[0].clone(), *lmid));
                        }
                    }
                    Mutation::Remove { row, .. } => {
                        txn.exec(
                            &format!(
                                "DELETE FROM {ROOM_CLIENT_MUTATIONS_TABLE} \
                                 WHERE doc = ?1 AND client_id = ?2"
                            ),
                            &[OwnedValue::str(doc), row[0].clone()],
                        )?;
                    }
                }
            }
            // Slice I-ii: the outcome co-edits land keyed `(doc, client_id, mid)` in this
            // SAME transaction as the lmid rows covering them — the soundness contract a
            // post-downgrade client resolves by (absent row under a covering lmid =
            // `applied`) only holds if the pair is atomic. Gated on the lifecycle: a
            // disabled daemon has no outcomes table and DISCARDS the rows (documented
            // above). Upserts are idempotent by PK, so a keyset-replayed batch re-lands
            // identical rows; a `Remove` retires one (the room never emits it — symmetry
            // with the ledger split).
            if self.cluster.realtime_lifecycle_enabled() {
                for m in muts
                    .iter()
                    .filter(|m| m.table() == ROOM_MUTATION_OUTCOMES_TABLE)
                {
                    match m {
                        Mutation::Truncate { .. } => {}
                        Mutation::Add { row, .. }
                        | Mutation::Edit { new: row, .. }
                        | Mutation::Upsert { row, .. } => {
                            // The wire row is the doc-less `(client_id, mid, kind, reason,
                            // name, args)` — a wrong width is a room-side protocol bug,
                            // refused loudly (whole-txn rollback), never partially stored.
                            if row.len() != 6 {
                                return Err(ReplicaError::Mutation(format!(
                                    "room outcome row must be 6-wide \
                                     (client_id, mid, kind, reason, name, args); got {}",
                                    row.len()
                                )));
                            }
                            let mut params = vec![OwnedValue::str(doc)];
                            params.extend(row.iter().cloned());
                            txn.exec(
                                &format!(
                                    "INSERT INTO {ROOM_MUTATION_OUTCOMES_TABLE} \
                                     (doc, client_id, mid, kind, reason, name, args) \
                                     VALUES(?1, ?2, ?3, ?4, ?5, ?6, ?7) \
                                     ON CONFLICT(doc, client_id, mid) DO UPDATE \
                                     SET kind = excluded.kind, reason = excluded.reason, \
                                         name = excluded.name, args = excluded.args"
                                ),
                                &params,
                            )?;
                        }
                        Mutation::Remove { row, .. } => {
                            txn.exec(
                                &format!(
                                    "DELETE FROM {ROOM_MUTATION_OUTCOMES_TABLE} \
                                     WHERE doc = ?1 AND client_id = ?2 AND mid = ?3"
                                ),
                                &[OwnedValue::str(doc), row[0].clone(), row[1].clone()],
                            )?;
                        }
                    }
                }
                // Retention, AFTER the upserts so an over-K backlog in one batch degrades
                // exactly like the shell's FIFO (oldest first): prune by lmid DISTANCE —
                // never time — for each `(doc, client)` this flush advanced. `mid ≤
                // lmid − K` keeps the K-deep window `(lmid − K, lmid]`; a pruned mid reads
                // as `applied` (the H-v-mirrored loss class, see OUTCOME_RETENTION_LMIDS).
                for (client, lmid) in advanced {
                    txn.exec(
                        &format!(
                            "DELETE FROM {ROOM_MUTATION_OUTCOMES_TABLE} \
                             WHERE doc = ?1 AND client_id = ?2 AND mid <= ?3"
                        ),
                        &[
                            OwnedValue::str(doc),
                            client,
                            OwnedValue::Int(lmid - OUTCOME_RETENTION_LMIDS),
                        ],
                    )?;
                }
            }
            // §4.2: the downgrade-fence watermark co-commits in the SAME transaction as
            // the flush it covers — one more row in a batch that already co-commits lmid
            // rows. Gated on the lifecycle being enabled (`realtime_lifecycle_enabled`),
            // so a daemon that never called `enable_realtime_lifecycle` applies room
            // flushes exactly as before. A CAS miss returned above (whole-txn rollback),
            // so the watermark can never land on a conflicted batch; the server's keyset
            // dedup (`run_already_applied`, rindle-server net.rs) short-circuits replays
            // before this path runs, and the monotone guard makes even a raw reordered
            // replica-level caller unable to regress the fence.
            if self.cluster.realtime_lifecycle_enabled() {
                // A room-plane offset IS the flush seq, zero-padded decimal (`padOffset`,
                // packages/room/src/shell.ts). A `doc`-flush offset that doesn't parse
                // back is a protocol violation — loud, never silently skipped.
                let flush_seq: i64 = offset.parse().map_err(|_| {
                    ReplicaError::Mutation(format!(
                        "room flush offset is not a flush seq (doc {doc:?}, offset {offset:?})"
                    ))
                })?;
                txn.exec(
                    &format!(
                        "INSERT INTO {ROOM_WATERMARK_TABLE} (doc, flush_seq) VALUES (?1, ?2) \
                         ON CONFLICT(doc) DO UPDATE SET flush_seq = excluded.flush_seq \
                         WHERE excluded.flush_seq > {ROOM_WATERMARK_TABLE}.flush_seq"
                    ),
                    &[OwnedValue::str(doc), OwnedValue::Int(flush_seq)],
                )?;
            }
        }
        upsert_source_offset_hashed(
            txn.core_mut(),
            source,
            offset,
            SOURCE_OFFSET_WHOLE_RUN,
            None,
            batch_hash,
        )?;
        let info = txn.commit_with_info()?;
        Ok(RoomFlushOutcome::Applied { cv: info.tx_id.0 })
    }

    /// The durably-stored cursor string for `source`, discarding the `chunk_seq` sub-position — for
    /// the string-only callers (snapshot-restore resume points, which are always run boundaries). The
    /// resume/dedup paths use [`source_checkpoint`](Self::source_checkpoint) for the full keyset.
    pub fn source_offset(&self, source: &str) -> Result<Option<String>, ReplicaError> {
        self.apply.source_offset(source)
    }

    /// Create the `_rindle_applied_ddl` idempotency journal (idempotent). Not registered for capture —
    /// daemon metadata, like `_rindle_source_offsets`. `actions` holds the entry's ordered apply
    /// report (design 227 fourth review pass), written in the same transaction as the marker.
    pub fn ensure_applied_ddl_table(&self) -> Result<(), ReplicaError> {
        self.apply.ensure_applied_ddl_table()
    }

    /// Whether a `ddl` entry keyed by `key` (migration id / offset) is already journaled — the
    /// crash-window-replay dedup, checked BEFORE re-applying.
    pub fn ddl_already_applied(&self, key: &str) -> Result<bool, ReplicaError> {
        self.apply.ddl_already_applied(key)
    }

    /// Apply a `ddl` entry's `statements` and journal its `key` atomically — one ordinary
    /// transaction, so retries can detect applied DDL from the durable marker. Delegates to
    /// [`Cluster::exec_ddl_with_marker`] against the [`APPLIED_DDL_TABLE`] journal.
    pub fn apply_ddl_with_marker(
        &self,
        key: &str,
        statements: &[String],
    ) -> Result<crate::DdlApplyReport, ReplicaError> {
        self.apply.apply_ddl_with_marker(key, statements)
    }

    /// Apply DDL, caller-owned per-statement bookkeeping effects, and the durable marker in one
    /// transaction. See [`Cluster::exec_ddl_with_marker_and_step_effects`].
    pub fn apply_ddl_with_marker_and_step_effects<F>(
        &self,
        key: &str,
        statements: &[String],
        apply_step_effects: F,
    ) -> Result<crate::DdlApplyReport, ReplicaError>
    where
        F: FnMut(&rusqlite::Connection, &crate::DdlStep) -> rusqlite::Result<()>,
    {
        self.apply
            .apply_ddl_with_marker_and_step_effects(key, statements, apply_step_effects)
    }

    /// The ordered apply report persisted with `key`'s marker (same transaction as the DDL), or
    /// `None` for entries marked before the report column existed (they degrade to the caller's
    /// end-state fallback). A replay consumes this instead of re-observing — the DDL does not
    /// re-run, so there is nothing to observe (design 227 fourth review pass).
    pub fn stored_ddl_report(
        &self,
        key: &str,
    ) -> Result<Option<crate::DdlApplyReport>, ReplicaError> {
        self.apply.stored_ddl_report(key)
    }

    /// Open one mutation's write transaction; run SQL against the handle, then
    /// [`ClusterMutationWrite::commit_with_lmid`] lands effects + lmid atomically (or
    /// [`commit`](ClusterMutationWrite::commit) for a foreign write).
    pub fn begin_mutation(&self) -> Result<ClusterMutationWrite, ReplicaError> {
        Ok(ClusterMutationWrite {
            txn: Some(self.cluster.write()?),
        })
    }

    /// Apply positional mutations to an open cluster write transaction (build_sql per row).
    pub fn apply_muts(
        &self,
        txn: &mut ClusterWriteTxn,
        muts: &[Mutation],
    ) -> Result<(), ReplicaError> {
        self.apply.apply_muts(txn.core_mut(), muts)
    }

    /// The flat schema of every base table the query's tree can surface (root + each
    /// `related`/EXISTS child table), for the publisher's `hello` + PK map. A relationship
    /// **aggregate** surfaces a SYNTHETIC table (`AGGREGATE-SYNC-DESIGN.md` §3.2) whose
    /// schema is derived from the AST (`agg_table_schemas`), not the DB registry — the
    /// child rows it replaces are never synced.
    pub fn normalized_table_schemas(
        &self,
        ast: &Ast,
    ) -> Result<Vec<TableWireSchema>, ReplicaError> {
        use crate::normalize::{agg_table_schemas, AggTable};
        use std::collections::HashMap;
        let synth: HashMap<Box<str>, AggTable> = agg_table_schemas(ast)
            .into_iter()
            .map(|t| (t.name.clone(), t))
            .collect();
        let mut names = std::collections::BTreeSet::new();
        collect_table_names(&table_tree(ast), &mut names);
        let tables = self.apply.tables();
        let mut out = Vec::with_capacity(names.len());
        for name in &names {
            if let Some(at) = synth.get(name.as_str()) {
                // Synthetic aggregate table: schema from the AST, PK = the leading group cols.
                out.push(TableWireSchema {
                    name: at.name.clone(),
                    columns: at.columns.clone(),
                    primary_key: (0..at.key_len as u32).collect(),
                });
            } else {
                let meta = tables.get(name).ok_or_else(|| {
                    ReplicaError::Schema(format!("unknown table in query tree: {name}"))
                })?;
                out.push(TableWireSchema {
                    name: name.as_str().into(),
                    columns: meta.columns.iter().map(|c| c.as_str().into()).collect(),
                    primary_key: meta.pk.iter().map(|&i| i as u32).collect(),
                });
            }
        }
        Ok(out)
    }
}

fn collect_table_names(node: &TableNode, out: &mut std::collections::BTreeSet<String>) {
    out.insert(node.table.to_string());
    // Pruned (`exists_noSync`) slots are `None`: their permission tables are never synced,
    // so they correctly drop out of the needed-tables set.
    for child in node.children.iter().flatten() {
        collect_table_names(child, out);
    }
}

/// One server-side mutation's open cluster transaction (see [`ClusterConsumer::begin_mutation`]).
/// Dropping without committing rolls back.
pub struct ClusterMutationWrite {
    txn: Option<ClusterWriteTxn>,
}

impl ClusterMutationWrite {
    fn txn(&mut self) -> Result<&mut ClusterWriteTxn, ReplicaError> {
        self.txn
            .as_mut()
            .ok_or_else(|| ReplicaError::Mutation("transaction already finished".into()))
    }

    /// Run one statement with positional parameters (the mutator's write path).
    pub fn exec(&mut self, sql: &str, params: &[OwnedValue]) -> Result<usize, ReplicaError> {
        self.txn()?.exec(sql, params)
    }

    /// Run one mutation statement under the shared writer time/VM budget. `guarded` additionally
    /// installs the public reserved-object authorizer for the statement.
    pub fn exec_bounded(
        &mut self,
        sql: &str,
        params: &[OwnedValue],
        guarded: bool,
    ) -> Result<usize, StatementRunError> {
        let txn = self.txn.as_mut().ok_or_else(|| {
            StatementRunError::Malformed("mutation transaction already finished".into())
        })?;
        txn.exec_bounded(sql, params, guarded)
    }

    /// Run one statement on the public v1 surface. Interactive transactions preserve the outer
    /// transaction on ordinary statement errors; one-shot batches set `preserve_on_error=false`
    /// and let the shared coordinator roll the complete unit back.
    pub fn public_statement(
        &mut self,
        statement: &SqlStatementRequest,
        preserve_on_error: bool,
        result_byte_limit: usize,
    ) -> Result<StatementResult, StatementRunError> {
        let txn = self.txn.as_mut().ok_or_else(|| {
            StatementRunError::Malformed("mutation transaction already finished".into())
        })?;
        let result = txn.public_statement(statement, preserve_on_error, result_byte_limit);
        if result.is_err() && !txn.is_open() {
            // A savepoint setup/release failure or SQLite auto-abort closed the outer transaction.
            // Drop the finished handle now so the next admitted request can open the writer.
            self.txn.take();
        }
        result
    }

    pub fn public_transaction_open(&self) -> bool {
        self.txn.as_ref().is_some_and(ClusterWriteTxn::is_open)
    }

    /// Run a read through the open transaction (sees its own uncommitted writes, §4.1).
    pub fn query(
        &mut self,
        sql: &str,
        params: &[OwnedValue],
    ) -> Result<Vec<Vec<OwnedValue>>, ReplicaError> {
        self.txn()?.query(sql, params)
    }

    /// [`query`](Self::query) plus ordered column names, enforcing SQLite's read-only verdict for
    /// the shared mutation-session contract.
    pub fn query_with_cols(
        &mut self,
        sql: &str,
        params: &[OwnedValue],
    ) -> Result<(Vec<String>, Vec<Vec<OwnedValue>>), ReplicaError> {
        self.txn()?.query_with_cols(sql, params)
    }

    /// Advance a foreign writer's producer watermark inside this open transaction, so the receipt
    /// commits with the effects it deduplicates or not at all.
    ///
    /// Written through the capture-aware [`exec`](ClusterWriteTxn::exec) — NOT the connection-only
    /// helper — because the watermark is replicated data (design 306 §3.3): an uncaptured row is
    /// absent from the journal, so a restored store would come back holding every effect and no
    /// dedup state, and the first retry after the restore would re-apply. The SQL itself is the
    /// shared one, so this cannot drift from [`writeplane::upsert_producer_seq`].
    pub fn upsert_producer_seq(
        &mut self,
        producer: &str,
        seq: u64,
    ) -> Result<(), BookkeepingError> {
        let txn = self.txn.as_mut().ok_or_else(|| {
            BookkeepingError::StoreFormat("mutation transaction already finished".into())
        })?;
        txn.exec(
            &writeplane::producer_offsets_upsert_sql(),
            &[OwnedValue::str(producer), OwnedValue::Int(seq as i64)],
        )
        .map(|_| ())
        .map_err(|error| BookkeepingError::StoreFormat(error.to_string()))
    }

    /// Append the captured apply-once marker at the tail of a standalone data migration.
    pub fn journal_data_migration(
        &mut self,
        id: &str,
        content_checksum: &str,
        applied_at: i64,
    ) -> Result<(), BookkeepingError> {
        let txn = self.txn.as_ref().ok_or_else(|| {
            BookkeepingError::StoreFormat("mutation transaction already finished".into())
        })?;
        writeplane::journal_data_migration(txn.connection(), id, content_checksum, applied_at)
    }

    /// Upsert `last_mutation_id = mid` for `client_id` **in this transaction**, commit, and
    /// return the commit version `cv` synchronously. The lmid row is ordinary captured
    /// data: it derives through the client's own system query and is released by the same
    /// `cv_min` as the commit's effects.
    pub fn commit_with_lmid(&mut self, client_id: &str, mid: u64) -> Result<u64, ReplicaError> {
        let mut txn = self
            .txn
            .take()
            .ok_or_else(|| ReplicaError::Mutation("transaction already finished".into()))?;
        txn.exec(
            &format!(
                "INSERT INTO {CLIENT_MUTATIONS_TABLE}(client_id, last_mutation_id) VALUES(?1, ?2) \
                 ON CONFLICT(client_id) DO UPDATE \
                 SET last_mutation_id = excluded.last_mutation_id"
            ),
            &[OwnedValue::str(client_id), OwnedValue::Int(mid as i64)],
        )?;
        let info = txn.commit_with_info()?;
        Ok(info.tx_id.0)
    }

    /// Commit WITHOUT an lmid advance (a foreign/system write: confirms no client mutation),
    /// returning the commit version `cv`.
    pub fn commit(&mut self) -> Result<u64, ReplicaError> {
        let txn = self
            .txn
            .take()
            .ok_or_else(|| ReplicaError::Mutation("transaction already finished".into()))?;
        let info = txn.commit_with_info()?;
        Ok(info.tx_id.0)
    }

    /// Store one public batch/transaction terminal outcome in the same SQLite transaction as its
    /// application effects. A captured unit renders the predicted next TxId into `cursor` before
    /// COMMIT; a zero-effect unit commits only the outcome metadata and does not advance TxId.
    pub fn commit_public_outcome(
        &mut self,
        outcome_key: &str,
        request_identity: Option<&str>,
        result_json: &str,
        now_ms: i64,
    ) -> Result<Option<String>, DdlMigrationError> {
        let txn = self
            .txn
            .take()
            .ok_or_else(|| ReplicaError::Mutation("transaction already finished".into()))?;
        let captured = txn.captured_user_event_count() != 0;
        let cursor = captured.then(|| format!("w:{:016x}", txn.next_tx_id().0));
        if let Err(error) = writeplane::insert_sql_outcome_with_cursor(
            txn.connection(),
            outcome_key,
            result_json,
            None,
            request_identity,
            cursor.as_deref(),
            now_ms,
        ) {
            txn.rollback();
            return Err(error.into());
        }
        if captured {
            let expected = txn.next_tx_id();
            let committed = txn.commit_with_info()?;
            debug_assert_eq!(committed.tx_id, expected);
        } else {
            txn.commit_metadata_only()?;
        }
        Ok(cursor)
    }

    /// Roll back: effects discarded, nothing delivered.
    pub fn rollback(&mut self) {
        if let Some(txn) = self.txn.take() {
            txn.rollback();
        }
    }
}