rostrum 14.0.1

use futures_util::{stream, StreamExt, TryStreamExt};
use std::{
    cmp::Ordering,
    collections::{HashMap, HashSet},
    sync::Arc,
};
use tokio::{join, task::JoinHandle};

use anyhow::{Context, Result};
use bitcoin_hashes::{hex::ToHex, Hash};
use bitcoincash::Txid;
use rayon::prelude::*;
use sha2::{Digest, Sha256};

use crate::{
    chaindef::{OutPointHash, ScriptHash, TokenID},
    indexes::{outputindex::OutputIndexRow, scripthashindex::ScriptHashIndexRow, DBRow},
    mempool::{ConfirmationState, Tracker},
    query::{
        queryutil::{outpoint_is_spent, token_from_outpoint},
        unspent::{listunspent_at_tip, lisunspent_full_search},
        BUFFER_SIZE, CHUNK_SIZE,
    },
    store::{DBContents, DBStore},
};

use super::{queryfilter::QueryFilter, queryutil::multi_get_utxo};
use crate::utilserialize::as_rpc_height;
#[cfg(bch)]
use crate::utilserialize::opt_as_hex;

#[derive(Serialize)]
pub struct HistoryItem {
    pub tx_hash: Txid,
    pub height: i32,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub fee: Option<u64>, // need to be set only for unconfirmed transactions (i.e. height <= 0)
}

#[derive(Serialize)]
pub struct UnspentItem {
    #[serde(rename = "tx_hash")]
    pub txid: Txid,
    #[serde(rename = "tx_pos")]
    pub vout: u32,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub outpoint_hash: Option<OutPointHash>,
    #[serde(serialize_with = "as_rpc_height")]
    pub height: u32,
    pub value: u64,

    #[cfg(nexa)]
    #[serde(skip_serializing_if = "Option::is_none", rename = "token_id_hex")]
    pub token_id: Option<TokenID>,

    #[cfg(bch)]
    #[serde(skip_serializing_if = "Option::is_none")]
    pub token_id: Option<TokenID>,

    #[cfg(bch)]
    #[serde(skip_serializing_if = "Option::is_none", serialize_with = "opt_as_hex")]
    pub commitment: Option<Vec<u8>>,

    #[serde(skip_serializing_if = "Option::is_none")]
    pub token_amount: Option<i64>,

    /// If this utxo has token in it. Useful when not querying for token info (which adds query cost).
    pub has_token: bool,
}

/**
 * For sorting history items by confirmation height (and then ID)
 */
pub(crate) fn by_block_height(a: &HistoryItem, b: &HistoryItem) -> Ordering {
    if a.height == b.height {
        // Order by little endian tx hash if height is the same,
        // in most cases, this order is the same as on the blockchain.
        return b.tx_hash.cmp(&a.tx_hash);
    }
    if a.height > 0 && b.height > 0 {
        return a.height.cmp(&b.height);
    }

    // mempool txs should be sorted last, so add to it a large number
    // per spec, mempool entries do not need to be sorted, but we do it
    // anyway so that statushash is deterministic
    let mut a_height = a.height;
    let mut b_height = b.height;
    if a_height <= 0 {
        a_height = 0xEE_EEEE + a_height.abs();
    }
    if b_height <= 0 {
        b_height = 0xEE_EEEE + b_height.abs();
    }
    a_height.cmp(&b_height)
}

/**
 * Find output rows given filter parameters.
 *
 * This is an expensive call used by  most blockchain.scripthash.* queries.
 *
 * Returns the utxos + bool indicating if utxo has token
 */
