rindle_cdc_apply/

consumer.rs

//! [`ApplyConsumer`] — the headless twin of `rindle-replica`'s `ClusterConsumer` apply
//! half (design 309 §3): everything needed to apply a rindle CDC stream to a SQLite
//! database — table registration, row apply, the co-transactional cursor discipline,
//! replicated DDL with idempotency markers, resume — with **no derivation**. The live
//! `ClusterConsumer` delegates every one of these methods here (one implementation,
//! moved, never forked — the byte-identity argument of design 309 §5); a headless host
//! (`rindle-backup-sqlite`'s portable replay, an external stream consumer) composes
//! this with [`NoFanout`](super::NoFanout) and nothing else.
//!
//! The type system replaces the old refusal rule: this consumer simply has no
//! derivation methods (`query_normalized`, `read_snapshot`, `view_schema` do not
//! exist here), so a host that cannot derive cannot express the call.

use std::collections::HashMap;
use std::path::Path;
use std::rc::Rc;
use std::sync::{Mutex, MutexGuard};

use rindle_value::value::OwnedValue;
use rusqlite::OptionalExtension;

use rindle_writeplane::schema_envelope::validate_default_expression;
use rindle_writeplane::table_shape::{
    build_mutation_sql, create_table_ddl, quote_ident, ColType, Mutation, TableMeta,
};
use rindle_writeplane::{
    writeplane, ForeignKeys, JournalMode, ReplicaError, SqlArgs, SqlStatementRequest,
    StatementResult, CLIENT_MUTATIONS_TABLE, SOURCE_OFFSETS_TABLE, SOURCE_OFFSET_WHOLE_RUN,
};

use super::fanout::{CommitFanout, NoFanout};
use super::store::{ApplyStore, DdlMigrationError, DEFAULT_WRITER_BEGIN_SQL};
use super::txn::ApplyTxn;

/// The pg-source type-metadata sidecar (`_rindle_columns`; pg-source `schema_map`, Amendment 2026-06-30).
/// Present only on a **PG-mirror follower**, where the mirror `CREATE TABLE` is affinity-only
/// (`INTEGER`/`TEXT`/`REAL`) and never marks a non-PK column `NOT NULL` — so the semantic type
/// (`boolean`/`json`) and the nullability both live here, not in the declared type a `PRAGMA` can read.
/// [`ApplyConsumer::register_existing_table`] consults it so `/schema` codegen is faithful (design 206).
/// Named by value, not imported: the apply plane does not depend on `rindle-pg-source`, and the tokens are
/// a persisted-file contract (they live in the follower's own SQLite), matched by string like the rows.
const RINDLE_COLUMNS_TABLE: &str = "_rindle_columns";
const RINDLE_TABLES_TABLE: &str = "_rindle_tables";

/// Recover the client [`ColType`] from a `_rindle_columns` row's `(rindle_type, pg_type_tag)` tokens.
///
/// `rindle_type` (the source profile's `RindleType::label()`) splits `boolean`/`integer`/`float`/
/// `string`/`json` (design 406 §11.1 writes `json` directly; a pre-406 sidecar reports a `json`/
/// `jsonb` column as `string`, which the stable `pg_type_tag` still distinguishes). `integer` is
/// the exact-i64 plane (design 226). Every other text-stored family (uuid, numeric, temporal,
/// bytea-hex, text) surfaces as `String`, matching the engine's own value type. Unknown tokens
/// fall through to `String`, the permissive default.
fn coltype_from_sidecar(rindle_type: &str, pg_type_tag: &str) -> ColType {
    match rindle_type {
        "boolean" => ColType::Boolean,
        "integer" => ColType::Int,
        "float" | "number" => ColType::Number,
        "json" => ColType::Json,
        // "string" (and any unrecognized label): text-stored. json/jsonb are the one text-stored kind the
        // client surfaces distinctly, recovered here from the type tag.
        _ => match pg_type_tag {
            "RINDLE_PG_JSON" | "RINDLE_PG_JSONB" => ColType::Json,
            _ => ColType::String,
        },
    }
}

/// The `_rindle_source_offsets` DDL — one string, shared with the snapshot/restore
/// stores that mint the table on a bare connection so it cannot drift from the
/// consumer's own [`ensure_source_offsets_table`](ApplyConsumer::ensure_source_offsets_table).
pub fn source_offsets_table_ddl() -> String {
    format!(
        "CREATE TABLE IF NOT EXISTS {SOURCE_OFFSETS_TABLE} \
             (source TEXT PRIMARY KEY, offset TEXT NOT NULL, \
              chunk_seq INTEGER NOT NULL DEFAULT {SOURCE_OFFSET_WHOLE_RUN}, run_id TEXT, \
              batch_hash TEXT, run_rows INTEGER, rows_cum INTEGER, committed_at INTEGER)"
    )
}

/// [`source_offsets_table_ddl`] applied to a caller-owned connection (the
/// initial-snapshot store writes it before any consumer exists).
pub fn ensure_source_offsets_table_on_conn(
    conn: &rusqlite::Connection,
) -> Result<(), ReplicaError> {
    conn.execute_batch(&source_offsets_table_ddl())
        .map_err(|error| ReplicaError::sqlite("create source offsets table", error))
}

