use crate::chain::Chain;
use crate::chain::NewHeader;
use crate::chaindef::Block;
use crate::chaindef::BlockHash;
use crate::chaindef::BlockHeader;
use crate::chaindef::OutPointHash;
use crate::chaindef::ScriptHash;
use crate::chaindef::{Transaction, TxIn};
use crate::config::Config;
use crate::daemon::p2pconnection::duration_to_seconds;
use crate::daemon::Daemon;
use crate::encode::compute_outpoint_hash;
use crate::indexes::headerindex::HeaderRow;
use crate::indexes::heightindex::HeightIndexRow;
use crate::indexes::inputindex::InputIndexRow;
use crate::indexes::outputindex::OutputIndexRow;
use crate::indexes::scripthashindex::OutputFlags;
use crate::indexes::scripthashindex::ScriptHashIndexRow;
use crate::indexes::unspentindex::UnspentIndexRow;
use crate::indexes::DBRow;
use crate::indexes::{
    outputtokenindex::OutputTokenIndexRow, tokenoutputindex::TokenOutputIndexRow,
};
use crate::metrics;
use crate::metrics::Metrics;
use crate::undo::{BlockUndoer, StoreBlockUndoer};

use crate::signal::Waiter;
use crate::store;
use crate::store::db_decode;
use crate::store::db_encode;
use crate::store::DBContents;
use crate::store::DBStore;
use crate::store::Row;
use crate::store::METADATA_LAST_INDEXED_BLOCK;
use crate::store::META_CF;
use crate::thread;
use crate::util;
use crate::writebatch::SpentUpdateBatch;
use crate::writebatch::WriteBatch;

use anyhow::Context;
use anyhow::Result;
use bitcoin_hashes::hex::ToHex;
use bitcoin_hashes::Hash;
use rayon::prelude::*;
use std::collections::HashMap;
use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;
use std::time::Instant;
use tokio::sync::Mutex;

struct Stats {
    update_duration: prometheus::HistogramVec,
    blocks: prometheus::IntCounter,
    txns: prometheus::IntCounter,
    vsize: prometheus::IntCounter,
    height: prometheus::IntGauge,
}

impl Stats {
    fn new(metrics: &Metrics) -> Stats {
        Stats {
            update_duration: metrics.histogram_vec(
                "rostrum_update_duration",
                "Indexing update duration (in seconds)",
                &["step"],
                metrics::default_duration_buckets(),
            ),
            blocks: metrics.counter_int(prometheus::Opts::new(
                "rostrum_index_blocks",
                "# of indexed blocks",
            )),
            txns: metrics.counter_int(prometheus::Opts::new(
                "rostrum_index_txns",
                "# of indexed transactions",
            )),
            vsize: metrics.counter_int(prometheus::Opts::new(
                "rostrum_index_vsize",
                "# of indexed vbytes",
            )),
            height: metrics.gauge_int(prometheus::Opts::new(
                "rostrum_index_height",
                "Last indexed block's height",
            )),
        }
    }

    fn update(&self, block: &Block, height: usize) {
        self.blocks.inc();
        self.txns.inc_by(block.txdata.len() as u64);
        for tx in &block.txdata {
            self.vsize.inc_by(tx.size() as u64);
        }
        self.update_height(height);
    }

    fn update_height(&self, height: usize) {
        self.height.set(height as i64);
    }

    fn start_timer(&self, step: &str) -> prometheus::HistogramTimer {
        self.update_duration
            .with_label_values(&[step])
            .start_timer()
    }

    async fn observe_chain(&self, chain: &Chain) {
        self.height.set(chain.height().await as i64)
    }
}

