rindle_replica/

lib.rs

//! # Embedded SQLite with incremental live queries
//!
//! [`Db`] provides one controlled SQL writer, ordinary SQL reads, and live queries
//! over file-backed SQLite. It captures row changes automatically and derives query
//! deltas through the `rindle` engine. Use it when an application owns its database
//! and needs live results without implementing change capture or transaction ordering.
//!
//! [`Cluster`] distributes queries across worker threads. Its channel exposes
//! provisional deltas and commit progress; [`ClusterConsumer`] owns the drain loop
//! and adds normalized subscriptions. The raw `rindle` crate is a lower-level choice
//! for applications that already own source storage and change capture.
//!
//! ## Transactions and delivery
//!
//! Every row write must pass through the controlled writer. The preupdate hook
//! captures changes, and the engine derives against a read-only pre-commit snapshot
//! plus an in-memory batch overlay. Derivation does not write SQL or replay triggers.
//!
//! [`Db`] derives before commit, commits the durable data, then calls subscribers
//! synchronously. [`Cluster`] workers can emit `Changed` slices before commit; callers
//! must stage them until the hosting worker emits [`ClusterEvent::Progressed`].
//! See [`Update`] and [`ClusterEvent`] for recovery and delivery boundaries.
//!
//! Query handles expose changes rather than maintaining an application view. Apply
//! changes in order, or use a higher-level consumer. Subscribe immediately after
//! registration and call [`Query::destroy`] when finished; dropping a handle does not
//! unregister it. `Db` and the `Cluster` coordinator are `!Send` and stay on one thread.
//!
//! ## Storage and schema
//!
//! Fresh embedded databases use ordinary WAL. Existing WAL/WAL2 files retain their
//! mode; [`OpenOptions`] selects a journal explicitly. Foreign keys are enforced by
//! default. Register every user table that a transaction can write, including cascade
//! targets. Captured foreign-key cascades become ordinary row changes.
//!
//! Triggers remain unsupported on registered tables, and generated columns are
//! rejected. Use the schema DDL methods before registration. To change a registered
//! table's shape, close the runtime, migrate the database, then reopen and register it.
//!
//! Number-domain INTEGER cells must round-trip exactly through `f64`; admission and
//! capture reject values that would round. The storage layer can preserve exact
//! `BIGINT`/`INT64` cells, but maintained queries reject a required exact-integer
//! column. See the [schema guide](https://rindle.sh/docs/schema) for surface limits.
//!
//! ```no_run
//! use rindle::table;
//! use rindle_replica::{Db, QueryId};
//!
//! # fn main() -> Result<(), rindle_replica::ReplicaError> {
//! let db = Db::open("app.db")?;
//! db.exec_ddl("CREATE TABLE IF NOT EXISTS issue (id INTEGER PRIMARY KEY, title TEXT NOT NULL)")?;
//! db.register_table("issue")?;
//! let query = db.query(QueryId(1), table("issue").build())?;
//! query.subscribe(|update| println!("{update:?}"));
//!
//! let mut tx = db.write()?;
//! tx.exec("INSERT INTO issue (title) VALUES ('First live query')", &[])?;
//! tx.commit()?; // subscribers run after the SQL commit
//! let count: i64 = db.read(|conn| conn.query_row("SELECT count(*) FROM issue", [], |row| row.get(0)))?;
//! assert!(count >= 1);
//! query.destroy();
//! # Ok(()) }
//! ```

use std::cell::Cell;
use std::cell::RefCell;
use std::collections::HashMap;
use std::path::Path;
use std::rc::Rc;

use rindle::Ast;
use rusqlite::{Connection, OptionalExtension};

mod analyze;
mod cluster;
mod consumer;
mod derivation_pool;
mod drain;
mod engine;
mod initial_snapshot;
mod maintenance;
mod mutations;
mod parallel;
mod progress;
mod query;
mod schema;
// Local, not re-exported: it adds the engine-coupled assembled-snapshot rendering to the write
// plane's codec. See the module doc.
pub mod wire_json;
mod writer;

// The normalized fold + protocol envelope moved to the wasm-clean `rindle-wire`
// (RINDLE-REALTIME-DESIGN.md §2.4) so the wasm room shares them. These module
// re-imports keep every internal `crate::normalize{,_protocol}::` path working, and the
// `pub use` blocks below keep the public surface byte-identical.
pub(crate) use rindle_wire::{normalize, normalize_protocol};

// The engine-free CDC apply plane moved to its own leaf crate (design 309): the store
// half of the cluster (connections + capture + the write-txn state machine + DDL), the
// CommitFanout seam the worker pool implements, and the headless ApplyConsumer that
// `rindle-backup-sqlite` (and any external stream consumer) drives without booting
// engines. This alias keeps every internal `crate::apply::` path working; the moved
// public items re-export at their old paths below and in `cluster`/`consumer`/
// `writer`/`maintenance`/`parallel`.
pub(crate) use rindle_cdc_apply as apply;

