use crate::bch::cashaddr::version_byte_flags as bch_flags;
use crate::chaindef::BlockHash;
use crate::chaindef::BlockHeader;
use crate::chaindef::ScriptHash;
use crate::chaindef::Transaction;
use crate::chaindef::TxOut;
use crate::doslimit::ConnectionLimits;
use crate::edition::Edition;
use crate::encode::compute_outpoint_hash_from_tx;
use crate::errors::rpc_invalid_params;
use crate::mempool::MEMPOOL_HEIGHT;
use crate::nexa::cashaddr::version_byte_flags as nexa_flags;
use crate::query::queryfilter::QueryFilter;
use crate::query::scripthash_subscriptions::ScripthashSubscriptions;
use crate::query::tx::get_addresses;
use crate::query::Query;
use crate::rpc::daemon::server::ConnectionId;
use crate::rpc::parseutil::{
    bool_from_value_or, hash_from_value, str_from_value, usize_from_value, usize_from_value_or,
};
use crate::rpc::rpcstats::RpcStats;
use crate::rpc::scripthash::{
    get_balance, get_first_spend, get_first_use, get_history, get_mempool, listunspent,
};
use crate::scripthash::addr_to_scripthash;
use crate::scripthash::decode_address;
use crate::signal::NetworkNotifier;
use anyhow::{Context, Result};
use bitcoincash::consensus::encode::{deserialize, serialize};
use bitcoincash::hash_types::Txid;
use bitcoincash::hashes::hex::ToHex;
use bitcoincash::Network;
use serde_json::Value;
use std::collections::HashMap;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use tokio::sync::Mutex;

use super::scripthash::get_volume;

struct Subscription {
    statushash: Option<[u8; 32]>,
    alias: Option<String>,
}

pub struct BlockchainRpc {
    query: Arc<Query>,
    stats: Arc<RpcStats>,
    subscriptions: Mutex<HashMap<ScriptHash, Subscription>>,
    last_header_entry: Mutex<Option<BlockHeader>>,
    doslimits: ConnectionLimits,

    // To notify own mempool of broadcasted transaction accepted by node.
    network_notifier: Arc<NetworkNotifier>,

    #[allow(dead_code)] // used in nex, but not bch
    network: Network,

    /* Resource tracking */
    alias_bytes_used: AtomicUsize,
    conn_id: ConnectionId,
    subscription_manager: Arc<ScripthashSubscriptions>,
    edition: Edition,
}

#[allow(clippy::too_many_arguments)]
impl BlockchainRpc {
    pub fn new(
        query: Arc<Query>,
        stats: Arc<RpcStats>,
        doslimits: ConnectionLimits,
        network: Network,
        network_notifier: Arc<NetworkNotifier>,
        conn_id: ConnectionId,
        subscription_manager: Arc<ScripthashSubscriptions>,
        edition: Edition,
    ) -> BlockchainRpc {
        BlockchainRpc {
            query,
            stats,
            subscriptions: Mutex::new(HashMap::new()),
            last_header_entry: Mutex::new(None), // disable header subscription for now
            doslimits,
            network_notifier,
            network,
            alias_bytes_used: AtomicUsize::new(0),
            conn_id,
            subscription_manager,
            edition,
        }
    }

    pub fn set_edition(&mut self, edition: Edition) {
        self.edition = edition;
    }

    pub fn edition(&self) -> Edition {
        self.edition
    }

    pub(crate) fn address_decode(&self, params: &[Value]) -> Result<Value> {
        let address = str_from_value(params.first(), "address")?;
        let (payload, payload_type) = decode_address(&address)?;

        assert_eq!(bch_flags::TYPE_P2PKH, nexa_flags::TYPE_P2PKH);
        assert_eq!(bch_flags::TYPE_P2SH, nexa_flags::TYPE_P2SH);

        let token_aware = [
            bch_flags::TYPE_P2PKH_TOKEN,
            bch_flags::TYPE_P2SH_TOKEN,
            nexa_flags::TYPE_SCRIPT_TEMPLATE,
            nexa_flags::TYPE_GROUP,
        ];

        Ok(json!({
            "payload": payload.to_hex(),
            "type": match payload_type {
                bch_flags::TYPE_P2PKH | bch_flags::TYPE_P2PKH_TOKEN => "p2pkh",
                bch_flags::TYPE_P2SH | bch_flags::TYPE_P2SH_TOKEN => "p2sh",
                nexa_flags::TYPE_SCRIPT_TEMPLATE => "scripttemplate",
                nexa_flags::TYPE_GROUP => "group",
                _=> "Unknown"
            },
            "is_token_aware": token_aware.contains(&payload_type)
        }))
    }

