miden-validator 0.17.3

use std::io;
use std::num::NonZeroUsize;
use std::path::{Path, PathBuf};

use miden_node_db::DatabaseError;
use miden_node_db::sqlite::{DbReader, DbWriter};
use miden_node_tracing::{info, miden_instrument};
use miden_protocol::block::{BlockHeader, BlockNumber};
use miden_protocol::protocol_config::ProtocolConfig;
use miden_protocol::transaction::TransactionId;

use crate::db::migrations::{bootstrap_database, migrate_database, verify_latest_schema};
use crate::metrics::InitialMetrics;
use crate::{COMPONENT, LOG_TARGET, StorageKeyEpoch, StoredPrivateRecord};

mod migrations;
mod queries;

pub(crate) use queries::{ListTransactionsParams, ListedTransaction};

// VALIDATOR DATABASE
// ================================================================================================

/// Read-only handle to the validator database.
///
/// Wraps the framework [`DbReader`] and exposes every read query as a method. Cloneable, and handed
/// to read-only components (the administration API); it has no write methods, so those components
/// cannot mutate the database.
#[derive(Clone)]
pub struct ValidatorDbReader {
    reader: DbReader,
}

impl ValidatorDbReader {
    /// Returns whether a transaction with the given id has already been validated.
    pub(crate) async fn transaction_exists(
        &self,
        tx_id: TransactionId,
    ) -> Result<bool, DatabaseError> {
        self.reader
            .read("transaction_exists", move |tx| queries::transaction_exists(tx, tx_id))
            .await
    }

    /// Returns the subset of `tx_ids` that this validator has not validated yet.
    ///
    /// An empty result means all supplied transaction ids have been validated in the past.
    pub(crate) async fn find_unvalidated_transactions(
        &self,
        tx_ids: Vec<TransactionId>,
    ) -> Result<Vec<TransactionId>, DatabaseError> {
        self.reader
            .read("find_unvalidated_transactions", move |tx| {
                queries::find_unvalidated_transactions(tx, &tx_ids)
            })
            .await
    }

    /// Loads the chain tip, or `None` if no block header has been persisted yet (i.e. bootstrap has
    /// not been run).
    #[miden_instrument(
        target = COMPONENT,
    )]
    pub(crate) async fn load_chain_tip(&self) -> Result<Option<BlockHeader>, DatabaseError> {
        self.reader.read("load_chain_tip", queries::load_chain_tip).await
    }

    /// Loads the block header at the given height, or `None` if no block header is stored there.
    pub(crate) async fn load_block_header(
        &self,
        block_num: BlockNumber,
    ) -> Result<Option<BlockHeader>, DatabaseError> {
        self.reader
            .read("load_block_header", move |tx| queries::load_block_header(tx, block_num))
            .await
    }

    /// Loads the protocol configuration with the given commitment.
    pub async fn load_protocol_config(
        &self,
        commitment: miden_protocol::Word,
    ) -> Result<Option<ProtocolConfig>, DatabaseError> {
        self.reader
            .read("load_protocol_config", move |tx| queries::load_protocol_config(tx, commitment))
            .await
    }

    /// Reads the values the server's in-memory counters start from, all within a single read
    /// transaction so that they describe one consistent database state.
    pub(crate) async fn load_initial_metrics(&self) -> Result<InitialMetrics, DatabaseError> {
        self.reader
            .read("load_initial_metrics", |tx| {
                Ok(InitialMetrics {
                    chain_tip: queries::load_chain_tip(tx)?
                        .map_or(0, |header| header.block_num().as_u32()),
                    validated_transactions: u64::try_from(queries::count_validated_transactions(
                        tx,
                    )?)
                    .unwrap_or(0),
                    signed_blocks: u64::try_from(queries::count_signed_blocks(tx)?).unwrap_or(0),
                })
            })
            .await
    }

    /// Returns the total number of validated transactions.
    ///
    /// Production code seeds its counter from [`Self::load_initial_metrics`] and tracks it in
    /// memory from there, so this standalone count only backs test assertions about what was
    /// actually persisted.
    #[cfg(test)]
    pub(crate) async fn count_validated_transactions(&self) -> Result<i64, DatabaseError> {
        self.reader
            .read("count_validated_transactions", queries::count_validated_transactions)
            .await
    }

    /// Loads one encrypted private record by transaction id.
    pub async fn load_private_record(
        &self,
        transaction_id: TransactionId,
    ) -> Result<Option<StoredPrivateRecord>, DatabaseError> {
        self.reader
            .read("load_private_record", move |tx| {
                queries::load_private_record(tx, transaction_id)
            })
            .await
    }

    /// Loads the encrypted private records sealed under one storage key epoch.
    pub async fn load_private_records_by_key_epoch(
        &self,
        key_epoch: StorageKeyEpoch,
    ) -> Result<Vec<StoredPrivateRecord>, DatabaseError> {
        self.reader
            .read("load_private_records_by_key_epoch", move |tx| {
                queries::load_private_records_by_key_epoch(tx, key_epoch)
            })
            .await
    }

    /// Loads the encrypted private records belonging to one Golden setup context.
    pub async fn load_private_records_by_setup_context(
        &self,
        setup_context_id: [u8; 32],
    ) -> Result<Vec<StoredPrivateRecord>, DatabaseError> {
        self.reader
            .read("load_private_records_by_setup_context", move |tx| {
                queries::load_private_records_by_setup_context(tx, setup_context_id)
            })
            .await
    }

    /// Loads one page of committed transactions in chronological order i.e. `(block_num,
    /// block_tx_index)`, with the chain tip and optional records from the same snapshot.
    pub(crate) async fn list_validated_transactions(
        &self,
        params: queries::ListTransactionsParams,
    ) -> Result<queries::ListedTransactionsPage, DatabaseError> {
        self.reader
            .read("list_validated_transactions", move |tx| {
                queries::list_validated_transactions(tx, &params)
            })
            .await
    }
}

