rostrum 14.0.1

use std::collections::HashSet;

use anyhow::Result;
use bitcoincash::Txid;
use futures::stream::StreamExt;
use futures_util::future;
#[cfg(nexa)]
use std::sync::Arc;

use crate::{
    chaindef::{TokenID, Transaction},
    indexes::{
        heightindex::HeightIndexRow, inputindex::InputIndexRow, outputindex::OutputIndexRow,
        tokenoutputindex::TokenOutputIndexRow, DBRow,
    },
    store::DBStore,
};

use super::{queryutil::get_row, tx::TxQuery, BUFFER_SIZE};

/**
 * Returns full transaction history for a token
 */
#[allow(unused_variables)]
pub(crate) async fn get_token_history(
    token_id: &TokenID,
    commitment: &Option<Vec<u8>>,
    store: &DBStore,
) -> Result<Vec<(Txid, u32)>> {
    let to_outpoint_filter = {
        #[cfg(bch)]
        {
            // If commitment is None, then include full history. For history only of parent ID,
            // then empty commitment can be passed. This would exclude genesis if it mints NFT.
            match commitment {
                Some(c) => TokenOutputIndexRow::filter_by_parent_and_subgroup(token_id, c),
                None => TokenOutputIndexRow::filter_by_parent_including_subroup(token_id),
            }
        }
        #[cfg(nexa)]
        TokenOutputIndexRow::filter_by_parent_and_subgroup(token_id)
    };

    let (query, token_stream) = store
        .scan(TokenOutputIndexRow::CF, to_outpoint_filter, None)
        .await;

    // Find all txids involved with tokens.
    // (spending_txid is in most cases going to be a duplicate; thus hashset. It's really only needed for token burns)
    let txids: HashSet<Txid> = token_stream
        .map(|r| TokenOutputIndexRow::from_row(&r))
        .map(|r| async move {
            let oph = r.outpoint_hash();
            let funding_txid =
                get_row::<OutputIndexRow>(store, OutputIndexRow::filter_by_outpointhash(&oph))
                    .await
                    .expect("txid not found for token output")
                    .txid();

            let spending_txid =
                get_row::<InputIndexRow>(store, InputIndexRow::filter_by_outpointhash(&oph))
                    .await
                    .map(|row| row.txid());

            std::iter::once(funding_txid).chain(spending_txid)
        })
        .buffer_unordered(BUFFER_SIZE)
        .flat_map(futures::stream::iter)
        .collect()
        .await;

    query.await??;

    // Add heights
    let mut h: Vec<(Txid, u32)> = futures::stream::iter(txids.into_iter())
        .map(|txid| async move {
            let height = get_row::<HeightIndexRow>(store, HeightIndexRow::filter_by_txid(&txid))
                .await
                .expect("height not found for txid")
                .height();

            (txid, height)
        })
        .buffer_unordered(BUFFER_SIZE)
        .collect()
        .await;

    // Sort by inverse height first, then txid.
    h.sort_unstable_by(|a, b| match b.1.cmp(&a.1) {
        std::cmp::Ordering::Less => std::cmp::Ordering::Less,
        std::cmp::Ordering::Equal => b.0.cmp(&a.0),
        std::cmp::Ordering::Greater => std::cmp::Ordering::Greater,
    });
    Ok(h)
}

/// Locates and returns the transaction that created a token.
#[cfg(bch)]
pub(crate) async fn find_token_genesis(
    token_id: &TokenID,
    confirmed_index: &DBStore,
    unconfirmed_index: &DBStore,
    txquery: &TxQuery,
) -> Result<(Transaction, i64)> {
    use anyhow::Context;
    use bitcoin_hashes::Hash;

    use crate::{encode::compute_outpoint_hash, errors::rpc_invalid_params};

    // Token genesis on BCH is the transaction spending the first output
    // of the transaction that has the same value for its TxID as the TokenID.
    let genesis_outpoint = compute_outpoint_hash(&Txid::from_inner(token_id.into_inner()), 0);
    let key = InputIndexRow::filter_by_outpointhash(&genesis_outpoint);
    let spender = match get_row::<InputIndexRow>(confirmed_index, key.clone()).await {
        Some(spender) => Some(spender),
        None => get_row::<InputIndexRow>(unconfirmed_index, key).await,
    }
    .context(rpc_invalid_params("Token genesis not found".to_string()))?;

    debug_assert!(spender.index() == 0);
    let genesis_txid = spender.txid();
    let height = txquery
        .get_confirmation_height(&genesis_txid)
        .await
        .context(rpc_invalid_params(
            "Genesis transaction not found".to_string(),
        ))?;
    Ok((
        txquery
            .get(
                &genesis_txid,
                None,
                if height > 0 {
                    Some(height as u32)
                } else {
                    None
                },
                false,
            )
            .await?,
        height,
    ))
}