    pub async fn address_get_balance(&self, params: &[Value]) -> Result<Value> {
        let addr = str_from_value(params.first(), "address")?;
        let filter = QueryFilter::from_param(params.get(1))?;
        let scripthash = addr_to_scripthash(&addr)?;
        get_balance(&self.query, scripthash, &filter).await
    }
    pub async fn address_get_first_use(&self, params: &[Value]) -> Result<Value> {
        let addr = str_from_value(params.first(), "address")?;
        let filter = QueryFilter::from_param(params.get(1))?;
        let scripthash = addr_to_scripthash(&addr)?;
        get_first_use(&self.query, scripthash, &filter, self.edition()).await
    }
    pub async fn address_get_first_spend(&self, params: &[Value]) -> Result<Value> {
        let addr = str_from_value(params.first(), "address")?;
        let filter = QueryFilter::from_param(params.get(1))?;
        let scripthash = addr_to_scripthash(&addr)?;
        get_first_spend(&self.query, scripthash, &filter).await
    }
    pub async fn address_get_history(&self, params: &[Value]) -> Result<Value> {
        let addr = str_from_value(params.first(), "address")?;
        let filter = QueryFilter::from_param(params.get(1))?;
        let scripthash = addr_to_scripthash(&addr)?;

        get_history(
            self.query.confirmed_index(),
            self.query.unconfirmed_index(),
            scripthash,
            filter,
        )
        .await
    }

    pub async fn address_get_mempool(&self, params: &[Value]) -> Result<Value> {
        let addr = str_from_value(params.first(), "address")?;
        let filter = QueryFilter::from_param(params.get(1))?;
        let scripthash = addr_to_scripthash(&addr)?;
        get_mempool(&self.query, scripthash, filter).await
    }

    pub fn address_get_scripthash(&self, params: &[Value]) -> Result<Value> {
        let scripthash = addr_to_scripthash(&str_from_value(params.first(), "address")?)?;
        Ok(json!(scripthash))
    }

    pub async fn address_get_volume(&self, params: &[Value]) -> Result<Value> {
        let addr = str_from_value(params.first(), "address")?;
        let filter = QueryFilter::from_param(params.get(1))?;
        let scripthash = addr_to_scripthash(&addr)?;
        get_volume(&self.query, scripthash, &filter).await
    }

    pub async fn address_listunspent(&self, params: &[Value]) -> Result<Value> {
        let addr = str_from_value(params.first(), "address")?;
        let filter = QueryFilter::from_param(params.get(1))?;
        let scripthash = addr_to_scripthash(&addr)?;
        listunspent(&self.query, scripthash, filter).await
    }

    pub async fn address_subscribe(&self, params: &[Value]) -> Result<Value> {
        let addr = str_from_value(params.first(), "address")?;
        let scripthash = addr_to_scripthash(&addr)?;
        self.remove_subscription(&scripthash).await;

        self.doslimits
            .check_subscriptions(self.get_num_subscriptions().await as u32 + 1)?;

        self.doslimits
            .check_alias_usage(self.alias_bytes_used.load(Ordering::Relaxed) + addr.len())?;

        // Subscribe - this will compute statushash if needed or reuse cached one
        let statushash = self
            .subscription_manager
            .subscribe(self.conn_id, scripthash, Some(addr.clone()))
            .await?;

        let result = statushash.map_or(Value::Null, |h| json!(hex::encode(h)));

        // We don't hold a lock on alias usage, so we could exceed limit here.
        // That's OK, it doesn't need to be a hard limit.
        self.alias_bytes_used
            .fetch_add(addr.len(), Ordering::Relaxed);
        self.subscriptions.lock().await.insert(
            scripthash,
            Subscription {
                statushash,
                alias: Some(addr),
            },
        );
        self.stats.subscriptions.inc();
        Ok(result)
    }

    pub async fn address_unsubscribe(&self, params: &[Value]) -> Result<Value> {
        let addr = str_from_value(params.first(), "address")?;
        let scripthash = addr_to_scripthash(&addr)?;
        Ok(json!(self.remove_subscription(&scripthash).await))
    }

