use crate::chaindef::{BlockHeader, ScriptHash, Transaction};
use crate::encode::outpoint_hash;
use crate::query::scripthash_subscriptions::ScripthashSubscriptions;
use crate::signal::{NetworkNotifier, Waiter};
use crate::util::Channel;
use anyhow::{Context, Result};
use bitcoincash::hash_types::Txid;
use futures_util::{stream, StreamExt};
use log::{info, trace, warn};
use std::collections::{HashMap, HashSet};
use std::net::SocketAddr;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::mpsc::error::TrySendError;
use tokio::sync::mpsc::Sender;
use tokio::sync::Mutex;

use crate::chaindef::BlockHash;
use crate::config::Config;
use crate::def::PROTOCOL_VERSION_MAX;
use crate::doslimit::{ConnectionLimits, GlobalLimits};
use crate::metrics::Metrics;
use crate::query::Query;

use crate::metrics;

use crate::rpc::rpcstats::RpcStats;

use super::restconnection::{start_http_server, start_https_server};
use super::tcpcommunication::start_tcp_server;
use super::tcpsslcommunication::start_tcp_ssl_server;
use super::wscommunication::start_ws_server;
use super::wsscommunication::start_wss_server;
use super::{Message, Notification};

#[cfg(bch)]
fn get_output_scripthash(txn: &Transaction, n: Option<usize>) -> Vec<ScriptHash> {
    if let Some(out) = n {
        vec![ScriptHash::from_script(&txn.output[out].script_pubkey)]
    } else {
        txn.output
            .iter()
            .map(|o| ScriptHash::from_script(&o.script_pubkey))
            .collect()
    }
}

#[cfg(nexa)]
fn get_output_scripthash(txn: &Transaction, n: Option<usize>) -> Vec<ScriptHash> {
    if let Some(out) = n {
        vec![match txn.output[out].scriptpubkey_without_token() {
            Ok(Some(s)) => ScriptHash::from_script(&s),
            _ => ScriptHash::from_script(&txn.output[out].script_pubkey),
        }]
    } else {
        txn.output
            .iter()
            .map(|o| match o.scriptpubkey_without_token() {
                Ok(Some(s)) => ScriptHash::from_script(&s),
                _ => ScriptHash::from_script(&o.script_pubkey),
            })
            .collect()
    }
}

/// Unique identifier for a connection
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct ConnectionId(pub(crate) u64);

impl ConnectionId {
    pub(crate) fn new() -> Self {
        static COUNTER: AtomicU64 = AtomicU64::new(0);
        ConnectionId(COUNTER.fetch_add(1, Ordering::Relaxed))
    }
}

pub struct Server {
    notification: tokio::sync::mpsc::Sender<Notification>,
    // so we can join the servers when dropping this object
    tcp_server: Option<tokio::task::JoinHandle<()>>,
    tcp_ssl_server: Option<tokio::task::JoinHandle<()>>,
    wss_server: Option<tokio::task::JoinHandle<()>>,
    ws_server: Option<tokio::task::JoinHandle<()>>,
    rest_server: Option<tokio::task::JoinHandle<()>>,
    https_server: Option<tokio::task::JoinHandle<()>>,
    query: Arc<Query>,
    subscription_manager: Arc<ScripthashSubscriptions>,
}

