#![allow(dead_code)]

use crate::p2p::notifications::{
    behaviour::ProtocolsData,
    messages::BlockAnnouncesHandshake,
    upgrades::{
        combine_upgrades::CombineUpgrades,
        handshake::{
            HandshakeInbound, HandshakeInboundSubstream, HandshakeOutbound,
            HandshakeOutboundSubstream,
        },
    },
};
use bytes::BytesMut;
use codec::Encode;
use futures::{channel::mpsc, prelude::*, SinkExt};
use libp2p::{
    core::ConnectedPoint,
    swarm::{
        handler::{ConnectionEvent, FullyNegotiatedInbound},
        ConnectionHandler, ConnectionHandlerEvent, KeepAlive, 
        Stream as NegotiatedSubstream, SubstreamProtocol,
    },
    PeerId,
};
use std::{
    collections::VecDeque,
    mem,
    pin::Pin,
    task::{Context, Poll},
};

const LOG_TARGET: &str = "ghost-walker-handler";

pub struct ProtocolDetails {
    pub name: String,
    pub handshake: Vec<u8>,
    pub upgrade: HandshakeInbound,
    pub state: State,
}

pub struct NotificationsHandler {
    protocols: Vec<ProtocolDetails>,

    pending_events: VecDeque<
        ConnectionHandlerEvent<
                HandshakeOutbound,
                usize,
                NotificationsHandlerToBehavior,
                NotificationsHandlerError,
            >,
        >,

        endpoint: ConnectedPoint,
        peer: PeerId,
}

#[derive(Debug, Clone)]
pub enum NotificationsHandlerFromBehavior {
    Open { index: usize },
    Close { index: usize },
}

#[derive(Debug, Clone)]
pub enum NotificationsHandlerToBehavior {
    HandshakeCompleted {
        index: usize,
        endpoint: ConnectedPoint,
        handshake: Vec<u8>,
        is_inbound: bool,
        sender: mpsc::Sender<Vec<u8>>,
    },
    HandshakeError {
        index: usize,
    },
    OpenDesiredByRemote {
        index: usize,
    },
    CloseDesired {
        index: usize,
    },
    Close {
        index: usize,
    },
    Notification {
        index: usize,
        bytes: BytesMut,
    },
}

pub enum State {
    Closed {
        pending_opening: bool,
    },
    OpenDesiredByRemote {
        inbound_substream: HandshakeInboundSubstream<NegotiatedSubstream>,
        pending_opening: bool,
    },
    Opening {
        inbound_substream: Option<HandshakeInboundSubstream<NegotiatedSubstream>>,
        inbound: bool,
    },
    Open {
        recv: stream::Peekable<mpsc::Receiver<Vec<u8>>>,
        inbound_substream: Option<HandshakeInboundSubstream<NegotiatedSubstream>>,
        outbound_substream: Option<HandshakeOutboundSubstream<NegotiatedSubstream>>,
    },
}

impl NotificationsHandler {
    pub fn new(peer: PeerId, endpoint: ConnectedPoint, data: ProtocolsData) -> Self {
        let genesis_string = hex::encode(data.genesis_hash);
        let blocks = format!("/{}/block-announces/1", genesis_string);

        let tx = format!("/{}/transactions/1", genesis_string);
        let block_announces = BlockAnnouncesHandshake::from_genesis(data.genesis_hash);

        let protocols = vec![
            ProtocolDetails {
                name: blocks.clone(),
                handshake: block_announces.encode(),
                upgrade: HandshakeInbound {
                    name: blocks.clone(),
                },
                state: State::Closed {
                    pending_opening: false,
                },
            },
            ProtocolDetails {
                name: tx.clone(),
                handshake: vec![data.node_role.encoded()],
                upgrade: HandshakeInbound { name: tx.clone() },
                state: State::Closed {
                    pending_opening: false,
                },
            },
        ];

        NotificationsHandler {
            peer,
            pending_events: VecDeque::with_capacity(16),
            endpoint,
            protocols,
        }
    }
}

#[derive(Debug, thiserror::Error)]
pub enum NotificationsHandlerError {}

impl ConnectionHandler for NotificationsHandler {
    // Received and submitted events.
    type FromBehaviour = NotificationsHandlerFromBehavior;
    type ToBehaviour = NotificationsHandlerToBehavior;

    type Error = NotificationsHandlerError;

    // Handle handshakes.
    type InboundProtocol = CombineUpgrades<HandshakeInbound>;
    type OutboundProtocol = HandshakeOutbound;

    // Extra information upon connections.
    type OutboundOpenInfo = usize;
    type InboundOpenInfo = ();

    fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, ()> {
        let protocol_upgrades: Vec<_> = self.protocols.iter().map(|p| p.upgrade.clone()).collect();
        let combine_upgrades = CombineUpgrades::from(protocol_upgrades);
        SubstreamProtocol::new(combine_upgrades, ())
    }

    fn on_connection_event(
        &mut self,
        event: ConnectionEvent<
        '_,
        Self::InboundProtocol,
        Self::OutboundProtocol,
        Self::InboundOpenInfo,
        Self::OutboundOpenInfo,
        >,
    ) {
        match event {
            ConnectionEvent::FullyNegotiatedInbound(FullyNegotiatedInbound {
                protocol, ..
            }) => {
                let (mut stream, index) = (protocol.data, protocol.index);

                tracing::debug!(target: LOG_TARGET,
                    "Handler negotiated inbound peer={:?} index={:?}",
                    self.peer,
                    index
                );

                let proto = &mut self.protocols[index];
                match proto.state {
                    State::Closed { pending_opening } => {
                        tracing::trace!(
                            target: LOG_TARGET,
                            "Handler negotiated inbound Closed -> OpenDesiredByRemote peer={:?} index={:?}",
                            self.peer,
                            index
                        );

                        self.pending_events
                            .push_back(ConnectionHandlerEvent::NotifyBehaviour(
                                    NotificationsHandlerToBehavior::OpenDesiredByRemote { index },
                            ));

                        proto.state = State::OpenDesiredByRemote {
                            inbound_substream: stream.substream,
                            pending_opening,
                        };
                    }
                    State::OpenDesiredByRemote { .. } => {
                        tracing::trace!(
                            target: LOG_TARGET,
                            "Handler negotiated inbound OpenDesiredByRemote peer={:?} index={:?}",
                            self.peer,
                            index
                        );
                    }
                    State::Opening {
                        ref mut inbound_substream,
                        ..
                    }
                    | State::Open {
                        ref mut inbound_substream,
                        ..
                    } => {
                        if inbound_substream.is_some() {
                            tracing::trace!(
                                target: LOG_TARGET,
                                "Handler negotiated inbound handshake already handled peer={:?} index={:?}",
                                self.peer,
                                index
                            );
                            return;
                        }

                        tracing::trace!(
                            target: LOG_TARGET,
                            "Handler negotiated inbound setup handshake peer={:?} index={:?}",
                            self.peer,
                            index
                        );

                        let handshake_message = proto.handshake.clone();
                        stream.substream.set_handshake(handshake_message);
                        *inbound_substream = Some(stream.substream);
                    }
                }
            }
            ConnectionEvent::FullyNegotiatedOutbound(outbound) => {
                let (opened, index) = (outbound.protocol, outbound.info);

                tracing::debug!(
                    target: LOG_TARGET,
                    "Handler negotiated outbound peer={:?} index={:?}",
                    self.peer,
                    index
                );

                let proto = &mut self.protocols[index];
                match proto.state {
                    State::Closed {
                        ref mut pending_opening,
                    }
                    | State::OpenDesiredByRemote {
                        ref mut pending_opening,
                        ..
                    } => {
                        tracing::trace!(
                            target: LOG_TARGET,
                            "Handler negotiated outbound Closed|OpenDesiredByRemote peer={:?} index={:?}",
                            self.peer,
                            index
                        );

                        *pending_opening = false;
                    }
                    State::Opening {
                        ref mut inbound_substream,
                        inbound,
                    } => {
                        tracing::trace!(
                            target: LOG_TARGET,
                            "Handler negotiated outbound Opening successful peer={:?} index={:?}",
                            self.peer,
                            index
                        );

                        let (send, recv) = mpsc::channel(1024);
                        proto.state = State::Open {
                            inbound_substream: inbound_substream.take(),
                            outbound_substream: Some(opened.substream),
                            recv: recv.peekable(),
                        };

                        self.pending_events
                            .push_back(ConnectionHandlerEvent::NotifyBehaviour(
                                    NotificationsHandlerToBehavior::HandshakeCompleted {
                                        index,
                                        endpoint: self.endpoint.clone(),
                                        handshake: opened.handshake,
                                        is_inbound: inbound,
                                        sender: send,
                                    },
                            ));
                    }
                    State::Open { .. } => {
                        tracing::trace!(
                            target: LOG_TARGET,
                            "Handler negotiated outbound Open missmatch-state peer={:?} index={:?}",
                            self.peer,
                            index
                        );
                    }
                }
            }
            ConnectionEvent::DialUpgradeError(err) => {
                tracing::debug!(
                    target: LOG_TARGET,
                    "Handler DialError peer={:?} index={:?} error={:?}",
                    self.peer,
                    err.info,
                    err.error,
                );

                let proto = &mut self.protocols[err.info];

                match proto.state {
                    State::Closed {
                        ref mut pending_opening,
                    }
                    | State::OpenDesiredByRemote {
                        ref mut pending_opening,
                        ..
                    } => {
                        tracing::trace!(
                            target: LOG_TARGET,
                            "Handler DialError Closed|OpenDesiredByRemote peer={:?} info={:?}",
                            self.peer,
                            err.info,
                        );

                        *pending_opening = false;
                    }
                    State::Opening { .. } => {
                        proto.state = State::Closed {
                            pending_opening: false,
                        };

                        tracing::trace!(
                            target: LOG_TARGET,
                            "Handler DialError Opening -> Closed peer={:?} info={:?}",
                            self.peer,
                            err.info,
                        );

                        self.pending_events
                            .push_back(ConnectionHandlerEvent::NotifyBehaviour(
                                    NotificationsHandlerToBehavior::HandshakeError { index: err.info },
                            ));
                    }
                    State::Open { .. } => {}
                }
            }
            _ => {}
        }
    }

