use std::future::Future;
use std::io;
use std::sync::Arc;

use futures::channel::{mpsc, oneshot};
use futures::prelude::*;
use futures::select;

use libp2p::kad::record::Key;
use libp2p::kad::BootstrapOk;
use libp2p::multiaddr::Protocol;
use libp2p::{Multiaddr, PeerId};
use log::*;
use tokio::task::JoinHandle;

use crate::graph::cryptograph::*;

macro_rules! node_control_field {
    ($in:ty) => {
        mpsc::Sender<$in>
    };
    ($in:ty => $out:ty) => {
        mpsc::Sender<($in, oneshot::Sender<$out>)>
    };
}

macro_rules! facade_control_fn {
    ($fn:ident, $field:ident, $in:ty) => {
        server_control_fn!($fn, $field, $in => ())
    };
    ($fn:ident, $field:ident, $in:ty => $out:ty) => {
        pub fn $fn(&mut self, input: $in) -> impl Future<Output=Result<$out, anyhow::Error>>
        {
            let mut field_tx = self.$field.clone();

            let (tx, rx) = oneshot::channel();
            async move {
                field_tx.send((input, tx)).await
                    .map_err(|e: mpsc::SendError| {
                        error!("Server has been shut down (field {:?})",
                                stringify!($field));
                        e
                    })?;
                rx.map_err(|e| {
                    error!("Transmission channel has been cancelled (field {:?})",
                            stringify!($field));
                    e.into()
                }).await
            }
        }
    };
}

#[derive(Clone)]
#[must_use = "ServerControl must be dropped or consumed."]
pub struct NodeFacade {
    pub bootstrap_tx: node_control_field!(() => ()),
    pub add_address_tx: node_control_field!((PeerId, Vec<Multiaddr>) => ()),
    // pub request_k_closest_tx: server_control_field!(router::ID => KClosestNodes),
    pub store_vertex_tx: node_control_field!(Vertex => ()),
    pub store_edge_tx: node_control_field!(Edge => ()),
    pub search_vertex_tx: node_control_field!(Key => Option<(Vertex, Vec<Edge>)>),
    pub connection_count_tx: node_control_field!(() => usize),
    pub local_addr_tx: node_control_field!(() => (PeerId, Vec<Multiaddr>)),

    pub(super) handle: Arc<JoinHandle<()>>,

    pub(super) local_peer_id: PeerId,
}

impl Drop for NodeFacade {
    fn drop(&mut self) {
        if let Some(handle) = Arc::get_mut(&mut self.handle) {
            log::debug!("Last ServerControl, aborting task.");
            handle.abort();
        }
    }
}

impl NodeFacade {
    facade_control_fn!(bootstrap, bootstrap_tx, () => ());
    facade_control_fn!(add_address, add_address_tx, (PeerId, Vec<Multiaddr>) => ());
    facade_control_fn!(store_vertex, store_vertex_tx, Vertex => ());
    facade_control_fn!(store_edge, store_edge_tx, Edge => ());
    facade_control_fn!(search_vertex, search_vertex_tx, Key => Option<(Vertex, Vec<Edge>)>);
    facade_control_fn!(connection_count, connection_count_tx, () => usize);
    facade_control_fn!(local_addr, local_addr_tx, () => (PeerId, Vec<Multiaddr>));

    pub async fn run_forever(mut self) -> Result<(), io::Error> {
        Arc::get_mut(&mut self.handle)
            .expect("run forever with only one handle")
            .await
            .unwrap();
        Ok(())
    }
}

