use super::{
    super::{
        config::ShardConfig,
        error::{Error, Result},
        event::Event,
        stage::Stage,
    },
    connect,
    emit,
    inflater::Inflater,
    session::Session,
    socket_forwarder::SocketForwarder,
};

use crate::{
    event::{DispatchEvent, GatewayEvent},
    listener::Listeners,
};
use twilight_model::gateway::payload::{
    identify::{Identify, IdentifyInfo, IdentifyProperties},
    resume::Resume,
};

use futures::{channel::mpsc::UnboundedReceiver, stream::StreamExt};
#[allow(unused_imports)]
use log::{debug, info, trace, warn};
use serde::Serialize;
use std::{env::consts::OS, ops::Deref, sync::Arc};
use tokio::sync::watch::{
    channel as watch_channel,
    Receiver as WatchReceiver,
    Sender as WatchSender,
};
use tokio_tungstenite::tungstenite::{
    protocol::{frame::coding::CloseCode, CloseFrame},
    Message,
};

use std::borrow::Cow;

#[cfg(feature = "metrics")]
use metrics::counter;

use std::error::Error as StdError;

/// Runs in the background and processes incoming events, and then broadcasts
/// to all listeners.
#[derive(Debug)]
pub struct ShardProcessor {
    pub config: Arc<ShardConfig>,
    pub listeners: Listeners<Event>,
    pub properties: IdentifyProperties,
    pub rx: UnboundedReceiver<Message>,
    pub session: Arc<Session>,
    inflater: Inflater,
    url: String,
    resume: Option<(u64, String)>,
    wtx: WatchSender<Arc<Session>>,
}

impl ShardProcessor {
    pub async fn new(config: Arc<ShardConfig>) -> Result<(Self, WatchReceiver<Arc<Session>>)> {
        debug!("[ShardProcessor {:?}] Queueing", config.shard());
        let shard_id = config.shard();
        config.queue.request(shard_id).await;
        debug!("[ShardProcessor {:?}] Finished queue", config.shard());

        let properties = IdentifyProperties::new("twilight.rs", "twilight.rs", OS, "", "");

        let mut url = config
            .http_client()
            .gateway()
            .await
            .map_err(|source| Error::GettingGatewayUrl {
                source,
            })?
            .url;
        url.push_str("?v=6&compress=zlib-stream");

        let stream = connect::connect(&url).await?;
        let (forwarder, rx, tx) = SocketForwarder::new(stream);
        tokio::spawn(async move {
            forwarder.run().await;
        });

        let session = Arc::new(Session::new(tx));

        let (wtx, wrx) = watch_channel(Arc::clone(&session));

        Ok((
            Self {
                config,
                listeners: Listeners::default(),
                properties,
                rx,
                session,
                inflater: Inflater::new(shard_id),
                url,
                resume: None,
                wtx,
            },
            wrx,
        ))
    }

    pub async fn run(mut self) {
        loop {
            let gateway_event = match self.next_event().await {
                Ok(ev) => ev,
                // Reconnect as this error is often fatal!
                Err(Error::Decompressing {
                    source,
                }) => {
                    warn!(
                        "[gateway] Decompressing error, clears buffers and reconnect! {:?}",
                        source
                    );

                    // Inflater gets reset in the reconnect call.
                    self.reconnect(true).await;
                    continue;
                },
                Err(err) => {
                    warn!("Error receiveing gateway event: {:?}", err.source());
                    continue;
                },
            };

            // The only reason for an error is if the sender couldn't send a
            // message or if the session didn't exist when it should, so do a
            // reconnect if this fails.
            if self.process(&gateway_event).await.is_err() {
                debug!("Error processing event; reconnecting");

                self.reconnect(true).await;

                continue;
            }

            emit::event(self.listeners.clone(), Event::from(gateway_event));
        }
    }

    /// Identifies with the gateway to create a new session.
    async fn identify(&mut self) -> Result<()> {
        self.session.set_stage(Stage::Identifying);

        let intents = self.config.intents().copied();

        let identify = Identify::new(IdentifyInfo {
            compression: false,
            guild_subscriptions: true,
            intents,
            large_threshold: 250,
            properties: self.properties.clone(),
            shard: Some(self.config.shard()),
            presence: None,
            token: self.config.token().to_owned(),
            v: 6,
        });

        self.send(identify).await
    }

