rindle_replica/

initial_snapshot.rs

//! One-connection, rollback-journal construction of an initial dark CDC snapshot.
//!
//! This is deliberately not a `Cluster` mode. It owns no graph, capture hook, worker, reader, or
//! query surface. The only durable states it can produce are an exact owned BUILDING database or a
//! complete snapshot plus its whole-run source checkpoint.

use std::collections::{BTreeMap, BTreeSet};
use std::marker::PhantomData;
use std::path::{Path, PathBuf};
use std::rc::Rc;

use rindle::value::OwnedValue;
use rusqlite::{params, params_from_iter, Connection, OptionalExtension};

use crate::apply::{ensure_source_offsets_table_on_conn, source_offsets_table_ddl};
use crate::sql::to_value;
use crate::table_shape::ColType;
use crate::ReplicaError;
use crate::{set_foreign_keys, ForeignKeys};
use crate::{SOURCE_OFFSETS_TABLE, SOURCE_OFFSET_WHOLE_RUN};

pub const CDC_BOOTSTRAP_TABLE: &str = "_rindle_cdc_bootstrap";
const BOOTSTRAP_FORMAT: i64 = 1;
const BOOTSTRAP_STATE: &str = "snapshot-building";

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct InitialSnapshotIdentity {
    pub source: String,
    pub generation: String,
    pub descriptor_hash: String,
    pub profile_fingerprint: String,
    pub schema_fingerprint: String,
    pub topic_id: String,
    pub created_at_ms: i64,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct InitialSnapshotColumn {
    pub name: String,
    pub ty: ColType,
    pub nullable: bool,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct InitialSnapshotTable {
    pub name: String,
    pub columns: Vec<InitialSnapshotColumn>,
    pub primary_key: Vec<String>,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredSourceCheckpoint {
    pub offset: String,
    pub chunk_seq: i64,
    pub run_id: Option<String>,
    pub run_rows: Option<u64>,
}

pub enum InitialSnapshotOpen {
    Building(Box<InitialSnapshotStore>),
    Committed(StoredSourceCheckpoint),
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct InitialSnapshotCommit {
    pub source: String,
    pub snapshot_id: String,
    pub offset: String,
    pub locator: String,
    pub rows: u64,
}

/// The single SQLite owner for an initial snapshot. `PhantomData<Rc<()>>` makes the lifecycle
/// explicitly thread-confined even on platforms where rusqlite's connection is movable.
pub struct InitialSnapshotStore {
    path: PathBuf,
    conn: Connection,
    identity: InitialSnapshotIdentity,
    tables: BTreeMap<String, InitialSnapshotTable>,
    open_snapshot_id: Option<String>,
    rows: u64,
    _not_send: PhantomData<Rc<()>>,
}

impl InitialSnapshotStore {
    /// Classify and open a generation-owned build. An unrelated data-bearing database is never
    /// cleared or adopted. SQLite performs hot-journal recovery while the existing schema is read.
    pub fn open_or_create(
        path: impl AsRef<Path>,
        identity: InitialSnapshotIdentity,
        tables: Vec<InitialSnapshotTable>,
    ) -> Result<InitialSnapshotOpen, ReplicaError> {
        validate_identity(&identity)?;
        let tables = validate_tables(tables)?;
        let path = path.as_ref().to_path_buf();
        let conn = Connection::open(&path)
            .map_err(|error| ReplicaError::sqlite("initial snapshot open", error))?;
        // An apply-plane store, so foreign keys are not enforced here — stated explicitly
        // rather than inherited from the vendored build's `SQLITE_DEFAULT_FOREIGN_KEYS=1`.
        // A dark snapshot is a bulk copy of an upstream's rows: they land table by table in
        // whatever order the source hands them over, so a child row routinely precedes its
        // parent and only the COMPLETED snapshot is referentially whole. Enforcing per row
        // would fail a snapshot that is correct the moment it commits. `PRAGMA
        // foreign_key_check` on the finished file is how a caller wanting proof gets it
        // (`rindle_writeplane::foreign_key_check`).
        set_foreign_keys(&conn, ForeignKeys::Unenforced)?;
        conn.busy_timeout(std::time::Duration::ZERO)
            .map_err(|error| ReplicaError::sqlite("initial snapshot busy timeout", error))?;

        let existing_tables = user_tables(&conn)?;
        let unexpected_objects = user_non_table_objects(&conn)?;
        if !unexpected_objects.is_empty() {
            return Err(state_error(format!(
                "initial snapshot database contains unexpected schema objects: {}",
                unexpected_objects.join(", ")
            )));
        }
        let marker_table_present = existing_tables.contains(CDC_BOOTSTRAP_TABLE);
        let marker_present = if marker_table_present {
            let rows: i64 = conn
                .query_row(
                    &format!("SELECT COUNT(*) FROM {CDC_BOOTSTRAP_TABLE}"),
                    [],
                    |row| row.get(0),
                )
                .map_err(|error| {
                    ReplicaError::sqlite("initial snapshot marker classification", error)
                })?;
            match rows {
                0 => false,
                1 => true,
                _ => return Err(state_error("bootstrap marker has more than one row")),
            }
        } else {
            false
        };
        let checkpoint = if existing_tables.contains(SOURCE_OFFSETS_TABLE) {
            verify_create_sql(
                &conn,
                SOURCE_OFFSETS_TABLE,
                &source_offsets_table_ddl(),
                "source-offset",
            )?;
            read_checkpoint(&conn, &identity.source)?
        } else {
            None
        };

        if !marker_present {
            if let Some(checkpoint) = checkpoint {
                if table_row_count(&conn, SOURCE_OFFSETS_TABLE, "checkpoint classification")? != 1 {
                    return Err(state_error(
                        "committed initial snapshot has more than one source checkpoint",
                    ));
                }
                verify_committed_build(&conn, &tables, &existing_tables, &checkpoint)?;
                return Ok(InitialSnapshotOpen::Committed(checkpoint));
            }
            if !existing_tables.is_empty() {
                return Err(state_error(
                    "database has tables but neither an owned bootstrap marker nor a source checkpoint",
                ));
            }
            configure_delete_exclusive(&conn)?;
            initialize_schema(&conn, &identity, &tables)?;
        } else {
            if checkpoint.is_some() {
                return Err(state_error(
                    "database contains both a bootstrap marker and a source checkpoint",
                ));
            }
            verify_owned_build(&conn, &identity, &tables, &existing_tables)?;
            configure_delete_exclusive(&conn)?;
        }

        Ok(InitialSnapshotOpen::Building(Box::new(Self {
            path,
            conn,
            identity,
            tables,
            open_snapshot_id: None,
            rows: 0,
            _not_send: PhantomData,
        })))
    }

    pub fn begin(&mut self, snapshot_id: &str) -> Result<(), ReplicaError> {
        if snapshot_id.is_empty() {
            return Err(state_error("snapshot ID must not be empty"));
        }
        if self.open_snapshot_id.is_some() {
            return Err(state_error(
                "an initial snapshot transaction is already open",
            ));
        }
        self.conn
            .execute_batch("BEGIN EXCLUSIVE")
            .map_err(|error| ReplicaError::sqlite("initial snapshot BEGIN EXCLUSIVE", error))?;
        self.open_snapshot_id = Some(snapshot_id.to_owned());
        self.rows = 0;
        Ok(())
    }

    /// Execute one strict positional INSERT immediately. The connection's prepared-statement cache
    /// retains the generated SQL per mapped table; no UPSERT/replace/ignore form is permitted.
    pub fn apply_add(&mut self, table: &str, row: &[OwnedValue]) -> Result<(), ReplicaError> {
        if self.open_snapshot_id.is_none() {
            return Err(state_error("snapshot add arrived without BEGIN"));
        }
        let table = self
            .tables
            .get(table)
            .ok_or_else(|| state_error("snapshot add names an unknown mapped table"))?;
        if row.len() != table.columns.len() {
            return Err(state_error(
                "snapshot add width differs from the mapped table",
            ));
        }
        let sql = insert_sql(table);
        let mut statement = self
            .conn
            .prepare_cached(&sql)
            .map_err(|error| ReplicaError::sqlite("initial snapshot prepare INSERT", error))?;
        statement
            .execute(params_from_iter(row.iter().map(to_value)))
            .map_err(|error| ReplicaError::sqlite("initial snapshot strict INSERT", error))?;
        self.rows = self
            .rows
            .checked_add(1)
            .ok_or_else(|| state_error("initial snapshot row count overflow"))?;
        Ok(())
    }

    /// Commit data, signed locator/checkpoint, accounting, and marker removal as one SQLite atom.
    pub fn commit(
        mut self,
        checkpoint: InitialSnapshotCommit,
    ) -> Result<StoredSourceCheckpoint, ReplicaError> {
        let open_id = self
            .open_snapshot_id
            .as_deref()
            .ok_or_else(|| state_error("initial snapshot commit arrived without BEGIN"))?;
        if checkpoint.source != self.identity.source
            || checkpoint.snapshot_id != open_id
            || checkpoint.rows != self.rows
            || checkpoint.offset.is_empty()
            || checkpoint.locator.is_empty()
        {
            return Err(state_error(
                "initial snapshot commit identity/count/checkpoint differs from the open build",
            ));
        }
        verify_marker(&self.conn, &self.identity)?;
        let rows = i64::try_from(checkpoint.rows)
            .map_err(|_| state_error("initial snapshot row count exceeds SQLite INTEGER"))?;
        self.conn
            .execute(
                &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, NULL, ?5, ?5, NULL)"
                ),
                params![
                    checkpoint.source,
                    checkpoint.offset,
                    SOURCE_OFFSET_WHOLE_RUN,
                    checkpoint.locator,
                    rows,
                ],
            )
            .map_err(|error| ReplicaError::sqlite("initial snapshot checkpoint INSERT", error))?;
        let removed = self
            .conn
            .execute(
                &format!("DELETE FROM {CDC_BOOTSTRAP_TABLE} WHERE singleton = 1"),
                [],
            )
            .map_err(|error| ReplicaError::sqlite("initial snapshot marker removal", error))?;
        if removed != 1 {
            return Err(state_error("initial snapshot BUILDING marker disappeared"));
        }
        self.conn
            .execute_batch("COMMIT")
            .map_err(|error| ReplicaError::sqlite("initial snapshot COMMIT", error))?;
        self.open_snapshot_id = None;
        Ok(StoredSourceCheckpoint {
            offset: checkpoint.offset,
            chunk_seq: SOURCE_OFFSET_WHOLE_RUN,
            run_id: Some(checkpoint.locator),
            run_rows: Some(checkpoint.rows),
        })
    }

    pub fn rollback(&mut self) {
        if self.open_snapshot_id.take().is_some() {
            let _ = self.conn.execute_batch("ROLLBACK");
            self.rows = 0;
        }
    }

    pub fn rows(&self) -> u64 {
        self.rows
    }

    pub fn path(&self) -> &Path {
        &self.path
    }

    pub fn journal_mode(&self) -> Result<String, ReplicaError> {
        pragma_text(&self.conn, "journal_mode")
    }

    pub fn locking_mode(&self) -> Result<String, ReplicaError> {
        pragma_text(&self.conn, "locking_mode")
    }

    pub fn synchronous(&self) -> Result<i64, ReplicaError> {
        self.conn
            .query_row("PRAGMA synchronous", [], |row| row.get(0))
            .map_err(|error| ReplicaError::sqlite("initial snapshot PRAGMA synchronous", error))
    }
}

impl Drop for InitialSnapshotStore {
    fn drop(&mut self) {
        self.rollback();
    }
}

fn validate_identity(identity: &InitialSnapshotIdentity) -> Result<(), ReplicaError> {
    if identity.created_at_ms < 0
        || [
            &identity.source,
            &identity.generation,
            &identity.descriptor_hash,
            &identity.profile_fingerprint,
            &identity.schema_fingerprint,
            &identity.topic_id,
        ]
        .into_iter()
        .any(|value| value.is_empty())
    {
        return Err(state_error("initial snapshot identity is incomplete"));
    }
    Ok(())
}

fn validate_tables(
    tables: Vec<InitialSnapshotTable>,
) -> Result<BTreeMap<String, InitialSnapshotTable>, ReplicaError> {
    let mut result = BTreeMap::new();
    for table in tables {
        if table.name.is_empty() || table.columns.is_empty() || table.primary_key.is_empty() {
            return Err(state_error("initial snapshot table identity is incomplete"));
        }
        let names = table
            .columns
            .iter()
            .map(|column| column.name.as_str())
            .collect::<BTreeSet<_>>();
        if names.len() != table.columns.len()
            || table
                .primary_key
                .iter()
                .any(|key| !names.contains(key.as_str()))
            || table
                .columns
                .iter()
                .any(|column| table.primary_key.contains(&column.name) && column.nullable)
        {
            return Err(state_error(
                "initial snapshot table has duplicate/unknown/nullable primary-key columns",
            ));
        }
        if result.insert(table.name.clone(), table).is_some() {
            return Err(state_error("initial snapshot maps a duplicate table name"));
        }
    }
    if result.is_empty() {
        return Err(state_error("initial snapshot has no mapped tables"));
    }
    Ok(result)
}

fn configure_delete_exclusive(conn: &Connection) -> Result<(), ReplicaError> {
    let journal = conn
        .query_row("PRAGMA journal_mode=DELETE", [], |row| {
            row.get::<_, String>(0)
        })
        .map_err(|error| ReplicaError::sqlite("initial snapshot journal_mode=DELETE", error))?;
    let locking = conn
        .query_row("PRAGMA locking_mode=EXCLUSIVE", [], |row| {
            row.get::<_, String>(0)
        })
        .map_err(|error| ReplicaError::sqlite("initial snapshot locking_mode=EXCLUSIVE", error))?;
    conn.execute_batch("PRAGMA synchronous=FULL")
        .map_err(|error| ReplicaError::sqlite("initial snapshot synchronous=FULL", error))?;
    if !journal.eq_ignore_ascii_case("delete")
        || !locking.eq_ignore_ascii_case("exclusive")
        || conn
            .query_row("PRAGMA synchronous", [], |row| row.get::<_, i64>(0))
            .map_err(|error| ReplicaError::sqlite("initial snapshot verify synchronous", error))?
            != 2
    {
        return Err(ReplicaError::Open(
            "initial snapshot requires journal_mode=delete, locking_mode=exclusive, synchronous=FULL"
                .into(),
        ));
    }
    Ok(())
}

fn initialize_schema(
    conn: &Connection,
    identity: &InitialSnapshotIdentity,
    tables: &BTreeMap<String, InitialSnapshotTable>,
) -> Result<(), ReplicaError> {
    conn.execute_batch("BEGIN EXCLUSIVE")
        .map_err(|error| ReplicaError::sqlite("initial snapshot schema BEGIN", error))?;
    let result = (|| {
        for table in tables.values() {
            conn.execute_batch(&create_table_sql(table))
                .map_err(|error| ReplicaError::sqlite("initial snapshot mapped schema", error))?;
        }
        ensure_source_offsets_table_on_conn(conn)?;
        conn.execute_batch(&bootstrap_table_ddl())
            .map_err(|error| ReplicaError::sqlite("initial snapshot marker schema", error))?;
        conn.execute(
            &format!(
                "INSERT INTO {CDC_BOOTSTRAP_TABLE} \
                 (singleton, format, state, source, generation, descriptor_hash, \
                  profile_fingerprint, schema_fingerprint, topic_id, created_at_ms) \
                 VALUES (1, ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)"
            ),
            params![
                BOOTSTRAP_FORMAT,
                BOOTSTRAP_STATE,
                identity.source,
                identity.generation,
                identity.descriptor_hash,
                identity.profile_fingerprint,
                identity.schema_fingerprint,
                identity.topic_id,
                identity.created_at_ms,
            ],
        )
        .map_err(|error| ReplicaError::sqlite("initial snapshot marker INSERT", error))?;
        conn.execute_batch("COMMIT")
            .map_err(|error| ReplicaError::sqlite("initial snapshot schema COMMIT", error))
    })();
    if result.is_err() {
        let _ = conn.execute_batch("ROLLBACK");
    }
    result
}

fn verify_owned_build(
    conn: &Connection,
    identity: &InitialSnapshotIdentity,
    tables: &BTreeMap<String, InitialSnapshotTable>,
    existing_tables: &BTreeSet<String>,
) -> Result<(), ReplicaError> {
    let mut expected = tables.keys().cloned().collect::<BTreeSet<_>>();
    expected.insert(SOURCE_OFFSETS_TABLE.to_owned());
    expected.insert(CDC_BOOTSTRAP_TABLE.to_owned());
    if &expected != existing_tables {
        return Err(state_error(
            "owned bootstrap database contains unexpected or missing tables",
        ));
    }
    verify_marker(conn, identity)?;
    verify_create_sql(
        conn,
        SOURCE_OFFSETS_TABLE,
        &source_offsets_table_ddl(),
        "source-offset",
    )?;
    verify_create_sql(
        conn,
        CDC_BOOTSTRAP_TABLE,
        &bootstrap_table_ddl(),
        "bootstrap-marker",
    )?;
    let checkpoints: i64 = conn
        .query_row(
            &format!("SELECT COUNT(*) FROM {SOURCE_OFFSETS_TABLE}"),
            [],
            |row| row.get(0),
        )
        .map_err(|error| {
            ReplicaError::sqlite("initial snapshot checkpoint classification", error)
        })?;
    if checkpoints != 0 {
        return Err(state_error("owned BUILDING database contains a checkpoint"));
    }
    for table in tables.values() {
        verify_table_schema(conn, table)?;
        verify_create_sql(conn, &table.name, &create_table_sql(table), "mapped-table")?;
        let row = conn
            .query_row(
                &format!("SELECT 1 FROM {} LIMIT 1", quote_ident(&table.name)),
                [],
                |_| Ok(()),
            )
            .optional()
            .map_err(|error| ReplicaError::sqlite("initial snapshot empty-table check", error))?;
        if row.is_some() {
            return Err(state_error("owned BUILDING database contains mapped rows"));
        }
    }
    Ok(())
}

fn verify_committed_build(
    conn: &Connection,
    tables: &BTreeMap<String, InitialSnapshotTable>,
    existing_tables: &BTreeSet<String>,
    checkpoint: &StoredSourceCheckpoint,
) -> Result<(), ReplicaError> {
    let mut expected = tables.keys().cloned().collect::<BTreeSet<_>>();
    expected.insert(SOURCE_OFFSETS_TABLE.to_owned());
    expected.insert(CDC_BOOTSTRAP_TABLE.to_owned());
    if &expected != existing_tables {
        return Err(state_error(
            "committed initial snapshot contains unexpected or missing tables",
        ));
    }
    verify_create_sql(
        conn,
        SOURCE_OFFSETS_TABLE,
        &source_offsets_table_ddl(),
        "source-offset",
    )?;
    verify_create_sql(
        conn,
        CDC_BOOTSTRAP_TABLE,
        &bootstrap_table_ddl(),
        "bootstrap-marker",
    )?;
    if table_row_count(conn, CDC_BOOTSTRAP_TABLE, "committed marker classification")? != 0 {
        return Err(state_error(
            "committed initial snapshot still contains a BUILDING marker",
        ));
    }
    if checkpoint.chunk_seq != SOURCE_OFFSET_WHOLE_RUN
        || checkpoint.offset.is_empty()
        || checkpoint.run_id.as_deref().is_none_or(str::is_empty)
    {
        return Err(state_error(
            "committed initial snapshot checkpoint is not a complete signed run",
        ));
    }

    let mut rows = 0_u64;
    for table in tables.values() {
        verify_table_schema(conn, table)?;
        verify_create_sql(conn, &table.name, &create_table_sql(table), "mapped-table")?;
        let table_rows = table_row_count(conn, &table.name, "committed mapped-row accounting")?;
        rows = rows
            .checked_add(
                u64::try_from(table_rows)
                    .map_err(|_| state_error("committed mapped-table row count is negative"))?,
            )
            .ok_or_else(|| state_error("committed mapped-table row count overflow"))?;
    }
    if checkpoint.run_rows != Some(rows) {
        return Err(state_error(
            "committed mapped rows differ from the source-checkpoint accounting",
        ));
    }
    Ok(())
}

fn verify_marker(
    conn: &Connection,
    identity: &InitialSnapshotIdentity,
) -> Result<(), ReplicaError> {
    let marker = conn
        .query_row(
            &format!(
                "SELECT format, state, source, generation, descriptor_hash, profile_fingerprint, \
                        schema_fingerprint, topic_id, created_at_ms \
                 FROM {CDC_BOOTSTRAP_TABLE} WHERE singleton = 1"
            ),
            [],
            |row| {
                Ok((
                    row.get::<_, i64>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, String>(4)?,
                    row.get::<_, String>(5)?,
                    row.get::<_, String>(6)?,
                    row.get::<_, String>(7)?,
                    row.get::<_, i64>(8)?,
                ))
            },
        )
        .optional()
        .map_err(|error| ReplicaError::sqlite("initial snapshot marker read", error))?;
    let expected = (
        BOOTSTRAP_FORMAT,
        BOOTSTRAP_STATE,
        identity.source.as_str(),
        identity.generation.as_str(),
        identity.descriptor_hash.as_str(),
        identity.profile_fingerprint.as_str(),
        identity.schema_fingerprint.as_str(),
        identity.topic_id.as_str(),
        identity.created_at_ms,
    );
    match marker {
        Some((
            format,
            state,
            source,
            generation,
            descriptor,
            profile,
            schema,
            topic,
            created_at_ms,
        )) if (
            format,
            state.as_str(),
            source.as_str(),
            generation.as_str(),
            descriptor.as_str(),
            profile.as_str(),
            schema.as_str(),
            topic.as_str(),
            created_at_ms,
        ) == expected =>
        {
            Ok(())
        }
        _ => Err(state_error(
            "bootstrap marker does not exactly match the configured descriptor identity",
        )),
    }
}

fn read_checkpoint(
    conn: &Connection,
    source: &str,
) -> Result<Option<StoredSourceCheckpoint>, ReplicaError> {
    let stored = conn
        .query_row(
            &format!(
                "SELECT offset, chunk_seq, run_id, run_rows, rows_cum FROM {SOURCE_OFFSETS_TABLE} \
                 WHERE source = ?1"
            ),
            [source],
            |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, i64>(1)?,
                    row.get::<_, Option<String>>(2)?,
                    row.get::<_, Option<i64>>(3)?,
                    row.get::<_, Option<i64>>(4)?,
                ))
            },
        )
        .optional()
        .map_err(|error| ReplicaError::sqlite("initial snapshot checkpoint read", error))?;
    stored
        .map(|(offset, chunk_seq, run_id, run_rows, rows_cum)| {
            let run_rows = run_rows
                .map(|rows| {
                    u64::try_from(rows)
                        .map_err(|_| state_error("stored initial snapshot row count is negative"))
                })
                .transpose()?;
            let rows_cum = rows_cum
                .map(|rows| {
                    u64::try_from(rows).map_err(|_| {
                        state_error("stored cumulative snapshot row count is negative")
                    })
                })
                .transpose()?;
            if run_rows != rows_cum {
                return Err(state_error(
                    "stored snapshot run and cumulative row accounting differ",
                ));
            }
            Ok(StoredSourceCheckpoint {
                offset,
                chunk_seq,
                run_id,
                run_rows,
            })
        })
        .transpose()
}