#[cfg(bch)]
#[inline]
pub(crate) fn is_coinbase_input(input: &TxIn) -> bool {
    use bitcoincash::Txid;
    let null_hash: Txid = Txid::all_zeros();
    input.previous_output.txid == null_hash
}
#[cfg(nexa)]
#[inline]
pub(crate) fn is_coinbase_input(_input: &TxIn) -> bool {
    // nexa does not have inputs in coinbase
    false
}
#[cfg(bch)]
#[inline]
fn get_outpoint_hash(input: &TxIn) -> [u8; 32] {
    compute_outpoint_hash(&input.previous_output.txid, input.previous_output.vout).into_inner()
}
#[cfg(nexa)]
#[inline]
fn get_outpoint_hash(input: &TxIn) -> [u8; 32] {
    input.previous_output.hash.into_inner()
}

#[cfg(nexa)]
fn index_outputs(writebatch: &WriteBatch, tx: &Transaction, height: u32) {
    let txid = tx.txid();
    let rows = tx.output.par_iter().enumerate().map(move |(i, output)| {
        OutputIndexRow::new(
            &txid,
            &compute_outpoint_hash(&tx.txidem(), i as u32),
            output.value,
            i as u32,
            height,
        )
        .to_row()
    });
    writebatch.insert(OutputIndexRow::CF, rows);
}

#[cfg(bch)]
fn index_outputs(writebatch: &WriteBatch, tx: &Transaction, height: u32) {
    let txid = tx.txid();
    let rows = tx.output.par_iter().enumerate().map(move |(i, output)| {
        OutputIndexRow::new(
            &txid,
            &compute_outpoint_hash(&txid, i as u32),
            output.value,
            i as u32,
            height,
        )
        .to_row()
    });
    writebatch.insert(OutputIndexRow::CF, rows);
}

#[cfg(nexa)]
fn index_scriptsig(writebatch: &WriteBatch, tx: &Transaction, height: u32) {
    use crate::indexes::outtoscriptindex::OutToScripthashIndex;

    let rows: Vec<(Row, Row)> = tx
        .output
        .par_iter()
        .enumerate()
        .map(move |(index, output)| {
            let scripthash = ScriptHash::normalized_from_txout(output);
            let oph = compute_outpoint_hash(&tx.txidem(), index as u32);

            let scripthash_to_out = ScriptHashIndexRow::new_funding(
                &scripthash,
                &oph,
                if output.has_token() {
                    OutputFlags::FundingHasTokens
                } else {
                    OutputFlags::FundingNone
                },
                tx.txid().into_inner(),
                index as u32,
                height,
            )
            .to_row();

            let out_to_scripthash =
                OutToScripthashIndex::new(scripthash, oph, output.has_token()).to_row();

            (scripthash_to_out, out_to_scripthash)
        })
        .collect();

    let (from_scripthash, to_scripthash): (Vec<Row>, Vec<Row>) = rows.into_iter().unzip();

    writebatch.insert(ScriptHashIndexRow::CF, from_scripthash);
    writebatch.insert(OutToScripthashIndex::CF, to_scripthash);
}

#[cfg(bch)]
fn index_scriptsig(writebatch: &WriteBatch, tx: &Transaction, height: u32) {
    use crate::indexes::outtoscriptindex::OutToScripthashIndex;

    let txid = tx.txid();
    let rows = tx
        .output
        .par_iter()
        .enumerate()
        .map(move |(index, output)| {
            let scripthash = ScriptHash::from_script(&output.script_pubkey);
            let oph = compute_outpoint_hash(&txid, index as u32);

            let scripthash_to_out = ScriptHashIndexRow::new_funding(
                &scripthash,
                &oph,
                if output.has_token() {
                    OutputFlags::FundingHasTokens
                } else {
                    OutputFlags::FundingNone
                },
                tx.txid().into_inner(),
                index as u32,
                height,
            )
            .to_row();

            let out_to_scripthash =
                OutToScripthashIndex::new(scripthash, oph, output.has_token()).to_row();

            (scripthash_to_out, out_to_scripthash)
        })
        .collect::<Vec<_>>();
    let (from_scripthash, to_scripthash): (Vec<Row>, Vec<Row>) = rows.into_iter().unzip();

    writebatch.insert(ScriptHashIndexRow::CF, from_scripthash);
    writebatch.insert(OutToScripthashIndex::CF, to_scripthash);
}