    async fn process(&mut self, event: &GatewayEvent) -> Result<()> {
        use GatewayEvent::{
            Dispatch,
            Heartbeat,
            HeartbeatAck,
            Hello,
            InvalidateSession,
            Reconnect,
        };

        match event {
            Dispatch(seq, dispatch) => {
                #[cfg(feature = "metrics")]
                counter!("GatewayEvent", 1, "GatewayEvent" => "Dispatch");
                self.session.set_seq(*seq);

                match dispatch.deref() {
                    DispatchEvent::Ready(ready) => {
                        self.session.set_stage(Stage::Connected);
                        self.session.set_id(&ready.session_id).await;
                    },
                    DispatchEvent::Resumed => {
                        self.session.set_stage(Stage::Connected);
                        self.session.heartbeats.receive().await;
                    },
                    _ => {},
                }
            },
            Heartbeat(seq) => {
                #[cfg(feature = "metrics")]
                counter!("GatewayEvent", 1, "GatewayEvent" => "Heartbeat");
                if *seq > self.session.seq() + 1 {
                    self.resume().await?;
                }

                if let Err(err) = self.session.heartbeat() {
                    warn!("Error sending heartbeat; reconnecting: {}", err);

                    self.reconnect(true).await;
                }
            },
            Hello(interval) => {
                #[cfg(feature = "metrics")]
                counter!("GatewayEvent", 1, "GatewayEvent" => "Hello");
                debug!("[EVENT] Hello({})", interval);

                if self.session.stage() == Stage::Resuming && self.resume.is_some() {
                    // Safe to unwrap so here as we have just checked that
                    // it is some.
                    let (seq, id) = self.resume.take().unwrap();
                    warn!("Resumeing with ({}, {})!", seq, id);
                    let payload = Resume::new(seq, &id, self.config.token());

                    // Set id so it is correct for next resume.
                    self.session.set_id(id).await;

                    if *interval > 0 {
                        self.session.set_heartbeat_interval(*interval);
                        self.session.start_heartbeater().await;
                    }

                    self.send(payload).await?;
                } else {
                    self.session.set_stage(Stage::Identifying);

                    if *interval > 0 {
                        self.session.set_heartbeat_interval(*interval);
                        self.session.start_heartbeater().await;
                    }

                    self.identify().await?;
                }
            },
            HeartbeatAck => {
                #[cfg(feature = "metrics")]
                counter!("GatewayEvent", 1, "GatewayEvent" => "HeartbeatAck");
                self.session.heartbeats.receive().await;
            },
            InvalidateSession(true) => {
                #[cfg(feature = "metrics")]
                counter!("GatewayEvent", 1, "GatewayEvent" => "InvalidateSessionTrue");
                debug!("[EVENT] InvalidateSession(true)");
                self.resume().await?;
            },
            InvalidateSession(false) => {
                #[cfg(feature = "metrics")]
                counter!("GatewayEvent", 1, "GatewayEvent" => "InvalidateSessionFalse");
                debug!("[EVENT] InvalidateSession(false)");
                self.reconnect(true).await;
            },
            Reconnect => {
                #[cfg(feature = "metrics")]
                counter!("GatewayEvent", 1, "GatewayEvent" => "Reconnect");
                debug!("[EVENT] Reconnect");
                let frame = CloseFrame {
                    code: CloseCode::Restart,
                    reason: Cow::Borrowed("Reconnecting"),
                };
                self.close(Some(frame)).await?;
                self.resume().await?;
            },
        }

        Ok(())
    }