fn user_tables(conn: &Connection) -> Result<BTreeSet<String>, ReplicaError> {
    conn.prepare(
        "SELECT name FROM sqlite_schema \
         WHERE type = 'table' AND name NOT LIKE 'sqlite_%' ORDER BY name",
    )
    .and_then(|mut statement| {
        statement
            .query_map([], |row| row.get::<_, String>(0))?
            .collect::<rusqlite::Result<BTreeSet<_>>>()
    })
    .map_err(|error| ReplicaError::sqlite("initial snapshot schema classification", error))
}

fn user_non_table_objects(conn: &Connection) -> Result<Vec<String>, ReplicaError> {
    conn.prepare(
        "SELECT type || ':' || name FROM sqlite_schema \
         WHERE type <> 'table' AND name NOT LIKE 'sqlite_%' ORDER BY type, name",
    )
    .and_then(|mut statement| {
        statement
            .query_map([], |row| row.get::<_, String>(0))?
            .collect::<rusqlite::Result<Vec<_>>>()
    })
    .map_err(|error| ReplicaError::sqlite("initial snapshot schema-object classification", error))
}

fn table_row_count(
    conn: &Connection,
    table: &str,
    context: &'static str,
) -> Result<i64, ReplicaError> {
    conn.query_row(
        &format!("SELECT COUNT(*) FROM {}", quote_ident(table)),
        [],
        |row| row.get(0),
    )
    .map_err(|error| ReplicaError::sqlite(context, error))
}