pub use analyze::{
    AnalyzeJoin, AnalyzePlan, AnalyzeReport, AnalyzeSource, AnalyzeTiming, AnalyzeTotals,
};
pub use cluster::{Cluster, ClusterWriteTxn, DdlMigrationError};
pub use consumer::{
    ClusterConsumer, ClusterMutationWrite, RoomFlushOutcome, SharedQueryInfo, SourceHead,
    APPLIED_DDL_TABLE,
};
// The base-table shape vocabulary moved to `table_shape` (write-plane extraction step 1); these
// keep `rindle_replica::{ColType, Mutation, create_table_ddl, ...}` byte-identical for every caller.
pub use drain::{ConnId, Drain, DrainHandle, DrainSink, SubId};
pub use rindle_writeplane::ReplicaError;
// The write plane's own modules, re-exported at their old paths: `rindle_replica::session`,
// `::sql` and `::writeplane` are what the daemon, the master and the wire suites import, and they
// must not move. `::wire_json` is a local module that re-exports the write plane's codec whole and
// adds the engine-coupled assembled-snapshot rendering on top, so that import path is unchanged
// too. `table_shape` was private here and stays crate-internal.
pub use initial_snapshot::{
    InitialSnapshotColumn, InitialSnapshotCommit, InitialSnapshotIdentity, InitialSnapshotOpen,
    InitialSnapshotStore, InitialSnapshotTable, StoredSourceCheckpoint, CDC_BOOTSTRAP_TABLE,
};
pub use maintenance::{MaintenanceOptions, MaintenanceReport, WalCheckpoint};
pub use mutations::{
    MutationEnvelope, MutationOutcome, MutationReject, MutationSql, MutatorRegistry, ServerMutator,
};
pub use progress::ProgressTracker;
pub use query::Query;
/// A parameterized query family's identity and template (design 310 §3), re-exported from
/// `rindle-wire` so a daemon groups subscriptions with the same types the engine binds on.
pub use rindle_wire::family_key::{
    extract_family, Binding, FamilyExtraction, FamilyKey, FamilyTemplate,
};
pub use rindle_wire::normalize::{
    agg_table_name, agg_table_schemas, rewrite_aggregates, rewrite_aggregates_with_local,
    table_tree, AggTable, NormalizeFold, NormalizedOp, PkMap, TableNode,
};
pub use rindle_wire::normalize_protocol::{
    build_query_parts, hello_from_parts, normalized_fp, unsafe_int_in_normalized_batch,
    NormalizedApplied, NormalizedBatch, NormalizedHello, NormalizedProtocolError,
    NormalizedPublisher, NormalizedSubscriber, ProgressFrame, ProjMap, TableWireSchema,
};
pub(crate) use rindle_writeplane::table_shape;
pub use rindle_writeplane::{
    build_mutation_sql, create_table_ddl, is_internal_schema_name, BaseColumn, BaseTableSchema,
    ColType, Mutation, ReplicatedColumn, ReplicatedTableSchema, TableMeta,
};
pub use rindle_writeplane::{
    canonical_migration_checksum, classify_statement, decode_wire_value, encode_wire_value,
    install_public_authorizer, normalize_migration_statements, run_ddl_statement, run_statement,
    split_migration_file, split_sql_script, split_sql_script_allow_empty, sql_cell_to_owned,
    statement_is_insert, PublicAuthorizerGuard, SqlArgs, SqlColumn, SqlStatementRequest,
    StatementClass, StatementResult, StatementRunError, SQL_BIND_LIMIT, SQL_BIND_VALUE_BYTE_LIMIT,
    SQL_RESULT_BYTE_LIMIT, SQL_RESULT_ROW_LIMIT, SQL_TEXT_LIMIT,
};
pub use rindle_writeplane::{
    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_OFFSETS_TABLE, SOURCE_OFFSET_WHOLE_RUN,
};
pub use rindle_writeplane::{
    introspect_replicated_table, introspect_schema_envelope, validate_schema_envelope,
    value_type_of, DdlActions, DdlApplyReport, DdlStep,
};
pub use rindle_writeplane::{schema_envelope, session, sql, writeplane};
pub use rindle_writeplane::{QueryId, TxId};
pub use rindle_writeplane::{
    ReadConn, SqlReadRows, READ_ROW_CAP, READ_TIME_BUDGET, READ_VM_BUDGET,
};
pub use writer::{CommitInfo, WriteTxn};