#[allow(clippy::clone_on_copy)]
pub async fn scan_for_outputs<'a>(
    store: &'a Arc<DBStore>,
    scripthash: ScriptHash,
    filter: &'a QueryFilter,
) -> (
    JoinHandle<Result<()>>,
    impl futures::stream::Stream<Item = (OutputIndexRow, bool)> + 'a,
) {
    let (scan_prefix, scan_bitmask) =
        ScriptHashIndexRow::filter_by_outputs(scripthash.into_inner(), filter);

    let (query, stream) = store
        .scan(ScriptHashIndexRow::CF, scan_prefix, scan_bitmask)
        .await;

    let stream =
        stream
            .map(|row| ScriptHashIndexRow::from_row(&row))
            .ready_chunks(CHUNK_SIZE)
            .then(move |rows| {
                let store = Arc::clone(store);
                // Bitmask filtering at scan level already filters to funding entries only
                let filtered_rows: Vec<ScriptHashIndexRow> = rows
                    .into_iter()
                    .filter(|row| filter.height_within(row.get_height()))
                    .collect();

                let outpoints: Vec<OutPointHash> =
                    filtered_rows.iter().map(|row| row.outpointhash()).collect();

                async move {
                    let utxos = multi_get_utxo(&store, &outpoints).await;

                    stream::iter(utxos.into_iter().zip(filtered_rows.into_iter()).map(
                        |(utxo, r)| {
                            let utxo = utxo.expect("missing utxo");
                            (utxo, r.has_token())
                        },
                    ))
                }
            })
            .flatten_unordered(BUFFER_SIZE);

    (query, stream)
}

// Used in .reduce for merging maps
fn merge_token_balance_maps(
    mut a: HashMap<TokenID, i64>,
    b: HashMap<TokenID, i64>,
) -> HashMap<TokenID, i64> {
    for (token_id, amount) in b {
        a.entry(token_id)
            .and_modify(|a| {
                *a += amount;
            })
            .or_insert(amount);
    }
    a
}

/**
 * Calculate token balance in a given store.
 */
async fn calc_token_balance(
    store: &Arc<DBStore>,
    scripthash: ScriptHash,
    filter_token: Option<TokenID>,
) -> Result<(HashMap<TokenID, i64>, Vec<OutPointHash>)> {
    let filter = QueryFilter::filter_token_only();
    let (outputs_query, outputs_stream) = scan_for_outputs(store, scripthash, &filter).await;

    let outpoints: Vec<(TokenID, i64, OutPointHash)> = outputs_stream
        .map(|(o, _)| o.take_hash())
        .map(|outpoint| async move {
            let (token_future, is_spent) = join!(
                token_from_outpoint(store, &outpoint),
                outpoint_is_spent(store, outpoint),
            );
            let (token_id, amount, _) = token_future?.context(format!(
                "token info not found for outpoint {}",
                outpoint.to_hex()
            ))?;

            let valid_amount = if amount < 0 || is_spent { 0 } else { amount };
            anyhow::Ok((token_id, valid_amount, outpoint))
        })
        .buffer_unordered(BUFFER_SIZE)
        .try_collect::<Vec<(TokenID, i64, OutPointHash)>>()
        .await?;

    outputs_query.await??;

    let outpoints = if let Some(filter) = filter_token {
        outpoints
            .into_iter()
            .filter(|(token_id, _, _)| token_id.eq(&filter))
            .collect()
    } else {
        outpoints
    };

    let outpoints_len = outpoints.len();

    let (balance, outputs) = outpoints.into_iter().fold(
        (
            HashMap::<TokenID, i64>::default(),
            Vec::with_capacity(outpoints_len),
        ),
        |(mut map, mut ops): (HashMap<TokenID, i64>, Vec<OutPointHash>),
         (token_id, amount, outpoint)| {
            map.entry(token_id)
                .and_modify(|a| *a += amount)
                .or_insert(amount);
            ops.push(outpoint);
            (map, ops)
        },
    );

    Ok((balance, outputs))
}

/**
 * Get the unconfirmed token balance
 */