    /**
     * Get the raw block at given height or hash
     */
    pub(crate) async fn block_get(&self, params: &[Value]) -> Result<Value> {
        if let Ok(blockhash) = hash_from_value::<BlockHash>(params.first()) {
            let block = self.query.get_block(&blockhash).await?;
            return Ok(json!(hex::encode(serialize(&block))));
        }
        if let Ok(height) = usize_from_value(params.first(), "height") {
            let header = self
                .query
                .header()
                .at_height(height)
                .await
                .ok_or_else(|| rpc_invalid_params("Invalid block height".to_string()))?;
            let block = self.query.get_block(&header.block_hash()).await?;
            return Ok(json!(hex::encode(serialize(&block))));
        }
        Err(rpc_invalid_params(
            "Argument must be hex encoded blockhash or numeric height".to_string(),
        ))
    }

    pub async fn block_header(&self, params: &[Value]) -> Result<Value> {
        let height = usize_from_value(params.first(), "height")?;
        let cp_height = usize_from_value_or(params.get(1), "cp_height", 0)?;

        let raw_header_hex: String = self
            .query
            .get_headers(&[height])
            .await
            .into_iter()
            .map(|entry| hex::encode(serialize(&entry)))
            .collect();

        if cp_height == 0 {
            return Ok(json!(raw_header_hex));
        }
        let (branch, root) = self
            .query
            .get_header_merkle_proof(height, cp_height)
            .await?;

        let branch_vec: Vec<String> = branch.into_iter().map(|b| b.to_hex()).collect();

        Ok(json!({
            "header": raw_header_hex,
            "root": root.to_hex(),
            "branch": branch_vec
        }))
    }

    pub async fn block_header_verbose(&self, params: &[Value]) -> Result<Value> {
        let height = match usize_from_value(params.first(), "height") {
            Ok(h) => h,
            Err(_) => {
                // it might be hash
                match hash_from_value::<BlockHash>(params.first()) {
                    Ok(hash) => match self.query.chain().get_block_height(&hash).await {
                        Some(h) => h as usize,
                        None => {
                            return Err(rpc_invalid_params(format!(
                                "block '{}' does not exist",
                                hash.to_hex()
                            )))
                        }
                    },
                    Err(_) => {
                        return Err(rpc_invalid_params(
                            "expected block height or block hash".into(),
                        ));
                    }
                }
            }
        };

        let header = match self.query.chain().get_block_header(height).await {
            Some(h) => h,
            None => {
                return Err(rpc_invalid_params(format!(
                    "block at height {} does not exist",
                    height
                )))
            }
        };

        let mtp = self
            .query
            .chain()
            .get_mtp(height)
            .await
            .context("unable to calculate mtp")?;

        #[cfg(bch)]
        {
            Ok(json!({

                    "version": header.version,
                    "previousblockhash": header.prev_blockhash.to_hex(),
                    "merkleroot": header.merkle_root.to_hex(),
                    "time": header.time,
                    "bits": header.bits,
                    "nonce": header.nonce,
                    "hash": header.block_hash(),
                    "height": height,
                    "hex": serialize(&header).to_hex(),
                    "mediantime": mtp,

            }))
        }

        #[cfg(nexa)]
        {
            Ok(json!({
                "previousblockhash": header.prev_blockhash.to_hex(),
                "bits": header.bits,
                "ancestorhash": header.ancestor.to_hex(),
                "merkleroot": header.merkle_root.to_hex(),
                "txfilter": header.tx_filter,
                "time": header.time,
                "height": header.height.0,
                "chainwork": header.chain_work,
                "size": header.size,
                "txcount": header.tx_count.0,
                "feepoolamt": header.fee_pool_amt.0,
                "utxocommitment": header.utxo_commitment.to_hex(),
                "minerdata": header.miner_data.to_hex(),
                "nonce": header.nonce.to_hex(),
                "mediantime": mtp,
                "hex": serialize(&header).to_hex(),
                "hash": header.block_hash()
            }))
        }
    }