pub use engine::{
    analyze_standalone, AnalyzeSpec, JoinPrecheckReport, OperatorStorage, StorageReport,
};
// `JournalMode` + the restore path moved to `rindle_writeplane::restore` (step 6): both are
// `rusqlite` + `std::fs`, and keeping them here made every restore caller link the engine.
pub use rindle_writeplane::{restore_bootstrap, restore_bootstrap_with_journal, JournalMode};
// The foreign-key enforcement posture (and its audit) lives beside them for the same reason:
// the write master and the restore paths choose one without linking the engine.
pub use rindle_writeplane::{
    foreign_key_check, set_foreign_keys, ForeignKeyAudit, ForeignKeyViolation, ForeignKeys,
    FOREIGN_KEY_AUDIT_ROW_CAP,
};

// --- The embedded-engine seam (design 308) --------------------------------------
// `rindle-sqlite-ext` — the SQL-native subscribe surface — drives the engine from a
// commit hook it installs on a HOST-owned writer connection, so it needs the
// building blocks the `Db`/`Cluster` openers wire together internally: the
// single-thread engine, the multi-thread worker pool ([`DerivationPool`] — the
// `Cluster` handshake for a host that owns its own writer + capture), schema
// discovery (+ the PK unique index `TableSource` requires), and the journal-aware
// connection opener. Narrow, documented re-exports for embedders that own their
// write path — not a general API; prefer [`Db`]/[`Cluster`] unless you are one.
pub use derivation_pool::{DerivationPool, PoolGate, PoolTxn};
pub use engine::Engine;
pub use parallel::open_journal;
use query::{QueryEntry, RegId};
use rindle_cdc::CaptureCtx;
pub use schema::{discover as discover_table_schema, ensure_unique_pk_index, TableSchema};

/// An owned, fully-materialized change event off the pipeline — the raw delta the
/// engine emits (re-exported from the engine, where it is `CaughtChange`). Nested
/// relationships are carried in [`NodeData::relationships`].
pub use rindle::CaughtChange as ChangeEvent;
/// An owned, fully-materialized node: a row plus its relationship subtrees (keyed by
/// relationship slot). Re-exported from the engine (where it is `CaughtNode`).
pub use rindle::CaughtNode as NodeData;

/// A query baseline or incremental changes. A `Hydrated` contains the full result
/// as `Add` events and replaces any previous baseline. `Changed` contains deltas to
/// apply in order. Delivery depends on the runtime:
///
/// - [`Db`] calls subscribers with `Changed` after the SQL commit.
/// - [`Cluster`] can emit several provisional `Changed` slices for one transaction,
///   before commit. Stage them until the hosting worker's [`ClusterEvent::Progressed`].
///
/// `Hydrated` can fire AGAIN after a **delta-overflow shed** (design 306 D4): a
/// transaction whose folded rows outgrow the derivation memory budget still
/// commits, but it cannot be derived incrementally — the engine rebuilds and every
/// query re-hydrates from the committed state. A re-delivered `Hydrated` REPLACES the
/// subscriber's entire view state (drop what you hold, apply the `Add`s); it is the
/// single-thread [`Db`]'s analogue of the cluster's `Faulted` → re-register →
/// re-hydrate cycle, folded into the existing subscription.
#[derive(Clone, Debug)]
pub enum Update {
    Hydrated {
        tx_id: TxId,
        changes: Vec<ChangeEvent>,
    },
    Changed {
        tx_id: TxId,
        changes: Vec<ChangeEvent>,
    },
    /// One **partition** of a parameterized query family hydrated (design 310 §5.2 / impl
    /// plan D8): the initial `Add` set for `binding`, tagged with the committed watermark
    /// it reflects — emitted once per bound partition (at family registration, and again
    /// for every later bind) in place of the singleton's `Hydrated`. A family's later
    /// `Changed` events carry every partition's deltas together; the consumer demuxes
    /// them by the root row's partition key (the drain does this for the daemon).
    PartitionHydrated {
        tx_id: TxId,
        binding: Binding,
        changes: Vec<ChangeEvent>,
    },
}

/// Events from a [`Cluster`]'s bounded channel. Registration emits a committed
/// `Hydrated` baseline. Later `Changed` slices can arrive before the transaction
/// commits. Buffer them until `Progressed` from the worker that hosts the query.
/// For a combined result across workers, wait for every relevant worker.
///
/// `Faulted` is terminal. Discard provisional changes for that query and re-register
/// for a fresh baseline. Derivation failures, worker loss, and rollback after
/// speculative delivery can fault a query; no further updates follow for that registration.
///
/// `Faulted` lives here, not as an [`Update`] variant, because it is an out-of-band
/// lifecycle signal, not an incremental data change — and the single-thread [`Db`]
/// never produces it (its ordinary derive errors abort the transaction; a
/// delta-overflow shed re-delivers `Update::Hydrated` instead).
#[derive(Clone, Debug)]
pub enum ClusterEvent {
    /// A query's incremental update.
    Update { query_id: QueryId, update: Update },
    /// A query was destroyed after a derivation fault on its worker; `reason`
    /// describes the underlying error and `cause` classifies it — a push-deadline
    /// bail must survive to the host distinguishable from an ordinary derive fault
    /// (FOLLOWER-LAG-SHED §6.6: it is shed trigger (b)). No further events arrive
    /// for this query.
    Faulted {
        query_id: QueryId,
        reason: String,
        cause: FaultCause,
    },
    /// Transaction `tx_id` committed, and `worker` emitted all its `Changed` slices
    /// before this marker on the same ordered channel. Receiving this event permits
    /// release of that worker's buffered changes. It does not acknowledge application
    /// callbacks or downstream transport delivery.
    ///
    /// A combined subscription waits for every relevant worker before advancing
    /// `cv_min`. The marker also advances queries whose results did not change.
    Progressed { worker: usize, tx_id: TxId },
}