#[cfg(nexa)]
fn index_token_outputs(writebatch: &WriteBatch, tx: &Transaction) {
    use crate::nexa::token::parse_token_from_scriptpubkey;
    tx.output.par_iter().enumerate().for_each(move |(i, out)| {
        if !out.has_token() {
            return;
        }
        if let Some((token, amount)) = parse_token_from_scriptpubkey(&out.script_pubkey) {
            let outpoint_hash = &compute_outpoint_hash(&tx.txidem(), i as u32);
            let out_token_row =
                OutputTokenIndexRow::new(outpoint_hash, token.clone(), amount).to_row();
            let token_out_row = TokenOutputIndexRow::new(token, outpoint_hash).to_row();
            writebatch.insert(OutputTokenIndexRow::CF, rayon::iter::once(out_token_row));
            writebatch.insert(TokenOutputIndexRow::CF, rayon::iter::once(token_out_row));
        };
    });
}

#[cfg(bch)]
fn index_token_outputs(writebatch: &WriteBatch, tx: &Transaction) {
    tx.output.par_iter().enumerate().for_each(move |(i, out)| {
        if let Some(t) = out.token.as_ref() {
            let outpoint_hash = compute_outpoint_hash(&tx.txid(), i as u32);
            let out_token_row = OutputTokenIndexRow::new(
                &outpoint_hash,
                t.id.clone(),
                t.commitment.clone(),
                t.amount,
            )
            .to_row();
            let token_out_row =
                TokenOutputIndexRow::new(t.id, t.commitment.clone(), &outpoint_hash).to_row();
            writebatch.insert(OutputTokenIndexRow::CF, rayon::iter::once(out_token_row));
            writebatch.insert(TokenOutputIndexRow::CF, rayon::iter::once(token_out_row));
        }
    });
}

fn index_unspent(
    write: &WriteBatch,
    spentupdatebatch: &SpentUpdateBatch,
    tx: &Transaction,
    height: u32,
) {
    let txid = tx.txid();

    // Collect spending metadata (which includes outpoints) in one pass
    for (vin, input) in tx.input.iter().enumerate() {
        if is_coinbase_input(input) {
            continue;
        }
        let outpoint = OutPointHash::from_inner(get_outpoint_hash(input));
        spentupdatebatch.insert_spending_metadata(outpoint, txid, vin as u32, height, None);
    }

    #[cfg(nexa)]
    let txid_or_txidem = tx.txidem();
    #[cfg(bch)]
    let txid_or_txidem = tx.txid();

    let rows = tx.output.par_iter().enumerate().map(|(index, out)| {
        let scripthash = ScriptHash::normalized_from_txout(out);
        let oph = compute_outpoint_hash(&txid_or_txidem, index as u32);

        UnspentIndexRow::new(&scripthash, &oph, out.has_token()).to_row()
    });
    write.insert(UnspentIndexRow::CF, rows);
}

pub fn index_transaction(
    writebatch: &WriteBatch,
    spentupdatebatch: &SpentUpdateBatch,
    txn: &Transaction,
    height: usize,
) {
    let txid = txn.txid();

    let inputs = txn
        .input
        .par_iter()
        .enumerate()
        .filter_map(move |(index, input)| {
            if is_coinbase_input(input) {
                None
            } else {
                let outpointhash = get_outpoint_hash(input);
                Some(InputIndexRow::new(outpointhash, txid, index as u32).to_row())
            }
        });
    writebatch.insert(InputIndexRow::CF, inputs);
    index_outputs(writebatch, txn, height as u32);
    index_unspent(writebatch, spentupdatebatch, txn, height as u32);

    index_scriptsig(writebatch, txn, height as u32);

    index_token_outputs(writebatch, txn);

    writebatch.insert(
        HeightIndexRow::CF,
        rayon::iter::once(HeightIndexRow::new(&txid, height as u32).to_row()),
    );
}

