use crate::log_util::setup_logging;
use bitcoin::hashes::hex::FromHex;
use bitcoin::hashes::Hash;
use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey};
use bitcoin::{Block, BlockHash, BlockHeader, FilterHash, FilterHeader, KeyPair, Network};
use bitcoind_client::bitcoind_client::bitcoind_client_from_url;
use bitcoind_client::{default_bitcoin_rpc_url, BitcoindClient, BlockSource};
use clap::Parser;
use log::*;
use rand::Rng;
use rayon::iter::{ParallelBridge, ParallelIterator};
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use std::cmp::max;
use std::collections::HashMap;
use std::fs;
use std::path::PathBuf;
use std::str::FromStr;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{SystemTime, UNIX_EPOCH};
use txoo::filter::BlockSpendFilter;
use txoo::source::{attestation_path, read_yaml_from_file, write_yaml_to_file};
use txoo::util::sign_attestation;
use txoo::{
    decode_checkpoint, Attestation, OracleSetup, SignedAttestation, CHECKPOINTS_BITCOIN,
    CHECKPOINTS_TESTNET,
};
use url::Url;

mod log_util;

#[derive(clap::Parser, Debug)]
#[clap(author, version, about, long_about = None)]
struct Opts {
    #[clap(short, long, default_value = "bitcoin")]
    network: Network,
    #[clap(short, long, help = "bitcoind RPC URL, must have http(s) schema")]
    rpc: Option<String>,
    #[clap(
        short,
        long,
        help = "data directory path.  if not set, will use ~/.txoo/<network>"
    )]
    datadir: Option<String>,
    #[clap(long, help = "preload blocks from the bitcoin blocks directory")]
    preload: bool,
    #[clap(long, help = "bitcoin blocks directory (defaults to ~/.bitcoin/...)")]
    blocks_dir: Option<String>,
    #[clap(
        long,
        help = "start computing commitments at height",
        default_value = "0"
    )]
    start_block: u32,
    // prune attestations older than about a year by default
    #[clap(
        long,
        help = "prune attestations deeper than this",
        default_value = "52560"
    )]
    prune_depth: u32,
    #[clap(long, help = "start at the genesis block")]
    no_checkpoint: bool,
    #[clap(long, help = "allow reorgs of any depth - for testnet blockstorms")]
    allow_deep_reorgs: bool,
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    handle_termination_signals();

    let opts: Opts = Opts::parse();
    let network = opts.network.clone();
    let rpc_s = default_bitcoin_rpc_url(opts.rpc.clone(), network);
    let rpc = Url::parse(&rpc_s).expect("rpc url");
    run_daemon(opts, network, rpc).await?;
    Ok(())
}

fn handle_termination_signals() -> () {
    // Work around the fact that process with PID 1 does not have a default
    // signal handler. We usually have PID 1 when run inside docker container.
    // We just exit the process without error as being interrupted is ok.
    ctrlc::set_handler(|| {
        eprintln!("Received termination signal. Shutting down.");
        std::process::exit(0);
    })
    .expect("Error setting termination signal handler");
}

async fn run_daemon(opts: Opts, network: Network, rpc: Url) -> anyhow::Result<()> {
    let datadir = opts.datadir.map(|s| s.into()).unwrap_or_else(|| {
        let home = dirs::home_dir().expect("cannot get cookie file if HOME is not set");
        home.join(".txoo").join(network.to_string())
    });

    fs::create_dir_all(datadir.join("public"))?;

    setup_logging(&datadir, "txood", "info");

    info!("bitcoind RPC {}", rpc);

    let client = bitcoind_client_from_url(rpc, network).await;

    assert!(opts.prune_depth > MAX_REORG_DEPTH);

    let blocks_dir = if opts.preload {
        let blocks_dir = opts
            .blocks_dir
            .map(|s| s.into())
            .unwrap_or_else(|| get_blocks_dir(network));
        Some(blocks_dir)
    } else {
        None
    };
    let mut daemon = Daemon::new(
        client,
        datadir,
        opts.start_block,
        opts.prune_depth,
        network,
        blocks_dir,
        !opts.no_checkpoint,
        opts.allow_deep_reorgs,
    )
    .await?;
    info!("starting at {:?}", daemon.tip());
    loop {
        let did_work = daemon.step().await?;
        if !did_work {
            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
        }
    }
}

fn get_blocks_dir(network: Network) -> PathBuf {
    let home = dirs::home_dir().expect("HOME is not set");
    let dir = home.join(".bitcoin");
    let dir = match network {
        Network::Bitcoin => dir,
        n => dir.join(n.to_string()),
    };
    dir.join("blocks")
}