/// Locates and returns the transaction that created a token.
#[cfg(nexa)]
pub(crate) async fn find_token_genesis(
    token_id: &TokenID,
    store: &Arc<DBStore>,
    txquery: &TxQuery,
) -> Result<(Transaction, i64)> {
    use futures_util::stream;

    use crate::{indexes::outputtokenindex::OutputTokenIndexRow, query::CHUNK_SIZE};

    let to_outpoint_filter = TokenOutputIndexRow::filter_by_parent_and_subgroup(token_id);

    let (query, token_stream) = store
        .scan(TokenOutputIndexRow::CF, to_outpoint_filter, None)
        .await;

    let lowest_height_txs: Vec<(Txid, u32)> = token_stream
        .chunks(CHUNK_SIZE)
        .then(|rows| {
            let store = Arc::clone(store);

            async move {
                let txid_keys: Vec<Vec<u8>> = rows
                    .iter()
                    .map(TokenOutputIndexRow::from_row)
                    .map(|r| OutputIndexRow::filter_by_outpointhash(&r.outpoint_hash()))
                    .collect();

                let txids: Vec<(Txid, u32)> = store
                    .multi_get(OutputIndexRow::CF, txid_keys)
                    .await
                    .into_iter()
                    .map(|row| {
                        OutputIndexRow::txid_and_height_from_value(
                            &row.expect("txid not found for token output"),
                        )
                        .expect("failed to parse txid for token output")
                    })
                    .collect();

                stream::iter(txids.into_iter())
            }
        })
        .flatten_unordered(BUFFER_SIZE)
        .fold(
            Vec::new(),
            |lowest: Vec<(Txid, u32)>, tx: (Txid, u32)| async move {
                if lowest.is_empty() {
                    vec![tx]
                } else {
                    match tx.1.cmp(&lowest[0].1) {
                        std::cmp::Ordering::Less => vec![tx],
                        std::cmp::Ordering::Equal => {
                            lowest.into_iter().chain(std::iter::once(tx)).collect()
                        }
                        std::cmp::Ordering::Greater => lowest,
                    }
                }
            },
        )
        .await;
    query.await??;

    // One of the txs at lowest block height must be genesis, but we need to check which one.
    for (txid, height) in lowest_height_txs {
        // Mempool txs are expected to use MEMPOOL_HEIGHT
        assert!(height > 0);

        let tx = txquery.get(&txid, None, Some(height), false).await?;

        // If any of the inputs contain token, then this tx is not the genesis.
        if stream::iter(tx.input.iter())
            .any(|i| async move {
                let outpoint_hash = &i.previous_output.hash;
                let key = OutputTokenIndexRow::filter_by_outpointhash_and_token(
                    outpoint_hash,
                    token_id.clone(),
                );
                store.exists(OutputTokenIndexRow::CF, key).await
            })
            .await
        {
            continue;
        } else {
            // No inputs had token, this tx must be genesis.
            return Ok((tx, height as i64));
        }
    }
    bail!("Token {} not found", token_id.to_hex())
}

/**
 * List all subtokens / commitments for a a token.
 */
pub(crate) async fn list_nft(
    token_id: &TokenID,
    store: &DBStore,
) -> Result<HashSet<([u8; 32], Vec<u8>)>> {
    let to_outpoint_filter = TokenOutputIndexRow::filter_by_parent_including_subroup(token_id);
    let (query, stream) = store
        .scan(TokenOutputIndexRow::CF, to_outpoint_filter, None)
        .await;

    let nfts = stream
        .map(|r| TokenOutputIndexRow::from_row(&r))
        .filter(|r| future::ready(r.is_subtoken()))
        .map(|r| r.into_parent_and_subgroup())
        .collect()
        .await;

    query.await??;

    Ok(nfts)
}