pub(crate) fn index_block(
    writebatch: &WriteBatch,
    spentupdatebatch: &SpentUpdateBatch,
    block: &Block,
    height: usize,
) {
    let blockhash = block.block_hash();
    let header_row = HeaderRow::new(&block.header).to_row();

    block
        .txdata
        .par_iter()
        .for_each(move |txn| index_transaction(writebatch, spentupdatebatch, txn, height));
    writebatch.insert(HeaderRow::CF, rayon::iter::once(header_row));
    writebatch.insert(META_CF, rayon::iter::once(last_indexed_block(&blockhash)));
}

pub(crate) fn last_indexed_block(blockhash: &BlockHash) -> Row {
    // Store last indexed block (i.e. all previous blocks were indexed)
    Row {
        key: METADATA_LAST_INDEXED_BLOCK.to_vec().into_boxed_slice(),
        value: db_encode(blockhash.as_inner()).unwrap().into_boxed_slice(),
    }
}

pub(crate) fn get_last_indexed_block(store: &DBStore) -> Option<BlockHash> {
    assert!(store.contents == DBContents::ConfirmedIndex);
    store
        .get_blocking(META_CF, METADATA_LAST_INDEXED_BLOCK)
        .map(|row| BlockHash::from_inner(db_decode(&row).expect("db error fetching blockhash")))
}

// In header database is corrupt, try to recover.
// Returns header chain that we have + a set of blocks we need to undo from our index.
async fn find_header_recovery(
    store: &DBStore,
    daemon: &Daemon,
    genesis: &BlockHash,
) -> Result<(Vec<BlockHeader>, HashSet<BlockHash>)> {
    info!("Searching for a reacovery point");

    let (_key, max_height) = store
        .get(META_CF, METADATA_LAST_INDEXED_BLOCK.to_vec())
        .await;

    let max_height: BlockHash = match max_height {
        Some(row) => BlockHash::from_inner(db_decode(&row).expect("db error fetching blockhash")),
        None => bail!("Unable to find starting point for recovery"),
    };

    let headers: Vec<BlockHeader> = HeaderRow::read_headers_unsorted(store).await;
    let headers: HashMap<BlockHash, BlockHeader> = headers
        .into_par_iter()
        .map(|h| (h.block_hash(), h))
        .collect();

    info!(
        "recovery: Our tip is {}; our db has {} headers",
        max_height.to_hex(),
        headers.len()
    );

    let mut max_height = daemon.getblockheader(&max_height).await?.1;
    let mut min_height: u64 = 0;
    let mut highest_match: Option<u64> = None;
    let mut highest_match_hash: Option<BlockHash> = None;

    let has_all_parent = |header: &BlockHeader| {
        let mut parent = header;
        loop {
            if parent.block_hash().eq(genesis) {
                return true;
            }

            parent = match headers.get(&parent.prev_blockhash) {
                Some(h) => h,
                None => {
                    return false;
                }
            }
        }
    };

    // Binary search for a header we actually have.
    loop {
        let mid = (min_height + max_height) / 2;

        info!(
            "recovery: searching for header at height {} (min: {}, max: {})",
            mid, min_height, max_height
        );
        if mid == 0 {
            break;
        }

        let header_at_mid: BlockHeader =
            match daemon.getblockheaders(&[mid as usize]).await?.first() {
                Some(h) => h.0.clone(),
                None => bail!("Failed to get header from daemon"),
            };

        if headers.contains_key(&header_at_mid.block_hash()) && has_all_parent(&header_at_mid) {
            highest_match = Some(mid);
            highest_match_hash = Some(header_at_mid.block_hash());
            min_height = mid + 1;
        } else {
            max_height = mid - 1;
        }

        if min_height > max_height {
            break;
        }
    }

    if highest_match.is_none() {
        bail!("Found no recovery point")
    };

    info!(
        "recovery: Highest match between us and daemon is {} ({})",
        highest_match.unwrap(),
        highest_match_hash.unwrap().to_hex()
    );

    // We found a recovery point. Let's produce a new valid header chain; list headers we need to undo.
    let mut new_headers: Vec<BlockHeader> = vec![];
    let mut parent = headers.get(&highest_match_hash.unwrap()).unwrap();
    loop {
        new_headers.push(parent.clone());
        if parent.block_hash().eq(genesis) {
            break;
        }
        parent = headers.get(&parent.prev_blockhash).unwrap();
    }

    new_headers.reverse();

    let new_headers_hashes: HashSet<BlockHash> =
        new_headers.iter().map(|h| h.block_hash()).collect();
    let old_headers_hashes: HashSet<BlockHash> = headers.keys().cloned().collect();
    let mut undo_blocks: HashSet<BlockHash> = old_headers_hashes
        .difference(&new_headers_hashes)
        .cloned()
        .collect();
    drop(new_headers_hashes);
    drop(old_headers_hashes);

    // belts-and-suspenders: let's push recovery further back
    for _ in 0..100 {
        if new_headers.len() == 1 {
            // We need to keep genesis
            break;
        }
        match new_headers.pop() {
            Some(h) => {
                undo_blocks.insert(h.block_hash());
            }
            None => break,
        }
    }
    info!(
        "recovery: New headers: {}, undo blocks: {}",
        new_headers.len(),
        undo_blocks.len()
    );
    Ok((new_headers, undo_blocks))
}