/// Write handle to the validator database.
///
/// Wraps the framework [`DbWriter`] and additionally holds a [`ValidatorDbReader`], so it exposes
/// the write queries directly and every read query through `Deref`. **Not `Clone`**: writes have a
/// single owner, matching SQLite's single-writer model.
pub struct ValidatorDbWriter {
    writer: DbWriter,
    reader: ValidatorDbReader,
}

impl std::ops::Deref for ValidatorDbWriter {
    type Target = ValidatorDbReader;

    fn deref(&self) -> &Self::Target {
        &self.reader
    }
}

impl ValidatorDbWriter {
    /// Returns a read-only handle onto the same connection pool, for handing to components that
    /// must not be able to write.
    pub fn reader(&self) -> ValidatorDbReader {
        self.reader.clone()
    }

    /// Inserts a validated transaction and its encrypted private record, returning the number of
    /// inserted rows. The count is zero if the transaction was already recorded.
    #[miden_instrument(
        target = COMPONENT,
    )]
    pub async fn insert_validated_private_transaction(
        &self,
        record: StoredPrivateRecord,
    ) -> Result<usize, DatabaseError> {
        self.writer
            .write("insert_validated_private_transaction", move |tx| {
                queries::insert_validated_private_transaction(tx, &record)
            })
            .await
    }

    /// Persists a block header and its configuration activation in one transaction.
    ///
    /// See [`record_protocol_config_activation`] for how the activation is recorded.
    /// Callers must validate block order before this method runs.
    ///
    /// The write replaces the header row. If the height holds a block, the `ON DELETE CASCADE` on
    /// `block_transactions` deletes the links of that block, and this method does not link the
    /// transactions of the new block. Only tests use this method. Server code uses
    /// [`Self::insert_signed_block`] and [`Self::replace_signed_block`].
    #[cfg(test)]
    #[miden_instrument(
        target = COMPONENT,
    )]
    pub(crate) async fn upsert_block_header_with_protocol_config(
        &self,
        header: BlockHeader,
        protocol_config: Option<ProtocolConfig>,
    ) -> Result<(), DatabaseError> {
        self.writer
            .write("upsert_block_header_with_protocol_config", move |tx| {
                record_protocol_config_activation(tx, &header, protocol_config)?;
                queries::upsert_block_header(tx, &header)
            })
            .await
    }

    /// Persists a signed block's header, its configuration activation, and the links from the
    /// block's transactions to their in-block positions, all in one database transaction so the
    /// three stay consistent.
    ///
    /// The height must not already hold a block; use [`Self::replace_signed_block`] to replace
    /// one.
    #[miden_instrument(
        target = COMPONENT,
    )]
    pub(crate) async fn insert_signed_block(
        &self,
        header: BlockHeader,
        protocol_config: ProtocolConfig,
        transactions: Vec<TransactionId>,
    ) -> Result<(), DatabaseError> {
        self.writer
            .write("insert_signed_block", move |tx| {
                persist_signed_block(tx, &header, protocol_config, &transactions)
            })
            .await
    }

    /// Replaces the block signed at `header`'s height, in one database transaction: deletes the
    /// replaced block — unlinking its transactions via the `ON DELETE CASCADE` on
    /// `block_transactions`, so transactions dropped by the replacement do not keep a stale link —
    /// then persists the new header and links exactly as [`Self::insert_signed_block`] does.
    #[miden_instrument(
        target = COMPONENT,
    )]
    pub(crate) async fn replace_signed_block(
        &self,
        header: BlockHeader,
        protocol_config: ProtocolConfig,
        transactions: Vec<TransactionId>,
    ) -> Result<(), DatabaseError> {
        self.writer
            .write("replace_signed_block", move |tx| {
                queries::delete_block(tx, header.block_num())?;
                persist_signed_block(tx, &header, protocol_config, &transactions)
            })
            .await
    }
}

/// Persists a signed block's header, records its configuration activation, and links its
/// transactions, within the caller's database transaction: the shared tail of
/// [`ValidatorDbWriter::insert_signed_block`] and [`ValidatorDbWriter::replace_signed_block`].
fn persist_signed_block(
    tx: &miden_node_db::sqlite::WriteTx<'_>,
    header: &BlockHeader,
    protocol_config: ProtocolConfig,
    transactions: &[TransactionId],
) -> Result<(), DatabaseError> {
    record_protocol_config_activation(tx, header, Some(protocol_config))?;
    queries::insert_block_header(tx, header)?;
    queries::link_block_transactions(tx, header.block_num(), transactions)
}

/// Records the configuration activation for a block, within the caller's database transaction.
///
/// An activation is recorded only if the configuration differs from the preceding activation.
/// A replacement at the current tip therefore retains its active configuration.
///
/// If `protocol_config` is absent, the configuration must already be stored, otherwise an error is
/// returned.
fn record_protocol_config_activation(
    tx: &miden_node_db::sqlite::WriteTx<'_>,
    header: &BlockHeader,
    protocol_config: Option<ProtocolConfig>,
) -> Result<(), DatabaseError> {
    let commitment = header.protocol_config_commitment();
    let config = if let Some(config) = protocol_config {
        let calculated = config.to_commitment();
        if calculated != commitment {
            return Err(invalid_protocol_config(format!(
                "protocol config commitment mismatch: expected {commitment}, got {calculated}"
            )));
        }
        config
    } else {
        queries::load_protocol_config(tx, commitment)?.ok_or_else(|| {
            invalid_protocol_config(format!("protocol config {commitment} is not stored"))
        })?
    };

    let block_number = header.block_num();
    let previous = queries::load_protocol_config_commitment_before(tx, block_number)?;
    if previous != Some(commitment) {
        queries::insert_protocol_config(tx, &config, block_number)?;
    }

    Ok(())
}

