use crate::{StageCheckpoint, StageId};
use alloy_primitives::{BlockHash, BlockNumber};
use futures_util::{Stream, StreamExt};
use reqwest::{Client, Url};
use reth_config::config::EtlConfig;
use reth_db_api::{table::Value, transaction::DbTxMut};
use reth_era::{
    common::file_ops::{EraFileType, StreamReader},
    era::file::EraReader,
    era1::file::Era1Reader,
    ere::file::EreReader,
};
use reth_era_downloader::{read_dir, read_era_dir, EraClient, EraMeta, EraStream, EraStreamConfig};
use reth_era_utils as era;
use reth_etl::Collector;
use reth_primitives_traits::{FullBlockBody, FullBlockHeader, NodePrimitives};
use reth_provider::{
    BlockReader, BlockWriter, DBProvider, StageCheckpointWriter, StaticFileProviderFactory,
    StaticFileWriter,
};
use reth_stages_api::{ExecInput, ExecOutput, Stage, StageError, UnwindInput, UnwindOutput};
use reth_static_file_types::StaticFileSegment;
use std::{
    fmt::{Debug, Formatter},
    iter,
    path::Path,
    task::{ready, Context, Poll},
};

type Item<Header, Body> =
    Box<dyn Iterator<Item = eyre::Result<(Header, Body)>> + Send + Sync + Unpin>;
type ThreadSafeEraStream<Header, Body> =
    Box<dyn Stream<Item = eyre::Result<Item<Header, Body>>> + Send + Sync + Unpin>;

/// The ERA history stage.
///
/// Imports block headers and bodies from
/// [ERA1](https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era1.md) files,
/// which hold pre-merge execution blocks, and
/// [ERE](https://github.com/eth-clients/e2store-format-specs/blob/main/formats/ere.md) files,
/// which hold both pre-merge and post-merge execution blocks, and consensus-layer
/// [ERA](https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era.md) files, whose
/// post-merge beacon blocks embed an execution payload. The reader is selected per file from its
/// extension (`.era` vs `.era1` vs `.ere`/`.erae`).
///
/// Imports block headers and bodies up to the configured target block. Receipts are generated by
/// execution, which is not done in this stage.
pub struct EraStage<Header, Body, StreamFactory> {
    /// The `source` creates `stream`.
    source: Option<StreamFactory>,
    /// A map of block hash to block height collected when processing headers and inserted into
    /// database afterward.
    hash_collector: Collector<BlockHash, BlockNumber>,
    /// Last extracted iterator of block `Header` and `Body` pairs.
    item: Option<Item<Header, Body>>,
    /// A stream of [`Item`]s, i.e. iterators over block `Header` and `Body` pairs.
    stream: Option<ThreadSafeEraStream<Header, Body>>,
}

trait EraStreamFactory<Header, Body> {
    fn create(self, input: ExecInput) -> Result<ThreadSafeEraStream<Header, Body>, StageError>;
}

impl<Header, Body> EraStreamFactory<Header, Body> for EraImportSource
where
    Header: FullBlockHeader + Value,
    Body: FullBlockBody<OmmerHeader = Header>,
{
    fn create(self, input: ExecInput) -> Result<ThreadSafeEraStream<Header, Body>, StageError> {
        match self {
            Self::Path(path) => {
                // Consensus `.era` files are slot-indexed and ship no `checksums.txt`, so they use
                // a different directory reader than the block-indexed `.era1`/`.ere` files.
                if EraFileType::from_dir(&path).map_err(|e| StageError::Fatal(e.into()))? ==
                    Some(EraFileType::Era)
                {
                    Self::convert(read_era_dir(path).map_err(|e| StageError::Fatal(e.into()))?)
                } else {
                    Self::convert(
                        read_dir(path, input.next_block())
                            .map_err(|e| StageError::Fatal(e.into()))?,
                    )
                }
            }
            Self::Url(url, folder) => {
                let _ = reth_fs_util::create_dir_all(&folder);
                // `start_from` maps a block number to a file index, valid only for block-indexed
                // `.era1`/`.ere`. Consensus `.era` files are slot-indexed, so stream from 0 and let
                // the pipeline skip already-imported blocks.
                let is_era = EraFileType::from_url(url.as_str()) == EraFileType::Era;
                let client = EraClient::new(Client::new(), url, folder);
                let mut config = EraStreamConfig::default();
                if !is_era {
                    config = config.start_from(input.next_block());
                }

                Self::convert(EraStream::new(client, config))
            }
        }
    }
}

