pub mod manager;
pub mod p2pconnection;
pub mod p2pmessage;
pub mod rpcconnection;

use crate::chain::Chain;
use crate::chaindef::BlockHash;

use crate::chaindef::Transaction;
use crate::config::Config;
use crate::daemon;
use crate::daemon::manager::BitcoindRPCConnectionManager;
use crate::errors::ConnectionError;
use crate::metrics;
use crate::signal::NetworkNotifier;
use bitcoin_hashes::Hash;
use bitcoincash::consensus::encode::{deserialize, serialize};
use bitcoincash::hash_types::Txid;
use bitcoincash::hashes::hex::FromHex;
use bitcoincash::network::constants::Network;
use serde_json::{from_value, Value};
use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use crate::cache::BlockTxIDsCache;
use crate::chaindef::Block;
use crate::chaindef::BlockHeader;

use crate::metrics::Metrics;
#[cfg(nexa)]
use crate::nexa::hash_types::TxIdem;
use crate::rpc::encoding::{blockhash_to_hex, hex_to_blockhash, hex_to_txid, txid_to_hex};
use crate::signal::Waiter;
use anyhow::{Context, Result};

use self::manager::BitcoindP2PConnectionManager;

fn header_from_value(value: Value) -> Result<BlockHeader> {
    let header_hex = value
        .as_str()
        .context(format!("non-string header: {}", value))?;
    let header_bytes = hex::decode(header_hex).context("non-hex header")?;
    deserialize(&header_bytes).context(format!("failed to parse header {}", header_hex))
}

fn block_from_value(value: Value) -> Result<Block> {
    let block_hex = value.as_str().context("non-string block")?;
    let block_bytes = hex::decode(block_hex).context("non-hex block")?;
    deserialize(&block_bytes).context(format!("failed to parse block {}", block_hex))
}

fn tx_from_value(value: Value) -> Result<Transaction> {
    let tx_hex = value.as_str().context("non-string tx")?;
    let tx_bytes = hex::decode(tx_hex).context("non-hex tx")?;
    deserialize(&tx_bytes).context(format!("failed to parse tx {}", tx_hex))
}

#[derive(Serialize, Deserialize, Debug)]
struct BlockchainInfo {
    chain: String,
    blocks: u32,
    headers: u32,
    verificationprogress: f64,
    bestblockhash: String,
    pruned: bool,
    initialblockdownload: bool,
}

#[derive(Serialize, Deserialize, Debug)]
struct NetworkInfo {
    version: u64,
    subversion: String,
    relayfee: f64, // in BTC
}

pub trait CookieGetter: Send + Sync {
    fn get(&self) -> Result<Vec<u8>>;
}

pub struct Daemon {
    daemon_dir: PathBuf,
    network: Network,
    rpc_pool: bb8::Pool<BitcoindRPCConnectionManager>,
    p2p_pool: bb8::Pool<BitcoindP2PConnectionManager>,
    blocktxids_cache: Arc<BlockTxIDsCache>,

    // monitoring
    latency: prometheus::HistogramVec,
    size: prometheus::HistogramVec,
}

impl Daemon {
    #[allow(clippy::too_many_arguments)]
    pub async fn new(
        config: &Config,
        network_notifier: Arc<NetworkNotifier>,
        metrics: &Metrics,
    ) -> Result<Daemon> {
        let rpc_manager = daemon::manager::BitcoindRPCConnectionManager::new(
            config.daemon_rpc_addr,
            Arc::clone(&config.cookie_getter),
        );

        let p2p_manager = daemon::manager::BitcoindP2PConnectionManager::new(
            config.network_type,
            config.daemon_p2p_addr,
            network_notifier,
            metrics,
        );

        let pool_size = config.daemon_rpc_connections;

        let rpc_pool = bb8::Pool::builder()
            .max_size(pool_size)
            .min_idle(Some(pool_size))
            .test_on_check_out(false)
            .build(rpc_manager)
            .await?;

        let p2p_pool = bb8::Pool::builder()
            .max_size(1)
            .min_idle(Some(1))
            .test_on_check_out(false)
            .build_unchecked(p2p_manager); // unchecked: we don't require p2p connections

        let blocktxids_cache = Arc::new(BlockTxIDsCache::new(
            config.blocktxids_cache_size as u64,
            metrics,
        ));

        let daemon = Daemon {
            daemon_dir: config.daemon_dir.to_path_buf(),
            network: config.network_type,
            rpc_pool,
            p2p_pool,
            blocktxids_cache,
            latency: metrics.histogram_vec(
                "rostrum_daemon_rpc",
                "Bitcoind RPC latency (in seconds)",
                &["method"],
                metrics::default_duration_buckets(),
            ),
            size: metrics.histogram_vec(
                "rostrum_daemon_bytes",
                "Bitcoind RPC size (in bytes)",
                &["method", "dir"],
                metrics::default_size_buckets(),
            ),
        };
        let network_info = daemon.getnetworkinfo().await?;
        info!("{:?}", network_info);
        let blockchain_info = daemon.getblockchaininfo().await?;
        info!("{:?}", blockchain_info);
        if blockchain_info.pruned {
            bail!("pruned node is not supported (use '-prune=0' bitcoind flag)".to_owned())
        }
        Ok(daemon)
    }