fn invalid_protocol_config(message: String) -> DatabaseError {
    DatabaseError::deserialization(
        "ProtocolConfig",
        io::Error::new(io::ErrorKind::InvalidData, message),
    )
}

/// Deletes a stored protocol configuration for a test.
#[cfg(test)]
pub(crate) async fn delete_protocol_config_for_test(
    db: &ValidatorDbWriter,
    commitment: miden_protocol::Word,
) -> Result<(), DatabaseError> {
    db.writer
        .write("delete_protocol_config_for_test", move |tx| {
            tx.execute("DELETE FROM protocol_configs WHERE commitment = ?1", &[&commitment])?;
            Ok::<_, DatabaseError>(())
        })
        .await
}

// LIFECYCLE
// ================================================================================================

/// Opens a connection pool after verifying that the database is at the latest schema version.
#[miden_instrument(
    target = COMPONENT,
)]
pub async fn load(database_filepath: PathBuf) -> Result<ValidatorDbWriter, DatabaseError> {
    load_with_pool_size(database_filepath, miden_node_db::default_connection_pool_size()).await
}

/// Opens a connection pool with a specific pool size after verifying that the database is at the
/// latest schema version.
#[miden_instrument(
    target = COMPONENT,
)]
pub async fn load_with_pool_size(
    database_filepath: PathBuf,
    connection_pool_size: NonZeroUsize,
) -> Result<ValidatorDbWriter, DatabaseError> {
    verify_latest_schema(&database_filepath)?;

    open_with_pool_size(&database_filepath, connection_pool_size)
}

/// Creates a new database, applies all migrations, and opens a connection pool.
#[miden_instrument(
    target = COMPONENT,
)]
pub async fn setup(database_filepath: PathBuf) -> Result<ValidatorDbWriter, DatabaseError> {
    setup_with_pool_size(database_filepath, miden_node_db::default_connection_pool_size()).await
}

/// Creates a new database with a specific pool size and applies all migrations.
#[miden_instrument(
    target = COMPONENT,
)]
async fn setup_with_pool_size(
    database_filepath: PathBuf,
    connection_pool_size: NonZeroUsize,
) -> Result<ValidatorDbWriter, DatabaseError> {
    bootstrap_database(&database_filepath)?;

    open_with_pool_size(&database_filepath, connection_pool_size)
}

/// Creates and initializes the database, then seeds it with the genesis block header as the chain
/// tip.
///
/// Returns an error if the database has already been bootstrapped.
#[miden_instrument(
    target = COMPONENT,
    fields(path = database_filepath),
    err,
)]
pub async fn bootstrap(
    database_filepath: PathBuf,
    connection_pool_size: NonZeroUsize,
    genesis_header: BlockHeader,
    protocol_config: ProtocolConfig,
) -> Result<(), DatabaseError> {
    let db = setup_with_pool_size(database_filepath, connection_pool_size).await?;

    db.insert_signed_block(genesis_header, protocol_config, Vec::new()).await
}

/// Applies all pending migrations to an existing DB.
#[miden_instrument(
    target = COMPONENT,
)]
pub fn migrate(database_filepath: impl AsRef<Path>) -> Result<(), DatabaseError> {
    migrate_database(database_filepath.as_ref())?;
    Ok(())
}

fn open_with_pool_size(
    database_filepath: &Path,
    connection_pool_size: NonZeroUsize,
) -> Result<ValidatorDbWriter, DatabaseError> {
    let (writer, reader) =
        miden_node_db::sqlite::open_with_pool_size(database_filepath, connection_pool_size)?;
    info!(
        target: LOG_TARGET,
        "Connected to the database",
        path = database_filepath,
        db.sqlite.connection_pool_size = connection_pool_size.get()
    );
    Ok(ValidatorDbWriter {
        writer,
        reader: ValidatorDbReader { reader },
    })
}

#[cfg(test)]
mod tests {
    mod protocol_config_history;

    use miden_node_utils::fee::{test_fee_params, test_protocol_config};
    use miden_protocol::Word;
    use miden_protocol::asset::AssetId;
    use miden_protocol::block::{BlockHeader, ValidatorConfig};
    use miden_protocol::crypto::dsa::ecdsa_k256_keccak::SigningKey;
    use miden_protocol::protocol_config::ProtocolConfig;
    use miden_protocol::testing::account_id::ACCOUNT_ID_PUBLIC_FUNGIBLE_FAUCET_1;
    use miden_protocol::utils::serde::Deserializable;
    use rand_chacha_03::ChaCha20Rng;
    use rand_chacha_03::rand_core::SeedableRng;

    use super::*;
    use crate::private_record::test_private_record_sealer;
    use crate::storage_key::tests::operator_keys;
    use crate::{
        PrivateRecordChainId,
        PrivateRecordCombiner,
        PrivateRecordContext,
        PrivateRecordError,
        PrivateRecordFormatVersion,
        PrivateRecordId,
        PrivateRecordSealer,
        PrivateRecordShareRequest,
    };