#[derive(Serialize, Deserialize, Debug)]
struct Tip {
    height: u32,
    hash: BlockHash,
    filter_header: FilterHeader,
    // keep track of how far back we've gone in a reorg.
    // normally zero.
    reorg_depth: u32,
}

impl Tip {
    fn new(height: u32, hash: BlockHash, filter_header: FilterHeader) -> Self {
        Self {
            height,
            hash,
            filter_header,
            reorg_depth: 0,
        }
    }

    /// Pop the tip during a reorg, given the header of the block that is being
    /// disconnected.
    /// Checks that the recorded hash matches the header hash.
    fn pop(&mut self, header: &BlockHeader, prev_filter_header: FilterHeader) {
        assert_eq!(self.hash, header.block_hash());
        self.height -= 1;
        self.hash = header.prev_blockhash;
        self.reorg_depth += 1;
        self.filter_header = prev_filter_header;
    }

    pub(crate) fn push(&mut self, header: &BlockHeader, filter_header: FilterHeader) {
        assert_eq!(self.hash, header.prev_blockhash);
        self.height += 1;
        self.hash = header.block_hash();
        self.filter_header = filter_header;
        if self.reorg_depth > 0 {
            self.reorg_depth -= 1;
        }
    }
}

const MAX_REORG_DEPTH: u32 = 144;

struct Daemon {
    client: BitcoindClient,
    network: Network,
    datadir: PathBuf,
    tip: Tip,
    prune_depth: u32,
    catchup_height: u32,
    keypair: KeyPair,
    secp: Secp256k1<bitcoin::secp256k1::All>,
    cache: Arc<Mutex<HashMap<BlockHash, FilterHash>>>,
    cache_miss: u32,
    checkpoints: HashMap<u32, (BlockHash, FilterHeader)>,
    allow_deep_reorgs: bool,
}

impl Daemon {
    async fn new(
        client: BitcoindClient,
        datadir: PathBuf,
        start_block: u32,
        prune_depth: u32,
        network: Network,
        blocks_dir: Option<PathBuf>,
        use_checkpoint: bool,
        allow_deep_reorgs: bool,
    ) -> anyhow::Result<Self> {
        let tip = Self::get_tip(&client, &datadir, start_block, network, use_checkpoint).await;
        let info = client.get_blockchain_info().await?;
        info!("{:?}", info);

        let secp = Secp256k1::new();

        let config: Option<OracleSetup> = read_yaml_from_file(&datadir, "public/config");
        let secret_key = if let Some(config) = config {
            // read secret key from disk
            let secret_key = SecretKey::from_str(
                &fs::read_to_string(datadir.join("secret_key")).expect("secret key on disk"),
            )
            .expect("secret key decode");
            let public_key = PublicKey::from_secret_key(&secp, &secret_key);

            if config.network != network {
                panic!("config network mismatch");
            }

            if config.start_block != start_block {
                panic!("config start block mismatch");
            }

            if config.public_key != public_key {
                panic!("config public key mismatch");
            }
            secret_key
        } else {
            let secret_key =
                SecretKey::from_slice(&rand::thread_rng().gen::<[u8; 32]>()).expect("secret key");
            fs::write(
                datadir.join("secret_key"),
                secret_key.display_secret().to_string(),
            )?;
            let public_key = PublicKey::from_secret_key(&secp, &secret_key);
            let config = OracleSetup {
                network,
                start_block,
                public_key,
            };
            write_yaml_to_file(&datadir, "public/config", &config);
            secret_key
        };

        let keypair = KeyPair::from_secret_key(&secp, &secret_key);
        let checkpoint_list = match network {
            Network::Bitcoin => CHECKPOINTS_BITCOIN,
            Network::Testnet => CHECKPOINTS_TESTNET,
            _ => &[],
        };
        let checkpoints = checkpoint_list
            .iter()
            .map(|(h, b, f, _)| {
                (
                    *h,
                    (
                        BlockHash::from_hex(b).unwrap(),
                        FilterHeader::from_hex(f).unwrap(),
                    ),
                )
            })
            .collect();
        let daemon = Self {
            client,
            network,
            datadir: datadir.clone(),
            tip,
            prune_depth,
            catchup_height: info.latest_height as u32,
            keypair,
            secp,
            cache: Arc::new(Mutex::new(HashMap::new())),
            cache_miss: 0,
            checkpoints,
            allow_deep_reorgs,
        };

        if let Some(blocks_dir) = blocks_dir {
            let preload_height = max(start_block, daemon.tip.height);
            daemon.preload(blocks_dir, preload_height);
        }
        daemon.prune();
        Ok(daemon)
    }

    pub fn tip(&self) -> &Tip {
        &self.tip
    }