    pub fn reconnect(&self) -> Result<Daemon> {
        Ok(Daemon {
            daemon_dir: self.daemon_dir.clone(),
            network: self.network,
            rpc_pool: self.rpc_pool.clone(),
            p2p_pool: self.p2p_pool.clone(),
            blocktxids_cache: Arc::clone(&self.blocktxids_cache),
            latency: self.latency.clone(),
            size: self.size.clone(),
        })
    }

    async fn retry_request_batch(
        &self,
        method: &str,
        params_list: &[Option<Value>],
    ) -> Result<Vec<Value>> {
        let mut connection = self.rpc_pool.get().await?;
        loop {
            Waiter::shutdown_check()?;
            match connection.multi_request(method, params_list).await {
                Err(e) => {
                    if let Some(e) = e.downcast_ref::<ConnectionError>() {
                        warn!("Full node connection broken: {}", e.msg);
                        drop(connection);
                        Waiter::wait_or_shutdown(Duration::from_secs(1)).await?;
                        connection = self.rpc_pool.get().await?;
                        continue;
                    }
                    return Err(e);
                }
                Ok(result) => return Ok(result),
            }
        }
    }

    async fn retry_request(&self, method: &str, params: &Option<Value>) -> Result<Value> {
        let mut connection = self.rpc_pool.get().await?;
        loop {
            match connection.request(method, params).await {
                Err(e) => {
                    if let Some(e) = e.downcast_ref::<ConnectionError>() {
                        warn!("Full node connection broken: {}", e.msg);
                        drop(connection);
                        Waiter::wait_or_shutdown(Duration::from_secs(1)).await?;
                        connection = self.rpc_pool.get().await?;
                        continue;
                    }
                    return Err(e);
                }
                Ok(result) => return Ok(result),
            }
        }
    }

    pub async fn request(&self, method: &str, params: &Option<Value>) -> Result<Value> {
        self.retry_request(method, params).await
    }

    pub async fn multi_request(
        &self,
        method: &str,
        params_list: &[Option<Value>],
    ) -> Result<Vec<Value>> {
        self.retry_request_batch(method, params_list).await
    }

    // bitcoind JSONRPC API:

    async fn getblockchaininfo(&self) -> Result<BlockchainInfo> {
        let info: Value = self.request("getblockchaininfo", &None).await?;
        from_value(info).context("invalid blockchain info")
    }

    async fn getnetworkinfo(&self) -> Result<NetworkInfo> {
        let info: Value = self.request("getnetworkinfo", &None).await?;
        from_value(info).context("invalid network info")
    }

    pub async fn get_subversion(&self) -> Result<String> {
        Ok(self.getnetworkinfo().await?.subversion)
    }

    pub async fn get_relayfee(&self) -> Result<f64> {
        Ok(self.getnetworkinfo().await?.relayfee)
    }

    pub async fn getbestblockhash(&self) -> Result<BlockHash> {
        hex_to_blockhash(&self.request("getbestblockhash", &None).await?)
            .context("invalid blockhash")
    }

    #[cfg(nexa)]
    pub async fn getblockheader(&self, blockhash: &BlockHash) -> Result<(BlockHeader, u64)> {
        let header = header_from_value(
            self.request(
                "getblockheader",
                &Some(json!([
                    blockhash_to_hex(blockhash),
                    /*verbose=*/ false,
                ])),
            )
            .await?,
        )?;
        let height = header.height();
        Ok((header, height))
    }