async fn read_indexed_headers(store: &DBStore) -> Result<Vec<BlockHeader>> {
    let (latest_blockhash, headers) = match store.get_blocking(META_CF, METADATA_LAST_INDEXED_BLOCK)
    {
        // latest blockheader persisted in the DB.
        Some(row) => {
            let tip: BlockHash =
                BlockHash::from_inner(db_decode(&row).expect("db error fetching blockhash"));
            let headers = HeaderRow::headers_up_to_tip(store, &tip)
                .await
                .context("Headers missing from database")?;
            (tip, headers)
        }
        None => (BlockHash::all_zeros(), vec![]),
    };
    info!("Latest indexed blockhash: {}", latest_blockhash.to_hex());

    assert_eq!(
        headers
            .first()
            .map(|h| h.prev_blockhash)
            .unwrap_or_else(BlockHash::all_zeros),
        BlockHash::all_zeros()
    );
    assert_eq!(
        headers
            .last()
            .map(BlockHeader::block_hash)
            .unwrap_or_else(BlockHash::all_zeros),
        latest_blockhash
    );
    Ok(headers)
}

pub struct Index {
    store: Arc<DBStore>,
    daemon: Arc<Daemon>,
    stats: Stats,
    batch_size: usize,
    chain: Chain,
    low_memory: bool,
    db_write_cache_entries: usize,
    last_flush: Mutex<Instant>,
    ignore_blocks_below: usize,
}

impl Index {
    pub async fn load(
        store: Arc<DBStore>,
        mut chain: Chain,
        daemon: Arc<Daemon>,
        metrics: &Metrics,
        config: &Config,
    ) -> Result<Self> {
        let (headers, undo_blocks) = match read_indexed_headers(&store).await {
            Ok(h) => (h, HashSet::default()),
            Err(e) => {
                warn!("Error loading headers; will attempt recovery: {}", e);
                find_header_recovery(&store, &daemon, &chain.genesis_hash().await).await?
            }
        };

        if !headers.is_empty() {
            let tip = headers.last().unwrap().block_hash();
            chain.load(headers, tip).await;
            let block_undoer = StoreBlockUndoer::new(store.clone(), daemon.clone());

            // Undo if needed
            for u in undo_blocks {
                let height = match daemon.getblockheader(&u).await {
                    Ok(header) => header.1,
                    Err(e) => {
                        // we need to give up, as we unspent index will be corrupted
                        bail!(
                            "Failed to undo block, failed to get header from daemon: {}",
                            e
                        );
                    }
                };
                if let Err(e) = block_undoer.undo_block(&u, height, &tip).await {
                    warn!("Failed to undo block: {}", e);
                }
            }

            chain
                .drop_last_headers(block_undoer, config.reindex_last_blocks as u64)
                .await;
            info!("Loaded {} headers from index", chain.height().await);
        }

        let stats = Stats::new(metrics);
        stats.observe_chain(&chain).await;
        Ok(Index {
            store,
            daemon,
            stats,
            batch_size: config.index_batch_size,
            chain,
            low_memory: config.low_memory,
            db_write_cache_entries: config.db_write_cache_entries,
            last_flush: Mutex::new(Instant::now()),
            ignore_blocks_below: config.ignore_blocks_below,
        })
    }