/// Why a query faulted (`ClusterEvent::Faulted`) — the classification the host's mode
/// machine consumes (FOLLOWER-LAG-SHED §6.6 item 3: today the error's identity died inside
/// the worker as a bare `faulted = true`, making a deadline bail indistinguishable from an
/// ordinary derive fault).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum FaultCause {
    /// A derive raised an error (bad state, SQLite I/O, an internal panic) — the ordinary
    /// fault; recovery is re-register + re-hydrate.
    Derive,
    /// A single push blew the host-armed deadline (`RindleError::PushDeadlineExceeded`) —
    /// the runaway-push bail, FOLLOWER-LAG-SHED §6.1 trigger (b): the host should consult
    /// its shed machinery BEFORE re-registering.
    PushDeadline,
    /// The worker thread itself was lost (died / unresponsive → detached + respawned);
    /// every query on the shard faulted with it.
    WorkerLost,
}

/// The shared single-thread state behind a [`Db`] handle. Held via `Rc` so cheap
/// `Db` clones, query handles, and write handles all reference the same engine + conns.
pub(crate) struct Inner {
    /// The ONE read-write connection. Every durable mutation goes through here; the
    /// preupdate hook observes them.
    pub(crate) writer: Rc<Connection>,
    /// A read-only connection exposed via [`Db::read`] for ad-hoc `SELECT`s.
    pub(crate) reader: Rc<Connection>,
    /// The per-thread IVM engine (one `Graph`, shared sources, the queries).
    pub(crate) engine: RefCell<Engine>,
    /// The preupdate-hook capture context (installed on `writer`). Observes every
    /// row mutation on the writer connection and turns it into a `SourceChange`.
    pub(crate) cdc: Rc<CaptureCtx>,
    /// Per-query subscriptions, keyed by the stable [`RegId`] (NOT the engine's sink
    /// `NodeId`, which a delta-overflow shed rebuild replaces; the current sink lives
    /// inside each entry).
    pub(crate) subs: RefCell<HashMap<RegId, QueryEntry>>,
    /// The next [`RegId`] to hand out.
    pub(crate) next_reg: Cell<u64>,
    /// Every registered table's discovered schema, cached so a delta-overflow shed
    /// (design 306 D4) can rebuild the engine with the sources re-registered — the
    /// single-thread twin of the worker threads' `tables` cache.
    pub(crate) tables: RefCell<Vec<(String, schema::TableSchema)>>,
    /// The last durably-committed tx (resume / fencing). 0 = nothing committed yet.
    pub(crate) committed_tx: Cell<u64>,
    /// Single-writer guard: true while a `WriteTxn` is open.
    pub(crate) in_write: Cell<bool>,
}

impl Drop for Inner {
    fn drop(&mut self) {
        // Remove the preupdate hook before the connection (and the capture context)
        // are torn down, so a stray late callback can't deref freed memory.
        rindle_cdc::uninstall(&self.writer);
    }
}