/// Upsert the durable `(offset, chunk_seq)` checkpoint for `source` into `txn` (co-transactional
/// with the effects that txn carries — a crash can never land data without its cursor, §4). Shared
/// by the whole-batch ([`ApplyConsumer::commit_normalized_with_offset`]) and streaming-follower
/// ([`ApplyConsumer::commit_follower_txn`]) commit paths. A whole-run commit passes
/// [`SOURCE_OFFSET_WHOLE_RUN`]; a mid-run segment passes the boundary chunk's `chunk_seq`.
/// `run_id` is the run's identity token from its `begin` frame (the fencing proof echoed on the
/// next subscribe — RELAY-CURSOR-EPOCH-FENCING-DESIGN.md §2); `None` (a pre-fence upstream or a
/// non-streaming source) stores NULL, which subscribes without a fence as before.
pub fn upsert_source_offset(
    txn: &mut ApplyTxn,
    source: &str,
    offset: &str,
    chunk_seq: i64,
    run_id: Option<&str>,
) -> Result<(), ReplicaError> {
    upsert_source_offset_hashed(txn, source, offset, chunk_seq, run_id, None)
}

/// [`upsert_source_offset`] carrying the §8.3 **batch identity**: the emitter-computed
/// `batch_hash` of the run's exact bytes, stored beside the cursor so a same-offset
/// resubmission with a different body is a loud error, never a silent dedup (T6). The
/// hash-less paths (follower stream, plain change sources) write `NULL` — once the
/// cursor advances, the previous offset's identity is dead anyway.
pub fn upsert_source_offset_hashed(
    txn: &mut ApplyTxn,
    source: &str,
    offset: &str,
    chunk_seq: i64,
    run_id: Option<&str>,
    batch_hash: Option<&str>,
) -> Result<(), ReplicaError> {
    upsert_source_offset_full(txn, source, offset, chunk_seq, run_id, batch_hash, None)
}

fn upsert_source_offset_full(
    txn: &mut ApplyTxn,
    source: &str,
    offset: &str,
    chunk_seq: i64,
    run_id: Option<&str>,
    batch_hash: Option<&str>,
    head: Option<SourceHead>,
) -> Result<(), ReplicaError> {
    let head = head.unwrap_or_default();
    let counter = |name: &str, value: Option<u64>| -> Result<OwnedValue, ReplicaError> {
        value
            .map(|value| {
                i64::try_from(value).map(OwnedValue::Int).map_err(|_| {
                    ReplicaError::Mutation(format!("source {name} exceeds SQLite INTEGER: {value}"))
                })
            })
            .transpose()
            .map(|value| value.unwrap_or(OwnedValue::Null))
    };
    let run_rows = counter("run_rows", head.run_rows)?;
    let rows_cum = counter("rows_cum", head.rows_cum)?;
    // `run_id` is COALESCE'd, not overwritten: an engine-written empty frame carries
    // run_id=None but does not change the timeline, so it must not erase the run
    // fence (RELAY-CURSOR-EPOCH-FENCING §2). A restore landing on an empty-cid head
    // would otherwise clear the fence to NULL and resume unfenced. The accounting
    // columns below are COALESCE'd for the same reason (preserve across empty advances).
    txn.exec(
        &format!(
            "INSERT INTO {SOURCE_OFFSETS_TABLE} \
             (source, offset, chunk_seq, run_id, batch_hash, run_rows, rows_cum, committed_at) \
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) \
             ON CONFLICT(source) DO UPDATE SET offset = excluded.offset, \
             chunk_seq = excluded.chunk_seq, \
             run_id = COALESCE(excluded.run_id, {SOURCE_OFFSETS_TABLE}.run_id), \
             batch_hash = excluded.batch_hash, \
             run_rows = COALESCE(excluded.run_rows, {SOURCE_OFFSETS_TABLE}.run_rows), \
             rows_cum = COALESCE(excluded.rows_cum, {SOURCE_OFFSETS_TABLE}.rows_cum), \
             committed_at = COALESCE(excluded.committed_at, {SOURCE_OFFSETS_TABLE}.committed_at)"
        ),
        &[
            OwnedValue::str(source),
            OwnedValue::str(offset),
            OwnedValue::Int(chunk_seq),
            match run_id {
                Some(run_id) => OwnedValue::str(run_id),
                None => OwnedValue::Null,
            },
            batch_hash.map_or(OwnedValue::Null, OwnedValue::str),
            run_rows,
            rows_cum,
            head.committed_at.map_or(OwnedValue::Null, OwnedValue::Int),
        ],
    )?;
    Ok(())
}

/// The follower's DDL idempotency journal: one row per applied `ddl` entry, keyed by the migration
/// id (or the entry offset when a source ships none). Unregistered bookkeeping like the offsets
/// table — never captured/fanned. A crash-window replay (the `ddl` re-delivered before its cursor
/// advanced) dedups against this BEFORE re-applying, so an already-applied reshape is skipped exactly
/// rather than re-run-and-inferred-from-the-error (see [`ApplyStore::exec_ddl_with_marker`]).
pub const APPLIED_DDL_TABLE: &str = "_rindle_applied_ddl";

/// Persisted source accounting at the applied cursor. Portable bases carry this
/// so lag/change-count stamps resume from restored history instead of zero.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct SourceHead {
    pub run_rows: Option<u64>,
    pub rows_cum: Option<u64>,
    pub committed_at: Option<i64>,
}

/// The registered-table metadata for [`CLIENT_MUTATIONS_TABLE`] — one literal, shared by
/// the headless enable and the live `ClusterConsumer`'s so the two can never drift.
pub fn client_mutations_table_meta() -> TableMeta {
    TableMeta {
        columns: vec!["client_id".to_string(), "last_mutation_id".to_string()],
        pk: vec![0],
        col_types: vec![ColType::String, ColType::Number],
        // Bookkeeping table (both columns NOT NULL); excluded from `base_table_schemas`.
        nullable: vec![false, false],
    }
}