impl EraImportSource {
    fn convert<Header, Body>(
        stream: impl Stream<Item = eyre::Result<impl EraMeta + Send + Sync + 'static + Unpin>>
            + Send
            + Sync
            + 'static
            + Unpin,
    ) -> Result<ThreadSafeEraStream<Header, Body>, StageError>
    where
        Header: FullBlockHeader + Value,
        Body: FullBlockBody<OmmerHeader = Header>,
    {
        Ok(Box::new(Box::pin(stream.map(|meta| {
            meta.and_then(|meta| {
                // The reader is selected per file from its extension: consensus `.era` (beacon
                // blocks with an embedded execution payload), `.ere`/`.erae`, or `.era1`.
                let file = reth_fs_util::open(meta.path())?;
                let iter = match meta
                    .path()
                    .file_name()
                    .and_then(|name| name.to_str())
                    .and_then(EraFileType::from_filename)
                {
                    Some(EraFileType::Era) => {
                        // `.era` blocks decode to an `Option` (pre-merge slots carry no payload and
                        // are skipped); `buf` is scratch space reused for the RLP re-encode.
                        let mut buf = Vec::new();
                        Box::new(
                            EraReader::new(file).iter().filter_map(move |block| {
                                era::Era::decode(block, &mut buf).transpose()
                            }),
                        ) as Item<Header, Body>
                    }
                    Some(EraFileType::Ere) => {
                        Box::new(EreReader::new(file).iter().map(era::Ere::decode))
                            as Item<Header, Body>
                    }
                    _ => Box::new(Era1Reader::new(file).iter().map(era::decode))
                        as Item<Header, Body>,
                };

                let iter = iter.chain(
                    iter::once_with(move || match meta.mark_as_processed() {
                        Ok(..) => None,
                        Err(e) => Some(Err(e)),
                    })
                    .flatten(),
                );

                Ok(Box::new(iter) as Item<Header, Body>)
            })
        }))))
    }
}

impl<Header: Debug, Body: Debug, F: Debug> Debug for EraStage<Header, Body, F> {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("EraStage")
            .field("source", &self.source)
            .field("hash_collector", &self.hash_collector)
            .field("item", &self.item.is_some())
            .field("stream", &"dyn Stream")
            .finish()
    }
}

impl<Header, Body, F> EraStage<Header, Body, F> {
    /// Creates a new [`EraStage`].
    pub fn new(source: Option<F>, etl_config: EtlConfig) -> Self {
        Self {
            source,
            item: None,
            stream: None,
            hash_collector: Collector::new(etl_config.file_size, etl_config.dir),
        }
    }
}