    const CHAIN_ID: PrivateRecordChainId = PrivateRecordChainId::new([1; 32]);
    const KEY_EPOCH: StorageKeyEpoch = StorageKeyEpoch::new([2; 32]);
    const SETUP_CONTEXT_ID: [u8; 32] = [4; 32];

    fn record_id(transaction_id: TransactionId) -> PrivateRecordId {
        let signer = SigningKey::read_from_bytes(&[7; 32]).unwrap();
        PrivateRecordId::new(transaction_id, &signer.public_key())
    }

    fn private_record(transaction_id: TransactionId, seed: u8) -> StoredPrivateRecord {
        let context = PrivateRecordContext::new(CHAIN_ID, KEY_EPOCH, transaction_id);
        let mut rng = ChaCha20Rng::from_seed([seed; 32]);
        test_private_record_sealer(KEY_EPOCH, SETUP_CONTEXT_ID)
            .seal(&mut rng, record_id(transaction_id), context, b"private transaction inputs")
            .unwrap()
    }

    fn genesis_header(config: &miden_protocol::protocol_config::ProtocolConfig) -> BlockHeader {
        miden_node_store::GenesisState::new(
            vec![],
            test_fee_params(),
            0,
            ValidatorConfig::new(vec![SigningKey::new().public_key()], 1).unwrap(),
            config.clone(),
        )
        .into_block()
        .unwrap()
        .inner()
        .header()
        .clone()
    }

    fn header_with_next_timestamp(header: &BlockHeader) -> BlockHeader {
        BlockHeader::new(
            header.prev_block_commitment(),
            header.block_num(),
            header.chain_commitment(),
            header.account_root(),
            header.nullifier_root(),
            header.note_root(),
            header.tx_commitment(),
            header.validator_config().clone(),
            header.fee_parameters().clone(),
            header.protocol_config_commitment(),
            header.next_protocol_config().cloned(),
            header.timestamp() + 1,
        )
    }

    #[test]
    fn migrate_rejects_missing_database() {
        let temp_dir = tempfile::tempdir().expect("failed to create temp directory");
        let db_path = temp_dir.path().join("validator.sqlite3");

        let err = migrate(db_path.clone()).expect_err("missing database should fail");

        assert!(matches!(err, DatabaseError::Migration(_)), "unexpected error: {err:?}");
        assert!(!db_path.exists());
    }

    /// The protocol configuration migration must preserve headers and private records.
    #[tokio::test]
    async fn migration_preserves_headers_and_private_records() {
        let temp_dir = tempfile::tempdir().unwrap();
        let db_path = temp_dir.path().join("validator.sqlite3");
        miden_node_db::migration::Migrator::builder()
            .unwrap()
            .push_sql("001_initial", include_str!("migrations/001_initial.sql"))
            .unwrap()
            .build()
            .unwrap()
            .bootstrap(&db_path)
            .unwrap();

        let config = test_protocol_config();
        let header = genesis_header(&config);
        let transaction_id = TransactionId::from_raw(Word::from([1u32, 2, 3, 4]));
        let record = private_record(transaction_id, 1);
        let db = open_with_pool_size(&db_path, NonZeroUsize::new(2).unwrap()).unwrap();
        let stored_header = header.clone();
        db.writer
            .write("seed legacy header", move |tx| queries::upsert_block_header(tx, &stored_header))
            .await
            .unwrap();
        db.insert_validated_private_transaction(record.clone()).await.unwrap();
        drop(db);

        assert!(load(db_path.clone()).await.is_err(), "the old schema requires migration");
        migrate(&db_path).unwrap();
        migrate(&db_path).expect("migration should also accept the latest schema");

        let db = load(db_path).await.unwrap();
        assert_eq!(db.load_chain_tip().await.unwrap(), Some(header.clone()));
        assert_eq!(db.load_private_record(transaction_id).await.unwrap(), Some(record));
        let migrated = db.load_private_record(transaction_id).await.unwrap().unwrap();
        assert_eq!(migrated.context().format_version(), PrivateRecordFormatVersion::V1);
        migrated.verify_encrypted_record_key().unwrap();
        assert_eq!(db.load_protocol_config(config.to_commitment()).await.unwrap(), None);

        db.upsert_block_header_with_protocol_config(header, Some(config.clone()))
            .await
            .unwrap();
        assert_eq!(
            protocol_config_history::history(&db).await,
            vec![(0, config.to_commitment(), config.clone())]
        );
        assert_eq!(db.load_protocol_config(config.to_commitment()).await.unwrap(), Some(config));
    }

    #[tokio::test]
    async fn setup_creates_database_that_load_accepts() {
        let temp_dir = tempfile::tempdir().expect("failed to create temp directory");
        let db_path = temp_dir.path().join("validator.sqlite3");

        setup(db_path.clone()).await.expect("setup should bootstrap the database");
        load(db_path).await.expect("load should accept a bootstrapped database");
    }

    #[tokio::test]
    async fn setup_creates_protocol_config_storage() {
        let temp_dir = tempfile::tempdir().expect("failed to create temp directory");
        let db = setup(temp_dir.path().join("validator.sqlite3")).await.unwrap();

        let row_count = db
            .reader
            .reader
            .read("protocol_config_storage", |tx| {
                Ok::<_, DatabaseError>(
                    tx.query("SELECT COUNT(*) FROM protocol_configs", &[], |row| {
                        row.get::<i64>(0)
                    })?
                    .into_iter()
                    .next()
                    .expect("COUNT always returns one row"),
                )
            })
            .await
            .unwrap();

        assert_eq!(row_count, 0);
    }