pub(crate) async fn unconfirmed_scripthash_token_balance(
    mempool: &Arc<DBStore>,
    index: &Arc<DBStore>,
    confirmed_outputs: Vec<OutPointHash>,
    scripthash: ScriptHash,
    filter_token: Option<TokenID>,
) -> Result<HashMap<TokenID, i64>> {
    assert!(mempool.contents == DBContents::MempoolIndex);
    assert!(index.contents == DBContents::ConfirmedIndex);

    // This finds the balance of entries that have been both created and spent
    // in the mempool. It cannot see spends of outputs thave have been confirmed.
    let (balance, _) = calc_token_balance(mempool, scripthash, filter_token).await?;

    // Find inputs that spends from confirmed outputs as well
    let spends_of_confirmed: Vec<(TokenID, i64)> = stream::iter(confirmed_outputs)
        .then(|outpoint| async move {
            if outpoint_is_spent(mempool, outpoint).await {
                let (token_id, amount, _) = token_from_outpoint(index, &outpoint)
                    .await?
                    .expect("token info found for outpoint {outpoint}");
                anyhow::Ok((token_id, -amount))
            } else {
                // This output was funded in the confirmed index, not mempool.
                // If it's not spent here, then ignore it.
                Ok((TokenID::all_zeros(), 0))
            }
        })
        .try_collect()
        .await?;

    let spends_of_confirmed = spends_of_confirmed.into_iter().fold(
        HashMap::<TokenID, i64>::default(),
        |mut map, (token_id, amount)| {
            map.entry(token_id)
                .and_modify(|a| {
                    *a += amount;
                })
                .or_insert(amount);
            map
        },
    );

    // Merge in spends of confirmed outputs
    let mut balance = merge_token_balance_maps(balance, spends_of_confirmed);
    balance.remove(&TokenID::all_zeros());

    Ok(balance)
}

/**
 * Get the confirmed token balance
 */
pub(crate) async fn confirmed_scripthash_token_balance(
    store: &Arc<DBStore>,
    scripthash: ScriptHash,
    filter_token: Option<TokenID>,
) -> Result<(HashMap<TokenID, i64>, Vec<OutPointHash>)> {
    calc_token_balance(store, scripthash, filter_token).await
}

/**
 * Find all transactions that spend or fund scripthash.
 *
 * Scans the ScriptHashIndex to find both funding and spending transactions.
 */
pub(crate) async fn scripthash_transactions(
    store: &Arc<DBStore>,
    scripthash: ScriptHash,
    filter: QueryFilter,
    filter_token: Option<TokenID>,
) -> Result<HashMap<Txid, u32>> {
    // Single scan of ScriptHashIndex for both funding and spending entries
    let (scan_prefix, scan_bitmask) =
        ScriptHashIndexRow::filter_by_outputs_and_inputs(scripthash.into_inner(), &filter);
    let (query, stream) = store
        .scan(ScriptHashIndexRow::CF, scan_prefix, scan_bitmask)
        .await;

    let mut txids = HashMap::new();

    let rows: Vec<ScriptHashIndexRow> = stream
        .map(|row| ScriptHashIndexRow::from_row(&row))
        .collect()
        .await;

    query.await??;

    // First pass: apply height filter (token filtering already done by bitmask at scan level)
    let candidate_rows: Vec<_> = rows
        .into_iter()
        .filter(|row| filter.height_within(row.get_height()))
        .collect();

    // Second pass: collect valid funding outpoints for specific token filtering
    let mut valid_funding_outpoints = HashSet::new();
    if let Some(requested_token) = &filter_token {
        for row in &candidate_rows {
            if !row.is_funding() {
                continue;
            }

            // Check the actual token for funding entries
            let outpoint = row.outpointhash();
            if let Ok(Some((actual_token, _, _))) = token_from_outpoint(store, &outpoint).await {
                if &actual_token == requested_token {
                    valid_funding_outpoints.insert(outpoint);
                }
            }
        }
    }

    // Third pass: include entries that pass all filters
    for row in candidate_rows {
        if filter_token.is_some() {
            if row.is_funding() {
                // For funding entries, check if their outpoint is in valid set
                let outpoint = row.outpointhash();
                if !valid_funding_outpoints.contains(&outpoint) {
                    continue;
                }
            } else {
                // For spending entries, check if they spend from valid funding outpoints
                let spent_outpoint = row.outpointhash();
                if !valid_funding_outpoints.contains(&spent_outpoint) {
                    continue;
                }
            }
        }

        let txid = Txid::from_inner(row.get_txid());
        txids.insert(txid, row.get_height());
    }

    Ok(txids)
}

