rostrum 14.0.1

use log::{debug, info, warn};
use std::{
    collections::HashMap,
    sync::{atomic::AtomicBool, Arc, Mutex},
    time::Duration,
};

use bitcoincash::Network;
use configure_me::toml::from_str;
use electrum_client_netagnostic::Param;
use serde_json::Value;

use crate::{
    config::{default_electrum_port, Config},
    metrics::Metrics,
    rpc::daemon::tcpsslcommunication::extract_hostname_from_cert,
    rpc::parseutil::parse_version_str,
    utilnet::get_global_ip_from_hostname,
};

use crate::{
    chaindef::BlockHash,
    def::{PROTOCOL_HASH_FUNCTION, PROTOCOL_VERSION_MAX, PROTOCOL_VERSION_MIN},
    edition::Edition,
    signal::Waiter,
};

use super::{
    client::ElectrumClient,
    discoverer::PeerDiscoverer,
    peer::{CandidatePeer, ServerPeer},
};
use crate::def::ROSTRUM_VERSION;
use anyhow::{Context, Result};
use std::net::IpAddr;

/**
 * Max number of candidates to consider from a 'server.peers.subscribe' response.
 */
const MAX_CANDIDATES_PER_REQ: usize = 50;

#[derive(Debug, Clone, Serialize)]
pub struct ServerFeatures {
    genesis_hash: BlockHash,
    hash_function: &'static str,
    protocol_max: &'static str,
    protocol_min: &'static str,
    server_version: String,
    hosts: Option<HashMap<String, HashMap<String, Option<u16>>>>,
    pruning: Option<u64>,
    tokens: bool,
    spent_utxo: bool,
    editions: Vec<&'static str>,
}

pub struct PeerAnnouncer {
    features: ServerFeatures,
    /// Thread responsible for announcing us to other server peers
    announce_thread: Mutex<Option<tokio::task::JoinHandle<()>>>,
    announce_thread_kill: Arc<AtomicBool>,
}

fn discover_our_hostname(network: &Network) -> Option<String> {
    if network == &Network::Regtest {
        return Some("localhost".to_string());
    }
    // TODO: Use other electrum servers to find IP, rather than external service.
    info!("Auto-detecting our public IP");
    let agent: ureq::Agent = ureq::Agent::config_builder()
        .timeout_global(Some(Duration::from_secs(5)))
        .build()
        .into();
    let response = agent.get("https://api.ipify.org").call();
    let body = match response {
        Ok(mut response) => response.body_mut().read_to_string().unwrap_or_default(),
        Err(e) => {
            debug!("Failed to request public IP: {e}.");
            String::default()
        }
    };
    let ip: Option<IpAddr> = body.parse().ok();
    if let Some(ip) = ip {
        debug!("Detected public as {}", ip);
        return Some(ip.to_string());
    }
    None
}

async fn select_hostname(config: &Config) -> Option<String> {
    if let Some(ref hostname) = config.announce_hostname {
        if get_global_ip_from_hostname(hostname).await.is_some() {
            info!("discover: Announcing ourselves using hostname {} in --announce-hostname configuration.", hostname);
            return Some(hostname.clone());
        } else {
            warn!("The hostname {} in --announce-hostname configuration is not global routable. Ignoring.", hostname);
        }
    }
    if let Some(cert_file) = &config.ssl_cert_file {
        if let Ok(Some(hostname)) = extract_hostname_from_cert(cert_file).await {
            info!("discover: Announcing ourselves using hostname {} from --ssl-cert-file configuration.", hostname);
            return Some(hostname);
        }
    }
    // Fallback to announcing ourselves using our public IP.
    let ip = discover_our_hostname(&config.network_type);
    if let Some(ip) = ip {
        info!("discover: Announcing ourselves using our public IP {}", ip);
        return Some(ip.to_string());
    }
    warn!("discover: No valid hostname found. Peer announcement disabled.");
    None
}