/// The engine-free CDC apply consumer. `!Send` — lives on one thread, like the store
/// under it. See the module docs; construction is [`open`](Self::open) (headless) or
/// [`from_parts`](Self::from_parts) (a derivation host composing over a shared store).
pub struct ApplyConsumer {
    store: Rc<ApplyStore>,
    fanout: Rc<dyn CommitFanout>,
    tables: Mutex<HashMap<String, TableMeta>>,
    /// What a running backfill hides (design 406 §10): tables and `(table, column)`s whose
    /// sidecar row says `visible = 0`. `TableMeta` keeps every physical column so positional
    /// mutations are unaffected; only the advertised schema is filtered.
    hidden: Mutex<Hidden>,
}

/// The visibility sidecars' hidden set (406 §10).
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Hidden {
    pub tables: std::collections::BTreeSet<String>,
    pub columns: std::collections::BTreeSet<(String, String)>,
}

/// Read the visibility sidecars. A store without them (the direct-SQLite topology, or a mirror
/// minted before the bit existed) hides nothing.
fn read_hidden(conn: &rusqlite::Connection) -> Result<Hidden, rusqlite::Error> {
    let mut hidden = Hidden::default();
    let has_tables_sidecar = conn
        .prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1")?
        .exists([RINDLE_TABLES_TABLE])?;
    if has_tables_sidecar {
        let mut stmt = conn.prepare(&format!(
            "SELECT \"table_name\" FROM {} WHERE \"visible\" = 0",
            quote_ident(RINDLE_TABLES_TABLE)
        ))?;
        let mut rows = stmt.query([])?;
        while let Some(row) = rows.next()? {
            hidden.tables.insert(row.get(0)?);
        }
    }
    let has_visible_bit = conn
        .prepare(&format!(
            "SELECT 1 FROM pragma_table_info({}) WHERE name = 'visible'",
            quote_str(RINDLE_COLUMNS_TABLE)
        ))?
        .exists([])?;
    if has_visible_bit {
        let mut stmt = conn.prepare(&format!(
            "SELECT \"table_name\", \"column_name\" FROM {} WHERE \"visible\" = 0",
            quote_ident(RINDLE_COLUMNS_TABLE)
        ))?;
        let mut rows = stmt.query([])?;
        while let Some(row) = rows.next()? {
            hidden.columns.insert((row.get(0)?, row.get(1)?));
        }
    }
    Ok(hidden)
}

fn quote_str(s: &str) -> String {
    format!("'{}'", s.replace('\'', "''"))
}

impl ApplyConsumer {
    /// Open a headless apply consumer over `path`. The caller owns the file's lifecycle.
    /// A fresh file gets plain `wal` (design 306 D5) and writes open `BEGIN IMMEDIATE` —
    /// exactly what `ClusterConsumer::open`'s defaults resolve to, so the headless
    /// applier and the default live follower produce byte-identical stores.
    ///
    /// Foreign keys are **not enforced** ([`ForeignKeys::Unenforced`]), which is the apply
    /// plane's posture and not a relaxation: the rows this consumer writes were validated
    /// by the authority that produced the stream, they arrive in the stream's order rather
    /// than a topological one, and the authority's `ON DELETE`/`ON UPDATE` actions are
    /// already in the stream as ordinary row changes — re-running them here would apply
    /// each cascade twice. [`rindle_writeplane::foreign_keys`] has the full argument, and
    /// [`foreign_key_audit`](Self::foreign_key_audit) is how a host proves referential
    /// integrity anyway. A host that really is the origin of its rows says so with
    /// [`open_with`](Self::open_with).
    pub fn open(path: &Path) -> Result<ApplyConsumer, ReplicaError> {
        Self::open_with_journal(path, JournalMode::default())
    }

    /// [`open`](Self::open) with an explicit fresh-file [`JournalMode`] — a wal2-fleet
    /// host passes [`JournalMode::Wal2`] (an existing wal/wal2 file keeps its mode).
    pub fn open_with_journal(
        path: &Path,
        journal: JournalMode,
    ) -> Result<ApplyConsumer, ReplicaError> {
        Self::open_with(path, journal, ForeignKeys::Unenforced)
    }

    /// [`open_with_journal`](Self::open_with_journal) with an explicit [`ForeignKeys`]
    /// posture — the escape hatch for a host that is the ORIGIN of the rows it writes
    /// through this consumer rather than a replayer of somebody else's. Everything else
    /// resolves exactly as [`open`](Self::open).
    pub fn open_with(
        path: &Path,
        journal: JournalMode,
        foreign_keys: ForeignKeys,
    ) -> Result<ApplyConsumer, ReplicaError> {
        let store = ApplyStore::open(path, journal, DEFAULT_WRITER_BEGIN_SQL, foreign_keys)?;
        Ok(Self::from_parts(Rc::new(store), Rc::new(NoFanout)))
    }

    /// Open a headless consumer over a **scratch** database — single-connection,
    /// `journal_mode=memory`, `synchronous=OFF`. See
    /// [`ApplyStore::open_scratch`](crate::ApplyStore::open_scratch) for what that trades
    /// and what the caller owes in return; the applied result is identical, only the crash
    /// posture differs.
    ///
    /// A scratch is by definition DERIVED from an authority that can rebuild it, so it is
    /// an apply-plane store and opens [`ForeignKeys::Unenforced`] like [`open`](Self::open).
    pub fn open_scratch(path: &Path) -> Result<ApplyConsumer, ReplicaError> {
        let store =
            ApplyStore::open_scratch(path, DEFAULT_WRITER_BEGIN_SQL, ForeignKeys::Unenforced)?;
        Ok(Self::from_parts(Rc::new(store), Rc::new(NoFanout)))
    }

