lssa/sequencer/core/src/lib.rs

use std::{path::Path, time::Instant};

use anyhow::{Context as _, Result, anyhow};
use common::{
    HashType,
    block::{BedrockStatus, Block, HashableBlockData},
    transaction::{NSSATransaction, clock_invocation},
};
use config::{GenesisAction, SequencerConfig};
use log::{error, info, warn};
use logos_blockchain_key_management_system_service::keys::{ED25519_SECRET_KEY_SIZE, Ed25519Key};
use mempool::{MemPool, MemPoolHandle};
#[cfg(feature = "mock")]
pub use mock::SequencerCoreWithMockClients;
use nssa::{AccountId, PublicTransaction, program::Program, public_transaction::Message};
use nssa_core::GENESIS_BLOCK_ID;
pub use storage::error::DbError;

use crate::{
    block_publisher::{BlockPublisherTrait, ZoneSdkPublisher},
    block_store::SequencerStore,
};

pub mod block_publisher;
pub mod block_store;
pub mod config;

#[cfg(feature = "mock")]
pub mod mock;

pub struct SequencerCore<BP: BlockPublisherTrait = ZoneSdkPublisher> {
    state: nssa::V03State,
    store: SequencerStore,
    mempool: MemPool<NSSATransaction>,
    sequencer_config: SequencerConfig,
    chain_height: u64,
    block_publisher: BP,
}

impl<BP: BlockPublisherTrait> SequencerCore<BP> {
    /// Starts the sequencer using the provided configuration.
    /// If an existing database is found, the sequencer state is loaded from it and
    /// assumed to represent the correct latest state consistent with Bedrock-finalized data.
    /// If no database is found, the sequencer performs a fresh start from genesis,
    /// initializing its state with the accounts defined in the configuration file.
    pub async fn start_from_config(
        config: SequencerConfig,
    ) -> (Self, MemPoolHandle<NSSATransaction>) {
        let signing_key = nssa::PrivateKey::try_new(config.signing_key).unwrap();

        let bedrock_signing_key =
            load_or_create_signing_key(&config.home.join("bedrock_signing_key"))
                .expect("Failed to load or create bedrock signing key");

        let db_path = config.home.join("rocksdb");
        let (store, state, genesis_block) = if db_path.exists() {
            let store =
                SequencerStore::open_db(&db_path, signing_key.clone()).unwrap_or_else(|err| {
                    panic!(
                        "Failed to open database at {} with error: {err}",
                        db_path.display()
                    )
                });
            let state = store
                .get_nssa_state()
                .expect("Failed to read state from store");
            let genesis_block = store
                .get_block_at_id(store.genesis_id())
                .expect("Failed to read genesis block from store")
                .expect("Genesis block not found in store");
            (store, state, genesis_block)
        } else {
            warn!(
                "Database not found at {}, starting from genesis",
                db_path.display()
            );

            // TODO: Remove msg_id from BlockMeta — it is no longer needed now that
            // zone-sdk manages L1 settlement state via its own checkpoint.
            let genesis_msg_id = [0; 32];
            let genesis_parent_msg_id = [0; 32];
            let (genesis_state, genesis_txs) = build_genesis_state(&config);

            let hashable_data = HashableBlockData {
                block_id: GENESIS_BLOCK_ID,
                transactions: genesis_txs,
                prev_block_hash: HashType([0; 32]),
                timestamp: 0,
            };
            let genesis_block =
                hashable_data.into_pending_block(&signing_key, genesis_parent_msg_id);

            let store = SequencerStore::create_db_with_genesis(
                &db_path,
                &genesis_block,
                genesis_msg_id,
                &genesis_state,
                signing_key,
            )
            .expect("Failed to create database with genesis block");

            (store, genesis_state, genesis_block)
        };

        let latest_block_meta = store
            .latest_block_meta()
            .expect("Failed to read latest block meta from store");

        let initial_checkpoint = store
            .get_zone_checkpoint()
            .expect("Failed to load zone-sdk checkpoint");
        let is_fresh_start = initial_checkpoint.is_none();

        let dbio_for_checkpoint = store.dbio();
        let on_checkpoint: block_publisher::CheckpointSink = Box::new(move |cp| {
            let bytes = match serde_json::to_vec(&cp) {
                Ok(b) => b,
                Err(err) => {
                    error!("Failed to serialize zone-sdk checkpoint: {err:#}");
                    return;
                }
            };
            if let Err(err) = dbio_for_checkpoint.put_zone_sdk_checkpoint_bytes(&bytes) {
                error!("Failed to persist zone-sdk checkpoint: {err:#}");
            }
        });

        let dbio_for_finalized = store.dbio();
        let on_finalized_block: block_publisher::FinalizedBlockSink = Box::new(move |block_id| {
            if let Err(err) = dbio_for_finalized.clean_pending_blocks_up_to(block_id) {
                error!("Failed to mark pending blocks finalized up to {block_id}: {err:#}");
            }
        });

        let block_publisher = BP::new(
            &config.bedrock_config,
            bedrock_signing_key,
            config.retry_pending_blocks_timeout,
            initial_checkpoint,
            on_checkpoint,
            on_finalized_block,
        )
        .await
        .expect("Failed to initialize Block Publisher");

        // On a truly fresh start (no checkpoint persisted yet), publish the
        // genesis block so the indexer can find the channel start. After the
        // first publish, zone-sdk's checkpoint persistence covers further
        // restarts.
        if is_fresh_start && let Err(err) = block_publisher.publish_block(&genesis_block).await {
            error!("Failed to publish genesis block: {err:#}");
        }

        let (mempool, mempool_handle) = MemPool::new(config.mempool_max_size);

        let sequencer_core = Self {
            state,
            store,
            mempool,
            chain_height: latest_block_meta.id,
            sequencer_config: config,
            block_publisher,
        };

        (sequencer_core, mempool_handle)
    }