    #[tokio::test]
    async fn block_header_and_protocol_config_are_persisted_together() {
        let temp_dir = tempfile::tempdir().expect("failed to create temp directory");
        let db = setup(temp_dir.path().join("validator.sqlite3")).await.unwrap();
        let config = test_protocol_config();
        let header = genesis_header(&config);

        db.upsert_block_header_with_protocol_config(header.clone(), Some(config.clone()))
            .await
            .unwrap();

        assert_eq!(db.load_chain_tip().await.unwrap(), Some(header));
        assert_eq!(db.load_protocol_config(config.to_commitment()).await.unwrap(), Some(config));
    }

    #[tokio::test]
    async fn duplicate_supplied_protocol_config_is_accepted() {
        let temp_dir = tempfile::tempdir().expect("failed to create temp directory");
        let db = setup(temp_dir.path().join("validator.sqlite3")).await.unwrap();
        let config = test_protocol_config();
        let header = genesis_header(&config);
        let replacement = header_with_next_timestamp(&header);

        db.upsert_block_header_with_protocol_config(header.clone(), Some(config.clone()))
            .await
            .unwrap();
        db.upsert_block_header_with_protocol_config(replacement.clone(), Some(config.clone()))
            .await
            .expect("a duplicate supplied protocol config should be accepted");

        assert_eq!(db.load_block_header(header.block_num()).await.unwrap(), Some(replacement));
        assert_eq!(db.load_protocol_config(config.to_commitment()).await.unwrap(), Some(config));
    }

    #[tokio::test]
    async fn known_protocol_config_can_be_omitted() {
        let temp_dir = tempfile::tempdir().expect("failed to create temp directory");
        let db = setup(temp_dir.path().join("validator.sqlite3")).await.unwrap();
        let config = test_protocol_config();
        let header = genesis_header(&config);
        let replacement = header_with_next_timestamp(&header);

        db.upsert_block_header_with_protocol_config(header.clone(), Some(config.clone()))
            .await
            .unwrap();
        db.upsert_block_header_with_protocol_config(replacement.clone(), None)
            .await
            .expect("a stored protocol config should not need to be supplied again");

        assert_eq!(db.load_block_header(header.block_num()).await.unwrap(), Some(replacement));
        assert_eq!(db.load_protocol_config(config.to_commitment()).await.unwrap(), Some(config));
    }

    #[tokio::test]
    async fn unknown_protocol_config_rolls_back_block_header() {
        let temp_dir = tempfile::tempdir().expect("failed to create temp directory");
        let db = setup(temp_dir.path().join("validator.sqlite3")).await.unwrap();
        let config = test_protocol_config();
        let header = genesis_header(&config);

        db.upsert_block_header_with_protocol_config(header.clone(), None)
            .await
            .expect_err("an unknown protocol config must reject the header");

        assert_eq!(db.load_block_header(header.block_num()).await.unwrap(), None);
        assert_eq!(db.load_protocol_config(config.to_commitment()).await.unwrap(), None);
    }

    #[tokio::test]
    async fn mismatched_protocol_config_rolls_back_block_header() {
        let temp_dir = tempfile::tempdir().expect("failed to create temp directory");
        let db = setup(temp_dir.path().join("validator.sqlite3")).await.unwrap();
        let expected = test_protocol_config();
        let header = genesis_header(&expected);
        let mismatched = ProtocolConfig::current(AssetId::new_fungible(
            ACCOUNT_ID_PUBLIC_FUNGIBLE_FAUCET_1.try_into().unwrap(),
        ))
        .unwrap();

        db.upsert_block_header_with_protocol_config(header.clone(), Some(mismatched.clone()))
            .await
            .expect_err("a mismatched config must reject the header transaction");

        assert_eq!(db.load_block_header(header.block_num()).await.unwrap(), None);
        assert_eq!(db.load_protocol_config(expected.to_commitment()).await.unwrap(), None);
        assert_eq!(db.load_protocol_config(mismatched.to_commitment()).await.unwrap(), None);
    }

    #[tokio::test]
    async fn block_header_insertion_failure_rolls_back_new_protocol_config() {
        let temp_dir = tempfile::tempdir().expect("failed to create temp directory");
        let db = setup(temp_dir.path().join("validator.sqlite3")).await.unwrap();
        let config = test_protocol_config();
        let commitment = config.to_commitment();
        let header = genesis_header(&config);

        db.writer
            .write("reject_block_header_inserts", |tx| {
                tx.execute(
                    "CREATE TRIGGER reject_block_header_insert
                     BEFORE INSERT ON block_headers
                     BEGIN
                         SELECT RAISE(ABORT, 'block header insertion rejected');
                     END;",
                    &[],
                )?;
                Ok::<_, DatabaseError>(())
            })
            .await
            .unwrap();

        db.upsert_block_header_with_protocol_config(header.clone(), Some(config))
            .await
            .expect_err("a block header insertion failure must reject the transaction");

        assert_eq!(db.load_block_header(header.block_num()).await.unwrap(), None);
        assert_eq!(db.load_protocol_config(commitment).await.unwrap(), None);
    }

    #[tokio::test]
    async fn transaction_exists_detects_validated_transactions() {
        let temp_dir = tempfile::tempdir().expect("failed to create temp directory");
        let db = setup(temp_dir.path().join("validator.sqlite3")).await.unwrap();

        let validated_id = TransactionId::from_raw(Word::try_from([1u64, 2, 3, 4]).unwrap());
        let unknown_id = TransactionId::from_raw(Word::try_from([5u64, 6, 7, 8]).unwrap());

        db.insert_validated_private_transaction(private_record(validated_id, 1))
            .await
            .unwrap();

        assert!(
            db.transaction_exists(validated_id).await.unwrap(),
            "an inserted transaction id should be reported as existing"
        );
        assert!(
            !db.transaction_exists(unknown_id).await.unwrap(),
            "an unknown transaction id should not be reported as existing"
        );
    }