    // return true if we did some work
    async fn step(&mut self) -> anyhow::Result<bool> {
        // get the next block after the tip
        let new_height = self.tip.height + 1;
        let hash_opt = self.client.get_block_hash(new_height).await?;
        if let Some(hash) = hash_opt {
            // check cache and dispatch accordingly
            let filter_hash_opt = {
                let cache = self.cache.lock().unwrap();
                if self.tip.height % 1000 == 0 {
                    if !cache.is_empty() {
                        info!("{} cache misses in last 1000", self.cache_miss);
                        self.cache_miss = 0;
                    }
                }
                let res = cache.get(&hash).cloned();
                if res.is_none() && !cache.is_empty() {
                    self.cache_miss += 1;
                }
                res
            };
            if let Some(filter_hash) = filter_hash_opt {
                let header = self.client.get_header(&hash).await?.header;
                if self.check_reorg(&header).await? {
                    return Ok(true);
                }
                assert_eq!(header.block_hash(), hash);
                self.handle_preloaded_block(&header, filter_hash)?;
            } else {
                let block = self.client.get_block(&hash).await?;
                if self.check_reorg(&block.header).await? {
                    return Ok(true);
                }
                assert_eq!(block.block_hash(), hash);
                self.handle_new_block(&block)?;
            }
            return Ok(true);
        } else {
            debug!("no new block");
            return Ok(false);
        }
    }

    fn check_checkpoint(&mut self, height: u32, hash: &BlockHash, filter_header: &FilterHeader) {
        if let Some((cp_hash, cp_filter_header)) = self.checkpoints.get(&height) {
            if cp_hash != hash || cp_filter_header != filter_header {
                panic!("checkpoint mismatch at {}", self.tip.height);
            }
        }
    }

    // if there is a reorg, handles it and returns true
    async fn check_reorg(&mut self, header: &BlockHeader) -> anyhow::Result<bool> {
        // if the block's prev doesn't point to our tip, we're in a reorg
        if header.prev_blockhash != self.tip.hash {
            info!(
                "reorg detected, height: {} new hash {}",
                self.tip.height, header.prev_blockhash
            );
            self.handle_reorg().await?;
            return Ok(true);
        }
        return Ok(false);
    }

    // pop the tip.  caller will try to move forward again, or continue popping if there is a multi
    // block reorg.
    async fn handle_reorg(&mut self) -> anyhow::Result<()> {
        if self.tip.reorg_depth >= MAX_REORG_DEPTH {
            if self.allow_deep_reorgs {
                warn!("deep reorg depth={}", self.tip.reorg_depth);
            } else {
                panic!("reorg depth exceeded");
            }
        }
        let header_data = self.client.get_header(&self.tip.hash).await?;
        assert_eq!(
            header_data.height, self.tip.height,
            "height mismatch at tip {}",
            self.tip.hash
        );
        let header = header_data.header;
        let prev_attestation = self
            .read_attestation(self.tip.height - 1, &header.prev_blockhash)
            .expect("attestation for prev block should exist");
        let prev_filter_header = prev_attestation.attestation.filter_header;
        self.tip.pop(&header, prev_filter_header);
        Ok(())
    }

    async fn get_tip(
        bitcoind_client: &BitcoindClient,
        datadir: &PathBuf,
        start_block: u32,
        network: Network,
        use_checkpoint: bool,
    ) -> Tip {
        if let Some(tip) = read_yaml_from_file(&datadir, "tip") {
            tip
        } else {
            if use_checkpoint && network == Network::Bitcoin {
                Self::tip_from_checkpoint(CHECKPOINTS_BITCOIN[0])
            } else if use_checkpoint && network == Network::Testnet {
                Self::tip_from_checkpoint(CHECKPOINTS_TESTNET[0])
            } else {
                // starting from scratch
                let hash = bitcoind_client
                    .get_block_hash(start_block)
                    .await
                    .expect("get block hash")
                    .expect("block");
                Tip::new(start_block, hash, FilterHeader::all_zeros())
            }
        }
    }

    fn tip_from_checkpoint(checkpoint: (u32, &str, &str, &str)) -> Tip {
        let (height, block_hash, filter_header, _) = decode_checkpoint(checkpoint);
        Tip::new(height, block_hash, filter_header)
    }

    fn write_attestation(
        &self,
        height: u32,
        hash: &BlockHash,
        signed_attestation: &SignedAttestation,
    ) {
        Self::do_write_attestation(&self.datadir, height, hash, signed_attestation);
    }

    fn do_write_attestation(
        datadir: &PathBuf,
        height: u32,
        hash: &BlockHash,
        signed_attestation: &SignedAttestation,
    ) {
        write_yaml_to_file(
            datadir,
            &attestation_path(height, hash),
            &signed_attestation,
        )
    }

    fn read_attestation(&self, height: u32, hash: &BlockHash) -> Option<SignedAttestation> {
        read_yaml_from_file(&self.datadir, &attestation_path(height, hash))
    }