impl<Provider, N, F> Stage<Provider> for EraStage<N::BlockHeader, N::BlockBody, F>
where
    Provider: DBProvider<Tx: DbTxMut>
        + StaticFileProviderFactory<Primitives = N>
        + BlockWriter<Block = N::Block>
        + BlockReader<Block = N::Block>
        + StageCheckpointWriter,
    F: EraStreamFactory<N::BlockHeader, N::BlockBody> + Send + Sync + Clone,
    N: NodePrimitives<BlockHeader: Value>,
{
    fn id(&self) -> StageId {
        StageId::Era
    }

    fn poll_execute_ready(
        &mut self,
        cx: &mut Context<'_>,
        input: ExecInput,
    ) -> Poll<Result<(), StageError>> {
        if input.target_reached() || self.item.is_some() {
            return Poll::Ready(Ok(()));
        }

        if self.stream.is_none() &&
            let Some(source) = self.source.clone()
        {
            self.stream.replace(source.create(input)?);
        }
        if let Some(stream) = &mut self.stream &&
            let Some(next) = ready!(stream.poll_next_unpin(cx))
                .transpose()
                .map_err(|e| StageError::Fatal(e.into()))?
        {
            self.item.replace(next);
        }

        Poll::Ready(Ok(()))
    }

    fn execute(&mut self, provider: &Provider, input: ExecInput) -> Result<ExecOutput, StageError> {
        let height = if let Some(era) = self.item.take() {
            let static_file_provider = provider.static_file_provider();

            // Consistency check of expected headers in static files vs DB is done on
            // provider::sync_gap when poll_execute_ready is polled.
            let last_header_number = static_file_provider
                .get_highest_static_file_block(StaticFileSegment::Headers)
                .unwrap_or_default();

            // Although headers were downloaded in reverse order, the collector iterates it in
            // ascending order
            let mut writer = static_file_provider.latest_writer(StaticFileSegment::Headers)?;

            // This stage never persists receipts, execution does that separately.
            let height = era::process_iter(
                era.map(|r| r.map(|(header, body)| (header, body, None))),
                &mut writer,
                None,
                provider,
                &mut self.hash_collector,
                last_header_number..=input.target(),
                era::ImportPolicy {
                    headers_tip: last_header_number,
                    is_receipt_verifiable: &|_| false,
                },
            )
            .map_err(|e| StageError::Fatal(e.into()))?;

            if !self.hash_collector.is_empty() {
                era::build_index(provider, &mut self.hash_collector)
                    .map_err(|e| StageError::Recoverable(e.into()))?;
                self.hash_collector.clear();
            }

            era::save_stage_checkpoints(
                provider,
                input.checkpoint().block_number,
                height,
                height,
                input.target(),
            )?;

            height
        } else {
            // No era files to process. Return the highest block we're aware of to avoid
            // limiting subsequent stages with an outdated checkpoint.
            //
            // This can happen when:
            // 1. Era import is complete (all pre-merge blocks imported)
            // 2. No era import source was configured
            //
            // We return max(checkpoint, highest_header, target) to ensure we don't return
            // a stale checkpoint that could limit subsequent stages like Headers.
            let highest_header = provider
                .static_file_provider()
                .get_highest_static_file_block(StaticFileSegment::Headers)
                .unwrap_or_default();

            let checkpoint = input.checkpoint().block_number;
            let from_target = input.target.unwrap_or(checkpoint);

            checkpoint.max(highest_header).max(from_target)
        };

        Ok(ExecOutput { checkpoint: StageCheckpoint::new(height), done: height >= input.target() })
    }

    fn unwind(
        &mut self,
        _provider: &Provider,
        input: UnwindInput,
    ) -> Result<UnwindOutput, StageError> {
        Ok(UnwindOutput { checkpoint: input.checkpoint.with_block_number(input.unwind_to) })
    }
}

/// Describes where to get the era files from.
#[derive(Debug, Clone)]
pub enum EraImportSource {
    /// Remote HTTP accessible host.
    Url(Url, Box<Path>),
    /// Local directory.
    Path(Box<Path>),
}