impl NodeFacade {
    pub async fn spawn(mut swarm: super::GlycosSwarm) -> anyhow::Result<NodeFacade> {
        use super::GlycosEvent;
        // Now move the swarm to another task, and create the ServerControl "facade"

        let (bootstrap_tx, mut bootstrap_rx) = mpsc::channel(2);
        let (add_address_tx, mut add_address_rx) = mpsc::channel(2);
        // let (request_k_closest_tx, request_k_closest_rx) = mpsc::channel(2);
        let (store_vertex_tx, mut store_vertex_rx) = mpsc::channel(2);
        let (store_edge_tx, mut store_edge_rx) = mpsc::channel(2);
        let (search_vertex_tx, mut search_vertex_rx) = mpsc::channel(2);
        let (connection_count_tx, mut connection_count_rx) = mpsc::channel(2);
        let (local_addr_tx, mut local_addr_rx) = mpsc::channel(2);

        macro_rules! unwrap_or_break {
            ($e:expr) => {
                match $e {
                    Some(x) => x,
                    None => break,
                }
            };
        }

        let local_peer_id = *swarm.local_peer_id();

        let (await_startup_tx, startup) = oneshot::channel();

        let mut puts = std::collections::HashMap::new();
        let mut gets = std::collections::HashMap::new();

        let ctrl = NodeFacade {
            bootstrap_tx,
            add_address_tx,
            store_vertex_tx,
            store_edge_tx,
            search_vertex_tx,
            connection_count_tx,
            local_addr_tx,

            handle: std::sync::Arc::new(tokio::spawn(async move {
                let mut startup = Some(await_startup_tx);
                loop {
                    let res: anyhow::Result<()> = select! {
                        request = bootstrap_rx.next() => {
                            let ((), respond) = unwrap_or_break!(request);

                            let boot = swarm.behaviour_mut().bootstrap().unwrap();
                            puts.insert(boot, respond);
                            Ok(())
                        },
                        request = add_address_rx.next() => {
                            let ((peer, addrs), respond) = unwrap_or_break!(request);

                            swarm.behaviour_mut().add_address(peer, addrs);
                            respond.send(()).unwrap();

                            Ok(())
                        },
                        // request_k_closest = request_k_closest_rx.next() => match request_k_closest? {
                        //     Some((target, on_ready)) => node_state.handle_request_k_closest(target, on_ready),
                        //     None => {
                        //         log::warn!("request_k_closest_rx stream returned None");
                        //         break
                        //     },
                        // },
                        store_vertex_rx = store_vertex_rx.next() => {
                            let (vertex, on_ready) = unwrap_or_break!(store_vertex_rx);
                            let id = swarm.behaviour_mut().store_vertex(vertex).unwrap();
                            puts.insert(id, on_ready);
                            Ok(())
                        },
                        store_edge_rx = store_edge_rx.next() => {
                            let (edge, on_ready) = unwrap_or_break!(store_edge_rx);
                            let id = swarm.behaviour_mut().store_edge(edge).unwrap();
                            puts.insert(id, on_ready);
                            Ok(())
                        },
                        search_vertex_rx = search_vertex_rx.next() => {
                            let (id, on_ready) = unwrap_or_break!(search_vertex_rx);
                            let id = swarm.behaviour_mut().search_vertex(&id).unwrap();
                            gets.insert(id, on_ready);
                            Ok(())
                        },
                        connection_count_rx = connection_count_rx.next() => {
                            let (_, on_ready) = unwrap_or_break!(connection_count_rx);
                            let count = swarm.behaviour_mut().kademlia_mut().kbuckets().map(|bucket| bucket.num_entries()).sum();
                            on_ready.send(count).unwrap();
                            Ok(())
                        },
                        local_addr = local_addr_rx.next() => {
                            let (_, on_ready) = unwrap_or_break!(local_addr);
                            let mut addrs = Vec::new();
                            for multiaddr in swarm.external_addresses().map(|x| &x.addr).chain(swarm.listeners()) {
                                let multiaddr = multiaddr.replace(0, |ip| match ip {
                                    Protocol::Ip4(x) if x.is_unspecified() => Some(Protocol::Ip4(std::net::Ipv4Addr::LOCALHOST)),
                                    Protocol::Ip6(x) if x.is_unspecified() => Some(Protocol::Ip6(std::net::Ipv6Addr::LOCALHOST)),
                                    Protocol::Ip4(x) if !x.is_loopback() => Some(Protocol::Ip4(*x)),
                                    Protocol::Ip6(x) if !x.is_loopback() => Some(Protocol::Ip6(*x)),
                                    _ => None,
                                });

                                if let Some(addr) = multiaddr {
                                    addrs.push(addr);
                                }
                            }
                            if addrs.is_empty() {
                                log::warn!("Not listening on any IPv4/IPv6 TCP sockets.");
                            }
                            on_ready.send((*swarm.local_peer_id(), addrs)).unwrap();
                            Ok(())
                        },
                        event = swarm.select_next_some() => {
                            use libp2p::swarm::SwarmEvent;
                            #[allow(clippy::single_match)]
                            match event {
                                SwarmEvent::NewListenAddr {
                                    ..
                                } => {
                                    if let Some(startup) = startup.take() {
                                        startup.send(()).unwrap();
                                    }
                                }
                                SwarmEvent::Behaviour(GlycosEvent::BootstrapResult {
                                    id,
                                    result: Ok(BootstrapOk {
                                        num_remaining: 0,
                                        ..
                                    }),
                                    stats: _,
                                }) => {
                                    puts.remove(&id).unwrap().send(()).expect("handler went away");
                                }
                                SwarmEvent::Behaviour(GlycosEvent::OutboundQueryCompleted {
                                    id,
                                    result: super::GlycosQueryResult::EdgeStored | super::GlycosQueryResult::VertexStored,
                                    stats: _,
                                }) => {
                                    puts.remove(&id).unwrap().send(()).expect("handler went away");
                                }
                                SwarmEvent::Behaviour(GlycosEvent::OutboundQueryCompleted {
                                    id,
                                    result: super::GlycosQueryResult::VertexRetrievalComplete {
                                        result,
                                    },
                                    stats: _,
                                }) => {
                                    match result {
                                        // puts.remove(&id).unwrap().send(()).expect("handler went away");
                                        Ok(super::VertexRetrievalOk {
                                            vertex,
                                            edges,
                                        }) => {
                                            gets.remove(&id).unwrap().send(Some((vertex, edges))).unwrap();
                                        }
                                        Err(e) => {
                                            log::warn!("Error retrieving vertex: {:?}", e);
                                            gets.remove(&id).unwrap().send(None).unwrap();
                                        }
                                    }
                                }
                                _ => {}
                            }
                            Ok(())
                        },
                    };
                    match res {
                        Ok(()) => continue,
                        Err(e) => log::error!("Error in handling stream: {}", e),
                    }
                }
                log::trace!("A stream ended; stopping node.");
            })),
            local_peer_id,
        };

        startup.await.unwrap();
        log::debug!("At least one listener is active, considering started.");

        Ok(ctrl)
    }
}