    fn on_behaviour_event(&mut self, message: NotificationsHandlerFromBehavior) {
        match message {
            NotificationsHandlerFromBehavior::Open { index } => {
                tracing::debug!(
                    target: LOG_TARGET,
                    "Handler from behaviour Open peer={:?} index={:?}",
                    self.peer,
                    index
                );

                let proto = &mut self.protocols[index];

                match &mut proto.state {
                    State::Closed { pending_opening } => {
                        if !*pending_opening {
                            let protocol = HandshakeOutbound {
                                name: proto.name.clone(),
                                handshake: proto.handshake.clone(),
                            };

                            tracing::trace!(
                                target: LOG_TARGET,
                                "Handler from behaviour Closed -> request new substream peer={:?} index={:?}",
                                self.peer,
                                index
                            );

                            self.pending_events.push_back(
                                ConnectionHandlerEvent::OutboundSubstreamRequest {
                                    protocol: SubstreamProtocol::new(protocol, index),
                                },
                            )
                        }

                        tracing::trace!(
                            target: LOG_TARGET,
                            "Handler from behaviour Closed -> Opening peer={:?} index={:?}",
                            self.peer,
                            index
                        );

                        proto.state = State::Opening {
                            inbound_substream: None,
                            inbound: false,
                        };
                    }
                    State::OpenDesiredByRemote {
                        inbound_substream,
                        pending_opening,
                    } => {
                        if !*pending_opening {
                            let protocol = HandshakeOutbound {
                                name: proto.name.clone(),
                                handshake: proto.handshake.clone(),
                            };

                            tracing::trace!(
                                target: LOG_TARGET,
                                "Handler from behaviour OpenDesiredByRemote -> request new substream peer={:?} index={:?}",
                                self.peer,
                                index
                            );

                            self.pending_events.push_back(
                                ConnectionHandlerEvent::OutboundSubstreamRequest {
                                    protocol: SubstreamProtocol::new(protocol, index),
                                },
                            )
                        }

                        tracing::trace!(
                            target: LOG_TARGET,
                            "Handler from behaviour OpenDesiredByRemote setup handshake peer={:?} index={:?}",
                            self.peer,
                            index
                        );

                        let handshake = proto.handshake.clone();
                        inbound_substream.set_handshake(handshake);

                        let inbound_substream = match mem::replace(
                            &mut proto.state,
                            State::Opening {
                                inbound_substream: None,
                                inbound: false,
                            },
                        ) {
                            State::OpenDesiredByRemote {
                                inbound_substream, ..
                            } => inbound_substream,
                            _ => unreachable!(),
                        };
                        proto.state = State::Opening {
                            inbound_substream: Some(inbound_substream),
                            inbound: true,
                        };
                    }
                    State::Opening { .. } | State::Open { .. } => {
                        tracing::trace!(
                            target: LOG_TARGET,
                            "Handler from behaviour Opening|Open statemissmatch peer={:?} index={:?}",
                            self.peer,
                            index
                        );
                    }
                }
            }

            NotificationsHandlerFromBehavior::Close { index } => {
                tracing::debug!(
                    target: LOG_TARGET,
                    "Handler from behaviour Close peer={:?} index={:?}",
                    self.peer,
                    index
                );

                let proto = &mut self.protocols[index];

                match proto.state {
                    State::Closed { .. } => {}
                    State::OpenDesiredByRemote {
                        pending_opening, ..
                    } => {
                        proto.state = State::Closed { pending_opening };
                    }
                    State::Opening { .. } => {
                        proto.state = State::Closed {
                            pending_opening: true,
                        };

                        tracing::trace!(
                            target: LOG_TARGET,
                            "Handler from behaviour Close with handshake in progress peer={:?} index={:?}",
                            self.peer,
                            index
                        );

                        self.pending_events
                            .push_back(ConnectionHandlerEvent::NotifyBehaviour(
                                    NotificationsHandlerToBehavior::HandshakeError { index },
                            ));
                    }
                    State::Open { .. } => {
                        proto.state = State::Closed {
                            pending_opening: false,
                        };
                    }
                }

                self.pending_events
                    .push_back(ConnectionHandlerEvent::NotifyBehaviour(
                            NotificationsHandlerToBehavior::Close { index },
                    ));
            }
        }
    }