/// Everything an opener can be configured with, in one struct-with-`Default` — so
/// every combination (planner × operator storage × journal) is expressible without
/// a ladder of positional `open_with_*` rungs. Taken by [`Db::open_with`],
/// [`Cluster::open_with`](crate::Cluster::open_with), and
/// [`ClusterConsumer::open_with`](crate::ClusterConsumer::open_with); the named
/// rungs (`open`, `open_with_planning`, `open_with_journal`, …) are conveniences
/// that fill one field each.
///
/// ```no_run
/// # use rindle_replica::{Db, JournalMode, OpenOptions};
/// // An explicitly-wal2 store with the defaults otherwise:
/// let db = Db::open_with(
///     "app.db",
///     OpenOptions {
///         journal: JournalMode::Wal2,
///         ..OpenOptions::default()
///     },
/// ).unwrap();
/// ```
#[derive(Clone, Copy, Debug)]
pub struct OpenOptions {
    /// Run the cost-based join-flip planner at query registration (default **true**,
    /// matching `open`). Result-preserving; the plan is frozen per registration.
    pub plan_queries: bool,
    /// Where stateful operators keep scratch state (default in-process memory).
    pub operator_storage: OperatorStorage,
    /// The journal mode a fresh (or explicitly-wal2) store gets (design 306 D5;
    /// default [`JournalMode::Wal`]). [`JournalMode::Wal2`] is a requirement, not a
    /// preference — the opener fails if the store does not end up in wal2.
    pub journal: JournalMode,
    /// `PRAGMA wal_autocheckpoint` for the **writer** connection, in pages (design 410
    /// §3.5). `None` (the default) leaves SQLite's own default of 1000 pages — what
    /// every server-side opener wants. `Some(0)` disables the automatic checkpoint
    /// entirely, so the WAL folds back only on an explicit
    /// [`maintain`](Db::maintain) pass: the embedded/mobile posture, where an
    /// unpredictable checkpoint landing inside whichever commit happens to cross the
    /// threshold is worse than a deliberate one at a moment the app picks (gesture end,
    /// idle, backgrounding). Connection-local, so it must be re-passed at every open.
    pub wal_autocheckpoint: Option<u32>,
    /// Whether SQLite enforces declared foreign keys on this store's read-write
    /// connections (default [`ForeignKeys::Enforced`], matching the vendored build's
    /// compile-time default).
    ///
    /// [`ForeignKeys::Enforced`] is the **origin** posture — right for an embedded
    /// `Db`, a directly-driven `Cluster`, and the standalone daemon's write plane,
    /// where a violated constraint is the application's bug. An **apply** plane — a
    /// follower replaying a master's journal — passes [`ForeignKeys::Unenforced`],
    /// because those rows were validated upstream, arrive in the stream's order rather
    /// than a topological one, and carry the authority's cascade effects as ordinary
    /// row changes that must not be re-run locally. See
    /// [`rindle_writeplane::foreign_keys`] for the full argument, and
    /// [`Cluster::foreign_key_audit`](crate::Cluster::foreign_key_audit) for the audit
    /// that proves referential integrity either way.
    pub foreign_keys: ForeignKeys,
}

impl Default for OpenOptions {
    fn default() -> OpenOptions {
        OpenOptions {
            plan_queries: true,
            operator_storage: OperatorStorage::default(),
            journal: JournalMode::default(),
            wal_autocheckpoint: None,
            foreign_keys: ForeignKeys::default(),
        }
    }
}

/// A single-thread SQLite runtime with live queries. Clones share the same writer,
/// engine, and registrations through `Rc`; they do not create independent databases.
/// This handle is `!Send`. Query and write handles keep the shared runtime alive.
#[derive(Clone)]
pub struct Db {
    inner: Rc<Inner>,
}

impl Db {
    /// Open a replica over a **file-backed** SQLite database (derivation needs WAL +
    /// multiple connections, which an in-memory DB cannot provide).
    ///
    /// A fresh file gets ordinary `journal_mode = wal` (design 306 D5 — the file stays
    /// openable by any stock SQLite build); an existing `wal` or `wal2` file keeps its
    /// mode. Asserts `sqlite3_threadsafe() != 0`, failing loud rather than silently
    /// degrading.
    pub fn open(path: impl AsRef<Path>) -> Result<Db, ReplicaError> {
        Self::open_with(path, OpenOptions::default())
    }

    /// Like [`open`](Self::open), but with explicit control over the cost-based join-flip
    /// planner. When `plan_queries` is true, [`query`](Self::query) runs the planner (a
    /// `SqliteCostModel` over the worker connection, cached) at registration to annotate
    /// `flip` before lowering. Result-preserving; the plan is frozen per registration.
    /// The planner is server-side (table-source) only and `open` enables it by default;
    /// pass `false` here to opt out (e.g. to pin the unplanned path).
    pub fn open_with_planning(
        path: impl AsRef<Path>,
        plan_queries: bool,
    ) -> Result<Db, ReplicaError> {
        Self::open_with(
            path,
            OpenOptions {
                plan_queries,
                ..OpenOptions::default()
            },
        )
    }

    /// Like [`open_with_planning`](Self::open_with_planning), but also selects the
    /// operator-scratch-state backend. Pass [`OperatorStorage::SqliteSpill`] to spill
    /// stateful-operator state (`take`/`cap`/`reduce`) to a private on-disk temp
    /// database instead of keeping it in RAM. This reduces operator scratch memory;
    /// it does not cap all memory used by the runtime or the application.
    pub fn open_with_options(
        path: impl AsRef<Path>,
        plan_queries: bool,
        operator_storage: OperatorStorage,
    ) -> Result<Db, ReplicaError> {
        Self::open_with(
            path,
            OpenOptions {
                plan_queries,
                operator_storage,
                ..OpenOptions::default()
            },
        )
    }