    async fn reconnect(&mut self, full_reconnect: bool) {
        info!("[reconnect] Reconnection started!");
        loop {
            // Await allowance if doing a full reconnect
            if full_reconnect {
                let shard = self.config.shard();
                self.config.queue.request(shard).await;
            }

            let new_stream = match connect::connect(&self.url).await {
                Ok(s) => s,
                Err(why) => {
                    warn!("Error reconnecting: {:?}", why);
                    continue;
                },
            };

            let (new_forwarder, new_rx, new_tx) = SocketForwarder::new(new_stream);
            tokio::spawn(async move {
                new_forwarder.run().await;
            });

            self.rx = new_rx;
            self.session = Arc::new(Session::new(new_tx));
            match self.wtx.broadcast(Arc::clone(&self.session)) {
                Ok(_) => (),
                Err(why) => {
                    warn!(
                        "Broadcast of new session failed, \
                         This should not happen, please open \
                         a issue on the repo. {}",
                        why
                    );
                    warn!(
                        "After this many of the commands on the \
                         shard will no longer work."
                    );
                },
            };

            if !full_reconnect {
                self.session.set_stage(Stage::Resuming);
            }

            self.inflater.reset();

            break;
        }
    }

    async fn resume(&mut self) -> Result<()> {
        info!("[resume] Resume started!");
        self.session.set_stage(Stage::Resuming);
        self.session.stop_heartbeater().await;

        let seq = self.session.seq();

        let id = if let Some(id) = self.session.id().await {
            id
        } else {
            warn!("Was not able to get the id, reconnecting.");
            self.reconnect(true).await;
            return Ok(());
        };

        self.resume = Some((seq, id));

        self.reconnect(false).await;

        Ok(())
    }

    pub async fn send(&mut self, payload: impl Serialize) -> Result<()> {
        match self.session.send(payload) {
            Ok(()) => Ok(()),
            Err(Error::PayloadSerialization {
                source,
            }) => {
                warn!("Failed to serialize message to send: {:?}", source);

                Err(Error::PayloadSerialization {
                    source,
                })
            },
            Err(Error::SendingMessage {
                source,
            }) => {
                warn!("Failed to send message: {:?}", source);
                info!("Reconnecting");

                self.reconnect(true).await;

                Ok(())
            },
            Err(other) => Err(other),
        }
    }

    async fn close(&mut self, close_frame: Option<CloseFrame<'static>>) -> Result<()> {
        self.session.close(close_frame)?;
        Ok(())
    }

    async fn next_event(&mut self) -> Result<GatewayEvent> {
        loop {
            // Returns None when the socket forwarder has ended, meaning the
            // connection was dropped.
            let msg = if let Some(msg) = self.rx.next().await {
                msg
            } else {
                if let Err(why) = self.resume().await {
                    warn!("Resume failed with {}, reconnecting", why);
                    self.reconnect(true).await;
                }
                continue;
            };

            match msg {
                Message::Binary(bin) => {
                    self.inflater.extend(&bin[..]);
                    let decompressed_msg =
                        self.inflater.msg().map_err(|source| Error::Decompressing {
                            source,
                        })?;
                    let msg_or_error = match decompressed_msg {
                        Some(json) => {
                            emit::bytes(self.listeners.clone(), json).await;

                            match serde_json::from_slice(json).map_err(|source| {
                                Error::PayloadSerialization {
                                    source,
                                }
                            }) {
                                Ok(ser) => Ok(ser),
                                Err(err) => {
                                    debug!("Broken JSON: {:?}", std::str::from_utf8(json));
                                    Err(err)
                                },
                            }
                        },
                        None => continue,
                    };
                    self.inflater.clear();
                    break msg_or_error;
                },
                Message::Close(code) => {
                    log::warn!("Got close code: {:?}.", code);
                    self.resume().await?;
                },
                Message::Ping(_) | Message::Pong(_) => {},
                Message::Text(text) => {
                    trace!("Text payload: {}", text);
                    emit::bytes(self.listeners.clone(), text.as_bytes()).await;

                    break match serde_json::from_str(&text).map_err(|source| {
                        Error::PayloadSerialization {
                            source,
                        }
                    }) {
                        Ok(ser) => Ok(ser),
                        Err(err) => {
                            debug!("Broken JSON: {:?}", &text);
                            Err(err)
                        },
                    };
                },
            }
        }
    }
}