impl Server {
    async fn start_notifier(
        notification: Channel<Notification>,
        senders: Arc<Mutex<HashMap<ConnectionId, Sender<Message>>>>,
        _subscription_manager: Arc<ScripthashSubscriptions>,
        acceptors: Vec<Sender<Option<(TcpStream, SocketAddr)>>>,
    ) {
        crate::thread::spawn_task("notification", async move {
            while let Some(msg) = notification.receiver().await.recv().await {
                let mut senders = senders.lock().await;
                match msg {
                    Notification::ScriptHashChange(connection_updates) => {
                        // Send pre-computed updates to each connection
                        for (conn_id, updates) in connection_updates {
                            if let Some(sender) = senders.get(&conn_id) {
                                match sender.try_send(Message::ScriptHashChange(updates)) {
                                    Ok(_) => {}
                                    Err(err) => match err {
                                        TrySendError::Closed(_) => {
                                            trace!("scripthash notify failed: peer disconnected");
                                            // Connection is closed, will be cleaned up
                                        }
                                        TrySendError::Full(_) => {
                                            trace!("scripthash notify failed: peer channel full");
                                        }
                                    },
                                }
                            }
                        }
                    }
                    Notification::ChainTipChange(hash) => {
                        // Chain tip changes go to all connections (for headers.subscribe)
                        senders.retain(|_, sender| {
                            match sender.try_send(Message::ChainTipChange(hash.clone())) {
                                Ok(_) => true,
                                Err(err) => match err {
                                    TrySendError::Closed(_) => {
                                        trace!("chaintip notify failed: peer disconnected");
                                        false
                                    }
                                    TrySendError::Full(_) => {
                                        trace!("chaintip notify failed: peer channel full");
                                        true // Keep the sender, just log the issue
                                    }
                                },
                            }
                        });
                    }
                    // mark acceptor as done
                    Notification::Exit => {
                        for a in &acceptors {
                            trace!("peer exit notification");
                            let _ = a.send(None).await;
                        }
                    }
                }
            }
            Ok(())
        });
    }

    async fn start_acceptor(
        addr: SocketAddr,
        acceptor_type: String,
    ) -> Channel<Option<(TcpStream, SocketAddr)>> {
        let chan = Channel::bounded(100_000);
        let acceptor = chan.sender();
        crate::thread::spawn_task("acceptor", async move {
            let listener = match TcpListener::bind(addr).await {
                Ok(l) => l,
                Err(e) => {
                    let errmsg = format!(
                        "Failed to start {} server - bind({}) failed: {}",
                        acceptor_type, addr, e
                    );
                    Waiter::shutdown(&errmsg);
                    bail!(errmsg)
                }
            };

            info!(
                "Electrum {} RPC server running on {} (protocol {})",
                acceptor_type, addr, PROTOCOL_VERSION_MAX
            );
            loop {
                let (stream, addr) = match listener.accept().await {
                    Ok(conn) => conn,
                    Err(e) => {
                        warn!("{} accept failed: {} - will retry", acceptor_type, e);
                        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
                        continue;
                    }
                };

                match acceptor.send(Some((stream, addr))).await {
                    Ok(_) => {}
                    Err(e) => trace!("Failed to send to client {:?}", e),
                }

                if Waiter::shutdown_check().is_err() {
                    break;
                }
            }
            Ok(())
        });
        chan
    }