    fn connection_keep_alive(&self) -> KeepAlive {
        if self
            .protocols
                .iter()
                .any(|p| !matches!(p.state, State::Closed { .. }))
        {
            return KeepAlive::Yes;
        }

        KeepAlive::No
    }

    #[allow(deprecated)]
    fn poll(
        &mut self,
        cx: &mut Context,
    ) -> Poll<ConnectionHandlerEvent<
        Self::OutboundProtocol,
        Self::OutboundOpenInfo,
        Self::ToBehaviour,
        Self::Error,
        >,
    > {
            if let Some(ev) = self.pending_events.pop_front() {
                return Poll::Ready(ev);
            }

            for index in 0..self.protocols.len() {
                if let State::Open {
                    outbound_substream: Some(outbound_substream),
                    recv,
                    ..
                } = &mut self.protocols[index].state
                {
                    loop {
                        match Pin::new(&mut *recv).as_mut().poll_peek(cx) {
                            Poll::Ready(Some(..)) => {}
                            _ => break,
                        };

                        match outbound_substream.poll_ready_unpin(cx) {
                            Poll::Ready(_) => {}
                            Poll::Pending => break,
                        };

                        let message = match recv.poll_next_unpin(cx) {
                            Poll::Ready(Some(message)) => message,
                            Poll::Ready(None) | Poll::Pending => {
                                debug_assert!(false);
                                break;
                            }
                        };

                        tracing::trace!(
                            target: LOG_TARGET,
                            "Handler poll send message peer={:?} index={:?} message={:?}",
                            self.peer,
                            index,
                            message
                        );

                        let _ = outbound_substream.start_send_unpin(message);
                    }
                }
            }

            for index in 0..self.protocols.len() {
                if let State::Open {
                    outbound_substream: outbound_substream @ Some(_),
                    ..
                } = &mut self.protocols[index].state
                {
                    match Sink::poll_flush(Pin::new(outbound_substream.as_mut().unwrap()), cx) {
                        Poll::Pending | Poll::Ready(Ok(())) => {}
                        Poll::Ready(Err(_)) => {
                            *outbound_substream = None;

                            return Poll::Ready(ConnectionHandlerEvent::NotifyBehaviour(
                                    NotificationsHandlerToBehavior::CloseDesired { index },
                            ));
                        }
                    }
                }
            }

            for index in 0..self.protocols.len() {
                match &mut self.protocols[index].state {
                    State::Open {
                        inbound_substream: inbound_substream @ Some(_),
                        ..
                    } => match Stream::poll_next(Pin::new(inbound_substream.as_mut().unwrap()), cx) {
                        Poll::Pending => {}
                        Poll::Ready(Some(Ok(bytes))) => {
                            let event = NotificationsHandlerToBehavior::Notification { index, bytes };
                            return Poll::Ready(ConnectionHandlerEvent::NotifyBehaviour(event));
                        }
                        Poll::Ready(None) | Poll::Ready(Some(Err(_))) => *inbound_substream = None,
                    },

                    State::OpenDesiredByRemote {
                        inbound_substream,
                        pending_opening,
                    } => match HandshakeInboundSubstream::poll_process(Pin::new(inbound_substream), cx)
                    {
                        Poll::Pending => {}
                        Poll::Ready(Ok(void)) => match void {},
                        Poll::Ready(Err(_)) => {
                            self.protocols[index].state = State::Closed {
                                pending_opening: *pending_opening,
                            };
                            return Poll::Ready(ConnectionHandlerEvent::NotifyBehaviour(
                                    NotificationsHandlerToBehavior::CloseDesired { index },
                            ));
                        }
                    },

                    State::Opening {
                        inbound_substream: inbound_substream @ Some(_),
                        ..
                    } => match HandshakeInboundSubstream::poll_process(
                        Pin::new(inbound_substream.as_mut().unwrap()),
                        cx,
                    ) {
                        Poll::Pending => {}
                        Poll::Ready(Ok(void)) => match void {},
                        Poll::Ready(Err(_)) => *inbound_substream = None,
                    },

                    _ => (),
                }
            }

            Poll::Pending
        }
}
