coven_database/store/store_session/

retained_replay.rs

//! Private accepted-history baselines and deterministic retained replay.

use crate::query_mapped_rows;
use crate::store::store_session::StoreRecords;
use std::collections::{BTreeMap, BTreeSet};

use rusqlite::{types::Value, Connection};
use serde::{Deserialize, Serialize};

use crate::{
    DbError, COVEN_INITIALIZED_STATE_KEY, COVEN_SCHEMA_MANIFEST_STATE_KEY,
    STORE_DEVICE_GENESIS_STATE_KEY, SYNC_ROUTING_CONTRACT_STATE_KEY, SYNC_ROUTING_HASH_STATE_KEY,
};
use coven_protocol::membership::OWNER_PUBKEY_STATE_KEY;
use coven_protocol::store_commit::{
    CommitFrontier, ObjectHash, RetainedReplaySnapshotAuthority, StoreBatchCommitRef,
    StoreDeviceRegistrationRef, StoreRootRef,
};

pub(crate) fn migrate_retained_replay_schema_on(
    conn: &Connection,
    store_dir: &coven_foundation::store_dir::StoreDir,
    policy: crate::CovenMigrationPolicy,
    migrations: &[crate::Migration],
    synced_tables: &[coven_protocol::synced_schema::SyncedTable],
) -> Result<(), crate::OpenError> {
    let records = StoreRecords::new(conn, store_dir);
    let Some(baseline) = load_replay_baseline_metadata_on(records)? else {
        return Ok(());
    };
    let mut image = Connection::open_in_memory().map_err(DbError::from)?;
    crate::connection_io::deserialize_database_image_into(
        &mut image,
        &baseline.image_bytes(conn, store_dir)?,
    )
    .map_err(|error| DbError::context("open retained replay database image", error))?;
    let routing = crate::database_open::load_coven_metadata(&image)?;
    let coven_schema_is_current = crate::database_open::initialized_coven_schema_is_current(
        &image,
        routing.has_scoped_graph(),
    )?;
    let host_schema_version = crate::ensure_schema_supported(&image, migrations)?;
    let target_host_schema_version = crate::supported_version(migrations);
    if coven_schema_is_current && host_schema_version == target_host_schema_version {
        return Ok(());
    }
    let had_schema_version = crate::get_protocol_state_on(
        &image,
        crate::coven_migration::COVEN_SCHEMA_VERSION_STATE_KEY,
    )?
    .is_some();
    let transaction = image.unchecked_transaction().map_err(DbError::from)?;
    crate::run_coven_migrations_in_transaction(&transaction, routing.has_scoped_graph(), policy)?;
    let migrated_host_schema_version =
        crate::run_migrations_in_transaction(&transaction, migrations)?;
    if !had_schema_version {
        crate::delete_protocol_state_on(
            &transaction,
            crate::coven_migration::COVEN_SCHEMA_VERSION_STATE_KEY,
        )?;
    }
    transaction.commit().map_err(DbError::from)?;
    let image_bytes = crate::connection_io::serialize_database_image(&image)?;
    let blob_decls = crate::BlobDecls::from_tables(&image, synced_tables).map_err(DbError::from)?;
    records.replace_retained_replay_image(
        &baseline,
        migrated_host_schema_version,
        &image_bytes,
        &blob_decls,
    )?;
    tracing::info!(
        previous_host_schema_version = baseline.schema_version,
        migrated_host_schema_version,
        coven_schema_migrated = !coven_schema_is_current,
        "Migrated retained replay baseline schema"
    );
    Ok(())
}

pub(crate) fn load_replay_baseline_on(
    records: StoreRecords<'_>,
) -> Result<Option<RetainedReplayBaseline>, DbError> {
    let Some(baseline) = load_replay_baseline_metadata_on(records)? else {
        return Ok(None);
    };
    records.validate_replay_baseline_image(&baseline)?;
    records.validate_replay_authority(&baseline)?;
    Ok(Some(baseline))
}