    pub async fn start(
        config: Arc<Config>,
        query: Arc<Query>,
        metrics: Arc<Metrics>,
        connection_limits: ConnectionLimits,
        global_limits: Arc<GlobalLimits>,
        network_notifier: Arc<NetworkNotifier>,
    ) -> Self {
        let stats = Arc::new(RpcStats {
            latency: metrics.histogram_vec(
                "rostrum_rpc_latency",
                "RPC latency (seconds)",
                &["method"],
                metrics::default_duration_buckets(),
            ),
            subscriptions: metrics.gauge_int(prometheus::Opts::new(
                "rostrum_scripthash_subscriptions",
                "# of scripthash subscriptions for node",
            )),
        });
        stats.subscriptions.set(0);

        let blockchain_event_queue = Channel::bounded(1_000_000);
        let blockchain_event_sender = blockchain_event_queue.sender();

        let tcp_acceptor = if !config.tcp {
            info!("TCP endpoint is not enabled in the server configuration");
            None
        } else {
            Some(Self::start_acceptor(config.electrum_rpc_addr, "tcp".to_string()).await)
        };
        let ws_acceptor = if !config.websocket {
            info!("WebSocket endpoint is not enabled in server the configuration.");
            None
        } else {
            Some(Self::start_acceptor(config.electrum_ws_addr, "websocket".to_string()).await)
        };
        let wss_acceptor = if !config.websocket_ssl {
            info!("WebSocket over SSL endpoint is not enabled in server the configuration.");
            None
        } else {
            Some(
                Self::start_acceptor(config.electrum_ws_ssl_addr, "websocket_ssl".to_string())
                    .await,
            )
        };
        let http_acceptor = if !config.http {
            info!("HTTP endpoint is not enabled in server the configuration.");
            None
        } else {
            Some(Self::start_acceptor(config.electrum_http_addr, "http".to_string()).await)
        };
        let https_acceptor = if !config.https {
            info!("HTTPS endpoint is not enabled in server the configuration.");
            None
        } else {
            Some(Self::start_acceptor(config.electrum_https_addr, "https".to_string()).await)
        };

        let ssl_acceptor = if !config.tcp_ssl {
            info!("TCP SSL endpoint is not enabled in the server configuration");
            None
        } else {
            Some(Self::start_acceptor(config.electrum_rpc_ssl_addr, "tcp_ssl".to_string()).await)
        };

        let rpc_queue_senders =
            Arc::new(Mutex::new(HashMap::<ConnectionId, Sender<Message>>::new()));
        let subscription_manager = Arc::new(ScripthashSubscriptions::new(
            query.clone(),
            config.statushash_max_elements,
        ));

        {
            let mut acceptors: Vec<Sender<Option<(TcpStream, SocketAddr)>>> = Vec::default();
            if let Some(a) = tcp_acceptor.as_ref() {
                acceptors.push(a.sender());
            }
            if let Some(a) = ws_acceptor.as_ref() {
                acceptors.push(a.sender());
            }
            if let Some(a) = ssl_acceptor.as_ref() {
                acceptors.push(a.sender());
            }
            Self::start_notifier(
                blockchain_event_queue,
                rpc_queue_senders.clone(),
                subscription_manager.clone(),
                acceptors,
            )
            .await;
        }

        // Create REST server
        let rest_server = http_acceptor.map(|acceptor| {
            crate::thread::spawn_task(
                "http_rpc",
                start_http_server(
                    acceptor,
                    global_limits.clone(),
                    config.clone(),
                    query.clone(),
                    stats.clone(),
                    connection_limits,
                    network_notifier.clone(),
                    config.rpc_idle_timeout,
                    subscription_manager.clone(),
                ),
            )
        });

        // Create HTTPS server
        let https_server = https_acceptor.map(|acceptor| {
            let cert_file = config
                .ssl_cert_file
                .as_ref()
                .context("SSL certificate file not specified")
                .unwrap();
            let key_file = config
                .ssl_key_file
                .as_ref()
                .context("SSL key file not specified")
                .unwrap();

            crate::thread::spawn_task(
                "https_rpc",
                start_https_server(
                    acceptor,
                    cert_file.clone(),
                    key_file.clone(),
                    global_limits.clone(),
                    config.clone(),
                    query.clone(),
                    stats.clone(),
                    connection_limits,
                    network_notifier.clone(),
                    config.rpc_idle_timeout,
                    subscription_manager.clone(),
                ),
            )
        });

        // Create WebSocket server
        let ws_server = ws_acceptor.map(|acceptor| {
            crate::thread::spawn_task(
                "ws_rpc",
                start_ws_server(
                    acceptor,
                    global_limits.clone(),
                    config.clone(),
                    query.clone(),
                    stats.clone(),
                    connection_limits,
                    network_notifier.clone(),
                    rpc_queue_senders.clone(),
                    subscription_manager.clone(),
                ),
            )
        });

        // Create TCP server
        let tcp_server = tcp_acceptor.map(|acceptor| {
            crate::thread::spawn_task(
                "tcp_rpc",
                start_tcp_server(
                    acceptor,
                    global_limits.clone(),
                    config.clone(),
                    query.clone(),
                    stats.clone(),
                    connection_limits,
                    network_notifier.clone(),
                    rpc_queue_senders.clone(),
                    subscription_manager.clone(),
                ),
            )
        });

        // Create SSL TCP server
        let tcp_ssl_server = ssl_acceptor.map(|acceptor| {
            let cert_file = config
                .ssl_cert_file
                .as_ref()
                .context("SSL certificate file not specified")
                .unwrap();
            let key_file = config
                .ssl_key_file
                .as_ref()
                .context("SSL key file not specified")
                .unwrap();

            crate::thread::spawn_task(
                "tcp_ssl_rpc",
                start_tcp_ssl_server(
                    acceptor,
                    cert_file.clone(),
                    key_file.clone(),
                    global_limits.clone(),
                    config.clone(),
                    query.clone(),
                    stats.clone(),
                    connection_limits,
                    network_notifier.clone(),
                    rpc_queue_senders.clone(),
                    subscription_manager.clone(),
                ),
            )
        });

        // Create WSS server
        let wss_server = wss_acceptor.map(|acceptor| {
            let cert_file = config
                .ssl_cert_file
                .as_ref()
                .context("SSL certificate file not specified")
                .unwrap();
            let key_file = config
                .ssl_key_file
                .as_ref()
                .context("SSL key file not specified")
                .unwrap();

            crate::thread::spawn_task(
                "wss_rpc",
                start_wss_server(
                    acceptor,
                    cert_file.clone(),
                    key_file.clone(),
                    global_limits.clone(),
                    config.clone(),
                    query.clone(),
                    stats.clone(),
                    connection_limits,
                    network_notifier.clone(),
                    rpc_queue_senders.clone(),
                    subscription_manager.clone(),
                ),
            )
        });

        Self {
            notification: blockchain_event_sender,
            query,
            tcp_server,
            tcp_ssl_server,
            wss_server,
            ws_server,
            rest_server,
            https_server,
            subscription_manager,
        }
    }