impl PeerAnnouncer {
    pub async fn new(genesis_hash: BlockHash, config: &Config) -> Self {
        let mut enabled = config.announce;
        let hostname = select_hostname(config).await;

        if hostname.is_none() {
            enabled = false;
        }

        let hosts: Option<HashMap<String, HashMap<String, Option<u16>>>> = if enabled {
            let hostname = hostname.unwrap();

            let tcp_port = if config.tcp {
                Some(config.electrum_rpc_addr.port())
            } else {
                None
            };

            let ssl_port = if config.tcp_ssl {
                Some(config.electrum_rpc_ssl_addr.port())
            } else {
                config.announce_ssl_port
            };

            let ws_port = if config.websocket {
                Some(config.electrum_ws_addr.port())
            } else {
                None
            };

            let wss_port = if config.websocket_ssl {
                Some(config.electrum_ws_ssl_addr.port())
            } else {
                config.announce_wss_port
            };

            let http_port = if config.http {
                Some(config.electrum_http_addr.port())
            } else {
                None
            };

            let https_port = if config.https {
                Some(config.electrum_https_addr.port())
            } else {
                None
            };

            let ports: HashMap<String, Option<u16>> = [
                ("tcp_port".to_string(), tcp_port),
                ("ssl_port".to_string(), ssl_port),
                ("ws_port".to_string(), ws_port),
                ("wss_port".to_string(), wss_port),
                ("http_port".to_string(), http_port),
                ("https_port".to_string(), https_port),
            ]
            .iter()
            .cloned()
            .collect();
            Some([(hostname, ports)].iter().cloned().collect())
        } else {
            None
        };

        let features = ServerFeatures {
            genesis_hash,
            hash_function: PROTOCOL_HASH_FUNCTION,
            protocol_min: PROTOCOL_VERSION_MIN,
            protocol_max: PROTOCOL_VERSION_MAX,
            server_version: format!("Rostrum {ROSTRUM_VERSION}"),
            hosts,
            pruning: None,
            tokens: true,
            spent_utxo: true,
            editions: Edition::supported(),
        };

        let thread_kill = Arc::new(AtomicBool::new(false));

        Self {
            features: features.clone(),
            announce_thread: Mutex::new(None),
            announce_thread_kill: Arc::clone(&thread_kill),
        }
    }

    pub fn server_features(&self) -> &ServerFeatures {
        &self.features
    }

    pub fn server_version(&self) -> &str {
        &self.features.server_version
    }

    pub async fn start_announce_thread(
        &self,
        discoverer: Arc<PeerDiscoverer>,
        config: &Config,
        metrics: &Metrics,
    ) {
        let enabled = config.announce;

        if !enabled {
            info!("discover: Announcer disabled");
            return;
        }

        let announce_requests = metrics.counter_int(prometheus::Opts::new(
            "rostrum_discover_announce_request",
            "How often we announced ourselfs (attempts) to other server peers",
        ));
        let subscribe_requests = metrics.counter_int(prometheus::Opts::new(
            "rostrum_discover_peer_subscribe_request",
            "How often we requested (attempts) good peers from other server peers",
        ));

        let network = config.network_type;
        let thread_kill = Arc::clone(&self.announce_thread_kill);
        let features = self.features.clone();

        let good_ask_interval =
            Duration::from_secs(if network == Network::Regtest { 10 } else { 300 });

        let announce_thread = crate::thread::spawn_task("announce", async move {
            loop {
                if Waiter::wait_or_shutdown(Duration::from_secs(1))
                    .await
                    .is_err()
                {
                    return Ok(());
                }
                if thread_kill.load(std::sync::atomic::Ordering::Relaxed) {
                    return Ok(());
                }
                if let Some(peer) = discoverer.next_announce().await {
                    announce_requests.inc();
                    announce_to_peer(&peer, &features).await;
                    continue;
                }

                // Chill, so we don't flood our good peers with requests
                if Waiter::wait_or_shutdown(good_ask_interval).await.is_err() {
                    return Ok(());
                }

                // Out of peers to announce to. Check if a random good peer knows of new candidates.
                if let Some(peer) = discoverer.random_good().await {
                    subscribe_requests.inc();
                    let candidates = ask_for_candidates(&peer, &network).await;
                    for peer in candidates.into_iter().take(MAX_CANDIDATES_PER_REQ) {
                        discoverer.maybe_add_announce_queue(peer).await;
                    }
                }
            }
        });
        info!("discover: Announce thread started");
        self.announce_thread
            .lock()
            .unwrap()
            .replace(announce_thread);
    }
}