pub(super) fn load_replay_baseline_metadata_on(
    records: StoreRecords<'_>,
) -> Result<Option<RetainedReplayBaseline>, DbError> {
    let stored = records.retained_replay_baseline_row()?;
    let Some(stored) = stored else {
        return Ok(None);
    };
    let schema_version = u32::try_from(stored.schema_version)
        .map_err(|_| DbError::Message("retained replay schema version exceeds u32".to_string()))?;
    let authority_hash = stored
        .authority_hash
        .parse()
        .map_err(|error| DbError::context("retained replay authority hash", error))?;
    let authority_bytes = records
        .verified_payload(authority_hash)
        .map_err(|error| DbError::context("read retained replay authority", error))?;
    let authority: RetainedReplayAuthority = serde_json::from_slice(&authority_bytes)
        .map_err(|error| DbError::context("retained replay authority", error))?;
    if serde_json::to_vec(&authority)
        .map_err(|error| DbError::context("serialize retained replay authority", error))?
        != authority_bytes
    {
        return Err(DbError::Message(
            "retained replay baseline metadata is not canonical".to_string(),
        ));
    }
    Ok(Some(RetainedReplayBaseline {
        schema_version,
        routing_hash: stored
            .routing_hash
            .parse()
            .map_err(|error| DbError::context("retained replay routing hash", error))?,
        image_payload_hash: stored
            .image_payload_hash
            .parse()
            .map_err(|error| DbError::context("retained replay image hash", error))?,
        authority,
    }))
}

pub(crate) fn install_generation_zero_replay_baseline_on(
    records: StoreRecords<'_>,
    schema_version: u32,
    routing_hash: ObjectHash,
    authority: RetainedReplayGenesisAuthority,
) -> Result<RetainedReplayBaseline, DbError> {
    if load_replay_baseline_on(records)?.is_some() {
        return Err(DbError::Message(
            "retained replay baseline already exists before founder activation".to_string(),
        ));
    }
    records.install_generation_zero_replay_baseline(schema_version, routing_hash, authority)
}

pub(crate) fn install_snapshot_replay_baseline_on(
    records: StoreRecords<'_>,
    schema_version: u32,
    routing_hash: ObjectHash,
    authority: RetainedReplaySnapshotAuthority,
    blob_decls: &crate::BlobDecls,
) -> Result<RetainedReplayBaseline, DbError> {
    if load_replay_baseline_on(records)?.is_some() {
        return Err(DbError::Message(
            "retained replay baseline already exists before snapshot bootstrap".to_string(),
        ));
    }
    records.install_snapshot_replay_baseline(schema_version, routing_hash, authority, blob_decls)
}

pub(crate) fn ensure_founder_replay_baseline_on(
    records: StoreRecords<'_>,
    schema_version: u32,
    routing_hash: ObjectHash,
    authority: RetainedReplayGenesisAuthority,
) -> Result<RetainedReplayBaseline, DbError> {
    if let Some(existing) = load_replay_baseline_on(records)? {
        let authority_matches = match &existing.authority {
            RetainedReplayAuthority::Genesis(existing) => existing == &authority,
            RetainedReplayAuthority::InstalledSnapshot(existing) => {
                existing.store_root == authority.store_root
                    && existing.founder_registration == authority.founder_registration
            }
        };
        if existing.schema_version != schema_version {
            return Err(DbError::Message(format!(
                "retained replay baseline schema version {} differs from database schema version {schema_version}",
                existing.schema_version,
            )));
        }
        if existing.routing_hash != routing_hash {
            return Err(DbError::Message(format!(
                "retained replay baseline routing hash {} differs from database routing hash {routing_hash}",
                existing.routing_hash,
            )));
        }
        if !authority_matches {
            return Err(DbError::Message(format!(
                "retained replay baseline authority {:?} differs from installed founder authority {:?}",
                existing.authority, authority,
            )));
        }
        return Ok(existing);
    }
    let accepted_history = records.accepted_history_count()?;
    if accepted_history != 0 {
        return Err(DbError::Message(
            "accepted Store history exists without a retained replay baseline".to_string(),
        ));
    }
    install_generation_zero_replay_baseline_on(records, schema_version, routing_hash, authority)
}

const GENESIS_PRESERVED_TABLES: &[&str] = &[
    "protocol_state",
    "store_protocol_root_authority",
    "store_device_registration_activations",
];

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ReplayTableDisposition {
    Replace,
    ReplaceWhenRouting,
    Preserve,
    ExactTransition,
}