    /**
     * All scripthash subscriptions that should be notified. This includes all
     * scripthashes that received new coins (outputs of this transactions), and
     * all the scripthashes that had coins spent from them (inputs of this transaction).
     */
    async fn get_scripthashes_affected_by_tx(
        &self,
        txid: &Txid,
        blockhash: Option<&BlockHash>,
    ) -> Result<Vec<ScriptHash>> {
        let txn = match self.query.tx().get(txid, blockhash, None, false).await {
            Ok(txn) => txn,
            Err(e) => {
                if e.to_string().contains("genesis") {
                    // ignore the edge case error "The genesis block coinbase is not considered an ordinary transaction and cannot be retrieved"
                    return Ok(Vec::default());
                }
                return Err(e);
            }
        };
        let mut scripthashes = get_output_scripthash(&txn, None);

        for txin in txn.input {
            if txin.previous_output.is_null() {
                continue;
            }
            let tx = self
                .query
                .get_tx_funding_prevout(&outpoint_hash(&txin.previous_output))
                .await;
            if let Ok(Some((tx, index, _))) = tx {
                scripthashes.extend(get_output_scripthash(&tx, Some(index as usize)));
            }
        }
        Ok(scripthashes)
    }

    /// Collect all scripthashes affected by the given headers and transactions
    async fn collect_affected_scripthashes(
        &self,
        headers_changed: &[BlockHeader],
        txs_changed: HashSet<Txid>,
    ) -> HashSet<ScriptHash> {
        let mut txn_done: HashSet<Txid> = HashSet::new();
        let mut scripthashes: HashSet<ScriptHash> = HashSet::new();

        async fn insert_for_txs(
            parent: &Server,
            txids: &mut dyn Iterator<Item = Txid>,
            blockhash: Option<BlockHash>,
            txn_done: &HashSet<Txid>,
        ) -> (HashSet<Txid>, Vec<ScriptHash>) {
            let new: HashSet<_> = txids.collect();
            let new: HashSet<_> = new.difference(txn_done).cloned().collect();

            let hashes = stream::iter(&new)
                .map(|tx| async move {
                    let effected = parent
                        .get_scripthashes_affected_by_tx(tx, blockhash.as_ref())
                        .await;
                    let hashes = match effected {
                        Ok(hashes) => hashes,
                        Err(e) => {
                            trace!("failed to get effected scripthashes for tx {}: {:?}", tx, e);
                            vec![]
                        }
                    };
                    stream::iter(hashes.into_iter())
                })
                .buffer_unordered(10)
                .flatten()
                .collect::<Vec<ScriptHash>>()
                .await;

            (new, hashes)
        }

        for header in headers_changed.iter() {
            let blockhash = header.block_hash();
            let mut txids = match self.query.getblocktxids(&blockhash).await {
                Ok(txids) => txids.into_iter(),
                Err(e) => {
                    warn!("Failed to get blocktxids for {}: {}", blockhash, e);
                    continue;
                }
            };
            let (new_txs, new_scripthashes) =
                insert_for_txs(self, &mut txids, Some(blockhash), &txn_done).await;
            txn_done.extend(new_txs.iter());
            scripthashes.extend(new_scripthashes.iter());
        }

        let mut txs_changed_iter = txs_changed.into_iter();
        let (new_txs, new_scripthashes) =
            insert_for_txs(self, &mut txs_changed_iter, None, &txn_done).await;
        txn_done.extend(new_txs.iter());
        scripthashes.extend(new_scripthashes.iter());

        scripthashes
    }