    /// The `rarray`-based lookup must return exactly the ids that are absent, preserving the order
    /// they were supplied in.
    #[tokio::test]
    async fn find_unvalidated_transactions_returns_only_missing_ids() {
        let temp_dir = tempfile::tempdir().expect("failed to create temp directory");
        let db = setup(temp_dir.path().join("validator.sqlite3")).await.unwrap();

        let ids = (1u64..=4)
            .map(|i| TransactionId::from_raw(Word::try_from([i, i, i, i]).unwrap()))
            .collect::<Vec<_>>();

        // Validate the second and fourth ids only.
        db.insert_validated_private_transaction(private_record(ids[1], 1))
            .await
            .unwrap();
        db.insert_validated_private_transaction(private_record(ids[3], 2))
            .await
            .unwrap();

        let unvalidated = db.find_unvalidated_transactions(ids.clone()).await.unwrap();
        assert_eq!(unvalidated, vec![ids[0], ids[2]]);

        // An empty request must not error and must return nothing.
        assert!(db.find_unvalidated_transactions(vec![]).await.unwrap().is_empty());
    }

    #[tokio::test]
    async fn load_initial_metrics_reports_persisted_state() {
        let temp_dir = tempfile::tempdir().expect("failed to create temp directory");
        let db = setup(temp_dir.path().join("validator.sqlite3")).await.unwrap();

        // A freshly bootstrapped database is empty.
        let metrics = db.load_initial_metrics().await.unwrap();
        assert_eq!(metrics.chain_tip, 0);
        assert_eq!(metrics.validated_transactions, 0);
        assert_eq!(metrics.signed_blocks, 0);
    }

    #[tokio::test]
    async fn private_record_indexes_work() {
        let temp_dir = tempfile::tempdir().expect("failed to create temp directory");
        let db = setup(temp_dir.path().join("validator.sqlite3")).await.unwrap();
        let transaction_id = TransactionId::from_raw(Word::from([5u32, 6, 7, 8]));
        let record = private_record(transaction_id, 9);

        let expected = record.clone();
        db.insert_validated_private_transaction(record).await.unwrap();

        let by_record = db.load_private_record(transaction_id).await.unwrap();
        assert_eq!(by_record, Some(expected.clone()));

        let by_epoch = db.load_private_records_by_key_epoch(KEY_EPOCH).await.unwrap();
        assert_eq!(by_epoch, vec![expected.clone()]);

        let by_setup = db.load_private_records_by_setup_context(SETUP_CONTEXT_ID).await.unwrap();
        assert_eq!(by_setup, vec![expected.clone()]);
    }

    #[tokio::test]
    async fn private_record_rejects_unsupported_formats() {
        let temp_dir = tempfile::tempdir().unwrap();
        let db = setup(temp_dir.path().join("validator.sqlite3")).await.unwrap();
        let transaction_id = TransactionId::from_raw(Word::from([5u32, 6, 7, 8]));
        db.insert_validated_private_transaction(private_record(transaction_id, 9))
            .await
            .unwrap();

        for format_version in [2_u32, 3, u32::MAX] {
            db.writer
                .write("set unsupported record format", move |tx| {
                    tx.execute(
                        "UPDATE validated_transactions SET format_version = ? WHERE id = ?",
                        &[&i64::from(format_version), &transaction_id],
                    )
                })
                .await
                .unwrap();

            let error = db.load_private_record(transaction_id).await.unwrap_err();
            assert!(matches!(
                error,
                DatabaseError::ConversionSqlToRust { inner: Some(source), .. }
                    if matches!(
                        source.downcast_ref::<PrivateRecordError>(),
                        Some(PrivateRecordError::UnsupportedFormat(version))
                            if *version == format_version,
                    ),
            ));
        }
    }

    /// Validated transactions that are not part of a signed block have no position in the committed
    /// order, so the listing does not surface them at all.
    #[tokio::test]
    async fn uncommitted_transactions_are_not_listed() {
        let temp_dir = tempfile::tempdir().expect("failed to create temp directory");
        let db = setup(temp_dir.path().join("validator.sqlite3")).await.unwrap();
        let transaction_ids = [
            TransactionId::from_raw(Word::from([9u32, 0, 0, 0])),
            TransactionId::from_raw(Word::from([1u32, 0, 0, 0])),
        ];
        for (transaction_id, seed) in transaction_ids.into_iter().zip([1u8, 2]) {
            db.insert_validated_private_transaction(private_record(transaction_id, seed))
                .await
                .unwrap();
        }

        let params = ListTransactionsParams {
            start: None,
            block_to: None,
            limit: 10,
            include_records: false,
        };
        assert!(db.list_validated_transactions(params).await.unwrap().transactions.is_empty());
        // They remain reachable by transaction id.
        assert!(db.load_private_record(transaction_ids[0]).await.unwrap().is_some());
    }