    /// Walk every declared foreign key and report violating rows — the opt-in audit that
    /// stands in for enforcement on this store. Delegates to
    /// [`ApplyStore::foreign_key_audit`](crate::ApplyStore::foreign_key_audit); see there
    /// for the cost and the cap.
    pub fn foreign_key_audit(
        &self,
        max_rows: usize,
    ) -> Result<rindle_writeplane::ForeignKeyAudit, ReplicaError> {
        self.store.foreign_key_audit(max_rows)
    }

    /// Compose a consumer over an existing store + fan-out — the seam a derivation host
    /// (`rindle_replica::ClusterConsumer`) uses to share ONE store between its cluster
    /// and its apply surface.
    pub fn from_parts(store: Rc<ApplyStore>, fanout: Rc<dyn CommitFanout>) -> ApplyConsumer {
        let hidden = store.read(read_hidden).unwrap_or_default();
        ApplyConsumer {
            store,
            fanout,
            tables: Mutex::new(HashMap::new()),
            hidden: Mutex::new(hidden),
        }
    }

    /// Re-read the visibility sidecars (after a `ddl` entry, design 406 §9 step 6). `true` when
    /// the hidden set changed — the caller bounces so the advertised schema moves with it.
    pub fn refresh_visibility(&self) -> Result<bool, ReplicaError> {
        let fresh = self.store.read(read_hidden)?;
        let mut hidden = self.hidden.lock().unwrap();
        let changed = *hidden != fresh;
        *hidden = fresh;
        Ok(changed)
    }

    /// The current hidden set (406 §10).
    pub fn hidden(&self) -> Hidden {
        self.hidden.lock().unwrap().clone()
    }

    /// The underlying store (controlled writes, ad-hoc reads, DDL, checkpoints).
    pub fn store(&self) -> &ApplyStore {
        &self.store
    }