impl Drop for PeerAnnouncer {
    fn drop(&mut self) {
        self.announce_thread_kill
            .store(true, std::sync::atomic::Ordering::Relaxed);
    }
}

/**
 * Connects to a server and attempts to announce ourself.
 */
pub async fn announce_to_peer(peer: &dyn ServerPeer, features: &ServerFeatures) {
    trace!(
        "discover: Announcing ourselfs to {}",
        peer.pretty_hostname()
    );
    let client = match ElectrumClient::connect(peer, super::client::Protocol::ANY).await {
        Ok(s) => s,
        Err(e) => {
            trace!("discover: Failed to announce to peer: {e}");
            return;
        }
    };

    match client
        .call("server.add_peer", vec![Param::Value(json!(features))])
        .await
    {
        Ok(response) => {
            trace!("discover: Success add_peer response: {}", response)
        }
        Err(e) => {
            trace!(
                "discover: Failed to add to peer {}: {}",
                peer.pretty_hostname(),
                e
            );
        }
    };
}

/**
 * Connects to a server and asks it for good candidates.
 */
pub async fn ask_for_candidates(peer: &dyn ServerPeer, network: &Network) -> Vec<CandidatePeer> {
    let client = match ElectrumClient::connect(peer, super::client::Protocol::ANY).await {
        Ok(s) => s,
        Err(e) => {
            trace!(
                "discover: Failed to request peers from {}: {e}",
                peer.pretty_hostname()
            );
            return Vec::default();
        }
    };

    let response = match client.call("server.peers.subscribe", vec![]).await {
        Ok(r) => r,
        Err(e) => {
            trace!(
                "discover: Failed to request peers from {}: {}",
                peer.pretty_hostname(),
                e
            );
            return vec![];
        }
    };

    match parse_candidate_response(response, network) {
        Ok(c) => c,
        Err(e) => {
            trace!("Invalid candidate response: {e}");
            Vec::default()
        }
    }
}

fn parse_candidate_response(result: Value, network: &Network) -> Result<Vec<CandidatePeer>> {
    let peers = result
        .as_array()
        .context("expected array of peers, got something else")?;
    let mut candidates: Vec<CandidatePeer> = Default::default();
    for peer in peers.iter() {
        let ip = peer
            .get(0)
            .context("ip missing")?
            .as_str()
            .context("ip is not a string")?;
        let hostname = peer
            .get(1)
            .context("hostname missing")?
            .as_str()
            .context("hostname is not a string")?;
        let properties = peer
            .get(2)
            .context("peer properties missing")?
            .as_array()
            .context("expected peer properties to be an array")?;
        let ip: IpAddr = match ip.parse() {
            Ok(ip) => ip,
            Err(_) => {
                // Cannot parse IP. Might be onion. Ignore peer.
                continue;
            }
        };

        let mut candidate = CandidatePeer {
            tcp_port: None,
            ssl_port: None,
            ws_port: None,
            wss_port: None,
            hostname: Some(hostname.to_string()),
            ip,
            version: None,
        };
        for p in properties {
            let p = p.as_str().context("peer property was not a string")?;
            match p.chars().next() {
                Some('t') => {
                    candidate.tcp_port = match from_str(&p[1..]) {
                        Ok(port) => Some(port),
                        Err(_) => {
                            // It's OK not to provide a port number. In this case it's the default port.
                            default_electrum_port(network)
                        }
                    };
                }
                Some('s') => {
                    candidate.ssl_port = match from_str(&p[1..]) {
                        Ok(port) => Some(port),
                        Err(_) => {
                            // It's OK not to provide a port number. In this case it's the default port.
                            // Hack: Assume +1 is the SSL port for all networks
                            default_electrum_port(network).and_then(|p| p.checked_add(1))
                        }
                    };
                }
                Some('v') => {
                    candidate.version = match parse_version_str(&p[1..]) {
                        Some(v) => Some(v),
                        None => {
                            trace!("invalid version string for candidate {}", ip);
                            continue;
                        }
                    }
                }
                _ => { /* ignore property */ }
            }
        }
        candidates.push(candidate);
    }

    Ok(candidates)
}