    /// A signed block links its transactions in block order; replacing the block at the same height
    /// deletes the replaced block's links along with its header.
    #[tokio::test]
    async fn insert_signed_block_links_and_relinks_transactions() {
        let temp_dir = tempfile::tempdir().expect("failed to create temp directory");
        let db = setup(temp_dir.path().join("validator.sqlite3")).await.unwrap();
        let transaction_ids = (1u64..=3)
            .map(|i| TransactionId::from_raw(Word::try_from([i, i, i, i]).unwrap()))
            .collect::<Vec<_>>();
        for (seed, transaction_id) in [1u8, 2, 3].into_iter().zip(&transaction_ids) {
            db.insert_validated_private_transaction(private_record(*transaction_id, seed))
                .await
                .unwrap();
        }

        let header = BlockHeader::mock(7, None, None, &[]);
        db.insert_signed_block(
            header.clone(),
            ProtocolConfig::mock(),
            vec![transaction_ids[0], transaction_ids[1]],
        )
        .await
        .unwrap();

        let params = ListTransactionsParams {
            start: None,
            block_to: None,
            limit: 10,
            include_records: false,
        };
        let listed = db.list_validated_transactions(params).await.unwrap().transactions;
        assert_eq!(
            listed
                .iter()
                .map(|item| (item.transaction_id, item.block_num, item.block_tx_index))
                .collect::<Vec<_>>(),
            vec![
                (transaction_ids[0], BlockNumber::from(7u32), 0),
                (transaction_ids[1], BlockNumber::from(7u32), 1),
            ],
        );

        // Replace the block at the same height with one that only includes the third transaction.
        // The two transactions the replacement drops go back to being uncommitted, and stop being
        // listed.
        db.replace_signed_block(header, ProtocolConfig::mock(), vec![transaction_ids[2]])
            .await
            .unwrap();

        let listed = db.list_validated_transactions(params).await.unwrap().transactions;
        assert_eq!(
            listed
                .iter()
                .map(|item| (item.transaction_id, item.block_num, item.block_tx_index))
                .collect::<Vec<_>>(),
            vec![(transaction_ids[2], BlockNumber::from(7u32), 0)],
        );
    }

    /// A listing keeps its records and chain tip consistent even when the tip is replaced and
    /// advanced while its read snapshot is open.
    #[tokio::test]
    async fn listing_snapshot_survives_tip_replacement_and_advance() {
        let directory = tempfile::tempdir().unwrap();
        let db = setup(directory.path().join("validator.sqlite3")).await.unwrap();
        let original_id = TransactionId::from_raw(Word::from([1u32, 0, 0, 0]));
        let replacement_id = TransactionId::from_raw(Word::from([2u32, 0, 0, 0]));
        let original_record = private_record(original_id, 1);
        let replacement_record = private_record(replacement_id, 2);
        for record in [&original_record, &replacement_record] {
            db.insert_validated_private_transaction(record.clone()).await.unwrap();
        }
        let header = BlockHeader::mock(1, None, None, &[]);
        db.insert_signed_block(header.clone(), ProtocolConfig::mock(), vec![original_id])
            .await
            .unwrap();

        let snapshot = db.reader.reader.begin_read().await.unwrap();
        let tip = snapshot.run("pin_listing_snapshot", queries::load_chain_tip).await.unwrap();
        assert_eq!(tip.unwrap().block_num(), BlockNumber::from(1u32));

        db.replace_signed_block(header, ProtocolConfig::mock(), vec![replacement_id])
            .await
            .unwrap();
        db.insert_signed_block(
            BlockHeader::mock(2, None, None, &[]),
            ProtocolConfig::mock(),
            Vec::new(),
        )
        .await
        .unwrap();

        let params = ListTransactionsParams {
            start: None,
            block_to: None,
            limit: 10,
            include_records: true,
        };
        let page = snapshot
            .run("list_snapshot", move |tx| queries::list_validated_transactions(tx, &params))
            .await
            .unwrap();
        assert_eq!(page.chain_tip, BlockNumber::from(1u32));
        assert_eq!(page.transactions.len(), 1);
        assert_eq!(page.transactions[0].transaction_id, original_id);
        assert_eq!(page.transactions[0].block_num, page.chain_tip);
        assert_eq!(page.transactions[0].record.as_ref(), Some(&original_record));
        snapshot.close().await.unwrap();

        let current = db.list_validated_transactions(params).await.unwrap();
        assert_eq!(current.chain_tip, BlockNumber::from(2u32));
        assert_eq!(current.transactions.len(), 1);
        assert_eq!(current.transactions[0].transaction_id, replacement_id);
        assert_eq!(current.transactions[0].block_num, BlockNumber::from(1u32));
        assert_eq!(current.transactions[0].record.as_ref(), Some(&replacement_record));
    }