    /// Blockchain headers
    pub fn chain(&self) -> &Chain {
        &self.chain
    }

    pub async fn update(&self) -> Result<(Vec<BlockHeader>, BlockHeader)> {
        let tip = self.daemon.getbestblockhash().await?;
        let tip_height = self.daemon.getblockheader(&tip).await?.1;

        let start = Instant::now();
        let mut new_headers = self.daemon.get_new_headers(&self.chain, &tip).await?;
        metrics::observe(
            &self.stats.update_duration,
            "new_headers",
            duration_to_seconds(start.elapsed()),
        );

        let block_undoer = StoreBlockUndoer::new(self.store.clone(), self.daemon.clone());

        // Store headers for all new headers before updating the chain
        // This ensures all headers in the chain are in the database, even if we skip indexing some blocks
        if !new_headers.is_empty() {
            let header_writebatch = WriteBatch::new();
            for (header, _) in &new_headers {
                let header_row = HeaderRow::new(header).to_row();
                header_writebatch.insert(HeaderRow::CF, rayon::iter::once(header_row));
            }
            self.store.write_batch(&header_writebatch);
        }

        // Update chain with all headers (chain needs all headers to maintain continuity)
        self.chain
            .update(
                block_undoer,
                new_headers
                    .iter()
                    .map(|(h, height)| NewHeader::from_header_and_height(h.clone(), *height))
                    .collect(),
                Some(tip_height),
            )
            .await;

        // Filter out blocks below ignore_blocks_below when fetching/indexing
        // (new_headers only contains headers we don't have, so they're not indexed yet)
        // Note: We update the chain with all headers above, but only fetch/index blocks above the threshold
        let (blockhashes, indexed_headers): (Vec<BlockHash>, Vec<(BlockHeader, u64)>) =
            if self.ignore_blocks_below > 0 {
                let filtered: Vec<_> = new_headers
                    .iter()
                    .filter(|(_, height)| *height >= self.ignore_blocks_below as u64)
                    .cloned()
                    .collect();
                let hashes: Vec<_> = filtered.iter().map(|(h, _)| h.block_hash()).collect();
                (hashes, filtered)
            } else {
                let hashes: Vec<_> = new_headers.iter().map(|(h, _)| h.block_hash()).collect();
                // When not filtering, we index all headers, so we can move new_headers
                let indexed = std::mem::take(&mut new_headers);
                (hashes, indexed)
            };

        let block_channel = util::Channel::<Option<Block>>::bounded(self.batch_size);
        let block_sender = block_channel.sender();
        let writer_channel = util::Channel::<Option<(WriteBatch, SpentUpdateBatch)>>::bounded(1);
        let writer_sender = writer_channel.sender();

        let daemon = self.daemon.clone();
        let block_fetcher = thread::spawn_task("fetcher", async move {
            for blockhash in blockhashes.iter() {
                Waiter::shutdown_check()?;
                block_sender
                    .send(Some(daemon.getblock(blockhash).await?))
                    .await
                    .context("failed sending blocks to be indexed")?;
            }
            block_sender
                .send(None)
                .await
                .context("failed sending explicit end of stream")?;
            Ok(())
        });

        let writer_store = Arc::clone(&self.store);
        let writer = thread::spawn_task("index_writer", async move {
            loop {
                Waiter::shutdown_check()?;
                let mut write_receiver = writer_channel.receiver().await;
                let (write, spentupdate): (WriteBatch, SpentUpdateBatch) =
                    match tokio::time::timeout(Duration::from_secs(10), write_receiver.recv()).await
                    {
                        Ok(Some(Some((write, erase)))) => (write, erase),
                        Ok(Some(None)) => break,
                        Ok(None) =>
                        /* done */
                        {
                            break
                        }
                        Err(_) =>
                        /* recvtimeout */
                        {
                            continue
                        }
                    };
                writer_store.write_batch(&write);
                writer_store.update_spends(&spentupdate);
            }
            Ok(())
        });

        let mut writebatch = WriteBatch::new();
        let mut spentupdatebatch = SpentUpdateBatch::new();

        loop {
            Waiter::shutdown_check()?;
            let mut block_receiver = block_channel.receiver().await;
            let timer = self.stats.start_timer("fetch");
            let block =
                match tokio::time::timeout(Duration::from_secs(10), block_receiver.recv()).await {
                    Ok(Some(Some(block))) => block,
                    Ok(Some(None)) =>
                    /* done */
                    {
                        break
                    }
                    Err(_) =>
                    /* timeout, poll waiter */
                    {
                        continue
                    }
                    Ok(None) => break,
                };

            timer.observe_duration();

            let blockhash = block.block_hash();
            let height = self
                .chain
                .get_block_height(&blockhash)
                .await
                .unwrap_or_else(|| panic!("missing header for block {}", blockhash));

            if height % 1000 == 0 && height != self.chain.height().await {
                let chain_height = self.chain.height().await;
                info!(
                    "Indexing block {height} out of {chain_height}. Progress: {:.2}%",
                    (height as f64 / chain_height as f64) * 100.
                )
            }

            let timer = self.stats.start_timer("index");
            index_block(&writebatch, &spentupdatebatch, &block, height as usize);
            timer.observe_duration();
            let timer = self.stats.start_timer("write");
            if self.low_memory || writebatch.len() >= self.db_write_cache_entries {
                if !self.low_memory {
                    debug!(
                        "Writing batch of {} entries to db at block {height}.",
                        self.db_write_cache_entries
                    );
                }
                writer_sender
                    .send(Some((writebatch, spentupdatebatch)))
                    .await
                    .context("failed to send batch for writing")?;
                writebatch = WriteBatch::new();
                spentupdatebatch = SpentUpdateBatch::new();
            }
            timer.observe_duration();
            self.stats.update(&block, height as usize);
            if self.last_flush.lock().await.elapsed() > Duration::from_secs(10 * 60) {
                // Flush every 10 minutes to avoid having too much to reindex.
                *self.last_flush.lock().await = Instant::now();
                info!("Flushing index to disk...");
                self.store.flush()?;
                info!("Flushing done");
            }
        }

        writer_sender
            .send(Some((writebatch, spentupdatebatch)))
            .await
            .context("failed to send batch for writing")?;
        writer_sender
            .send(None)
            .await
            .context("failed to EOF to index writer")?;

        let timer = self.stats.start_timer("flush");
        timer.observe_duration();

        if let Err(e) = block_fetcher.await {
            warn!("failed to complete block fetcher: {}", e);
        }
        if let Err(e) = writer.await {
            warn!("failed to complete index writer: {}", e);
        }
        let tip_header = self.chain.tip().await;
        assert_eq!(&tip, &tip_header.block_hash());
        self.stats.observe_chain(&self.chain).await;
        Ok((
            indexed_headers.into_iter().map(|(h, _)| h).collect(),
            tip_header,
        ))
    }

    pub fn read_store(&self) -> &Arc<store::DBStore> {
        &self.store
    }
}