fn verify_create_sql(
    conn: &Connection,
    table: &str,
    expected: &str,
    kind: &str,
) -> Result<(), ReplicaError> {
    let actual = conn
        .query_row(
            "SELECT sql FROM sqlite_schema WHERE type = 'table' AND name = ?1",
            [table],
            |row| row.get::<_, String>(0),
        )
        .optional()
        .map_err(|error| ReplicaError::sqlite("initial snapshot CREATE SQL validation", error))?;
    // SQLite intentionally omits the idempotence clause from the canonical SQL it stores in
    // `sqlite_schema`; compare against that representation while keeping every column, constraint,
    // and ordering byte exact.
    let expected = expected.replacen("CREATE TABLE IF NOT EXISTS ", "CREATE TABLE ", 1);
    if actual.as_deref() != Some(expected.as_str()) {
        return Err(state_error(format!(
            "owned BUILDING database {kind} schema differs from the current bootstrap format"
        )));
    }
    Ok(())
}

fn verify_table_schema(
    conn: &Connection,
    table: &InitialSnapshotTable,
) -> Result<(), ReplicaError> {
    let sql = format!("PRAGMA table_info({})", quote_ident(&table.name));
    let mut actual = conn
        .prepare(&sql)
        .and_then(|mut statement| {
            statement
                .query_map([], |row| {
                    Ok((
                        row.get::<_, String>(1)?,
                        row.get::<_, String>(2)?,
                        row.get::<_, i64>(3)? != 0,
                        row.get::<_, i64>(5)?,
                    ))
                })?
                .collect::<rusqlite::Result<Vec<_>>>()
        })
        .map_err(|error| {
            ReplicaError::sqlite("initial snapshot mapped schema validation", error)
        })?;
    let mut actual_pk = actual
        .iter()
        .filter(|(_, _, _, ordinal)| *ordinal > 0)
        .map(|(name, _, _, ordinal)| (*ordinal, name.clone()))
        .collect::<Vec<_>>();
    actual_pk.sort_by_key(|(ordinal, _)| *ordinal);
    let actual_pk = actual_pk
        .into_iter()
        .map(|(_, name)| name)
        .collect::<Vec<_>>();
    if actual.len() != table.columns.len()
        || actual
            .iter_mut()
            .zip(&table.columns)
            .any(|((name, ty, not_null, _), expected)| {
                name != &expected.name
                    || !ty.eq_ignore_ascii_case(sql_type(expected.ty))
                    || *not_null == expected.nullable
            })
        || actual_pk != table.primary_key
    {
        return Err(state_error(
            "owned BUILDING database mapped schema differs from the descriptor",
        ));
    }
    Ok(())
}