const REPLAY_TABLES: &[(&str, ReplayTableDisposition)] = &[
    ("active_store_publication", ReplayTableDisposition::Preserve),
    ("activated_circle_acks", ReplayTableDisposition::Replace),
    ("activated_store_acks", ReplayTableDisposition::Replace),
    // Blob provenance belongs to accepted object lifetime, including orphaned
    // ciphertext whose row version a replay omits. Exact activation and reclaim
    // transitions update this index alongside remote_objects.
    ("blob_locators", ReplayTableDisposition::ExactTransition),
    ("blob_make_remote_intents", ReplayTableDisposition::Preserve),
    ("circle_access_cache", ReplayTableDisposition::Replace),
    (
        "circle_bootstrap_coverage",
        ReplayTableDisposition::Preserve,
    ),
    ("circle_close_exclusions", ReplayTableDisposition::Preserve),
    (
        "circle_control_activations",
        ReplayTableDisposition::Replace,
    ),
    ("circle_current_state", ReplayTableDisposition::Replace),
    ("circle_operation_uploads", ReplayTableDisposition::Preserve),
    ("circle_operations", ReplayTableDisposition::Preserve),
    ("cloud_outbox", ReplayTableDisposition::Preserve),
    ("local_blob_refs", ReplayTableDisposition::Preserve),
    ("local_cleanup_intents", ReplayTableDisposition::Preserve),
    (
        "local_store_device_registration",
        ReplayTableDisposition::Preserve,
    ),
    (
        "local_owner_recovery_publication",
        ReplayTableDisposition::Preserve,
    ),
    (
        "local_store_founder_graph",
        ReplayTableDisposition::Preserve,
    ),
    (
        "local_store_protocol_root",
        ReplayTableDisposition::Preserve,
    ),
    // A commit's position is a fact accepted at commit time and is never rebuilt by a
    // projection replay. A filtered replay (one that omits covered or
    // beyond-cutoff Circle packages) would re-derive the commit's retained-input
    // hash from its filtered packages and drift from the preserved
    // `retained_merge_materializations` row, breaking the `materialized_commits`
    // foreign key.
    ("materialized_commits", ReplayTableDisposition::Preserve),
    (
        "outbound_membership_mutation",
        ReplayTableDisposition::Preserve,
    ),
    ("outbound_circle_acks", ReplayTableDisposition::Preserve),
    ("outbound_circle_snapshot", ReplayTableDisposition::Preserve),
    ("outbound_store_acks", ReplayTableDisposition::Preserve),
    (
        "outbound_store_device_exclusion",
        ReplayTableDisposition::Preserve,
    ),
    ("outbound_store_snapshot", ReplayTableDisposition::Preserve),
    ("payload_cleanup", ReplayTableDisposition::Preserve),
    ("payload_owners", ReplayTableDisposition::Preserve),
    ("payload_storage", ReplayTableDisposition::Preserve),
    (
        "protocol_inert_objects",
        ReplayTableDisposition::ExactTransition,
    ),
    ("protocol_state", ReplayTableDisposition::Preserve),
    (
        "published_blob_drop_intents",
        ReplayTableDisposition::Preserve,
    ),
    ("published_circle_acks", ReplayTableDisposition::Preserve),
    (
        "published_circle_snapshot",
        ReplayTableDisposition::Preserve,
    ),
    ("published_store_acks", ReplayTableDisposition::Preserve),
    ("published_store_snapshot", ReplayTableDisposition::Preserve),
    ("reclaimed_store_packages", ReplayTableDisposition::Preserve),
    ("remote_objects", ReplayTableDisposition::ExactTransition),
    (
        "retained_merge_materializations",
        ReplayTableDisposition::Preserve,
    ),
    (
        "retained_replay_baselines",
        ReplayTableDisposition::Preserve,
    ),
    (
        "retained_replay_blob_leases",
        ReplayTableDisposition::Preserve,
    ),
    ("retained_replay_objects", ReplayTableDisposition::Preserve),
    ("row_blob_locators", ReplayTableDisposition::Replace),
    ("snapshot_coverage", ReplayTableDisposition::Preserve),
    (
        "store_author_exclusion_activations",
        ReplayTableDisposition::Replace,
    ),
    (
        "store_device_registration_activations",
        ReplayTableDisposition::Replace,
    ),
    // Preserved for the same reason as `materialized_commits`: a commit's
    // derived device state is accepted at commit time, never recomputed by a
    // filtered replay; only explicit retraction removes it.
    ("store_device_states", ReplayTableDisposition::Preserve),
    (
        "store_device_state_snapshots",
        ReplayTableDisposition::Preserve,
    ),
    (
        "store_protocol_root_authority",
        ReplayTableDisposition::Preserve,
    ),
    (
        "store_publication_current",
        ReplayTableDisposition::Preserve,
    ),
    (
        "store_publication_entries",
        ReplayTableDisposition::Preserve,
    ),
    ("store_reclaim_operations", ReplayTableDisposition::Preserve),
    ("store_write_blob_leases", ReplayTableDisposition::Preserve),
    ("store_write_blobs", ReplayTableDisposition::Preserve),
    ("store_write_packages", ReplayTableDisposition::Preserve),
    ("store_write_partitions", ReplayTableDisposition::Preserve),
    ("store_writes", ReplayTableDisposition::Preserve),
    ("stream_activations", ReplayTableDisposition::Replace),
    (
        "_coven_audience",
        ReplayTableDisposition::ReplaceWhenRouting,
    ),
    (
        "_coven_row_routes",
        ReplayTableDisposition::ReplaceWhenRouting,
    ),
];