impl EraImportSource {
    /// Maybe constructs a new `EraImportSource` depending on the arguments.
    ///
    /// Only one of `url` or `path` should be provided, but upholding this invariant is delegated
    /// above so that both parameters can be accepted.
    ///
    /// # Arguments
    /// * The `path` uses a directory as the import source. It and its contents must be readable.
    /// * The `url` uses an HTTP client to list and download files.
    /// * The `default` gives the default [`Url`] if none of the previous parameters are provided.
    /// * For any [`Url`] the `folder` is used as the download directory for storing files
    ///   temporarily. It and its contents must be readable and writable.
    pub fn maybe_new(
        path: Option<Box<Path>>,
        url: Option<Url>,
        default: impl FnOnce() -> Option<Url>,
        folder: impl FnOnce() -> Box<Path>,
    ) -> Option<Self> {
        path.map(Self::Path).or_else(|| url.or_else(default).map(|url| Self::Url(url, folder())))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::test_utils::{
        stage_test_suite, ExecuteStageTestRunner, StageTestRunner, UnwindStageTestRunner,
    };
    use alloy_consensus::{BlockBody, Header};
    use alloy_primitives::B256;
    use assert_matches::assert_matches;
    use futures_util::stream;
    use reth_db_api::tables;
    use reth_era::{
        common::file_ops::{EraFileFormat, StreamWriter},
        ere::{
            file::{EreFile, EreWriter},
            types::{
                execution::{BlockTuple, CompressedBody, CompressedHeader},
                group::{DynamicBlockIndex, EreGroup, EreId},
            },
        },
    };
    use reth_ethereum_primitives::{Block, TransactionSigned};
    use reth_primitives_traits::SealedBlock;
    use reth_provider::BlockHashReader;
    use reth_testing_utils::generators::{
        self, random_block_range, random_header, BlockRangeParams,
    };
    use std::{fs::File, path::PathBuf};
    use test_runner::EraTestRunner;

    #[tokio::test]
    async fn test_era_range_ends_below_target() {
        let era_cap = 2;
        let target = 20000;

        let mut runner = EraTestRunner::default();

        let input = ExecInput { target: Some(era_cap), checkpoint: None };
        runner.seed_execution(input).unwrap();

        let input = ExecInput { target: Some(target), checkpoint: None };
        let output = runner.execute(input).await.unwrap();

        runner.commit();

        assert_matches!(
            output,
            Ok(ExecOutput {
                checkpoint: StageCheckpoint { block_number, stage_checkpoint: None },
                done: false
            }) if block_number == era_cap
        );

        let output = output.unwrap();
        let validation_output = runner.validate_execution(input, Some(output.clone()));

        assert_matches!(validation_output, Ok(()));

        runner.take_responses();

        let input = ExecInput { target: Some(target), checkpoint: Some(output.checkpoint) };
        let output = runner.execute(input).await.unwrap();

        runner.commit();

        assert_matches!(
            output,
            Ok(ExecOutput {
                checkpoint: StageCheckpoint { block_number, stage_checkpoint: None },
                done: true
            }) if block_number == target
        );

        let validation_output = runner.validate_execution(input, output.ok());

        assert_matches!(validation_output, Ok(()));
    }

    /// Writes `blocks` to a header+body-only ERE file at `path`.
    fn write_ere_file(path: &Path, blocks: &[SealedBlock<Block>]) {
        let tuples = blocks
            .iter()
            .map(|block| {
                BlockTuple::new(
                    CompressedHeader::from_header(block.header()).unwrap(),
                    CompressedBody::from_body(block.body()).unwrap(),
                )
            })
            .collect::<Vec<_>>();

        let start = blocks[0].number;
        // Header + body only is component-count 2. The index offsets are stored verbatim and not
        // interpreted when assembling blocks, so placeholder values are fine.
        let component_count = 2;
        let offsets = vec![0; blocks.len() * component_count as usize];
        let index = DynamicBlockIndex::new(start, component_count, offsets);
        let group = EreGroup::new(tuples, None, index);
        let file = EreFile::new(group, EreId::new("mainnet", start, blocks.len() as u32));

        EreWriter::new(File::create(path).unwrap()).write_file(&file).unwrap();
    }

    #[derive(Debug)]
    struct EreTestMeta {
        path: PathBuf,
    }

    impl EraMeta for EreTestMeta {
        fn mark_as_processed(&self) -> eyre::Result<()> {
            Ok(())
        }

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

    #[tokio::test]
    async fn convert_decodes_ere_files() {
        let mut rng = generators::rng();
        let blocks = random_block_range(
            &mut rng,
            1..=3,
            BlockRangeParams { tx_count: 1..3, ..Default::default() },
        );

        let dir = tempfile::tempdir().unwrap();

        // ERE files are published with either the `.ere` or `.erae` extension; both must be
        // recognized and routed to the ERE reader rather than decoded as era1.
        for ext in ["ere", "erae"] {
            let path = dir.path().join(format!("mainnet-00000-abcd1234.{ext}"));
            write_ere_file(&path, &blocks);

            let stream =
                stream::iter(vec![Ok::<_, eyre::Error>(EreTestMeta { path: path.clone() })]);
            let mut stream =
                EraImportSource::convert::<Header, BlockBody<TransactionSigned>>(stream).unwrap();

            let item = stream.next().await.expect("a file to decode").expect("decoding to succeed");
            let decoded = item.collect::<eyre::Result<Vec<_>>>().unwrap();

            assert_eq!(
                decoded.len(),
                blocks.len(),
                "ERE file with `.{ext}` extension should decode every block"
            );
            for ((header, body), block) in decoded.iter().zip(&blocks) {
                assert_eq!(header, block.header());
                assert_eq!(body, block.body());
            }
        }
    }

    mod test_runner {
        use super::*;
        use crate::test_utils::{TestRunnerError, TestStageDB};
        use alloy_consensus::{BlockBody, Header};
        use futures_util::stream;
        use reth_db_api::{
            cursor::DbCursorRO,
            models::{StoredBlockBodyIndices, StoredBlockOmmers},
            transaction::DbTx,
        };
        use reth_ethereum_primitives::TransactionSigned;
        use reth_primitives_traits::{SealedBlock, SealedHeader};
        use reth_provider::{BlockNumReader, HeaderProvider, TransactionsProvider};
        use reth_testing_utils::generators::{
            random_block_range, random_signed_tx, BlockRangeParams,
        };
        use tokio::sync::watch;

        pub(crate) struct EraTestRunner {
            channel: (watch::Sender<B256>, watch::Receiver<B256>),
            db: TestStageDB,
            responses: Option<Vec<(Header, BlockBody<TransactionSigned>)>>,
        }

        impl Default for EraTestRunner {
            fn default() -> Self {
                Self {
                    channel: watch::channel(B256::ZERO),
                    db: TestStageDB::default(),
                    responses: Default::default(),
                }
            }
        }

        impl StageTestRunner for EraTestRunner {
            type S = EraStage<Header, BlockBody<TransactionSigned>, StubResponses>;

            fn db(&self) -> &TestStageDB {
                &self.db
            }

            fn stage(&self) -> Self::S {
                EraStage::new(self.responses.clone().map(StubResponses), EtlConfig::default())
            }
        }

        impl ExecuteStageTestRunner for EraTestRunner {
            type Seed = Vec<SealedBlock<reth_ethereum_primitives::Block>>;

            fn seed_execution(&mut self, input: ExecInput) -> Result<Self::Seed, TestRunnerError> {
                let start = input.checkpoint().block_number;
                let end = input.target();

                let static_file_provider = self.db.factory.static_file_provider();

                let mut rng = generators::rng();

                // Static files do not support gaps in headers, so we need to generate 0 to end
                let blocks = random_block_range(
                    &mut rng,
                    0..=end,
                    BlockRangeParams {
                        parent: Some(B256::ZERO),
                        tx_count: 0..2,
                        ..Default::default()
                    },
                );
                self.db.insert_headers(blocks.iter().map(|block| block.sealed_header()))?;
                if let Some(progress) = blocks.get(start as usize) {
                    // Insert last progress data
                    {
                        let tx = self.db.factory.provider_rw()?.into_tx();
                        let mut static_file_producer = static_file_provider
                            .get_writer(start, StaticFileSegment::Transactions)?;

                        let body = StoredBlockBodyIndices {
                            first_tx_num: 0,
                            tx_count: progress.transaction_count() as u64,
                        };

                        static_file_producer.set_block_range(0..=progress.number);

                        body.tx_num_range().try_for_each(|tx_num| {
                            let transaction = random_signed_tx(&mut rng);
                            static_file_producer.append_transaction(tx_num, &transaction).map(drop)
                        })?;

                        if body.tx_count != 0 {
                            tx.put::<tables::TransactionBlocks>(
                                body.last_tx_num(),
                                progress.number,
                            )?;
                        }

                        tx.put::<tables::BlockBodyIndices>(progress.number, body)?;

                        if !progress.ommers_hash_is_empty() {
                            tx.put::<tables::BlockOmmers>(
                                progress.number,
                                StoredBlockOmmers { ommers: progress.body().ommers.clone() },
                            )?;
                        }

                        static_file_producer.commit()?;
                        tx.commit()?;
                    }
                }
                self.responses.replace(
                    blocks.iter().map(|v| (v.header().clone(), v.body().clone())).collect(),
                );
                Ok(blocks)
            }

            /// Validate stored headers and bodies
            fn validate_execution(
                &self,
                input: ExecInput,
                output: Option<ExecOutput>,
            ) -> Result<(), TestRunnerError> {
                let initial_checkpoint = input.checkpoint().block_number;
                match output {
                    Some(output) if output.checkpoint.block_number > initial_checkpoint => {
                        let provider = self.db.factory.provider()?;

                        for block_num in initial_checkpoint..
                            output
                                .checkpoint
                                .block_number
                                .min(self.responses.as_ref().map(|v| v.len()).unwrap_or_default()
                                    as BlockNumber)
                        {
                            // look up the header hash
                            let hash = provider.block_hash(block_num)?.expect("no header hash");

                            // validate the header number
                            assert_eq!(provider.block_number(hash)?, Some(block_num));

                            // validate the header
                            let header = provider.header_by_number(block_num)?;
                            assert!(header.is_some());
                            let header = SealedHeader::seal_slow(header.unwrap());
                            assert_eq!(header.hash(), hash);
                        }

                        self.validate_db_blocks(
                            output.checkpoint.block_number,
                            output.checkpoint.block_number,
                        )?;
                    }
                    _ => self.check_no_header_entry_above(initial_checkpoint)?,
                };
                Ok(())
            }

            async fn after_execution(&self, headers: Self::Seed) -> Result<(), TestRunnerError> {
                let tip = if headers.is_empty() {
                    let tip = random_header(&mut generators::rng(), 0, None);
                    self.db.insert_headers(iter::once(&tip))?;
                    tip.hash()
                } else {
                    headers.last().unwrap().hash()
                };
                self.send_tip(tip);
                Ok(())
            }
        }

        impl UnwindStageTestRunner for EraTestRunner {
            fn validate_unwind(&self, _input: UnwindInput) -> Result<(), TestRunnerError> {
                Ok(())
            }
        }

        impl EraTestRunner {
            pub(crate) fn check_no_header_entry_above(
                &self,
                block: BlockNumber,
            ) -> Result<(), TestRunnerError> {
                self.db
                    .ensure_no_entry_above_by_value::<tables::HeaderNumbers, _>(block, |val| val)?;
                self.db.ensure_no_entry_above::<tables::CanonicalHeaders, _>(block, |key| key)?;
                self.db.ensure_no_entry_above::<tables::Headers, _>(block, |key| key)?;
                Ok(())
            }

            pub(crate) fn send_tip(&self, tip: B256) {
                self.channel.0.send(tip).expect("failed to send tip");
            }

            /// Validate that the inserted block data is valid
            pub(crate) fn validate_db_blocks(
                &self,
                prev_progress: BlockNumber,
                highest_block: BlockNumber,
            ) -> Result<(), TestRunnerError> {
                let static_file_provider = self.db.factory.static_file_provider();

                self.db.query(|tx| {
                    // Acquire cursors on body related tables
                    let mut bodies_cursor = tx.cursor_read::<tables::BlockBodyIndices>()?;
                    let mut ommers_cursor = tx.cursor_read::<tables::BlockOmmers>()?;
                    let mut tx_block_cursor = tx.cursor_read::<tables::TransactionBlocks>()?;

                    let first_body_key = match bodies_cursor.first()? {
                        Some((key, _)) => key,
                        None => return Ok(()),
                    };

                    let mut prev_number: Option<BlockNumber> = None;


                    for entry in bodies_cursor.walk(Some(first_body_key))? {
                        let (number, body) = entry?;

                        // Validate sequentiality only after prev progress,
                        // since the data before is mocked and can contain gaps
                        if number > prev_progress
                            && let Some(prev_key) = prev_number {
                                assert_eq!(prev_key + 1, number, "Body entries must be sequential");
                            }

                        // Validate that the current entry is below or equals to the highest allowed block
                        assert!(
                            number <= highest_block,
                            "We wrote a block body outside of our synced range. Found block with number {number}, highest block according to stage is {highest_block}",
                        );

                        let header = static_file_provider.header_by_number(number)?.expect("to be present");
                        // Validate that ommers exist if any
                        let stored_ommers =  ommers_cursor.seek_exact(number)?;
                        if header.ommers_hash_is_empty() {
                            assert!(stored_ommers.is_none(), "Unexpected ommers entry");
                        } else {
                            assert!(stored_ommers.is_some(), "Missing ommers entry");
                        }

                        let tx_block_id = tx_block_cursor.seek_exact(body.last_tx_num())?.map(|(_,b)| b);
                        if body.tx_count == 0 {
                            assert_ne!(tx_block_id,Some(number));
                        } else {
                            assert_eq!(tx_block_id, Some(number));
                        }

                        for tx_id in body.tx_num_range() {
                            assert!(static_file_provider.transaction_by_id(tx_id)?.is_some(), "Transaction is missing.");
                        }

                        prev_number = Some(number);
                    }
                    Ok(())
                })?;
                Ok(())
            }

            pub(crate) fn take_responses(&mut self) {
                self.responses.take();
            }

            pub(crate) fn commit(&self) {
                self.db.factory.static_file_provider().commit().unwrap();
            }
        }

        #[derive(Clone)]
        pub(crate) struct StubResponses(Vec<(Header, BlockBody<TransactionSigned>)>);

        impl EraStreamFactory<Header, BlockBody<TransactionSigned>> for StubResponses {
            fn create(
                self,
                _input: ExecInput,
            ) -> Result<ThreadSafeEraStream<Header, BlockBody<TransactionSigned>>, StageError>
            {
                let stream = stream::iter(vec![self.0]);

                Ok(Box::new(Box::pin(stream.map(|meta| {
                    Ok(Box::new(meta.into_iter().map(Ok))
                        as Item<Header, BlockBody<TransactionSigned>>)
                }))))
            }
        }
    }

    stage_test_suite!(EraTestRunner, era);
}