    pub async fn block_headers(&self, params: &[Value]) -> Result<Value> {
        let start_height = usize_from_value(params.first(), "start_height")?;
        let count = usize_from_value(params.get(1), "count")?;
        let cp_height = usize_from_value_or(params.get(2), "cp_height", 0)?;
        let heights: Vec<usize> = (start_height..(start_height + count)).collect();
        let headers: Vec<String> = self
            .query
            .get_headers(&heights)
            .await
            .into_iter()
            .map(|entry| hex::encode(serialize(&entry)))
            .collect();

        if count == 0 || cp_height == 0 {
            return Ok(json!({
                "count": headers.len(),
                "hex": headers.join(""),
                "max": 2016,
            }));
        }

        let (branch, root) = self
            .query
            .get_header_merkle_proof(start_height + (count - 1), cp_height)
            .await?;

        let branch_vec: Vec<String> = branch.into_iter().map(|b| b.to_hex()).collect();

        Ok(json!({
            "count": headers.len(),
            "hex": headers.join(""),
            "max": 2016,
            "root": root.to_hex(),
            "branch" : branch_vec
        }))
    }

    pub async fn estimatefee(&self, params: &[Value]) -> Result<Value> {
        let blocks_count = usize_from_value(params.first(), "blocks_count")?;
        let fee_rate = self.query.estimate_fee(blocks_count).await; // in BCH/kB
        Ok(json!(fee_rate.max(self.query.get_relayfee().await?)))
    }

    pub async fn headers_tip(&self) -> Result<Value> {
        let (height, header) = self.query.get_best_header().await?;
        let hex = hex::encode(serialize(&header));
        Ok(json!({"hex":hex, "height":height}))
    }

    pub async fn headers_subscribe(&self) -> Result<Value> {
        let (height, header) = self.query.get_best_header().await?;
        let hex = hex::encode(serialize(&header));
        let result = json!({"hex":hex, "height":height});
        let mut last_entry = self.last_header_entry.lock().await;
        *last_entry = Some(header);
        Ok(result)
    }

    pub async fn relayfee(&self) -> Result<Value> {
        Ok(json!(self.query.get_relayfee().await?)) // in BTC/kB
    }

    pub async fn scripthash_get_balance(&self, params: &[Value]) -> Result<Value> {
        let scripthash = hash_from_value::<ScriptHash>(params.first())?;
        let filter = QueryFilter::from_param(params.get(1))?;
        get_balance(&self.query, scripthash, &filter).await
    }

    pub async fn scripthash_get_first_use(&self, params: &[Value]) -> Result<Value> {
        let scripthash = hash_from_value::<ScriptHash>(params.first())?;
        let filter = QueryFilter::from_param(params.get(1))?;
        get_first_use(&self.query, scripthash, &filter, self.edition()).await
    }

    pub async fn scripthash_get_first_spend(&self, params: &[Value]) -> Result<Value> {
        let scripthash = hash_from_value::<ScriptHash>(params.first())?;
        let filter = QueryFilter::from_param(params.get(1))?;
        get_first_spend(&self.query, scripthash, &filter).await
    }

    pub async fn scripthash_get_history(&self, params: &[Value]) -> Result<Value> {
        let scripthash = hash_from_value::<ScriptHash>(params.first())?;
        let filter = QueryFilter::from_param(params.get(1))?;
        get_history(
            self.query.confirmed_index(),
            self.query.unconfirmed_index(),
            scripthash,
            filter,
        )
        .await
    }

    pub async fn scripthash_get_mempool(&self, params: &[Value]) -> Result<Value> {
        let scripthash = hash_from_value::<ScriptHash>(params.first())?;
        let filter = QueryFilter::from_param(params.get(1))?;
        get_mempool(&self.query, scripthash, filter).await
    }

    pub async fn scripthash_get_volume(&self, params: &[Value]) -> Result<Value> {
        let scripthash = hash_from_value::<ScriptHash>(params.first())?;
        let filter = QueryFilter::from_param(params.get(1))?;
        get_volume(&self.query, scripthash, &filter).await
    }

    pub async fn scripthash_listunspent(&self, params: &[Value]) -> Result<Value> {
        let scripthash = hash_from_value::<ScriptHash>(params.first())?;
        let filter = QueryFilter::from_param(params.get(1))?;
        listunspent(&self.query, scripthash, filter).await
    }

    pub async fn scripthash_subscribe(&self, params: &[Value]) -> Result<Value> {
        let scripthash = hash_from_value::<ScriptHash>(params.first())?;
        self.remove_subscription(&scripthash).await;

        self.doslimits
            .check_subscriptions(self.get_num_subscriptions().await as u32 + 1)?;

        // Subscribe - this will compute statushash if needed or reuse cached one
        let statushash = self
            .subscription_manager
            .subscribe(self.conn_id, scripthash, None)
            .await?;

        let result = statushash.map_or(Value::Null, |h| json!(hex::encode(h)));
        self.subscriptions.lock().await.insert(
            scripthash,
            Subscription {
                statushash,
                alias: None,
            },
        );
        self.stats.subscriptions.inc();
        Ok(result)
    }