    /// Produces a new block from mempool transactions and publishes it via zone-sdk.
    pub async fn produce_new_block(&mut self) -> Result<u64> {
        let block = self
            .build_block_from_mempool()
            .context("Failed to build block from mempool transactions")?;

        // TODO: Remove msg_id from store.update — it is no longer needed now that
        // zone-sdk manages L1 settlement state via its own checkpoint.
        let placeholder_msg_id = [0_u8; 32];

        if let Err(err) = self.block_publisher.publish_block(&block).await {
            error!("Failed to publish block to Bedrock with error: {err:#}");
        }
        self.store.update(&block, placeholder_msg_id, &self.state)?;

        Ok(self.chain_height)
    }

    /// Builds a new block from transactions in the mempool.
    /// Does NOT publish or store the block — the caller is responsible for that.
    pub fn build_block_from_mempool(&mut self) -> Result<Block> {
        let now = Instant::now();

        let new_block_height = self.next_block_id();

        let mut valid_transactions = vec![];

        let max_block_size = usize::try_from(self.sequencer_config.max_block_size.as_u64())
            .expect("`max_block_size` should fit into usize");

        let latest_block_meta = self
            .store
            .latest_block_meta()
            .context("Failed to get latest block meta from store")?;

        let new_block_timestamp = u64::try_from(chrono::Utc::now().timestamp_millis())
            .expect("Timestamp must be positive");

        // Pre-create the mandatory clock tx so its size is included in the block size check.
        let clock_tx = clock_invocation(new_block_timestamp);
        let clock_nssa_tx = NSSATransaction::Public(clock_tx.clone());

        while let Some(tx) = self.mempool.pop() {
            let tx_hash = tx.hash();

            // Check if block size exceeds limit (including the mandatory clock tx).
            let temp_valid_transactions = [
                valid_transactions.as_slice(),
                std::slice::from_ref(&tx),
                std::slice::from_ref(&clock_nssa_tx),
            ]
            .concat();
            let temp_hashable_data = HashableBlockData {
                block_id: new_block_height,
                transactions: temp_valid_transactions,
                prev_block_hash: latest_block_meta.hash,
                timestamp: new_block_timestamp,
            };

            let block_size = borsh::to_vec(&temp_hashable_data)
                .context("Failed to serialize block for size check")?
                .len();

            if block_size > max_block_size {
                // Block would exceed size limit, remove last transaction and push back
                warn!(
                    "Transaction with hash {tx_hash} deferred to next block: \
                     block size {block_size} bytes would exceed limit of {max_block_size} bytes",
                );

                self.mempool.push_front(tx);
                break;
            }

            let validated_diff = match tx.validate_on_state(
                &self.state,
                new_block_height,
                new_block_timestamp,
            ) {
                Ok(diff) => diff,
                Err(err) => {
                    error!(
                        "Transaction with hash {tx_hash} failed execution check with error: {err:#?}, skipping it",
                    );
                    continue;
                }
            };

            self.state.apply_state_diff(validated_diff);

            valid_transactions.push(tx);
            info!("Validated transaction with hash {tx_hash}, including it in block");
            if valid_transactions.len() >= self.sequencer_config.max_num_tx_in_block {
                break;
            }
        }

        // Append the Clock Program invocation as the mandatory last transaction.
        self.state
            .transition_from_public_transaction(&clock_tx, new_block_height, new_block_timestamp)
            .context("Clock transaction failed. Aborting block production.")?;
        valid_transactions.push(clock_nssa_tx);

        let hashable_data = HashableBlockData {
            block_id: new_block_height,
            transactions: valid_transactions,
            prev_block_hash: latest_block_meta.hash,
            timestamp: new_block_timestamp,
        };

        // TODO: Remove bedrock_parent_id from Block — it is no longer needed now
        // that zone-sdk manages the inscription parent chain internally.
        let placeholder_parent_id = [0_u8; 32];
        let block = hashable_data
            .clone()
            .into_pending_block(self.store.signing_key(), placeholder_parent_id);

        self.chain_height = new_block_height;

        log::info!(
            "Created block with {} transactions in {} seconds",
            hashable_data.transactions.len(),
            now.elapsed().as_secs()
        );
        Ok(block)
    }