    #[allow(unused)]
    fn read_data<T: DeserializeOwned>(&self, name: &str) -> Option<T> {
        read_yaml_from_file(&self.datadir, name)
    }

    fn write_data<T: Serialize>(&self, name: &str, data: T) {
        write_yaml_to_file(&self.datadir, name, data);
    }

    fn now() -> u64 {
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("time")
            .as_secs()
    }

    fn handle_new_block(&mut self, block: &Block) -> anyhow::Result<()> {
        let filter = BlockSpendFilter::from_block(&block);
        let filter_header = filter.filter_header(&self.tip.filter_header);

        self.do_new_block(&block.header, filter_header);

        Ok(())
    }

    fn handle_preloaded_block(
        &mut self,
        header: &BlockHeader,
        filter_hash: FilterHash,
    ) -> anyhow::Result<()> {
        let filter_header = filter_hash.filter_header(&self.tip.filter_header);

        self.do_new_block(&header, filter_header);

        Ok(())
    }

    fn do_new_block(&mut self, header: &BlockHeader, filter_header: FilterHeader) {
        let new_height = self.tip.height + 1;
        let block_hash = header.block_hash();
        self.check_checkpoint(new_height, &block_hash, &filter_header);

        // Only write out attestations if we passed the prune depth,
        // but write them out anyway every 1000 blocks.
        if self.prune_depth == 0
            || new_height + self.prune_depth > self.catchup_height
            || new_height % 1000 == 0
        {
            let attestation = Attestation {
                block_hash,
                block_height: new_height,
                filter_header,
                time: Self::now(),
            };
            let signed_attestation = sign_attestation(attestation, &self.keypair, &self.secp);

            self.write_attestation(new_height, &block_hash, &signed_attestation);
            self.prune();
        }

        self.tip.push(&header, filter_header);

        if new_height >= self.catchup_height {
            info!(
                "processed {} block {} filter {}",
                new_height, block_hash, filter_header
            );
        } else if new_height % 1000 == 0 {
            info!(
                "processed (catchup) {} block {} filter {}",
                new_height, block_hash, filter_header
            );
        }

        self.write_data("tip", &self.tip);
    }

    fn prune(&self) {
        if self.prune_depth == 0 {
            return;
        }
        for entry in fs::read_dir(self.datadir.join("public")).expect("read_dir") {
            let entry = entry.expect("entry");
            let name_os = entry.file_name();
            let name = name_os.to_str().unwrap();
            if name.ends_with(".sa") {
                let (height_s, _) = name.split_once("-").expect(".sa file with no dash");
                let height: u32 = height_s.parse().unwrap();
                if height + self.prune_depth <= self.tip.height && height % 1000 != 0 {
                    fs::remove_file(entry.path()).expect("remove_file");
                }
            }
        }
    }

    fn preload(&self, blocks_dir: PathBuf, start_height: u32) {
        let network = self.network;
        let cache = self.cache.clone();
        thread::spawn(move || {
            Self::do_preload(&*cache, network, blocks_dir, start_height);
        });
    }

    fn do_preload(
        cache: &Mutex<HashMap<BlockHash, FilterHash>>,
        network: Network,
        blocks_dir: PathBuf,
        start_height: u32,
    ) {
        info!(
            "preloading from starting at block {} from {}",
            start_height,
            blocks_dir.display()
        );
        let config = blocks_iterator::Config {
            blocks_dir,
            network,
            skip_prevout: true,
            max_reorg: 144,
            channels_size: 100,
            start_at_height: start_height,
            stop_at_height: None,
        };
        blocks_iterator::iter(config)
            .par_bridge()
            .for_each(|block| {
                if block.height % 1000 == 0 {
                    info!("preloaded {} blocks", block.height);
                }
                let filter = BlockSpendFilter::from_block(&block.block);
                cache
                    .lock()
                    .unwrap()
                    .insert(block.block_hash, filter.filter_hash());
            });
    }
}

#[cfg(test)]
mod tests {
    use bitcoin::hashes::Hash;
    use bitcoin::secp256k1::{Secp256k1, SecretKey};
    use bitcoin::{BlockHash, FilterHeader, KeyPair};

    #[test]
    fn test_sign() {
        let attestation = crate::Attestation {
            block_hash: BlockHash::all_zeros(),
            block_height: 0,
            filter_header: FilterHeader::all_zeros(),
            time: 0,
        };
        let secp = Secp256k1::new();
        let secret_key = SecretKey::from_slice(&[1; 32]).unwrap();
        let keypair = KeyPair::from_secret_key(&secp, &secret_key);
        let signed_attestation = super::sign_attestation(attestation, &keypair, &secp);
        assert!(signed_attestation.verify(&keypair.public_key(), &secp));
    }
}