    /// The full-combination opener: every [`OpenOptions`] field is honored, so any
    /// planner × operator-storage × journal combination is one call — the named
    /// rungs above are conveniences over this.
    pub fn open_with(path: impl AsRef<Path>, opts: OpenOptions) -> Result<Db, ReplicaError> {
        let OpenOptions {
            plan_queries,
            operator_storage,
            journal,
            wal_autocheckpoint,
            foreign_keys,
        } = opts;
        // The engine confines a connection to one thread, but the C library must still
        // be compiled threadsafe (serialized) for one-connection-per-thread to be sound.
        if unsafe { rusqlite::ffi::sqlite3_threadsafe() } == 0 {
            return Err(ReplicaError::NotThreadsafe);
        }

        let path = path.as_ref();
        // Shared per-connection setup (the journal mode + maintenance/hardening pragmas)
        // lives in `parallel`; the public read callback gets a physically read-only handle.
        let open_one = || -> Result<Rc<Connection>, ReplicaError> {
            crate::parallel::open_journal(path, journal, foreign_keys).map(Rc::new)
        };

        let writer = open_one()?;
        // The WAL-growth knob, on the writer only (the sole connection that appends
        // frames) and BEFORE the preupdate hook is installed, so the pragma cannot be
        // observed as part of a captured transaction.
        if let Some(pages) = wal_autocheckpoint {
            crate::parallel::set_wal_autocheckpoint(&writer, pages)?;
        }
        let worker = open_one()?;
        let reader = Rc::new(crate::parallel::open_journal_read_only(path)?);

        // Our bookkeeping: a single-row watermark table carrying the committed tx id.
        // Created before the hook is installed; its "__"-prefixed name is skipped by CDC.
        writer
            .execute_batch(
                "CREATE TABLE IF NOT EXISTS __replica_meta (id INTEGER PRIMARY KEY, tx_id INTEGER NOT NULL)",
            )
            .map_err(|e| ReplicaError::sqlite("create __replica_meta", e))?;
        let committed: u64 = writer
            .query_row("SELECT tx_id FROM __replica_meta WHERE id = 0", [], |r| {
                r.get::<_, i64>(0)
            })
            .optional()
            .map_err(|e| ReplicaError::sqlite("read watermark", e))?
            .map(|v| v as u64)
            .unwrap_or(0);

        // Observe every row mutation on the writer via the preupdate hook (raw FFI).
        let cdc = CaptureCtx::new();
        rindle_cdc::install(&writer, &cdc);

        let engine = Engine::new(worker, plan_queries, operator_storage)?;
        Ok(Db {
            inner: Rc::new(Inner {
                writer,
                reader,
                engine: RefCell::new(engine),
                cdc,
                subs: RefCell::new(HashMap::new()),
                next_reg: Cell::new(0),
                tables: RefCell::new(Vec::new()),
                committed_tx: Cell::new(committed),
                in_write: Cell::new(false),
            }),
        })
    }

    /// Run one **online maintenance pass** on the writer connection — see
    /// [`Cluster::maintain`](crate::Cluster::maintain) for the full contract. Best-effort and
    /// bounded; **skipped** (reporting `MaintenanceReport::skipped()`) while a write transaction
    /// is open so it never disturbs in-flight work.
    pub fn maintain(&self, opts: &MaintenanceOptions) -> Result<MaintenanceReport, ReplicaError> {
        if self.inner.in_write.get() {
            return Ok(MaintenanceReport::skipped());
        }
        maintenance::run(&self.inner.writer, opts)
    }

    /// Run a **full** `ANALYZE` on the writer to build deep planner statistics (`sqlite_stat4`) —
    /// the single-thread parity for [`Cluster::analyze_full`](crate::Cluster::analyze_full). Call
    /// once after a bulk load so the planner can seek (not scan) an `ORDER BY … LIMIT n` view's
    /// displacement re-fetch (GitHub #68). Restores the bounded `analysis_limit` afterward.
    /// Rejected while a write transaction is open.
    pub fn analyze_full(&self) -> Result<(), ReplicaError> {
        if self.inner.in_write.get() {
            return Err(ReplicaError::Open(
                "cannot run a full ANALYZE while a write transaction is open".into(),
            ));
        }
        maintenance::analyze_full(&self.inner.writer)
    }