    /**
     * Fetch block header for Bitcoin Cash. We also need the height, which is only available with
     * the verbose output. To avoid two requests, we parse the verbose output into a header.
     */
    #[cfg(bch)]
    pub async fn getblockheader(&self, blockhash: &BlockHash) -> Result<(BlockHeader, u64)> {
        use bitcoincash::TxMerkleNode;

        let h = self
            .request(
                "getblockheader",
                &Some(json!([blockhash_to_hex(blockhash), /*verbose=*/ true,])),
            )
            .await?;

        let version = h
            .get("version")
            .context("version missing")?
            .as_i64()
            .context("version is not numeric")? as i32;
        let prev_blockhash = BlockHash::from_hex(
            h.get("previousblockhash")
                .unwrap_or(&json!(BlockHash::all_zeros()))
                .as_str()
                .context("previousblockhash not a string")?,
        )
        .context("invalid prev_blockhash")?;
        let merkle_root = TxMerkleNode::from_hex(
            h.get("merkleroot")
                .context("merkleroot missing")?
                .as_str()
                .context("merkleroot not a hex")?,
        )
        .context("invalid merkleroot")?;
        let time = h
            .get("time")
            .context("time missing")?
            .as_u64()
            .context("time is not numeric")? as u32;
        let bits = h
            .get("bits")
            .context("bits missing")?
            .as_str()
            .context("bits is not a string")?;
        let bits = u32::from_str_radix(bits, 16).context("invalid bits")?;
        let nonce = h
            .get("nonce")
            .context("nonce missing")?
            .as_u64()
            .context("nonce is not numeric")? as u32;

        let header = BlockHeader {
            version,
            prev_blockhash,
            merkle_root,
            time,
            bits,
            nonce,
        };

        debug_assert!(&header.block_hash() == blockhash);

        let height = h
            .get("height")
            .context("height missing for header")?
            .as_u64()
            .context("height was not numerical")?;
        Ok((header, height))
    }

    pub async fn getblockheaders(&self, heights: &[usize]) -> Result<Vec<(BlockHeader, u64)>> {
        let heights_json: Vec<Option<Value>> = heights
            .iter()
            .map(|height| Some(json!([height, false])))
            .collect();

        let result = self
            .multi_request("getblockheader", &heights_json)
            .await?
            .into_iter()
            .zip(heights.iter())
            .map(|(response, height)| {
                let header = header_from_value(response)?;
                Ok((header, *height as u64))
            })
            .collect();
        result
    }

    pub async fn getblock(&self, blockhash: &BlockHash) -> Result<Block> {
        let block = block_from_value(
            self.request(
                "getblock",
                &Some(json!([blockhash_to_hex(blockhash), /*verbose=*/ 0,])),
            )
            .await?,
        )?;
        assert_eq!(block.block_hash(), *blockhash);
        Ok(block)
    }

    async fn load_blocktxids(&self, blockhash: &BlockHash) -> Result<Vec<Txid>> {
        let block = self
            .request(
                "getblock",
                &Some(json!([blockhash_to_hex(blockhash), /*verbose=*/ 1])),
            )
            .await?;

        block
            .get("tx")
            .or_else(|| block.get("txid")) // NEXA variant of the keyword
            .context("block missing txids")?
            .as_array()
            .context("invalid block txids")?
            .iter()
            .map(hex_to_txid)
            .collect::<Result<Vec<Txid>>>()
    }

    pub async fn getblocktxids(&self, blockhash: &BlockHash) -> Result<Vec<Txid>> {
        self.blocktxids_cache
            .get_or_else(
                blockhash,
                async move { self.load_blocktxids(blockhash).await },
            )
            .await
    }

    pub async fn gettransaction(
        &self,
        txid: &Txid,
        blockhash: Option<BlockHash>,
    ) -> Result<Transaction> {
        let tx = self
            .gettransaction_raw(txid, blockhash.as_ref(), false)
            .await?;
        tx_from_value(tx)
    }

    pub async fn gettransaction_raw(
        &self,
        txid: &Txid,
        blockhash: Option<&BlockHash>,
        verbose: bool,
    ) -> Result<Value> {
        let txid = txid_to_hex(txid);
        if let Some(h) = blockhash {
            return self
                .request(
                    "getrawtransaction",
                    &Some(json!([txid, verbose, blockhash_to_hex(h)])),
                )
                .await;
        }
        self.request("getrawtransaction", &Some(json!([txid])))
            .await
    }

    fn txids_to_hashset(&self, txids: Value) -> Result<HashSet<Txid>> {
        let mut result = HashSet::new();
        for value in txids.as_array().context("non-array result")? {
            result.insert(hex_to_txid(value).context("invalid txid")?);
        }
        Ok(result)
    }

    #[cfg(nexa)]
    pub async fn getmempooltxids(&self) -> Result<HashSet<Txid>> {
        let txids: Value = self
            .request(
                "getrawtxpool",
                &Some(json!([/*verbose=*/ false, /* id or idem */ "id"])),
            )
            .await?;
        self.txids_to_hashset(txids)
    }