    pub const fn state(&self) -> &nssa::V03State {
        &self.state
    }

    pub const fn block_store(&self) -> &SequencerStore {
        &self.store
    }

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

    pub const fn sequencer_config(&self) -> &SequencerConfig {
        &self.sequencer_config
    }

    /// Marks all pending blocks with `block_id <= last_finalized_block_id` as
    /// finalized. Idempotent. Production callers don't invoke this directly —
    /// it's wired up in `start_from_config` to the publisher's
    /// `on_finalized_block` sink, which fires on `Event::TxsFinalized` /
    /// `Event::FinalizedInscriptions`. Kept on the type for tests.
    // TODO: Delete blocks instead of marking them as finalized. Current
    // approach is used because we still have `GetBlockDataRequest`.
    pub fn clean_finalized_blocks_from_db(&self, last_finalized_block_id: u64) -> Result<()> {
        info!("Clearing pending blocks up to id: {last_finalized_block_id}");
        self.store
            .dbio()
            .clean_pending_blocks_up_to(last_finalized_block_id)?;
        Ok(())
    }

    /// Returns the list of stored pending blocks.
    pub fn get_pending_blocks(&self) -> Result<Vec<Block>> {
        Ok(self
            .store
            .get_all_blocks()
            .collect::<block_store::DbResult<Vec<Block>>>()?
            .into_iter()
            .filter(|block| matches!(block.bedrock_status, BedrockStatus::Pending))
            .collect())
    }

    pub fn block_publisher(&self) -> BP {
        self.block_publisher.clone()
    }

    fn next_block_id(&self) -> u64 {
        self.chain_height
            .checked_add(1)
            .unwrap_or_else(|| panic!("Max block height reached: {}", self.chain_height))
    }
}

/// Builds the initial genesis state from `testnet_initial_state` plus configured genesis
/// transactions. Returns the final state and the list of [`NSSATransaction`]s that should be
/// committed to the genesis block so external observers can replay them.
fn build_genesis_state(config: &SequencerConfig) -> (nssa::V03State, Vec<NSSATransaction>) {
    #[cfg(not(feature = "testnet"))]
    let mut state = testnet_initial_state::initial_state();

    #[cfg(feature = "testnet")]
    let mut state = testnet_initial_state::initial_state_testnet();

    let genesis_txs = config
        .genesis
        .iter()
        .map(|genesis_tx| match genesis_tx {
            GenesisAction::SupplyAccount {
                account_id,
                balance,
            } => build_supply_account_genesis_transaction(account_id, *balance),
        })
        .chain(std::iter::once(clock_invocation(0)))
        .inspect(|tx| {
            state
                .transition_from_public_transaction(tx, GENESIS_BLOCK_ID, 0)
                .expect("Failed to execute genesis transaction");
        })
        .map(NSSATransaction::Public)
        .collect();

    (state, genesis_txs)
}

fn build_supply_account_genesis_transaction(
    account_id: &AccountId,
    balance: u128,
) -> PublicTransaction {
    let vault_program_id = Program::vault().id();
    let recipient_vault_id = vault_core::compute_vault_account_id(vault_program_id, *account_id);

    let message = Message::try_new(
        vault_program_id,
        vec![nssa::SYSTEM_FAUCET_ACCOUNT_ID, recipient_vault_id],
        vec![],
        vault_core::Instruction::Transfer {
            recipient_id: *account_id,
            amount: balance,
        },
    )
    .expect("Failed to serialize genesis transfer instruction");
    let witness_set = nssa::public_transaction::WitnessSet::from_raw_parts(vec![]);

    PublicTransaction::new(message, witness_set)
}

