use crate::inbound::{RPCReceiver, RequestHandler, RequestReceiver};

use crate::listener::Listener;

use crate::outbound::{ConnectionError, RPCSender, RequestSender};

use crate::peermanager::PeerManager;

use lyquor_primitives::NodeID;

use lyquor_tls::TlsConfig;

use std::sync::Arc;

use thiserror::Error;

use tokio::sync::mpsc;

use tokio_util::sync::CancellationToken;

#[derive(Debug, Error)]

pub enum HubError {

    #[error("failed to bind listener on {addr}: {source}")]

    BindAddr {

        addr: String,

        #[source]

        source: Box<dyn std::error::Error + Send + Sync>,

    },

    #[error("peer not exist")]

    PeerNotExist,

    #[error("mismatched node id from peer: got {got:?}, expected {expected:?}")]

    PeerIdMismatch { got: NodeID, expected: NodeID },

    #[error("invalid peer TLS certificate")]

    PeerTlsCertificateError,

    #[error("address {addr} is already mapped to {existing:?}, cannot map to {requested:?}")]

    AddressConflict {

        addr: String,

        existing: NodeID,

        requested: NodeID,

    },

    #[error("self connection is not allowed")]

    SelfConnection,

    #[error("connection error: {0}")]

    ConnectionError(#[source] ConnectionError),

    #[error("TLS error: {0}")]

    TlsError(#[from] lyquor_tls::TlsError),

    #[error("Inbound is not ready")]

    InboundError,

    #[error("Outbound is not ready")]

    OutboundError,

    #[error("unknown error")]

    Unknown,

}

type Result<T> = std::result::Result<T, HubError>;

pub struct Hub {

    id: NodeID,

    peer_manager: Arc<PeerManager>,

    control: Control,

    listener: crate::listener::Listener,

    shutdown: CancellationToken,

    tls_config: TlsConfig,

}

impl std::fmt::Debug for Hub {

    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {

        f.debug_struct("Hub").field("id", &self.id).finish_non_exhaustive()

    }

}

impl Hub {

    #[tracing::instrument(level = "trace", skip(tls_config))]

    pub async fn new(tls_config: TlsConfig, listen_addr: String, shutdown: CancellationToken) -> anyhow::Result<Self> {

        let id = tls_config.node_id();

        let server_config = tls_config.server_config().map_err(HubError::TlsError)?;

        let client_config = tls_config.client_config().map_err(HubError::TlsError)?;

        let control = Control::new();

        let peer_manager = Arc::new(PeerManager::new(

            id.clone(),

            client_config,

            shutdown.child_token(),

            control.rpc_sender(),

        ));

        let listener = Listener::new(server_config, listen_addr, shutdown.child_token(), peer_manager.clone()).await?;

        Ok(Hub {

            id,

            peer_manager,

            control,

            listener,

            shutdown,

            tls_config,

        })

    }

    pub fn get_id(&self) -> NodeID {

        self.id

    }

    pub fn signing_key(&self) -> Arc<lyquor_tls::NodeSigningKey> {

        self.tls_config.signing_key()

    }

    pub async fn wait_for_shutdown(&self) {

        self.listener.wait().await;

        let _ = self.peer_manager.remove_all_peers().await;

    }

    #[tracing::instrument(level = "trace", ret, err)]

    pub async fn add_peer(&self, id: &NodeID, addr: &str) -> Result<()> {

        tracing::info!("add peer for {:?}: {:?} -> {:?}", self.id, id, addr);

        self.peer_manager.add_peer(id, addr).await

    }

    #[tracing::instrument(level = "trace", err(level = "debug"))]

    pub async fn wait_inbound_ready(&self, id: &NodeID, timeout: std::time::Duration) -> Result<RequestReceiver> {

        let deadline = tokio::time::Instant::now() + timeout;

        while tokio::time::Instant::now() < deadline {

            if let Ok(receiver) = self.peer_manager.inbound(id) {

                return Ok(receiver);

            }

            if self.shutdown.is_cancelled() {

                return Err(HubError::Unknown);

            }

            tokio::time::sleep(std::time::Duration::from_millis(100)).await;

        }

        Err(HubError::InboundError)

    }

    #[tracing::instrument(level = "trace", err(level = "debug"))]

    pub async fn wait_outbound_ready(&self, id: &NodeID, timeout: std::time::Duration) -> Result<RequestSender> {

        let deadline = tokio::time::Instant::now() + timeout;

        while tokio::time::Instant::now() < deadline {

            if let Ok(sender) = self.peer_manager.outbound(id) {

                return Ok(sender);

            }

            if self.shutdown.is_cancelled() {

                return Err(HubError::Unknown);

            }

            tokio::time::sleep(std::time::Duration::from_millis(100)).await;

        }

        Err(HubError::OutboundError)

    }

    #[tracing::instrument(level = "trace", ret, err(level = "debug"))]

    pub async fn remove_peer(&self, id: &NodeID) -> Result<()> {

        self.peer_manager.remove_peer(id).await

    }

    #[tracing::instrument(level = "trace", err(level = "debug"))]

    pub fn outbound(&self, id: &NodeID) -> Result<RequestSender> {

        self.peer_manager.outbound(id)

    }

    #[tracing::instrument(level = "trace", err(level = "debug"))]

    pub fn inbound(&self, id: &NodeID) -> Result<RequestReceiver> {

        self.peer_manager.inbound(id)

    }

    #[tracing::instrument(level = "trace", err(level = "debug"))]

    pub fn rpc_outbound(&self, id: &NodeID) -> Result<RPCSender> {

        self.peer_manager.rpc_outbound(id)

    }

    #[tracing::instrument(level = "trace", err(level = "debug"))]

    pub fn rpc_inbound(&self) -> Result<RPCReceiver> {

        Ok(self.control.rpc_inbound())

    }

    pub async fn get_peers(&self) -> Vec<NodeID> {

        self.peer_manager.get_peers().await

    }

}

#[derive(Clone)]

pub struct Control {

    rpc_sender: mpsc::Sender<RequestHandler>,

    rpc_receiver: RPCReceiver,

}

impl Control {

    pub fn new() -> Self {

        let (tx, rx) = mpsc::channel(100);

        Control {

            rpc_sender: tx,

            rpc_receiver: RPCReceiver::new(rx),

        }

    }

    pub fn rpc_inbound(&self) -> RPCReceiver {

        self.rpc_receiver.clone()

    }

    pub fn rpc_sender(&self) -> mpsc::Sender<RequestHandler> {

        self.rpc_sender.clone()

    }

}