    pub async fn scripthash_unsubscribe(&self, params: &[Value]) -> Result<Value> {
        let scripthash = hash_from_value::<ScriptHash>(params.first())?;
        Ok(json!(self.remove_subscription(&scripthash).await))
    }

    fn parse_broadcast_opt(&self, param: Option<&Value>) -> Result<bool> {
        let opt = match param {
            Some(p) => p,
            None => return Ok(false),
        };

        if let Some(p2p) = opt.get("p2p_only") {
            let p2p = p2p.as_bool().context(rpc_invalid_params(
                "option 'p2p_only' not a boolean".to_string(),
            ))?;

            if p2p {
                if self.network != Network::Regtest {
                    // Regtest testing only for now; we don't want the peer to ban us due to external bad input
                    bail!("p2p broadcast only supported on {}", Network::Regtest)
                }
                return Ok(true);
            }
        }

        Ok(false)
    }

    pub async fn transaction_broadcast(&self, params: &[Value]) -> Result<Value> {
        let tx = params
            .first()
            .context(rpc_invalid_params("missing tx".to_string()))?;
        let tx = tx
            .as_str()
            .context(rpc_invalid_params("non-string tx".to_string()))?;
        let tx = hex::decode(tx).context(rpc_invalid_params("non-hex tx".to_string()))?;
        let tx: Transaction =
            deserialize(&tx).context(rpc_invalid_params("failed to parse tx".to_string()))?;

        let p2p_broadcast = self.parse_broadcast_opt(params.get(1))?;
        if p2p_broadcast {
            #[cfg(nexa)]
            let txidentifier = tx.txidem();
            #[cfg(bch)]
            let txidentifier = tx.txid();

            debug!("p2p broadcast of tx {}", txidentifier);
            self.query
                .broadcast_p2p(tx)
                .await
                .context("p2p broadcast failed")?;
            return Ok(json!(txidentifier.to_hex()));
        }

        let txid = self
            .query
            .broadcast(&tx)
            .await
            .context(rpc_invalid_params("rejected by network".to_string()))?;

        if let Err(e) = self.network_notifier.full_tx_channel.sender().try_send(tx) {
            debug!("Failed to notify self of broadcasted tx: {}", e);
        }

        Ok(json!(txid.to_hex()))
    }

    pub async fn transaction_get(&self, params: &[Value]) -> Result<Value> {
        let tx_hash = hash_from_value::<Txid>(params.first())?;
        let verbose = match params.get(1) {
            Some(value) => value.as_bool().context("non-bool verbose value")?,
            None => false,
        };
        if !verbose {
            let tx = self.query.tx().get(&tx_hash, None, None, true).await?;
            Ok(json!(hex::encode(serialize(&tx))))
        } else {
            self.query.tx().get_verbose(&tx_hash, true).await
        }
    }

    pub async fn transaction_get_confirmed_blockhash(&self, params: &[Value]) -> Result<Value> {
        let tx_hash: Txid = hash_from_value(params.first()).context("bad tx_hash")?;
        self.query.get_confirmed_blockhash(&tx_hash).await
    }

    pub async fn transaction_get_merkle(&self, params: &[Value]) -> Result<Value> {
        let tx_hash = hash_from_value::<Txid>(params.first())?;
        let height = if params.get(1).is_some() {
            usize_from_value(params.get(1), "height")
        } else {
            let header = self.query.header().get_by_txid(&tx_hash, None).await;
            match header {
                Ok(Some(header)) => {
                    let height = self
                        .query
                        .header()
                        .get_height(&header)
                        .await
                        .context(anyhow!("Block height for {} missing", header.block_hash()))?;
                    Ok(height as usize)
                }
                _ => Err(rpc_invalid_params(
                    "Transaction is not confirmed in a block".to_string(),
                )),
            }
        }?;

        let (merkle, pos) = self
            .query
            .get_merkle_proof(&tx_hash, height)
            .await
            .context("cannot create merkle proof")?;
        let merkle: Vec<String> = merkle.into_iter().map(|txid| txid.to_hex()).collect();
        Ok(json!({
                "block_height": height,
                "merkle": merkle,
                "pos": pos}))
    }