pub fn projection_table_names(include_routing: bool) -> Vec<String> {
    REPLAY_TABLES
        .iter()
        .filter_map(|(table, disposition)| match disposition {
            ReplayTableDisposition::Replace => Some((*table).to_string()),
            ReplayTableDisposition::ReplaceWhenRouting if include_routing => {
                Some((*table).to_string())
            }
            ReplayTableDisposition::ReplaceWhenRouting
            | ReplayTableDisposition::Preserve
            | ReplayTableDisposition::ExactTransition => None,
        })
        .collect()
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RetainedReplayGenesisAuthority {
    pub store_root: StoreRootRef,
    pub founder_registration: StoreDeviceRegistrationRef,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case", deny_unknown_fields)]
pub enum RetainedReplayAuthority {
    Genesis(RetainedReplayGenesisAuthority),
    /// The device installed a snapshot image and replays forward from it. Named
    /// for what happened, not for a property of the snapshot: installing one
    /// never required the store's other devices to have acknowledged it.
    InstalledSnapshot(RetainedReplaySnapshotAuthority),
}

/// The database image a replay starts from, and the authority that says which
/// history it covers.
///
/// `image_payload_hash` names the image bytes in the payload store, as
/// `authority_hash` on the row names the authority's canonical bytes. The row
/// holds the facts without carrying either payload in its own columns.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RetainedReplayBaseline {
    pub schema_version: u32,
    pub routing_hash: ObjectHash,
    pub image_payload_hash: ObjectHash,
    pub authority: RetainedReplayAuthority,
}

impl RetainedReplayBaseline {
    /// Accepted history covered by the authority this image starts from.
    pub fn coverage(&self) -> &CommitFrontier {
        match &self.authority {
            RetainedReplayAuthority::Genesis(_) => {
                static EMPTY: CommitFrontier = CommitFrontier(BTreeMap::new());
                &EMPTY
            }
            RetainedReplayAuthority::InstalledSnapshot(authority) => &authority.metadata.coverage,
        }
    }

    pub fn canonical_authority_bytes(&self) -> Result<Vec<u8>, DbError> {
        serde_json::to_vec(&self.authority)
            .map_err(|error| DbError::context("serialize retained replay authority", error))
    }

    /// The verified bytes of the baseline image. Replays open a private,
    /// writable connection from these bytes inside the replay capability.
    pub(super) fn image_bytes(
        &self,
        conn: &Connection,
        store_dir: &coven_foundation::store_dir::StoreDir,
    ) -> Result<Vec<u8>, DbError> {
        crate::payload_store::read_verified_payload_blocking(
            conn,
            store_dir,
            self.image_payload_hash,
        )
        .map_err(|error| DbError::context("read retained replay image", error))
    }

    pub(crate) fn validate_image(
        &self,
        conn: &Connection,
        store_dir: &coven_foundation::store_dir::StoreDir,
    ) -> Result<(), DbError> {
        let mut image = Connection::open_in_memory().map_err(DbError::from)?;
        crate::connection_io::deserialize_database_image_into(
            &mut image,
            &self.image_bytes(conn, store_dir)?,
        )
        .map_err(|error| DbError::context("open retained replay database image", error))?;
        self.validate_open_image(&image, store_dir)
    }

    pub(super) fn validate_open_image(
        &self,
        image: &Connection,
        store_dir: &coven_foundation::store_dir::StoreDir,
    ) -> Result<(), DbError> {
        match &self.authority {
            RetainedReplayAuthority::Genesis(_) => {
                self.validate_image_metadata(image)?;
                let protocol_keys = protocol_state_keys(image)?;
                let founder_membership_cursor = founder_membership_cursor_key(image)?;
                if protocol_keys
                    .iter()
                    .any(|key| !generation_zero_protocol_key(&founder_membership_cursor, key))
                    || !required_generation_zero_protocol_keys()
                        .iter()
                        .all(|key| protocol_keys.contains(*key))
                    || !protocol_keys.contains(&founder_membership_cursor)
                {
                    return Err(DbError::Message(
                        "retained replay image protocol state is not the generation-zero set"
                            .to_string(),
                    ));
                }
                for table in crate::user_table_names(image).map_err(DbError::from)? {
                    let count: i64 = image
                        .query_row(
                            &format!("SELECT COUNT(*) FROM {}", crate::quote_ident(&table)),
                            [],
                            |row| row.get(0),
                        )
                        .map_err(DbError::from)?;
                    let expected = match table.as_str() {
                        "protocol_state" => None,
                        "store_protocol_root_authority"
                        | "store_device_registration_activations" => Some(1),
                        _ => Some(0),
                    };
                    if expected.is_some_and(|expected| count != expected) {
                        return Err(DbError::Message(format!(
                            "generation-zero retained replay image table {table:?} has {count} rows"
                        )));
                    }
                }
                validate_replay_image_foreign_keys(image)?;
            }
            RetainedReplayAuthority::InstalledSnapshot(authority) => {
                authority.validate()?;
                self.validate_image_metadata(image)?;
                let mut actual = BTreeMap::new();
                let rows = query_mapped_rows(
                    image,
                    "SELECT device_id, seq, commit_ref FROM snapshot_coverage ORDER BY device_id",
                    [],
                    |row| {
                        Ok((
                            row.get::<_, String>(0)?,
                            row.get::<_, i64>(1)?,
                            row.get::<_, String>(2)?,
                        ))
                    },
                )?;
                for (stream_id, sequence, encoded) in rows {
                    let reference: StoreBatchCommitRef =
                        serde_json::from_str(&encoded).map_err(|error| {
                            DbError::context("snapshot replay coverage reference", error)
                        })?;
                    if sequence < 0
                        || u64::try_from(sequence).ok() != Some(reference.coord.sequence())
                    {
                        return Err(DbError::Message(
                            "snapshot replay coverage sequence differs from its exact reference"
                                .to_string(),
                        ));
                    }
                    if actual.insert(stream_id, reference).is_some() {
                        return Err(DbError::Message(
                            "snapshot replay coverage repeats a Store stream".to_string(),
                        ));
                    }
                }
                if actual != self.coverage().clone().into_refs() {
                    return Err(DbError::Message(
                        "snapshot replay image coverage differs from its baseline".to_string(),
                    ));
                }
                let materialized_commits: i64 = image
                    .query_row("SELECT COUNT(*) FROM materialized_commits", [], |row| {
                        row.get(0)
                    })
                    .map_err(DbError::from)?;
                if materialized_commits != 0 {
                    return Err(DbError::Message(
                        "snapshot replay baseline contains materialized_commits rows".to_string(),
                    ));
                }
                let mut verified_authority =
                    super::VerifiedStoreAuthority::for_replay_baseline(self.clone());
                crate::StoreDatabase::validate_snapshot_retained_inputs_on(
                    StoreRecords::new(image, store_dir),
                    &mut verified_authority,
                    &authority.store_root,
                    self.coverage(),
                )?;
                validate_replay_image_foreign_keys(image)?;
            }
        }
        let routing = crate::database_open::load_coven_metadata(image)?;
        if routing.hash() != self.routing_hash {
            return Err(DbError::Message(
                "retained replay image routing contract differs from its baseline".to_string(),
            ));
        }
        crate::database_open::validate_initialized_coven_schema(image, routing.has_scoped_graph())?;
        crate::store_authority_records::validate_replay_authority_on(image, self)
    }

    fn validate_image_metadata(&self, image: &Connection) -> Result<(), DbError> {
        let stored_schema_version: u32 = image
            .pragma_query_value(None, "user_version", |row| row.get(0))
            .map_err(DbError::from)?;
        if stored_schema_version != self.schema_version {
            return Err(DbError::Message(format!(
                "retained replay image schema version is {stored_schema_version}, expected {}",
                self.schema_version
            )));
        }
        let stored_routing_hash =
            crate::required_protocol_state_on(image, SYNC_ROUTING_HASH_STATE_KEY)?;
        if stored_routing_hash != self.routing_hash.to_string() {
            return Err(DbError::Message(
                "retained replay image routing hash differs from its baseline".to_string(),
            ));
        }
        Ok(())
    }
}

pub(super) struct GenerationZeroReplayImage {
    image: Connection,
}

impl GenerationZeroReplayImage {
    pub(super) fn database_bytes(&self) -> Result<Vec<u8>, DbError> {
        crate::connection_io::serialize_database_image(&self.image)
    }

    pub(super) fn validate(
        &self,
        baseline: &RetainedReplayBaseline,
        store_dir: &coven_foundation::store_dir::StoreDir,
    ) -> Result<(), DbError> {
        baseline.validate_open_image(&self.image, store_dir)
    }
}

/// Copy `table` from `source` into `target`. With `ignore_existing`, a row whose
/// unique key already exists in `target` is left untouched instead of failing —
/// used when installing a Circle image onto a Store image that already carries the
/// shared, deterministic audience-routing rows.
pub(crate) fn copy_table_with_conflicts(
    source: &Connection,
    target: &Connection,
    table: &str,
    ignore_existing: bool,
) -> Result<(), DbError> {
    let pragma = format!("PRAGMA table_info({})", crate::quote_ident(table));
    let columns = query_mapped_rows(source, &pragma, [], |row| row.get::<_, String>(1))?;
    if columns.is_empty() {
        return Err(DbError::Message(format!(
            "retained replay projection table {table:?} is absent"
        )));
    }
    let quoted_columns = columns
        .iter()
        .map(|column| crate::quote_ident(column))
        .collect::<Vec<_>>()
        .join(", ");
    let select = format!("SELECT {quoted_columns} FROM {}", crate::quote_ident(table));
    let rows = query_mapped_rows(source, &select, [], |row| {
        (0..columns.len())
            .map(|index| row.get::<_, Value>(index))
            .collect::<rusqlite::Result<Vec<_>>>()
    })?;
    let placeholders = (1..=columns.len())
        .map(|index| format!("?{index}"))
        .collect::<Vec<_>>()
        .join(", ");
    let verb = if ignore_existing {
        "INSERT OR IGNORE"
    } else {
        "INSERT"
    };
    let insert = format!(
        "{verb} INTO {} ({quoted_columns}) VALUES ({placeholders})",
        crate::quote_ident(table)
    );
    for values in rows {
        target
            .execute(&insert, rusqlite::params_from_iter(values))
            .map_err(DbError::from)?;
    }
    Ok(())
}

pub(super) fn project_generation_zero_image(
    source: &Connection,
) -> Result<GenerationZeroReplayImage, DbError> {
    let source_bytes = crate::connection_io::serialize_database_image(source)?;
    let mut image = Connection::open_in_memory().map_err(DbError::from)?;
    crate::connection_io::deserialize_database_image_into(&mut image, &source_bytes)
        .map_err(|error| DbError::context("open retained replay database image", error))?;
    image
        .pragma_update(None, "foreign_keys", "OFF")
        .map_err(DbError::from)?;
    let transaction = image.unchecked_transaction().map_err(DbError::from)?;
    let founder_membership_cursor = founder_membership_cursor_key(&transaction)?;
    for table in crate::user_table_names(&transaction).map_err(DbError::from)? {
        if GENESIS_PRESERVED_TABLES.contains(&table.as_str()) {
            continue;
        }
        transaction
            .execute_batch(&format!("DELETE FROM {}", crate::quote_ident(&table)))
            .map_err(DbError::from)?;
    }
    let protocol_keys = protocol_state_keys(&transaction)?;
    for key in protocol_keys {
        if !generation_zero_protocol_key(&founder_membership_cursor, &key) {
            crate::delete_protocol_state_on(&transaction, &key)?;
        }
    }
    transaction
        .execute("DELETE FROM sqlite_sequence", [])
        .map_err(DbError::from)?;
    transaction.commit().map_err(DbError::from)?;
    image.execute_batch("VACUUM").map_err(DbError::from)?;
    image
        .pragma_update(None, "foreign_keys", "ON")
        .map_err(DbError::from)?;
    let violations: bool = image
        .query_row(
            "SELECT EXISTS(SELECT 1 FROM pragma_foreign_key_check)",
            [],
            |row| row.get(0),
        )
        .map_err(DbError::from)?;
    if violations {
        return Err(DbError::Message(
            "generation-zero retained replay image violates foreign keys".to_string(),
        ));
    }
    Ok(GenerationZeroReplayImage { image })
}

fn validate_replay_image_foreign_keys(image: &Connection) -> Result<(), DbError> {
    let violations: bool = image
        .query_row(
            "SELECT EXISTS(SELECT 1 FROM pragma_foreign_key_check)",
            [],
            |row| row.get(0),
        )
        .map_err(DbError::from)?;
    if violations {
        return Err(DbError::Message(
            "retained replay image violates foreign keys".to_string(),
        ));
    }
    Ok(())
}

fn protocol_state_keys(connection: &Connection) -> Result<BTreeSet<String>, DbError> {
    let mut statement = connection
        .prepare("SELECT key FROM protocol_state ORDER BY key")
        .map_err(DbError::from)?;
    let keys = statement
        .query_map([], |row| row.get::<_, String>(0))
        .map_err(DbError::from)?
        .collect::<rusqlite::Result<BTreeSet<_>>>()
        .map_err(DbError::from)?;
    Ok(keys)
}

fn required_generation_zero_protocol_keys() -> &'static [&'static str] {
    &[
        COVEN_INITIALIZED_STATE_KEY,
        COVEN_SCHEMA_MANIFEST_STATE_KEY,
        OWNER_PUBKEY_STATE_KEY,
        STORE_DEVICE_GENESIS_STATE_KEY,
        SYNC_ROUTING_CONTRACT_STATE_KEY,
        SYNC_ROUTING_HASH_STATE_KEY,
    ]
}

fn founder_membership_cursor_key(connection: &Connection) -> Result<String, DbError> {
    let bytes: Vec<u8> = connection
        .query_row(
            "SELECT store_protocol_root_bytes
             FROM store_protocol_root_authority WHERE singleton = 1",
            [],
            |row| row.get(0),
        )
        .map_err(DbError::from)?;
    let root = coven_protocol::store_commit::StoreProtocolRoot::parse(&bytes)
        .map_err(|error| DbError::context("retained replay Store root", error))?;
    let stream = coven_protocol::membership::derive_founder_stream_id(
        &root.descriptor.store_root_id().to_string(),
        &root.descriptor.founder_pubkey,
    );
    Ok(
        crate::InitialStoreMembershipAuthority::cursor_state_key_for_stream(
            &root.descriptor.founder_grant,
            stream,
        ),
    )
}

fn generation_zero_protocol_key(founder_membership_cursor: &str, key: &str) -> bool {
    required_generation_zero_protocol_keys().contains(&key) || founder_membership_cursor == key
}

#[cfg(test)]
#[path = "retained_replay_test.rs"]
mod tests;