/// Load signing key from file or generate a new one if it doesn't exist.
fn load_or_create_signing_key(path: &Path) -> Result<Ed25519Key> {
    if path.exists() {
        let key_bytes = std::fs::read(path)?;

        let key_array: [u8; ED25519_SECRET_KEY_SIZE] = key_bytes
            .try_into()
            .map_err(|_bytes| anyhow!("Found key with incorrect length"))?;

        Ok(Ed25519Key::from_bytes(&key_array))
    } else {
        let mut key_bytes = [0_u8; ED25519_SECRET_KEY_SIZE];
        rand::RngCore::fill_bytes(&mut rand::thread_rng(), &mut key_bytes);
        // Create parent directory if it doesn't exist
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        std::fs::write(path, key_bytes)?;
        Ok(Ed25519Key::from_bytes(&key_bytes))
    }
}

#[cfg(test)]
#[cfg(feature = "mock")]
mod tests {
    #![expect(clippy::shadow_unrelated, reason = "We don't care about it in tests")]

    use std::{pin::pin, time::Duration};

    use common::{
        HashType,
        block::HashableBlockData,
        test_utils::sequencer_sign_key_for_testing,
        transaction::{NSSATransaction, clock_invocation},
    };
    use logos_blockchain_core::mantle::ops::channel::ChannelId;
    use mempool::MemPoolHandle;
    use tempfile::tempdir;
    use testnet_initial_state::{initial_accounts, initial_pub_accounts_private_keys};

    use crate::{
        block_store::SequencerStore,
        build_genesis_state,
        config::{BedrockConfig, SequencerConfig},
        mock::SequencerCoreWithMockClients,
    };

    fn setup_sequencer_config() -> SequencerConfig {
        let tempdir = tempfile::tempdir().unwrap();
        let home = tempdir.path().to_path_buf();

        SequencerConfig {
            home,
            max_num_tx_in_block: 10,
            max_block_size: bytesize::ByteSize::mib(1),
            mempool_max_size: 10000,
            block_create_timeout: Duration::from_secs(1),
            signing_key: *sequencer_sign_key_for_testing().value(),
            bedrock_config: BedrockConfig {
                channel_id: ChannelId::from([0; 32]),
                node_url: "http://not-used-in-unit-tests".parse().unwrap(),
                auth: None,
            },
            retry_pending_blocks_timeout: Duration::from_mins(4),
            genesis: vec![],
        }
    }

    fn create_signing_key_for_account1() -> nssa::PrivateKey {
        initial_pub_accounts_private_keys()[0].pub_sign_key.clone()
    }

    fn create_signing_key_for_account2() -> nssa::PrivateKey {
        initial_pub_accounts_private_keys()[1].pub_sign_key.clone()
    }

    async fn common_setup() -> (SequencerCoreWithMockClients, MemPoolHandle<NSSATransaction>) {
        let config = setup_sequencer_config();
        common_setup_with_config(config).await
    }

    async fn common_setup_with_config(
        config: SequencerConfig,
    ) -> (SequencerCoreWithMockClients, MemPoolHandle<NSSATransaction>) {
        let (mut sequencer, mempool_handle) =
            SequencerCoreWithMockClients::start_from_config(config).await;

        let tx = common::test_utils::produce_dummy_empty_transaction();
        mempool_handle.push(tx).await.unwrap();

        sequencer.produce_new_block().await.unwrap();

        (sequencer, mempool_handle)
    }

    #[tokio::test]
    async fn start_from_config() {
        let config = setup_sequencer_config();
        let (sequencer, _mempool_handle) =
            SequencerCoreWithMockClients::start_from_config(config.clone()).await;

        assert_eq!(sequencer.chain_height, 1);
        assert_eq!(sequencer.sequencer_config.max_num_tx_in_block, 10);

        let acc1_account_id = initial_accounts()[0].account_id;
        let acc2_account_id = initial_accounts()[1].account_id;

        let balance_acc_1 = sequencer.state.get_account_by_id(acc1_account_id).balance;
        let balance_acc_2 = sequencer.state.get_account_by_id(acc2_account_id).balance;

        assert_eq!(10000, balance_acc_1);
        assert_eq!(20000, balance_acc_2);
    }