    /// Register a base table with the engine: discover its columns + primary key from
    /// the schema, ensure the PK has the UNIQUE index `TableSource` requires, and build
    /// the shared source. Idempotent per table. Rejects BLOB-typed columns and tables
    /// without a primary key.
    ///
    /// The table must be **plain** — SQL triggers and generated columns are rejected (see the
    /// crate-level "supported usage" docs). Foreign-key cascades are represented by the explicit
    /// row deltas observed by the preupdate hook.
    pub fn register_table(&self, table: &str) -> Result<(), ReplicaError> {
        if self.inner.engine.borrow().has_source(table) {
            return Ok(());
        }
        // Registering a NEW table while a write txn is open would run its
        // `CREATE UNIQUE INDEX` inside the uncommitted transaction (invisible to the
        // engine's separate worker connection until commit) — reject the nonsensical
        // sequence rather than build a source against a not-yet-visible index.
        if self.inner.in_write.get() {
            return Err(ReplicaError::Open(
                "cannot register a new table while a write transaction is open".into(),
            ));
        }
        let ts = schema::discover(&self.inner.writer, table)?;
        schema::ensure_unique_pk_index(&self.inner.writer, table, &ts)?;
        // Teach the capture hook this table's column types (for leaf-matching coercion).
        let col_types: Vec<_> = ts.columns.iter().map(|c| c.ty).collect();
        self.inner
            .cdc
            .set_table_with_primary_key(table, &col_types, &ts.primary_key);
        self.inner
            .engine
            .borrow_mut()
            .register_table(table, ts.clone())?;
        // Cache the schema AFTER a successful registration, so a delta-overflow shed can
        // rebuild the engine with exactly the sources it had (the `has_source` guard
        // above keeps this dup-free).
        self.inner.tables.borrow_mut().push((table.to_string(), ts));
        Ok(())
    }

    /// Register a live query from a Zero-wire [`Ast`] (build it with `rindle::table(..)`
    /// or deserialize via [`Db::query_json`]) under the caller-supplied [`QueryId`] tag
    /// (echoed via [`Query::id`]; the engine never interprets it — see [`QueryId`]).
    /// Lowers the AST into the shared engine, hydrates it, and returns a [`Query`] handle
    /// to subscribe to. `BuildError` (unknown table/column, unsupported shape) is
    /// surfaced synchronously here.
    ///
    /// Each call builds its **own** pipeline — the engine does not de-duplicate, even
    /// for an identical `query_id`/AST. A caller that wants to share one pipeline across
    /// requesters dedups at its own layer (it holds the per-requester hydration state the
    /// raw change stream does not). Subscribe before any later writes: the handle's
    /// initial hydration is cached at registration, not kept current. Explicitly call
    /// [`Query::destroy`] to unregister; dropping the handle does not stop the query.
    pub fn query(&self, query_id: QueryId, ast: Ast) -> Result<Query, ReplicaError> {
        let (sink, initial) = self
            .inner
            .engine
            .borrow_mut()
            .register_query(query_id, &ast)?;
        let tx = self.inner.committed_tx.get();
        let reg = RegId(self.inner.next_reg.get());
        self.inner.next_reg.set(reg.0 + 1);
        self.inner.subs.borrow_mut().insert(
            reg,
            QueryEntry {
                sink,
                query_id,
                ast,
                hydrated: initial,
                hydrated_tx: tx,
                callbacks: Vec::new(),
            },
        );
        Ok(Query {
            inner: self.inner.clone(),
            reg,
            query_id,
        })
    }

    /// Convenience: parse a Zero-wire AST from JSON, then [`Db::query`] under `query_id`.
    pub fn query_json(&self, query_id: QueryId, json: &str) -> Result<Query, ReplicaError> {
        let ast: Ast = serde_json::from_str(json)
            .map_err(|e| ReplicaError::Schema(format!("invalid AST JSON: {e}")))?;
        self.query(query_id, ast)
    }

    /// Analyze `ast` COLD and report where its time and rows go
    /// (`ANALYZE-QUERY-DESIGN.md`) — the single-thread parity for
    /// [`Cluster::analyze_query`](crate::Cluster::analyze_query). Read-only: it builds and
    /// drops a throwaway instrumented pipeline over a pinned read snapshot, registering no
    /// materialization and leaving every live query untouched. `BuildError` (unknown
    /// table/column, unsupported shape) surfaces as `Err`. Rejected while a write
    /// transaction is open (its snapshot is not yet committed).
    pub fn analyze_query(&self, ast: &Ast) -> Result<AnalyzeReport, ReplicaError> {
        if self.inner.in_write.get() {
            return Err(ReplicaError::Open(
                "cannot analyze a query while a write transaction is open".into(),
            ));
        }
        self.inner.engine.borrow().analyze_query(ast)
    }

    /// Tune the **derivation memory budget** (design 306 D4): the estimated bytes of
    /// folded rows one transaction may hold — across every table it touches, not per
    /// table — before its derivation gives up. The default is
    /// [`rindle_sqlite::DEFAULT_MAX_DELTA_BYTES`] (256 MiB).
    ///
    /// Overflow is a **shed**, not a failure: the transaction still commits; the engine
    /// rebuilds and every registered query re-hydrates from the committed state
    /// (subscribers receive a fresh [`Update::Hydrated`] — see [`Update`]). Lower it to
    /// bound derivation memory harder (shedding sooner); raise it to keep huge bulk
    /// loads deriving incrementally at the cost of memory. Applies to every registered
    /// table, now and later, and survives the shed rebuild itself.
    ///
    /// The accounting is an upper-bound estimate of the delta's own heap — row buffers
    /// plus per-index bookkeeping — described at [`rindle_sqlite::DeltaBudget`]. It is
    /// not a process RSS limit: SQLite's page cache, the query pipelines, and the
    /// re-hydration this shed triggers are all outside it.
    pub fn set_max_delta_bytes(&self, max_bytes: usize) {
        self.inner.engine.borrow().set_max_delta_bytes(max_bytes);
    }