    pub async fn notify_scripthash_subscriptions(
        &self,
        headers_changed: &[BlockHeader],
        txs_changed: HashSet<Txid>,
    ) {
        // 1. Collect affected scripthashes
        let scripthashes = self
            .collect_affected_scripthashes(headers_changed, txs_changed)
            .await;

        if scripthashes.is_empty() {
            return;
        }

        // 2. Delegate to ScripthashSubscriptions to compute statushash updates
        let connection_updates = match self
            .subscription_manager
            .notify_scripthashes_changed(scripthashes)
            .await
        {
            Ok(updates) => updates,
            Err(e) => {
                warn!("Failed to compute subscription updates: {}", e);
                return;
            }
        };

        if connection_updates.is_empty() {
            return;
        }

        // 3. Send notifications
        if let Err(e) = self
            .notification
            .send(Notification::ScriptHashChange(connection_updates))
            .await
        {
            trace!("Scripthash change notification failed: {}", e);
        }
    }

    pub async fn notify_subscriptions_chaintip(&self, header: BlockHeader) {
        if let Err(e) = self
            .notification
            .send(Notification::ChainTipChange(Box::new(header)))
            .await
        {
            trace!("Failed to notify about chaintip change {}", e);
        }
    }

    pub async fn disconnect_clients(&self) {
        trace!("disconncting clients");
        let _ = self.notification.send(Notification::Exit).await;
    }
}

impl Drop for Server {
    fn drop(&mut self) {
        trace!("stop accepting new RPCs");
        let n = self.notification.clone();
        crate::thread::spawn_task("drop_rpc", async move {
            let _ = n.send(Notification::Exit).await;
            Ok(())
        });
        if let Some(handle) = self.tcp_server.take() {
            info!("sending abort to tcp");
            handle.abort();
            info!("abort sent");
        }
        if let Some(handle) = self.ws_server.take() {
            handle.abort();
        }
        if let Some(handle) = self.rest_server.take() {
            handle.abort();
        }
        if let Some(handle) = self.https_server.take() {
            handle.abort();
        }
        if let Some(handle) = self.tcp_ssl_server.take() {
            handle.abort();
        }
        if let Some(handle) = self.wss_server.take() {
            handle.abort();
        }
        trace!("RPC server is stopped");
    }
}
