miden-validator 0.17.0-rc.2

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;

// 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 all validated private transactions in insertion order.
    pub(crate) async fn load_all_transactions(
        &self,
    ) -> Result<Vec<StoredPrivateRecord>, DatabaseError> {
        self.reader.read("load_all_transactions", queries::load_all_transactions).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.
    ///
    /// Records an activation if the configuration differs from the preceding activation.
    /// A replacement at the current tip must retain its active configuration.
    /// Callers must validate block order before this method runs.
    ///
    /// If `protocol_config` is absent, the configuration must already be stored
    /// otherwise an error is returned.
    #[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| {
                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)?;
                }

                queries::upsert_block_header(tx, &header)
            })
            .await
    }
}

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.upsert_block_header_with_protocol_config(genesis_header, Some(protocol_config))
        .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_all_transactions().await.unwrap(), vec![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,
                    ),
            ));
        }
    }

    #[tokio::test]
    async fn validated_private_transactions_are_loaded_in_insertion_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 = [
            TransactionId::from_raw(Word::from([9u32, 0, 0, 0])),
            TransactionId::from_raw(Word::from([1u32, 0, 0, 0])),
            TransactionId::from_raw(Word::from([5u32, 0, 0, 0])),
        ];
        let records = transaction_ids
            .into_iter()
            .zip([1u8, 2, 3])
            .map(|(transaction_id, seed)| private_record(transaction_id, seed))
            .collect::<Vec<_>>();

        for record in records.clone() {
            db.insert_validated_private_transaction(record).await.unwrap();
        }

        let loaded = db.load_all_transactions().await.unwrap();

        assert_eq!(loaded, records);
    }

    #[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 = 'validated_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"));
    }
}