    /// The registered-table metadata registry — one entry per table this consumer can
    /// apply into, exactly what [`apply_muts`](Self::apply_muts) builds its SQL from. A
    /// derivation host reads and extends it (its lifecycle tables carry meta too); most
    /// callers never touch it directly.
    pub fn tables(&self) -> MutexGuard<'_, HashMap<String, TableMeta>> {
        self.tables.lock().unwrap()
    }

    /// Open one write transaction on the store (the follower's streaming apply drives
    /// [`apply_muts`](Self::apply_muts) into it per chunk frame, then
    /// [`commit_follower_txn_with_head`](Self::commit_follower_txn_with_head)).
    pub fn begin(&self) -> Result<ApplyTxn, ReplicaError> {
        ApplyTxn::begin(self.store.clone(), self.fanout.clone())
    }

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

    /// [`register_table`](Self::register_table) with the registration step swapped out —
    /// the derivation-host seam: `ClusterConsumer` passes its engine-inclusive
    /// `Cluster::register_table` so the capture half AND the worker sources build in the
    /// one place they always did, while the DDL and the meta bookkeeping stay here.
    pub fn register_table_via(
        &self,
        table: &str,
        columns: &[String],
        pk: &[usize],
        col_types: &[ColType],
        register: impl FnOnce() -> Result<(), ReplicaError>,
    ) -> Result<(), ReplicaError> {
        if self.tables.lock().unwrap().contains_key(table) {
            return Ok(());
        }
        let ddl = create_table_ddl(table, columns, pk, col_types);
        self.store.exec_ddl(&ddl)?;
        register()?;
        self.tables.lock().unwrap().insert(
            table.to_string(),
            TableMeta {
                columns: columns.to_vec(),
                pk: pk.to_vec(),
                col_types: col_types.to_vec(),
                // Typed registration leaves ordinary fields nullable, while PK fields are emitted
                // `NOT NULL` because every replicated row must have stable identity.
                nullable: (0..columns.len()).map(|i| !pk.contains(&i)).collect(),
            },
        );
        Ok(())
    }

    /// 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 (a derivation host adds
    /// its sources through the `_via` seam).
    pub fn register_existing_table(&self, table: &str) -> Result<(), ReplicaError> {
        self.register_existing_table_via(table, || {
            self.store.register_table_capture(table).map(|_| ())
        })
    }

    /// [`register_existing_table`](Self::register_existing_table) with the registration
    /// step swapped out (see [`register_table_via`](Self::register_table_via)).
    pub fn register_existing_table_via(
        &self,
        table: &str,
        register: impl FnOnce() -> Result<(), ReplicaError>,
    ) -> Result<(), ReplicaError> {
        if self.tables.lock().unwrap().contains_key(table) {
            return Ok(());
        }
        let meta = self.store.read(|conn| {
            let mut stmt = conn.prepare(&format!("PRAGMA table_info({})", quote_ident(table)))?;
            let mut columns = Vec::new();
            let mut col_types = Vec::new();
            let mut nullable = Vec::new();
            let mut pk_pos: Vec<(i64, usize)> = Vec::new();
            let mut rows = stmt.query([])?;
            let mut defaults: Vec<Option<String>> = Vec::new();
            while let Some(row) = rows.next()? {
                let name: String = row.get(1)?;
                // PRAGMA table_info column 2 is the declared type; ColType::from_sqlite_decl
                // recovers BOOLEAN/JSON from the declared name, else falls back to affinity.
                let decl: String = row.get(2)?;
                // Column 3 is `notnull` (0 ⇒ nullable) — carried for `/schema` codegen (design 206);
                // the same bit table introspection reads into the engine's `ColumnDef.optional`.
                let notnull: i64 = row.get(3)?;
                // Column 4 is `dflt_value` — read so exact-i64 columns get default
                // admission (below); other columns keep this surface's permissiveness.
                defaults.push(row.get(4)?);
                let pk_rank: i64 = row.get(5)?;
                if pk_rank > 0 {
                    pk_pos.push((pk_rank, columns.len()));
                }
                columns.push(name);
                col_types.push(ColType::from_sqlite_decl(&decl));
                nullable.push(notnull == 0);
            }
            pk_pos.sort();

            // PG-mirror topology: if the source minted a `_rindle_columns` sidecar, it carries the semantic
            // type + nullability the affinity-only mirror DDL drops (pg-source Amendment 2026-06-30).
            // Override the PRAGMA-derived facts per column so `/schema` codegen emits `boolean()`/`json()`
            // and `.nullable()` faithfully instead of collapsing to `number`/`string` and the mirror's
            // permissive all-nullable. Absent (the direct-SQLite topology) ⇒ the PRAGMA facts stand — the
            // app's own migration DDL already carries the real types + `NOT NULL` (design 206).
            let has_sidecar = conn
                .prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1")?
                .exists([RINDLE_COLUMNS_TABLE])?;
            if has_sidecar {
                let mut by_name: HashMap<String, (ColType, bool)> = HashMap::new();
                let mut sc_stmt = conn.prepare(&format!(
                    "SELECT \"column_name\", \"rindle_type\", \"pg_type_tag\", \"not_null\" \
                     FROM {} WHERE \"table_name\" = ?1",
                    quote_ident(RINDLE_COLUMNS_TABLE)
                ))?;
                let mut sc_rows = sc_stmt.query([table])?;
                while let Some(row) = sc_rows.next()? {
                    let cname: String = row.get(0)?;
                    let rtype: String = row.get(1)?;
                    let tag: String = row.get(2)?;
                    let not_null: i64 = row.get(3)?;
                    by_name.insert(cname, (coltype_from_sidecar(&rtype, &tag), not_null == 0));
                }
                for (i, name) in columns.iter().enumerate() {
                    if let Some(&(ty, is_nullable)) = by_name.get(name) {
                        col_types[i] = ty;
                        nullable[i] = is_nullable;
                    }
                }
            }

            // Exact-i64 columns whose DDL carries a default (post-sidecar types): these
            // get the same default admission the master's envelope runs, below.
            let int_defaults: Vec<(String, String)> = columns
                .iter()
                .zip(&col_types)
                .zip(&defaults)
                .filter(|((_, ty), _)| **ty == ColType::Int)
                .filter_map(|((name, _), d)| d.clone().map(|d| (name.clone(), d)))
                .collect();

            Ok((
                TableMeta {
                    columns,
                    pk: pk_pos.into_iter().map(|(_, i)| i).collect(),
                    col_types,
                    nullable,
                },
                int_defaults,
            ))
        })?;
        let (meta, int_defaults) = meta;
        if meta.columns.is_empty() {
            return Err(ReplicaError::Schema(format!("unknown table: {table}")));
        }
        // Design 226 §4.1 on the embedded/napi surface too: an exact-i64 column's
        // default must evaluate INTEGER-or-NULL and be time-invariant. Without this,
        // an off-plane default (`BIGINT DEFAULT 2.5`) registers fine and every later
        // omitted-column insert fails at capture with a storage-class message that
        // never names the default — refuse at admission, like the master's envelope.
        if !int_defaults.is_empty() {
            self.store.read(|conn| {
                Ok(int_defaults.iter().try_for_each(|(column, default_sql)| {
                    validate_default_expression(
                        conn,
                        table,
                        column,
                        default_sql,
                        rindle_value::value::ValueType::Int,
                    )
                }))
            })??;
        }
        register()?;
        self.tables.lock().unwrap().insert(table.to_string(), meta);
        Ok(())
    }

    /// One-time setup for client mutations on a headless applier: the
    /// `_rindle_client_mutations` (and room-scoped) ledger tables are created and the
    /// former registered for capture — its rows are **replicated data** on the master's
    /// stream, so an applier that skips this stalls on the first lmid row with
    /// `unknown table`. The meta joins the table map like any table's.
    pub fn enable_client_mutations(&self) -> Result<(), ReplicaError> {
        self.enable_client_mutations_via(|| {
            self.store.create_client_mutations_tables()?;
            self.store
                .register_table_capture(CLIENT_MUTATIONS_TABLE)
                .map(|_| ())
        })
    }

    /// [`enable_client_mutations`](Self::enable_client_mutations) with the DDL +
    /// registration step swapped out — `ClusterConsumer` passes
    /// `Cluster::enable_client_mutations` (the same DDL plus engine hosting) so the meta
    /// bookkeeping stays here either way.
    pub fn enable_client_mutations_via(
        &self,
        enable: impl FnOnce() -> Result<(), ReplicaError>,
    ) -> Result<(), ReplicaError> {
        enable()?;
        self.tables.lock().unwrap().insert(
            CLIENT_MUTATIONS_TABLE.to_string(),
            client_mutations_table_meta(),
        );
        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<rindle_writeplane::BaseTableSchema> {
        let tables = self.tables.lock().unwrap();
        let hidden = self.hidden.lock().unwrap();
        let mut out: Vec<rindle_writeplane::BaseTableSchema> = tables
            .iter()
            .filter(|(name, _)| {
                !name.starts_with("_rindle_")
                    && !name.starts_with("__")
                    && !name.starts_with("sqlite_")
                    && !hidden.tables.contains(*name)
            })
            .map(|(name, meta)| {
                let pk_set: std::collections::BTreeSet<usize> = meta.pk.iter().copied().collect();
                let columns = meta
                    .columns
                    .iter()
                    .zip(&meta.col_types)
                    .zip(&meta.nullable)
                    .enumerate()
                    // A column mid-backfill is not advertised (406 §10); the mirror still
                    // holds it positionally.
                    .filter(|(_, ((column, _), _))| {
                        !hidden
                            .columns
                            .contains(&((*name).clone(), (*column).clone()))
                    })
                    .map(
                        |(i, ((name, &ty), &nullable))| rindle_writeplane::BaseColumn {
                            name: name.clone(),
                            ty,
                            // A PK column is row identity — never null, whatever the DDL says (SQLite
                            // doesn't enforce NOT NULL on rowid-table PKs), so its type stays `T`.
                            nullable: nullable && !pk_set.contains(&i),
                        },
                    )
                    .collect();
                let primary_key = meta.pk.iter().map(|&i| meta.columns[i].clone()).collect();
                rindle_writeplane::BaseTableSchema {
                    name: (*name).clone(),
                    columns,
                    primary_key,
                }
            })
            .collect();
        out.sort_by(|a, b| a.name.cmp(&b.name));
        out
    }

    /// Apply positional mutations to an open write transaction (build_sql per row).
    pub fn apply_muts(&self, txn: &mut ApplyTxn, muts: &[Mutation]) -> Result<(), ReplicaError> {
        let tables = self.tables.lock().unwrap();
        for m in muts {
            let table = m.table();
            let meta = tables
                .get(table)
                .ok_or_else(|| ReplicaError::Schema(format!("unknown table: {table}")))?;
            let (sql, params) = build_mutation_sql(&meta.columns, &meta.pk, m)?;
            txn.exec(&sql, &params)?;
        }
        Ok(())
    }

    /// 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> {
        let mut txn = self.begin()?;
        self.apply_muts(&mut txn, muts)?;
        let info = txn.commit_with_info()?;
        Ok(info.tx_id.0)
    }

    /// 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> {
        let mut txn = self.begin()?;
        self.apply_muts(&mut txn, muts)?;
        upsert_source_offset(&mut txn, source, offset, chunk_seq, run_id)?;
        let info = txn.commit_with_info()?;
        Ok(info.tx_id.0)
    }

    /// Terminal step of the **streaming-follower apply** (`REPLICATOR-PRECOMMIT-STREAMING-DESIGN.md`
    /// §7): the caller has opened ONE [`ApplyTxn`](super::ApplyTxn) via [`begin`](Self::begin) 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,
        mut txn: ApplyTxn,
        source: &str,
        offset: &str,
        chunk_seq: i64,
        run_id: Option<&str>,
    ) -> Result<u64, ReplicaError> {
        upsert_source_offset(&mut txn, source, offset, chunk_seq, run_id)?;
        let info = txn.commit_with_info()?;
        Ok(info.tx_id.0)
    }

    /// [`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,
        mut txn: ApplyTxn,
        source: &str,
        offset: &str,
        chunk_seq: i64,
        run_id: Option<&str>,
        head: SourceHead,
    ) -> Result<u64, ReplicaError> {
        upsert_source_offset_full(
            &mut txn,
            source,
            offset,
            chunk_seq,
            run_id,
            None,
            Some(head),
        )?;
        let info = txn.commit_with_info()?;
        Ok(info.tx_id.0)
    }

    /// 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> {
        let mut txn = self.begin()?;
        let changed = txn.exec(
            &format!(
                "UPDATE {SOURCE_OFFSETS_TABLE} SET run_id = ?1 \
                 WHERE source = ?2 AND offset = ?3 AND chunk_seq = ?4 AND run_id = ?5"
            ),
            &[
                OwnedValue::str(next_run_id),
                OwnedValue::str(source),
                OwnedValue::str(offset),
                OwnedValue::Int(SOURCE_OFFSET_WHOLE_RUN),
                OwnedValue::str(previous_run_id),
            ],
        )?;
        if changed != 1 {
            return Err(ReplicaError::Mutation(format!(
                "CDC locator refresh lost its checkpoint compare-and-swap for source {source:?}"
            )));
        }
        let info = txn.commit_with_info()?;
        Ok(info.tx_id.0)
    }

    /// 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.store.exec_ddl(&source_offsets_table_ddl())?;
        // Idempotent forward-migration for a pre-`chunk_seq` table (RELAY-DDL §6.6): add the
        // column with the whole-run default so every cursor written before this slice reads back
        // as a run-boundary checkpoint (which it was). `ADD COLUMN` is a metadata-only change (the
        // PK is unchanged), so no table rebuild is needed — unlike the relay's `_rindle_change_log`.
        let cols = self.store.read(|conn| {
            conn.prepare(&format!("PRAGMA table_info({SOURCE_OFFSETS_TABLE})"))?
                .query_map([], |r| r.get::<_, String>(1))?
                .collect::<rusqlite::Result<Vec<String>>>()
        })?;
        if !cols.iter().any(|c| c == "chunk_seq") {
            self.store.exec_ddl(&format!(
                "ALTER TABLE {SOURCE_OFFSETS_TABLE} \
                 ADD COLUMN chunk_seq INTEGER NOT NULL DEFAULT {SOURCE_OFFSET_WHOLE_RUN}"
            ))?;
        }
        // Same idempotent forward-migration for the pre-fence table: the run identity token
        // (RELAY-CURSOR-EPOCH-FENCING-DESIGN.md §2), nullable — a NULL cursor row subscribes
        // without a fence, exactly the pre-upgrade behavior.
        if !cols.iter().any(|c| c == "run_id") {
            self.store.exec_ddl(&format!(
                "ALTER TABLE {SOURCE_OFFSETS_TABLE} ADD COLUMN run_id TEXT"
            ))?;
        }
        // The §8.3 batch-identity column (nullable — only hash-bearing sources,
        // i.e. room flushes, populate it). Same metadata-only forward-migration.
        if !cols.iter().any(|c| c == "batch_hash") {
            self.store.exec_ddl(&format!(
                "ALTER TABLE {SOURCE_OFFSETS_TABLE} ADD COLUMN batch_hash TEXT"
            ))?;
        }
        for column in ["run_rows", "rows_cum", "committed_at"] {
            if !cols.iter().any(|existing| existing == column) {
                self.store.exec_ddl(&format!(
                    "ALTER TABLE {SOURCE_OFFSETS_TABLE} ADD COLUMN {column} INTEGER"
                ))?;
            }
        }
        Ok(())
    }

    /// 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.store.read(|conn| {
            conn.query_row(
                &format!(
                    "SELECT offset, chunk_seq, run_id FROM {SOURCE_OFFSETS_TABLE} \
                     WHERE source = ?1"
                ),
                [source],
                |r| {
                    Ok((
                        r.get::<_, String>(0)?,
                        r.get::<_, i64>(1)?,
                        r.get::<_, Option<String>>(2)?,
                    ))
                },
            )
            .optional()
        })
    }

    /// Persisted accounting carried beside the source checkpoint.
    pub fn source_head(&self, source: &str) -> Result<Option<SourceHead>, ReplicaError> {
        let stored = self.store.read(|conn| {
            conn.query_row(
                &format!(
                    "SELECT run_rows, rows_cum, committed_at \
                     FROM {SOURCE_OFFSETS_TABLE} WHERE source = ?1"
                ),
                [source],
                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
            )
            .optional()
        })?;
        let Some((run_rows, rows_cum, committed_at)) = stored else {
            return Ok(None);
        };
        let counter = |name: &str, value: Option<i64>| -> Result<Option<u64>, ReplicaError> {
            value
                .map(|value| {
                    u64::try_from(value).map_err(|_| {
                        ReplicaError::Mutation(format!("stored source {name} is negative: {value}"))
                    })
                })
                .transpose()
        };
        Ok(Some(SourceHead {
            run_rows: counter("run_rows", run_rows)?,
            rows_cum: counter("rows_cum", rows_cum)?,
            committed_at,
        }))
    }

    /// 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.store.read(|conn| {
            conn.query_row(
                &format!("SELECT batch_hash FROM {SOURCE_OFFSETS_TABLE} WHERE source = ?1"),
                [source],
                |r| r.get::<_, Option<String>>(0),
            )
            .optional()
            .map(Option::flatten)
        })
    }

    /// 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> {
        Ok(self.source_checkpoint(source)?.map(|(offset, _, _)| offset))
    }

    /// 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.store.exec_ddl(&writeplane::producer_offsets_ddl())
    }

    /// 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.store.exec_ddl(&format!(
            "CREATE TABLE IF NOT EXISTS {APPLIED_DDL_TABLE} \
                 (key TEXT PRIMARY KEY, applied_at INTEGER NOT NULL DEFAULT (unixepoch()), \
                  actions TEXT)"
        ))?;
        // Additive upgrade for pre-report files: markers written before the column existed keep
        // NULL actions and degrade to the caller's end-state fallback on replay.
        let has_actions = self.store.read(|conn| {
            let mut stmt = conn.prepare(&format!(
                "SELECT name FROM pragma_table_info('{APPLIED_DDL_TABLE}')"
            ))?;
            let names = stmt
                .query_map([], |row| row.get::<_, String>(0))?
                .collect::<rusqlite::Result<Vec<String>>>()?;
            Ok(names.iter().any(|name| name == "actions"))
        })?;
        if !has_actions {
            self.store.exec_ddl(&format!(
                "ALTER TABLE {APPLIED_DDL_TABLE} ADD COLUMN actions TEXT"
            ))?;
        }
        Ok(())
    }

    /// 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.store.read(|conn| {
            conn.query_row(
                &format!("SELECT 1 FROM {APPLIED_DDL_TABLE} WHERE key = ?1"),
                [key],
                |_| Ok(()),
            )
            .optional()
            .map(|row| row.is_some())
        })
    }

    /// Apply a `ddl` entry's `statements` and journal its `key` atomically — one ordinary
    /// transaction, since DDL can't ride the `BEGIN CONCURRENT` cursor advance. Delegates to
    /// [`ApplyStore::exec_ddl_with_marker`] against the [`APPLIED_DDL_TABLE`] journal.
    pub fn apply_ddl_with_marker(
        &self,
        key: &str,
        statements: &[String],
    ) -> Result<rindle_writeplane::DdlApplyReport, ReplicaError> {
        self.store
            .exec_ddl_with_marker(APPLIED_DDL_TABLE, key, statements)
    }

    /// Apply DDL, caller-owned per-statement bookkeeping effects, and the durable marker in one
    /// transaction. See [`ApplyStore::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<rindle_writeplane::DdlApplyReport, ReplicaError>
    where
        F: FnMut(&rusqlite::Connection, &rindle_writeplane::DdlStep) -> rusqlite::Result<()>,
    {
        self.store.exec_ddl_with_marker_and_step_effects(
            APPLIED_DDL_TABLE,
            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<rindle_writeplane::DdlApplyReport>, ReplicaError> {
        let json: Option<Option<String>> = self.store.read(|conn| {
            conn.query_row(
                &format!("SELECT actions FROM {APPLIED_DDL_TABLE} WHERE key = ?1"),
                [key],
                |row| row.get::<_, Option<String>>(0),
            )
            .optional()
        })?;
        Ok(json
            .flatten()
            .and_then(|json| serde_json::from_str(&json).ok()))
    }

    /// 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<
        (
            writeplane::PublicOperationCommit,
            rindle_writeplane::DdlApplyReport,
        ),
        DdlMigrationError,
    >
    where
        F: FnMut(&rusqlite::Connection, &rindle_writeplane::DdlStep) -> rusqlite::Result<()>,
    {
        let (results, report, tx_id) = self.store.exec_public_ddl_unit(
            std::slice::from_ref(statement),
            declared_tables,
            "DDL operations may not create or mutate table rows",
            apply_step_effects,
            |conn, tx_id, results| {
                let wire = results
                    .iter()
                    .map(StatementResult::to_wire_json)
                    .collect::<Result<Vec<_>, _>>()?;
                let stored = serde_json::to_string(&wire)
                    .map_err(|error| writeplane::BookkeepingError::Wire(error.to_string()))?;
                if stored.len() > result_byte_limit {
                    return Err(writeplane::sql_result_limit_error());
                }
                let cursor = format!("w:{:016x}", tx_id.0);
                writeplane::insert_sql_outcome_with_cursor(
                    conn,
                    outcome_key,
                    &stored,
                    None,
                    Some(request_identity),
                    Some(&cursor),
                    now_ms,
                )
            },
        )?;
        let wire = results
            .iter()
            .map(StatementResult::to_wire_json)
            .collect::<Result<Vec<_>, _>>()
            .map_err(writeplane::BookkeepingError::from)?;
        Ok((
            writeplane::PublicOperationCommit {
                results: wire,
                cursor: Some(format!("w:{:016x}", tx_id.0)),
            },
            report,
        ))
    }

    /// 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, rindle_writeplane::DdlApplyReport), DdlMigrationError>
    where
        F: FnMut(&rusqlite::Connection, &rindle_writeplane::DdlStep) -> rusqlite::Result<()>,
    {
        let requests = normalized
            .iter()
            .map(|sql| SqlStatementRequest {
                sql: sql.clone(),
                args: SqlArgs::default(),
                want_rows: false,
            })
            .collect::<Vec<_>>();
        let (_results, report, tx_id) = self.store.exec_public_ddl_unit(
            &requests,
            declared_tables,
            "migrations may not create or mutate table rows",
            apply_step_effects,
            |conn, tx_id, _| {
                let cursor = format!("w:{:016x}", tx_id.0);
                writeplane::journal_migration_with_cursor(
                    conn,
                    id,
                    supplied_checksum,
                    content_checksum,
                    None,
                    identity_json,
                    Some(&cursor),
                    now_ms,
                )
            },
        )?;
        Ok((format!("w:{:016x}", tx_id.0), report))
    }
}

#[cfg(test)]
mod tests {
    use super::{coltype_from_sidecar, ColType};

    #[test]
    fn coltype_from_sidecar_maps_rindle_type_then_json_by_tag() {
        // boolean/integer/float come straight off `rindle_type`; the tag is only consulted to split JSON
        // out of the text-stored `rindle_type = "string"` (the one kind `rindle_type` collapses).
        assert_eq!(
            coltype_from_sidecar("boolean", "RINDLE_PG_BOOL"),
            ColType::Boolean
        );
        // `integer` is the exact-i64 plane (design 226 / 406 §11.1); `float`/legacy `number` are f64.
        assert_eq!(
            coltype_from_sidecar("integer", "RINDLE_PG_INT4"),
            ColType::Int
        );
        assert_eq!(
            coltype_from_sidecar("number", "RINDLE_PG_INT4"),
            ColType::Number
        );
        assert_eq!(
            coltype_from_sidecar("json", "RINDLE_PG_JSONB"),
            ColType::Json
        );
        assert_eq!(
            coltype_from_sidecar("float", "RINDLE_PG_FLOAT8"),
            ColType::Number
        );
        assert_eq!(
            coltype_from_sidecar("string", "RINDLE_PG_JSON"),
            ColType::Json
        );
        assert_eq!(
            coltype_from_sidecar("string", "RINDLE_PG_JSONB"),
            ColType::Json
        );
        // Every other text-stored family surfaces as a string (uuid, numeric, temporal, bytea-hex, text).
        assert_eq!(
            coltype_from_sidecar("string", "RINDLE_PG_TEXT"),
            ColType::String
        );
        assert_eq!(
            coltype_from_sidecar("string", "RINDLE_PG_UUID"),
            ColType::String
        );
        assert_eq!(
            coltype_from_sidecar("string", "RINDLE_PG_NUMERIC"),
            ColType::String
        );
        // Unknown tokens fall through to the permissive string default.
        assert_eq!(
            coltype_from_sidecar("mystery", "RINDLE_PG_FUTURE"),
            ColType::String
        );
    }
}