    /// A full sweep pages through committed transactions in committed order, honoring the row limit
    /// exactly, with each page resuming one position past the last row of the previous one.
    #[tokio::test]
    async fn list_validated_transactions_pages_in_committed_order() {
        let temp_dir = tempfile::tempdir().expect("failed to create temp directory");
        let db = setup(temp_dir.path().join("validator.sqlite3")).await.unwrap();
        let transaction_ids = (1u64..=5)
            .map(|i| TransactionId::from_raw(Word::try_from([i, i, i, i]).unwrap()))
            .collect::<Vec<_>>();
        for (seed, transaction_id) in (1u8..=5).zip(&transaction_ids) {
            db.insert_validated_private_transaction(private_record(*transaction_id, seed))
                .await
                .unwrap();
        }
        // Blocks 1 and 2 include two transactions each; the fifth is never committed. The later
        // insertion is committed in the earlier block, to prove the listing follows committed order
        // rather than insertion order.
        let block_1 = BlockHeader::mock(1, None, None, &[]);
        let block_2 = BlockHeader::mock(2, None, None, &[]);
        db.insert_signed_block(
            block_1,
            ProtocolConfig::mock(),
            vec![transaction_ids[3], transaction_ids[0]],
        )
        .await
        .unwrap();
        db.insert_signed_block(
            block_2,
            ProtocolConfig::mock(),
            vec![transaction_ids[1], transaction_ids[2]],
        )
        .await
        .unwrap();
        let expected_order =
            [transaction_ids[3], transaction_ids[0], transaction_ids[1], transaction_ids[2]];

        // Sweep with a limit of three: the first page ends mid-block, and the next page resumes one
        // position past the last row returned.
        let mut swept = Vec::new();
        let mut start = None;
        loop {
            let params = ListTransactionsParams {
                start,
                block_to: None,
                limit: 3,
                include_records: false,
            };
            let page = db.list_validated_transactions(params).await.unwrap().transactions;
            let Some(last) = page.last() else { break };
            assert!(page.len() <= 3, "a page must honor the row limit");
            start = Some((last.block_num, last.block_tx_index + 1));
            swept.extend(page.into_iter().map(|item| item.transaction_id));
        }
        assert_eq!(swept, expected_order);

        // Bounding by block number: `start` excludes block 1, `block_to` excludes block 2.
        let from_block_2 = db
            .list_validated_transactions(ListTransactionsParams {
                start: Some((BlockNumber::from(2u32), 0)),
                block_to: None,
                limit: 10,
                include_records: false,
            })
            .await
            .unwrap();
        assert_eq!(
            from_block_2
                .transactions
                .iter()
                .map(|item| item.transaction_id)
                .collect::<Vec<_>>(),
            vec![transaction_ids[1], transaction_ids[2]],
        );
        let up_to_block_1 = db
            .list_validated_transactions(ListTransactionsParams {
                start: None,
                block_to: Some(BlockNumber::from(1u32)),
                limit: 10,
                include_records: false,
            })
            .await
            .unwrap();
        assert_eq!(
            up_to_block_1
                .transactions
                .iter()
                .map(|item| item.transaction_id)
                .collect::<Vec<_>>(),
            vec![transaction_ids[3], transaction_ids[0]],
        );
    }

    /// Linking a transaction this validator never validated fails the foreign key on
    /// `block_transactions`, so a signed block cannot silently reference unknown transactions.
    #[tokio::test]
    async fn insert_signed_block_rejects_unvalidated_transactions() {
        let temp_dir = tempfile::tempdir().expect("failed to create temp directory");
        let db = setup(temp_dir.path().join("validator.sqlite3")).await.unwrap();
        let header = BlockHeader::mock(1, None, None, &[]);
        let unknown = TransactionId::from_raw(Word::from([1u32, 0, 0, 0]));

        let result = db.insert_signed_block(header, ProtocolConfig::mock(), vec![unknown]).await;
        assert!(result.is_err(), "linking an unvalidated transaction must fail");
    }

    #[tokio::test]
    async fn stored_private_record_opens_with_threshold_shares() {
        let temp_dir = tempfile::tempdir().expect("failed to create temp directory");
        let db = setup(temp_dir.path().join("validator.sqlite3")).await.unwrap();
        let operators = operator_keys();
        let transaction_id = TransactionId::from_raw(Word::from([9u32, 10, 11, 12]));
        let context = PrivateRecordContext::new(CHAIN_ID, operators[0].key_epoch(), transaction_id);
        let plaintext = b"private transaction inputs";
        let mut seal_rng = ChaCha20Rng::from_seed([40; 32]);
        let record = PrivateRecordSealer::from_operator_key(&operators[0])
            .seal(&mut seal_rng, record_id(transaction_id), context, plaintext)
            .unwrap();
        db.insert_validated_private_transaction(record).await.unwrap();

        let stored = db.load_private_record(transaction_id).await.unwrap().unwrap();
        let request = PrivateRecordShareRequest::for_record(&stored);
        let mut first_rng = ChaCha20Rng::from_seed([41; 32]);
        let mut second_rng = ChaCha20Rng::from_seed([42; 32]);
        let shares = [
            operators[0]
                .issue_private_record_share(&mut first_rng, &request, &stored)
                .unwrap(),
            operators[1]
                .issue_private_record_share(&mut second_rng, &request, &stored)
                .unwrap(),
        ];

        let opened = PrivateRecordCombiner::from_operator_key(&operators[2])
            .unwrap()
            .open(&request, &stored, &shares)
            .unwrap();
        assert_eq!(opened.as_slice(), plaintext);
    }

    #[tokio::test]
    async fn private_record_schema_has_required_indexes() {
        let temp_dir = tempfile::tempdir().expect("failed to create temp directory");
        let db = setup(temp_dir.path().join("validator.sqlite3")).await.unwrap();

        let schema = db
            .reader
            .reader
            .read("private_record_schema", |tx| {
                tx.query(
                    "SELECT sql FROM sqlite_schema \
                     WHERE tbl_name IN ('validated_transactions', 'block_transactions') \
                       AND sql IS NOT NULL \
                     ORDER BY name",
                    &[],
                    |row| row.get::<String>(0),
                )
            })
            .await
            .unwrap()
            .join("\n");

        assert!(schema.contains("insertion_sequence    INTEGER PRIMARY KEY AUTOINCREMENT"));
        assert!(schema.contains("id                    BLOB NOT NULL UNIQUE"));
        assert!(schema.contains("idx_validated_transactions_key_epoch"));
        assert!(schema.contains("idx_validated_transactions_setup_context_id"));
        assert!(schema.contains("PRIMARY KEY (block_num, block_tx_index)"));
    }
}