/**
 * Generate list of HistoryItem for a scripthashes confirmed history.
 *
 * Also returns confirmed outputs that can be used to get unconfirmed spends
 * of confirmed utxos.
 */
async fn confirmed_history(
    store: &Arc<DBStore>,
    scripthash: ScriptHash,
    filter: QueryFilter,
    filter_token: Option<TokenID>,
) -> Result<Vec<HistoryItem>> {
    let txids = scripthash_transactions(store, scripthash, filter, filter_token).await?;

    let history: Vec<HistoryItem> = txids
        .into_par_iter()
        .map(|(txid, height)| HistoryItem {
            tx_hash: txid,
            height: height as i32,
            fee: None,
        })
        .collect();

    Ok(history)
}

/**
 * Generate list of HistoryItem for a scripthash's unconfirmed history
 */
pub async fn unconfirmed_history(
    mempool: &Tracker,
    scripthash: ScriptHash,
    filter: QueryFilter,
    filter_token: Option<TokenID>,
) -> Result<Vec<HistoryItem>> {
    let txids = scripthash_transactions(mempool.index(), scripthash, filter, filter_token).await?;

    Ok(stream::iter(txids)
        .map(|(txid, _)| async move {
            let height = match mempool.tx_confirmation_state(&txid, None).await {
                ConfirmationState::InMempool => 0,
                ConfirmationState::UnconfirmedParent => -1,
                ConfirmationState::Indeterminate => {
                    debug_assert!(false, "Mempool tx's state cannot be indeterminate");
                    0
                }
                ConfirmationState::Confirmed => {
                    debug_assert!(false, "Mempool tx's state cannot be confirmed");
                    0
                }
            };
            HistoryItem {
                tx_hash: txid,
                height,
                fee: mempool.get_fee(&txid).await,
            }
        })
        .buffer_unordered(BUFFER_SIZE)
        .collect()
        .await)
}

pub async fn scripthash_history(
    store: &Arc<DBStore>,
    mempool: &Tracker,
    scripthash: ScriptHash,
    filter: QueryFilter,
    filter_token: Option<TokenID>,
) -> Result<Vec<HistoryItem>> {
    let mut history = confirmed_history(store, scripthash, filter, filter_token.clone()).await?;

    history.extend(unconfirmed_history(mempool, scripthash, filter, filter_token).await?);

    let mut history = tokio::task::spawn_blocking(|| async move {
        history.par_sort_unstable_by(by_block_height);
        history
    })
    .await
    .unwrap()
    .await;

    filter.filter_limit_offset(&mut history);
    Ok(history)
}

/**
 * Generate a hash of scripthash history as defined in electrum spec
 */
pub fn hash_scripthash_history(history: &Vec<HistoryItem>) -> Option<[u8; 32]> {
    if history.is_empty() {
        None
    } else {
        let mut sha2 = Sha256::new();
        let parts: Vec<String> = history
            .into_par_iter()
            .map(|t| format!("{}:{}:", t.tx_hash.to_hex(), t.height))
            .collect();

        parts.into_iter().for_each(|p| {
            sha2.update(p.as_bytes());
        });
        Some(sha2.finalize().into())
    }
}

pub(crate) async fn scripthash_listunspent(
    store: &Arc<DBStore>,
    mempool: &Arc<DBStore>,
    scripthash: ScriptHash,
    filter: QueryFilter,
    filter_token: Option<TokenID>,
) -> Result<Vec<UnspentItem>> {
    assert!(store.contents == DBContents::ConfirmedIndex);
    assert!(mempool.contents == DBContents::MempoolIndex);

    if filter_token.is_some() {
        assert!(filter.token_only)
    }

    if !filter.has_height_filter() {
        // we can use the faster unspent index
        return listunspent_at_tip(store, mempool, &scripthash, &filter, &filter_token).await;
    }

    // Do a slower historical lookup; "unspent at blockheight X"
    lisunspent_full_search(store, mempool, &scripthash, &filter, &filter_token).await
}