    /// Open the single-writer transaction. Run ordinary SQL through the returned
    /// [`WriteTxn`]; `commit` derives + delivers each query's incremental events,
    /// `rollback` (or drop) leaves every view untouched. Errors if a write txn is
    /// already open (there is exactly one writer).
    pub fn write(&self) -> Result<WriteTxn, ReplicaError> {
        writer::begin(self.inner.clone())
    }

    /// The last durably-committed global tx id (0 if none yet).
    pub fn committed_tx_id(&self) -> TxId {
        TxId(self.inner.committed_tx.get())
    }

    /// Run SQL against a physically read-only connection using SQLite snapshot semantics.
    /// This separate connection cannot see an open [`WriteTxn`]'s uncommitted writes.
    /// `SQLITE_OPEN_READ_ONLY` prevents a callback from bypassing capture with writes.
    /// Schema changes belong in [`Self::exec_ddl`], and row mutations in [`Self::write`].
    pub fn read<T>(
        &self,
        f: impl FnOnce(&Connection) -> rusqlite::Result<T>,
    ) -> Result<T, ReplicaError> {
        f(&self.inner.reader).map_err(|e| ReplicaError::sqlite("read", e))
    }

    /// Walk every declared foreign key and report violating rows — the opt-in audit.
    ///
    /// `PRAGMA foreign_key_check` reports what is in the file, not what SQLite would have
    /// refused, so it answers the same question whatever
    /// [`OpenOptions::foreign_keys`] this replica was opened with. It is a full scan of
    /// the referencing tables — a deliberate operation, never a per-commit step. Pass
    /// [`FOREIGN_KEY_AUDIT_ROW_CAP`] unless you have a reason not to, or `0` for no cap.
    pub fn foreign_key_audit(&self, max_rows: usize) -> Result<ForeignKeyAudit, ReplicaError> {
        foreign_key_check(&self.inner.reader, max_rows)
    }

    /// The hierarchical view [`rindle::value::Schema`] `ast` materializes to over the registered
    /// tables (shape + sort + `singular` + in-view relationships) — derived identically to
    /// [`query`](Db::query). A layer that ships views to a remote (the flat-change/wire
    /// schema + fingerprint) needs this without registering the query. `BuildError`
    /// (unknown table/column) surfaces as `Err`.
    pub fn view_schema(&self, ast: &Ast) -> Result<rindle::value::Schema, ReplicaError> {
        self.inner.engine.borrow().view_schema(ast)
    }

    /// Fetch the registered query through its pipeline as a fresh set of hydration
    /// `Add` events. This traverses current sources; it does not read a cached view
    /// assembled from prior `Changed` events. It does not consume subscriber events.
    /// An unregistered query yields an empty snapshot. Read failures return `Err`.
    /// Use distinct query IDs when reading snapshots by ID.
    pub fn read_snapshot(&self, query_id: QueryId) -> Result<Vec<ChangeEvent>, ReplicaError> {
        Ok(self
            .inner
            .engine
            .borrow()
            .read_snapshot(query_id)?
            .unwrap_or_default())
    }

    /// Run schema DDL (`CREATE TABLE …`) against the writer connection in autocommit —
    /// the supported way to define the plain base tables you then [`Self::register_table`] and
    /// write through (crate-level "supported usage"). Rejected while a write transaction is
    /// open (DDL there would be invisible to the engine's separate worker until commit).
    /// Accepts schema DDL (`CREATE`/`ALTER`/`DROP`/`REINDEX`) plus the historical bounded
    /// `ANALYZE` maintenance call; row-changing statements are rejected. The complete batch and
    /// schema-envelope validation commit atomically, and row-producing DDL is rejected too.
    pub fn exec_ddl(&self, sql: &str) -> Result<(), ReplicaError> {
        if self.inner.in_write.get() {
            return Err(ReplicaError::Open(
                "cannot run DDL while a write transaction is open".into(),
            ));
        }
        schema::ensure_embedded_ddl_batch(sql)?;
        schema::transactionally_apply_schema(
            &self.inner.writer,
            &self.inner.cdc,
            "exec_ddl",
            true,
            false,
            || {
                self.inner
                    .writer
                    .execute_batch(sql)
                    .map_err(|error| ReplicaError::sqlite("exec_ddl", error))
            },
        )
    }
}