    #[tokio::test]
    async fn start_from_config_opens_existing_db_if_it_exists() {
        let config = setup_sequencer_config();
        let temp_dir = tempdir().unwrap();
        let mut config = config;
        config.home = temp_dir.path().to_path_buf();

        let signing_key = nssa::PrivateKey::try_new(config.signing_key).unwrap();
        let (genesis_state, genesis_txs) = build_genesis_state(&config);
        let genesis_hashable_data = HashableBlockData {
            block_id: 1,
            transactions: genesis_txs,
            prev_block_hash: HashType([0; 32]),
            timestamp: 0,
        };
        let genesis_block = genesis_hashable_data.into_pending_block(&signing_key, [0; 32]);

        let expected_msg_id = [7; 32];
        SequencerStore::create_db_with_genesis(
            &config.home.join("rocksdb"),
            &genesis_block,
            expected_msg_id,
            &genesis_state,
            signing_key,
        )
        .unwrap();

        let (sequencer, _mempool_handle) =
            SequencerCoreWithMockClients::start_from_config(config).await;
        let latest_meta = sequencer.store.latest_block_meta().unwrap();

        assert_eq!(latest_meta.msg_id, expected_msg_id);
        assert_eq!(sequencer.chain_height, 1);
    }

    #[should_panic(expected = "Failed to open database")]
    #[tokio::test]
    async fn start_from_config_panics_when_db_open_returns_non_not_found_error() {
        let mut config = setup_sequencer_config();
        let temp_dir = tempdir().unwrap();
        config.home = temp_dir.path().to_path_buf();

        let db_path = config.home.join("rocksdb");

        std::fs::create_dir_all(&config.home).unwrap();
        // Force RocksDB open to fail with an IO error by placing a file at DB path.
        std::fs::write(&db_path, b"not-a-directory").unwrap();

        let _ = SequencerCoreWithMockClients::start_from_config(config).await;
    }