    #[cfg(bch)]
    pub async fn getmempooltxids(&self) -> Result<HashSet<Txid>> {
        let txids: Value = self
            .request("getrawmempool", &Some(json!([/*verbose=*/ false])))
            .await?;
        self.txids_to_hashset(txids)
    }

    #[cfg(bch)]
    pub async fn broadcast(&self, tx: &Transaction) -> Result<Txid> {
        let tx = hex::encode(serialize(tx));
        let txid = self
            .request("sendrawtransaction", &Some(json!([tx])))
            .await?;
        Txid::from_hex(txid.as_str().context("non-string txid")?).context("failed to parse txid")
    }
    #[cfg(nexa)]
    pub async fn broadcast(&self, tx: &Transaction) -> Result<TxIdem> {
        let tx = hex::encode(serialize(tx));
        let txidem = self
            .request("sendrawtransaction", &Some(json!([tx])))
            .await?;
        TxIdem::from_hex(txidem.as_str().context("non-string txidem")?)
            .context("failed to parse txidem")
    }

    pub async fn broadcast_p2p(&self, tx: Transaction) -> Result<()> {
        self.p2p_pool.get().await?.send_tx(tx).await
    }

    async fn get_all_headers(&self, tip: &BlockHash) -> Result<Vec<(BlockHeader, u64)>> {
        let info: Value = self
            .request("getblockheader", &Some(json!([blockhash_to_hex(tip)])))
            .await?;
        let tip_height = info
            .get("height")
            .expect("missing height")
            .as_u64()
            .expect("non-numeric height") as usize;
        let all_heights: Vec<usize> = (0..=tip_height).collect();
        let chunk_size = 100_000;
        let mut result = vec![];
        let null_hash = BlockHash::all_zeros();
        for heights in all_heights.chunks(chunk_size) {
            Waiter::shutdown_check()?;
            info!("Downloading {} block headers", heights.len());
            let mut headers = self.getblockheaders(heights).await?;
            assert!(headers.len() == heights.len());
            result.append(&mut headers);
        }

        let mut blockhash = null_hash;
        let mut prev_header: Option<BlockHeader> = None;
        #[allow(clippy::clone_on_copy)]
        for (header, _height) in &result {
            assert_eq!(
                header.prev_blockhash,
                blockhash,
                "Prev block hex: {}, this block hex: {}",
                hex::encode(serialize(&prev_header.unwrap())),
                hex::encode(serialize(&header)),
            );
            blockhash = header.block_hash();
            prev_header = Some(header.clone());
        }
        assert_eq!(blockhash, *tip);
        Ok(result)
    }

    // Returns a list of BlockHeaders in ascending height (i.e. the tip is last).
    pub async fn get_new_headers(
        &self,
        chain: &Chain,
        bestblockhash: &BlockHash,
    ) -> Result<Vec<(BlockHeader, u64)>> {
        // Iterate back over headers until known blockash is found:
        if chain.height().await == 0 {
            info!("Fetching all headers");
            return self.get_all_headers(bestblockhash).await;
        }
        info!(
            "Downloading new block headers ({} already indexed) from {}",
            chain.height().await,
            bestblockhash,
        );
        let mut new_headers: Vec<(BlockHeader, u64)> = vec![];
        let null_hash = BlockHash::all_zeros();
        let mut blockhash = *bestblockhash;
        while blockhash != null_hash {
            Waiter::shutdown_check()?;
            if new_headers.len().is_multiple_of(1000) {
                info!(
                    "Downloading headers progress: {} fetched... ",
                    new_headers.len()
                );
            }
            if chain.contains(&blockhash).await {
                break;
            }

            let header = self
                .getblockheader(&blockhash)
                .await
                .context(format!("failed to get {} header", blockhash))?;
            blockhash = header.0.prev_blockhash;
            new_headers.push(header);
        }
        trace!("downloaded {} block headers", new_headers.len());
        new_headers.reverse(); // so the tip is the last vector entry
        Ok(new_headers)
    }

    pub async fn get_genesis(&self) -> Result<Block> {
        let hash = self.request("getblockhash", &Some(json!([0]))).await?;
        let response = self.request("getblock", &Some(json!([hash, 0]))).await?;
        block_from_value(response)
    }

    pub async fn get_block_count(&self) -> Result<u64> {
        self.request("getblockcount", &None)
            .await?
            .as_u64()
            .context("expected blockheight")
    }

    /// Trigger bb8 to check if p2p connection is broken
    pub async fn p2p_keepalive(&self) {
        let parent = self;
        let _ = tokio::time::timeout(Duration::from_millis(50), parent.p2p_pool.get()).await;
    }
}