fn create_table_sql(table: &InitialSnapshotTable) -> String {
    let columns = table
        .columns
        .iter()
        .map(|column| {
            format!(
                "{} {}{}",
                quote_ident(&column.name),
                sql_type(column.ty),
                if column.nullable { "" } else { " NOT NULL" }
            )
        })
        .collect::<Vec<_>>()
        .join(", ");
    let primary_key = table
        .primary_key
        .iter()
        .map(|name| quote_ident(name))
        .collect::<Vec<_>>()
        .join(", ");
    format!(
        "CREATE TABLE {} ({columns}, PRIMARY KEY ({primary_key}))",
        quote_ident(&table.name)
    )
}

fn bootstrap_table_ddl() -> String {
    format!(
        "CREATE TABLE {CDC_BOOTSTRAP_TABLE} (\
             singleton INTEGER PRIMARY KEY CHECK (singleton = 1),\
             format INTEGER NOT NULL CHECK (format = {BOOTSTRAP_FORMAT}),\
             state TEXT NOT NULL CHECK (state = '{BOOTSTRAP_STATE}'),\
             source TEXT NOT NULL, generation TEXT NOT NULL, descriptor_hash TEXT NOT NULL,\
             profile_fingerprint TEXT NOT NULL, schema_fingerprint TEXT NOT NULL,\
             topic_id TEXT NOT NULL, created_at_ms INTEGER NOT NULL\
         )"
    )
}

fn insert_sql(table: &InitialSnapshotTable) -> String {
    let columns = table
        .columns
        .iter()
        .map(|column| quote_ident(&column.name))
        .collect::<Vec<_>>()
        .join(", ");
    let parameters = (1..=table.columns.len())
        .map(|index| format!("?{index}"))
        .collect::<Vec<_>>()
        .join(", ");
    format!(
        "INSERT INTO {} ({columns}) VALUES ({parameters})",
        quote_ident(&table.name)
    )
}

fn sql_type(ty: ColType) -> &'static str {
    match ty {
        ColType::String => "TEXT",
        ColType::Number => "REAL",
        ColType::Boolean => "BOOLEAN",
        ColType::Json => "JSON",
        ColType::Int => "BIGINT",
    }
}

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

fn pragma_text(conn: &Connection, name: &'static str) -> Result<String, ReplicaError> {
    conn.query_row(&format!("PRAGMA {name}"), [], |row| row.get(0))
        .map_err(|error| ReplicaError::sqlite("initial snapshot PRAGMA read", error))
}

fn state_error(message: impl Into<String>) -> ReplicaError {
    ReplicaError::Open(message.into())
}