    pub async fn transaction_id_from_pos(&self, params: &[Value]) -> Result<Value> {
        let height = usize_from_value(params.first(), "height")?;
        let tx_pos = usize_from_value(params.get(1), "tx_pos")?;
        let want_merkle = bool_from_value_or(params.get(2), "merkle", false)?;

        let (txid, merkle) = self
            .query
            .get_id_from_pos(height, tx_pos, want_merkle)
            .await?;

        if !want_merkle {
            return Ok(json!(txid.to_hex()));
        }

        let merkle_vec: Vec<String> = merkle.into_iter().map(|entry| entry.to_hex()).collect();

        Ok(json!({
            "tx_hash" : txid.to_hex(),
            "merkle" : merkle_vec}))
    }

    #[cfg(nexa)]
    pub async fn utxo_get(&self, params: &[Value]) -> Result<Value> {
        use crate::{
            nexa::transaction::{parse_script_template, TxOutType},
            rpc::parseutil::param_to_outpointhash,
        };
        let outpoint = param_to_outpointhash(params.first())?;
        let (tx, output_index, _) = match self.query.get_tx_funding_prevout(&outpoint).await? {
            Some(found) => found,
            None => return Err(rpc_invalid_params("Outpoint not found".to_string())),
        };

        let (mut value, utxo) = self.utxo_get_inner(&tx, output_index).await?;
        {
            let v = value.as_object_mut().unwrap();
            v.insert("tx_idem".to_string(), json!(tx.txidem().to_hex()));
            v.insert("tx_hash".to_string(), json!(tx.txid().to_hex()));
            v.insert("tx_pos".to_string(), json!(output_index));

            let template = if utxo.txout_type == TxOutType::TEMPLATE as u8 {
                parse_script_template(&utxo.script_pubkey)
            } else {
                None
            };

            v.insert(
                "group".to_string(),
                json!(template.as_ref().map(|t| t
                    .token
                    .as_ref()
                    .map(|token| token.as_cashaddr(self.network)))),
            );
            v.insert(
                "token_id_hex".to_string(),
                json!(template
                    .as_ref()
                    .map(|t| t.token.as_ref().map(|token| token.to_hex()))),
            );
            v.insert(
                "group_quantity".to_string(),
                json!(template.as_ref().map(|t| t.quantity)),
            );
            v.insert(
                "group_authority".to_string(),
                json!(template.as_ref().map(|t| t.group_authority().unwrap_or(0))),
            );
            v.insert(
                "template_scripthash".to_string(),
                json!(template.as_ref().map(|t| t.scripthash_human())),
            );
            v.insert(
                "template_argumenthash".to_string(),
                json!(template.map(|t| t.argumenthash.to_hex())),
            );
        }

        Ok(value)
    }

    #[cfg(bch)]
    pub async fn utxo_get(&self, params: &[Value]) -> Result<Value> {
        let txid = hash_from_value::<Txid>(params.first())?;
        let out_n = usize_from_value(params.get(1), "out_n")?;
        if out_n > u32::MAX as usize {
            return Err(rpc_invalid_params(format!(
                "Too large value for out_n parameter ({} > {})",
                out_n,
                u32::MAX
            )));
        }
        let utxo_creation_tx = self.query.tx().get(&txid, None, None, false).await?;
        let (mut value, utxo) = self.utxo_get_inner(&utxo_creation_tx, out_n as u32).await?;
        {
            let v = value.as_object_mut().unwrap();
            v.insert(
                "token_id".to_string(),
                json!(utxo.token.as_ref().map(|t| t.id.to_hex())),
            );
            v.insert(
                "commitment".to_string(),
                json!(utxo.token.as_ref().map(|t| t.commitment.to_hex())),
            );
            v.insert(
                "token_amount".to_string(),
                json!(utxo.token.as_ref().map(|t| t.amount)),
            );
            v.insert(
                "token_bitfield".to_string(),
                json!(utxo.token.as_ref().map(|t| t.bitfield)),
            );
        }
        Ok(value)
    }