    #[test]
    fn transaction_pre_check_pass() {
        let tx = common::test_utils::produce_dummy_empty_transaction();
        let result = tx.transaction_stateless_check();

        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn transaction_pre_check_native_transfer_valid() {
        let (_sequencer, _mempool_handle) = common_setup().await;

        let acc1 = initial_accounts()[0].account_id;
        let acc2 = initial_accounts()[1].account_id;

        let sign_key1 = create_signing_key_for_account1();

        let tx = common::test_utils::create_transaction_native_token_transfer(
            acc1, 0, acc2, 10, &sign_key1,
        );
        let result = tx.transaction_stateless_check();

        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn transaction_pre_check_native_transfer_other_signature() {
        let (mut sequencer, _mempool_handle) = common_setup().await;

        let acc1 = initial_accounts()[0].account_id;
        let acc2 = initial_accounts()[1].account_id;

        let sign_key2 = create_signing_key_for_account2();

        let tx = common::test_utils::create_transaction_native_token_transfer(
            acc1, 0, acc2, 10, &sign_key2,
        );

        // Signature is valid, stateless check pass
        let tx = tx.transaction_stateless_check().unwrap();

        // Signature is not from sender. Execution fails
        let result = tx.execute_check_on_state(&mut sequencer.state, 0, 0);

        assert!(matches!(
            result,
            Err(nssa::error::NssaError::ProgramExecutionFailed(_))
        ));
    }

    #[tokio::test]
    async fn transaction_pre_check_native_transfer_sent_too_much() {
        let (mut sequencer, _mempool_handle) = common_setup().await;

        let acc1 = initial_accounts()[0].account_id;
        let acc2 = initial_accounts()[1].account_id;

        let sign_key1 = create_signing_key_for_account1();

        let tx = common::test_utils::create_transaction_native_token_transfer(
            acc1, 0, acc2, 10_000_000, &sign_key1,
        );

        let result = tx.transaction_stateless_check();

        // Passed pre-check
        assert!(result.is_ok());

        let result = result
            .unwrap()
            .execute_check_on_state(&mut sequencer.state, 0, 0);
        let is_failed_at_balance_mismatch = matches!(
            result.err().unwrap(),
            nssa::error::NssaError::ProgramExecutionFailed(_)
        );

        assert!(is_failed_at_balance_mismatch);
    }

    #[tokio::test]
    async fn transaction_execute_native_transfer() {
        let (mut sequencer, _mempool_handle) = common_setup().await;

        let acc1 = initial_accounts()[0].account_id;
        let acc2 = initial_accounts()[1].account_id;

        let sign_key1 = create_signing_key_for_account1();

        let tx = common::test_utils::create_transaction_native_token_transfer(
            acc1, 0, acc2, 100, &sign_key1,
        );

        tx.execute_check_on_state(&mut sequencer.state, 0, 0)
            .unwrap();

        let bal_from = sequencer.state.get_account_by_id(acc1).balance;
        let bal_to = sequencer.state.get_account_by_id(acc2).balance;

        assert_eq!(bal_from, 9900);
        assert_eq!(bal_to, 20100);
    }

    #[tokio::test]
    async fn push_tx_into_mempool_blocks_until_mempool_is_full() {
        let config = SequencerConfig {
            mempool_max_size: 1,
            ..setup_sequencer_config()
        };
        let (mut sequencer, mempool_handle) = common_setup_with_config(config).await;

        let tx = common::test_utils::produce_dummy_empty_transaction();

        // Fill the mempool
        mempool_handle.push(tx.clone()).await.unwrap();

        // Check that pushing another transaction will block
        let mut push_fut = pin!(mempool_handle.push(tx.clone()));
        let poll = futures::poll!(push_fut.as_mut());
        assert!(poll.is_pending());

        // Empty the mempool by producing a block
        sequencer.produce_new_block().await.unwrap();

        // Resolve the pending push
        assert!(push_fut.await.is_ok());
    }

    #[tokio::test]
    async fn build_block_from_mempool() {
        let (mut sequencer, mempool_handle) = common_setup().await;
        let genesis_height = sequencer.chain_height;

        let tx = common::test_utils::produce_dummy_empty_transaction();
        mempool_handle.push(tx).await.unwrap();

        let result = sequencer.build_block_from_mempool();
        assert!(result.is_ok());
        assert_eq!(sequencer.chain_height, genesis_height + 1);
    }

    #[tokio::test]
    async fn replay_transactions_are_rejected_in_the_same_block() {
        let (mut sequencer, mempool_handle) = common_setup().await;

        let acc1 = initial_accounts()[0].account_id;
        let acc2 = initial_accounts()[1].account_id;

        let sign_key1 = create_signing_key_for_account1();

        let tx = common::test_utils::create_transaction_native_token_transfer(
            acc1, 0, acc2, 100, &sign_key1,
        );

        let tx_original = tx.clone();
        let tx_replay = tx.clone();
        // Pushing two copies of the same tx to the mempool
        mempool_handle.push(tx_original).await.unwrap();
        mempool_handle.push(tx_replay).await.unwrap();

        // Create block
        sequencer.produce_new_block().await.unwrap();
        let block = sequencer
            .store
            .get_block_at_id(sequencer.chain_height)
            .unwrap()
            .unwrap();

        // Only one user tx should be included; the clock tx is always appended last.
        assert_eq!(
            block.body.transactions,
            vec![
                tx.clone(),
                NSSATransaction::Public(clock_invocation(block.header.timestamp))
            ]
        );
    }

    #[tokio::test]
    async fn replay_transactions_are_rejected_in_different_blocks() {
        let (mut sequencer, mempool_handle) = common_setup().await;

        let acc1 = initial_accounts()[0].account_id;
        let acc2 = initial_accounts()[1].account_id;

        let sign_key1 = create_signing_key_for_account1();

        let tx = common::test_utils::create_transaction_native_token_transfer(
            acc1, 0, acc2, 100, &sign_key1,
        );

        // The transaction should be included the first time
        mempool_handle.push(tx.clone()).await.unwrap();
        sequencer.produce_new_block().await.unwrap();
        let block = sequencer
            .store
            .get_block_at_id(sequencer.chain_height)
            .unwrap()
            .unwrap();
        assert_eq!(
            block.body.transactions,
            vec![
                tx.clone(),
                NSSATransaction::Public(clock_invocation(block.header.timestamp))
            ]
        );

        // Add same transaction should fail
        mempool_handle.push(tx.clone()).await.unwrap();
        sequencer.produce_new_block().await.unwrap();
        let block = sequencer
            .store
            .get_block_at_id(sequencer.chain_height)
            .unwrap()
            .unwrap();
        // The replay is rejected, so only the clock tx is in the block.
        assert_eq!(
            block.body.transactions,
            vec![NSSATransaction::Public(clock_invocation(
                block.header.timestamp
            ))]
        );
    }

    #[tokio::test]
    async fn restart_from_storage() {
        let config = setup_sequencer_config();
        let acc1_account_id = initial_accounts()[0].account_id;
        let acc2_account_id = initial_accounts()[1].account_id;
        let balance_to_move = 13;

        // In the following code block a transaction will be processed that moves `balance_to_move`
        // from `acc_1` to `acc_2`. The block created with that transaction will be kept stored in
        // the temporary directory for the block storage of this test.
        {
            let (mut sequencer, mempool_handle) =
                SequencerCoreWithMockClients::start_from_config(config.clone()).await;
            let signing_key = create_signing_key_for_account1();

            let tx = common::test_utils::create_transaction_native_token_transfer(
                acc1_account_id,
                0,
                acc2_account_id,
                balance_to_move,
                &signing_key,
            );

            mempool_handle.push(tx.clone()).await.unwrap();
            sequencer.produce_new_block().await.unwrap();
            let block = sequencer
                .store
                .get_block_at_id(sequencer.chain_height)
                .unwrap()
                .unwrap();
            assert_eq!(
                block.body.transactions,
                vec![
                    tx.clone(),
                    NSSATransaction::Public(clock_invocation(block.header.timestamp))
                ]
            );
        }

        // Instantiating a new sequencer from the same config. This should load the existing block
        // with the above transaction and update the state to reflect that.
        let (sequencer, _mempool_handle) =
            SequencerCoreWithMockClients::start_from_config(config.clone()).await;
        let balance_acc_1 = sequencer.state.get_account_by_id(acc1_account_id).balance;
        let balance_acc_2 = sequencer.state.get_account_by_id(acc2_account_id).balance;

        // Balances should be consistent with the stored block
        assert_eq!(
            balance_acc_1,
            initial_accounts()[0].balance - balance_to_move
        );
        assert_eq!(
            balance_acc_2,
            initial_accounts()[1].balance + balance_to_move
        );
    }

    #[tokio::test]
    async fn get_pending_blocks() {
        let config = setup_sequencer_config();
        let (mut sequencer, _mempool_handle) =
            SequencerCoreWithMockClients::start_from_config(config).await;
        sequencer.produce_new_block().await.unwrap();
        sequencer.produce_new_block().await.unwrap();
        sequencer.produce_new_block().await.unwrap();
        assert_eq!(sequencer.get_pending_blocks().unwrap().len(), 4);
    }

    #[tokio::test]
    async fn delete_blocks() {
        let config = setup_sequencer_config();
        let (mut sequencer, _mempool_handle) =
            SequencerCoreWithMockClients::start_from_config(config).await;
        sequencer.produce_new_block().await.unwrap();
        sequencer.produce_new_block().await.unwrap();
        sequencer.produce_new_block().await.unwrap();

        let last_finalized_block = 3;
        sequencer
            .clean_finalized_blocks_from_db(last_finalized_block)
            .unwrap();

        assert_eq!(sequencer.get_pending_blocks().unwrap().len(), 1);
    }

    #[tokio::test]
    async fn produce_block_with_correct_prev_meta_after_restart() {
        let config = setup_sequencer_config();
        let acc1_account_id = initial_accounts()[0].account_id;
        let acc2_account_id = initial_accounts()[1].account_id;

        // Step 1: Create initial database with some block metadata
        let expected_prev_meta = {
            let (mut sequencer, mempool_handle) =
                SequencerCoreWithMockClients::start_from_config(config.clone()).await;

            let signing_key = create_signing_key_for_account1();

            // Add a transaction and produce a block to set up block metadata
            let tx = common::test_utils::create_transaction_native_token_transfer(
                acc1_account_id,
                0,
                acc2_account_id,
                100,
                &signing_key,
            );

            mempool_handle.push(tx).await.unwrap();
            sequencer.produce_new_block().await.unwrap();

            // Get the metadata of the last block produced
            sequencer.store.latest_block_meta().unwrap()
        };

        // Step 2: Restart sequencer from the same storage
        let (mut sequencer, mempool_handle) =
            SequencerCoreWithMockClients::start_from_config(config.clone()).await;

        // Step 3: Submit a new transaction
        let signing_key = create_signing_key_for_account1();
        let tx = common::test_utils::create_transaction_native_token_transfer(
            acc1_account_id,
            1, // Next nonce
            acc2_account_id,
            50,
            &signing_key,
        );

        mempool_handle.push(tx.clone()).await.unwrap();

        // Step 4: Produce new block
        sequencer.produce_new_block().await.unwrap();

        // Step 5: Verify the new block has correct previous block metadata
        let new_block = sequencer
            .store
            .get_block_at_id(sequencer.chain_height)
            .unwrap()
            .unwrap();

        assert_eq!(
            new_block.header.prev_block_hash, expected_prev_meta.hash,
            "New block's prev_block_hash should match the stored metadata hash"
        );
        assert_eq!(
            new_block.body.transactions,
            vec![
                tx,
                NSSATransaction::Public(clock_invocation(new_block.header.timestamp))
            ],
            "New block should contain the submitted transaction and the clock invocation"
        );
    }

    #[tokio::test]
    async fn transactions_touching_clock_account_are_dropped_from_block() {
        let (mut sequencer, mempool_handle) = common_setup().await;

        // Canonical clock invocation and a crafted variant with a different timestamp — both must
        // be dropped because their diffs touch the clock accounts.
        let crafted_clock_tx = {
            let message = nssa::public_transaction::Message::try_new(
                nssa::program::Program::clock().id(),
                nssa::CLOCK_PROGRAM_ACCOUNT_IDS.to_vec(),
                vec![],
                42_u64,
            )
            .unwrap();
            NSSATransaction::Public(nssa::PublicTransaction::new(
                message,
                nssa::public_transaction::WitnessSet::from_raw_parts(vec![]),
            ))
        };
        mempool_handle
            .push(NSSATransaction::Public(clock_invocation(0)))
            .await
            .unwrap();
        mempool_handle.push(crafted_clock_tx).await.unwrap();
        sequencer.produce_new_block().await.unwrap();

        let block = sequencer
            .store
            .get_block_at_id(sequencer.chain_height)
            .unwrap()
            .unwrap();

        // Both transactions were dropped. Only the system-appended clock tx remains.
        assert_eq!(
            block.body.transactions,
            vec![NSSATransaction::Public(clock_invocation(
                block.header.timestamp
            ))]
        );
    }

    #[tokio::test]
    async fn user_tx_that_chain_calls_clock_is_dropped() {
        let (mut sequencer, mempool_handle) = common_setup().await;

        // Deploy the clock_chain_caller test program.
        let deploy_tx =
            NSSATransaction::ProgramDeployment(nssa::ProgramDeploymentTransaction::new(
                nssa::program_deployment_transaction::Message::new(
                    test_program_methods::CLOCK_CHAIN_CALLER_ELF.to_vec(),
                ),
            ));
        mempool_handle.push(deploy_tx).await.unwrap();
        sequencer.produce_new_block().await.unwrap();

        // Build a user transaction that invokes clock_chain_caller, which in turn chain-calls the
        // clock program with the clock accounts. The sequencer should detect that the resulting
        // state diff modifies clock accounts and drop the transaction.
        let clock_chain_caller_id =
            nssa::program::Program::new(test_program_methods::CLOCK_CHAIN_CALLER_ELF.to_vec())
                .unwrap()
                .id();
        let clock_program_id = nssa::program::Program::clock().id();
        let timestamp: u64 = 0;

        let message = nssa::public_transaction::Message::try_new(
            clock_chain_caller_id,
            nssa::CLOCK_PROGRAM_ACCOUNT_IDS.to_vec(),
            vec![], // no signers
            (clock_program_id, timestamp),
        )
        .unwrap();
        let user_tx = NSSATransaction::Public(nssa::PublicTransaction::new(
            message,
            nssa::public_transaction::WitnessSet::from_raw_parts(vec![]),
        ));

        mempool_handle.push(user_tx).await.unwrap();
        sequencer.produce_new_block().await.unwrap();

        let block = sequencer
            .store
            .get_block_at_id(sequencer.chain_height)
            .unwrap()
            .unwrap();

        // The user tx must have been dropped; only the mandatory clock invocation remains.
        assert_eq!(
            block.body.transactions,
            vec![NSSATransaction::Public(clock_invocation(
                block.header.timestamp
            ))]
        );
    }

    #[tokio::test]
    async fn block_production_aborts_when_clock_account_data_is_corrupted() {
        let (mut sequencer, mempool_handle) = common_setup().await;

        // Corrupt the clock 01 account data so the clock program panics on deserialization.
        let clock_account_id = nssa::CLOCK_01_PROGRAM_ACCOUNT_ID;
        let mut corrupted = sequencer.state.get_account_by_id(clock_account_id);
        corrupted.data = vec![0xff; 3].try_into().unwrap();
        sequencer
            .state
            .force_insert_account(clock_account_id, corrupted);

        // Push a dummy transaction so the mempool is non-empty.
        let tx = common::test_utils::produce_dummy_empty_transaction();
        mempool_handle.push(tx).await.unwrap();

        // Block production must fail because the appended clock tx cannot execute.
        let result = sequencer.produce_new_block().await;
        assert!(
            result.is_err(),
            "Block production should abort when clock account data is corrupted"
        );
    }
}