rostrum 14.0.1

use std::sync::Arc;

use crate::{
    chaindef::{OutPointHash, ScriptHash, TokenID},
    indexes::outputindex::OutputIndexRow,
    query::{
        queryutil::{outpoints_are_spent, token_from_outpoint},
        tip::get_utxos_at_tip,
        BUFFER_SIZE, CHUNK_SIZE,
    },
    store::DBStore,
    util::Channel,
};

use super::{
    queryfilter::QueryFilter,
    status::{scan_for_outputs, UnspentItem},
    tip::UtxoEntry,
};
use anyhow::Result;
use futures::TryStreamExt;
use futures::{stream, StreamExt};
use rayon::slice::ParallelSliceMut;

/// When no height filter is used, we can use the faster unspent output index.
pub async fn listunspent_at_tip(
    confirmed: &Arc<DBStore>,
    mempool: &Arc<DBStore>,
    scripthash: &ScriptHash,
    filter: &QueryFilter,
    filter_token: &Option<TokenID>,
) -> Result<Vec<UnspentItem>> {
    let channel: Channel<Option<UtxoEntry>> = Channel::bounded(1000);
    let sender = channel.sender();

    let unspent_task = tokio::spawn(async move {
        let mut res = Vec::new();
        let mut receiver = channel.receiver().await;
        while let Some(Some((unspent, utxo, token))) = receiver.recv().await {
            res.push(UnspentItem {
                txid: utxo.txid(),
                vout: utxo.index(),
                outpoint_hash: Some(unspent.outpointhash()),
                height: utxo.height(),
                value: utxo.value(),
                token_amount: token.as_ref().map(|t| t.1),
                has_token: unspent.has_token(),
                #[cfg(bch)]
                commitment: token.as_ref().and_then(|t| t.2.clone()),
                token_id: token.map(|t| t.0),
            });
        }
        res
    });
    get_utxos_at_tip(confirmed, mempool, scripthash, filter, filter_token, sender).await?;
    let mut utxos = unspent_task.await?;
    utxos.par_sort_by_key(|u| u.height);
    Ok(utxos)
}

pub async fn lisunspent_full_search(
    confirmed: &Arc<DBStore>,
    mempool: &Arc<DBStore>,
    scripthash: &ScriptHash,
    filter: &QueryFilter,
    filter_token: &Option<TokenID>,
) -> Result<Vec<UnspentItem>> {
    let (confirmed_query, confirmed_stream) =
        scan_for_outputs(confirmed, *scripthash, filter).await;

    let confirmed_utxos = confirmed_stream
        .ready_chunks(CHUNK_SIZE)
        .then(|outputs| {
            // Filter out spends
            let outpoints: Vec<OutPointHash> = outputs.iter().map(|(o, _)| o.hash()).collect();
            let confirmed = confirmed.clone();
            let unconfirmed = mempool.clone();
            async move {
                // spent in confirmed
                let is_spent = outpoints_are_spent(&confirmed, &outpoints).await;
                let outputs: Vec<(OutputIndexRow, bool)> = outputs
                    .into_iter()
                    .zip(is_spent)
                    .filter_map(|(o, is_spent)| if is_spent { None } else { Some(o) })
                    .collect();

                // spent in unconfirmed
                let outpoints: Vec<OutPointHash> = outputs.iter().map(|(o, _)| o.hash()).collect();
                let is_spent = outpoints_are_spent(&unconfirmed, &outpoints).await;

                // forward filtered
                stream::iter(
                    outputs
                        .into_iter()
                        .zip(is_spent)
                        .filter_map(|(o, is_spent)| if is_spent { None } else { Some(o) }),
                )
            }
        })
        .flatten_unordered(BUFFER_SIZE);

    let (unconfirmed_query, unconfirmed_stream) =
        scan_for_outputs(mempool, *scripthash, filter).await;

    let unconfirmed =
        unconfirmed_stream
            .ready_chunks(CHUNK_SIZE)
            .then(|outputs| {
                let outpoints: Vec<OutPointHash> = outputs.iter().map(|(o, _)| o.hash()).collect();
                let mempool = Arc::clone(mempool);
                async move {
                    let is_spent = outpoints_are_spent(&mempool, &outpoints).await;
                    stream::iter(outputs.into_iter().zip(is_spent).map(
                        |((o, has_token), is_spent)| {
                            if is_spent {
                                None
                            } else {
                                Some((o, has_token))
                            }
                        },
                    ))
                }
            })
            .flatten_unordered(BUFFER_SIZE)
            .filter_map(|item| async move { item });

    let unspent_outputs = confirmed_utxos.chain(unconfirmed);

    // Fetch token info & filter if token_filter is set.
    let unspent_outputs = unspent_outputs.filter_map(|(output, has_token)| {
        let filter_token = filter_token.clone();

        // Take a copies for async call if user requested token info or filters on token.
        let (confirmed, unconfirmed) = if has_token && filter.token_only {
            (Some(Arc::clone(confirmed)), Some(Arc::clone(mempool)))
        } else {
            (None, None)
        };

        async move {
            if !has_token || !filter.token_only {
                // No filter
                return Some(Ok((output, None, None, has_token, None)));
            }

            let outpointhash = output.hash();
            let token = match token_from_outpoint(&confirmed.unwrap(), &outpointhash).await {
                std::result::Result::Ok(t) => t,
                Err(e) => return Some(Err(e)),
            };
            let (token_id, amount, commitment) = if let Some(t) = token {
                t
            } else {
                // Try mempool.
                let t = match token_from_outpoint(&unconfirmed.unwrap(), &outpointhash).await {
                    std::result::Result::Ok(t) => t,
                    Err(e) => return Some(Err(e)),
                };
                match t {
                    Some(t) => t,
                    None => return Some(Err(anyhow!("expected to find token output"))),
                }
            };

            if let Some(filter) = &filter_token {
                if filter == &token_id {
                    Some(Ok((
                        output,
                        Some(token_id),
                        Some(amount),
                        has_token,
                        commitment,
                    )))
                } else {
                    None
                }
            } else {
                // Did not filter on specific token.
                Some(Ok((
                    output,
                    Some(token_id),
                    Some(amount),
                    has_token,
                    commitment,
                )))
            }
        }
    });

    #[allow(unused_variables)] // commitment unused on nexa (it's part of token_id)
    #[allow(clippy::type_complexity)]
    let unspent_outputs = unspent_outputs.map(|output| {
        let (out, token_id, token_amount, has_token, commitment): (
            OutputIndexRow,
            Option<TokenID>,
            Option<i64>,
            bool,
            Option<Vec<u8>>,
        ) = output?;
        anyhow::Ok({
            UnspentItem {
                txid: out.txid(),
                vout: out.index(),
                outpoint_hash: Some(out.hash()),
                height: out.height(),
                value: out.value(),
                token_id,
                token_amount,
                has_token,
                #[cfg(bch)]
                commitment,
            }
        })
    });

    let mut unspent: Vec<UnspentItem> = unspent_outputs.try_collect().await?;
    confirmed_query.await??;
    unconfirmed_query.await??;

    unspent.par_sort_by_key(|u| u.height);
    Ok(unspent)
}