    async fn utxo_get_inner(
        &self,
        utxo_creation_tx: &Transaction,
        out_n: u32,
    ) -> Result<(Value, TxOut)> {
        // We want to provide the utxo amount regardless of if it's spent or
        // unspent.

        let utxo = match utxo_creation_tx.output.get(out_n as usize) {
            Some(utxo) => utxo,
            None => {
                bail!(rpc_invalid_params(format!(
                    "out_n {} does not exist on tx {}, the transaction has {} outputs",
                    out_n,
                    utxo_creation_tx.txid(),
                    utxo_creation_tx.output.len()
                )));
            }
        };

        // Fetch the spending transaction (if the utxo is spent).
        let utxo_spending_tx = self
            .query
            .get_tx_spending_prevout(&compute_outpoint_hash_from_tx(utxo_creation_tx, out_n))
            .await?;

        let status = if utxo_spending_tx.is_some() {
            "spent"
        } else {
            "unspent"
        };

        let spent_json = match utxo_spending_tx {
            Some((tx, input_index, height)) => {
                json!({
                    "tx_hash": Some(tx.txid().to_string()),
                    "tx_pos": Some(input_index),
                    "height": Some(if height == MEMPOOL_HEIGHT { 0 } else { height}),
                })
            }
            None => {
                json!({
                    "tx_hash": None::<String>,
                    "tx_pos": None::<u32>,
                    "height": None::<i64>,
                })
            }
        };

        let utxo_confirmation_height = self
            .query
            .tx()
            .get_confirmation_height(&utxo_creation_tx.txid())
            .await;
        let utxo_scripthash = ScriptHash::from_script(&utxo.script_pubkey);

        Ok((
            json!({
                "status": status,
                "amount": utxo.value,
                "scriptpubkey": utxo.script_pubkey,
                "scripthash": utxo_scripthash,
                "height": utxo_confirmation_height,
                "spent": spent_json,
                "addresses": get_addresses(utxo, self.network),
            }),
            utxo.clone(),
        ))
    }

    pub async fn on_chaintip_change(&self, chaintip: BlockHeader) -> Result<Option<Value>> {
        let timer = self
            .stats
            .latency
            .with_label_values(&["chaintip_update"])
            .start_timer();

        let mut last_entry = self.last_header_entry.lock().await;
        if last_entry.is_none() {
            return Ok(None);
        }
        if last_entry.as_ref() == Some(&chaintip) {
            return Ok(None);
        }

        let height = self
            .query
            .header()
            .get_height(&chaintip)
            .await
            .context("header missing")?;
        *last_entry = Some(chaintip);
        let hex_header = hex::encode(serialize(&last_entry.as_ref().unwrap()));
        let header = json!({"hex": hex_header, "height": height});
        timer.observe_duration();
        Ok(Some(json!({
            "jsonrpc": "2.0",
            "method": "blockchain.headers.subscribe",
            "params": [header]})))
    }

    pub async fn on_scripthash_change(
        &self,
        updates: Vec<super::daemon::ScriptHashUpdate>,
    ) -> Vec<Value> {
        if updates.is_empty() {
            return Vec::default();
        }

        let timer = self
            .stats
            .latency
            .with_label_values(&["statushash_update"])
            .start_timer();

        let mut notifications: Vec<Value> = Vec::default();
        let mut subscriptions = self.subscriptions.lock().await;

        for update in updates {
            let subscription_name = update
                .alias
                .clone()
                .unwrap_or_else(|| update.scripthash.to_hex());

            let method = if update.alias.is_some() {
                "blockchain.address.subscribe"
            } else {
                "blockchain.scripthash.subscribe"
            };

            let new_statushash_hex = update
                .new_statushash
                .map_or(Value::Null, |h| json!(hex::encode(h)));

            notifications.push(json!({
                "jsonrpc": "2.0",
                "method": method,
                "params": [subscription_name, new_statushash_hex]
            }));

            // Update stored statushash
            if let Some(s) = subscriptions.get_mut(&update.scripthash) {
                s.statushash = update.new_statushash;
            }
        }

        timer.observe_duration();
        notifications
    }

    pub async fn get_num_subscriptions(&self) -> i64 {
        self.subscriptions.lock().await.len() as i64
    }

    async fn remove_subscription(&self, scripthash: &ScriptHash) -> bool {
        let removed = self.subscriptions.lock().await.remove(scripthash);
        match removed {
            Some(subscription) => {
                if let Some(alias) = subscription.alias {
                    self.alias_bytes_used
                        .fetch_sub(alias.len(), Ordering::Relaxed);
                }
                self.subscription_manager
                    .unsubscribe(self.conn_id, *scripthash)
                    .await;
                self.stats.subscriptions.dec();
                true
            }
            None => false,
        }
    }
}
