use crate::connection::Connection;
use crate::router::Router;
use crate::server::{UpstreamConfig, VarDiffConfig};
use crate::BanManager;
pub use crate::MinerList;
use crate::{Error, Result};
use async_std::net::TcpStream;
use async_std::sync::Arc;
use futures::channel::mpsc::{unbounded, UnboundedReceiver};
use futures::io::BufReader;
use futures::io::ReadHalf;
use futures::io::WriteHalf;
use futures::io::{AsyncBufReadExt, AsyncReadExt};
use futures::AsyncWriteExt;
use futures::SinkExt;
use futures::StreamExt;
use log::{debug, info, warn};
use serde_json::{Map, Value};
use std::net::SocketAddr;

//@todo might make sene to wrap a lot of these into one param called "ConnectionConfig" and then
//just pass that along, but we'll see.
#[allow(clippy::too_many_arguments)]
pub async fn handle_connection<
    State: Clone + Send + Sync + 'static,
    CState: Clone + Send + Sync + 'static,
>(
    ban_manager: Arc<BanManager>,
    mut addr: SocketAddr,
    connection_list: Arc<MinerList<CState>>,
    router: Arc<Router<State, CState>>,
    upstream_router: Arc<Router<State, CState>>,
    upstream_config: UpstreamConfig,
    state: State,
    stream: TcpStream,
    // mut wh: BufReader<ReadHalf<TcpStream>>,
    // rh: WriteHalf<TcpStream>,
    var_diff_config: VarDiffConfig,
    initial_difficulty: f64,
    connection_state: CState,
    proxy: bool,
    expected_port: u16,
) {
    let (rh, wh) = stream.split();

    let mut buffer_stream = BufReader::new(rh);

    //@todo wrap this in a function.
    if proxy {
        let mut buf = String::new();
        //@todo handle unwrap here.
        buffer_stream.read_line(&mut buf).await.unwrap();

        //Buf will be of the format "PROXY TCP4 92.118.161.17 172.20.42.228 55867 8080\r\n"
        //Trim the \r\n off
        let buf = buf.trim();
        //Might want to not be ascii whitespace and just normal here.
        // let pieces = buf.split_ascii_whitespace();

        let pieces: Vec<&str> = buf.split(' ').collect();

        let attempted_port: u16 = pieces[5].parse().unwrap();

        //Check that they were trying to connect to us.
        if attempted_port != expected_port {
            //@todo Warn error here
            return;
        }

        addr = format!("{}:{}", pieces[2], pieces[4]).parse().unwrap();
    }

    if !ban_manager.check_banned(&addr).await {
        let (tx, rx) = unbounded();
        let (utx, urx) = unbounded();
        //@todo name...
        let (mut urtx, urrx) = unbounded();

        let connection = Arc::new(Connection::new(
            addr,
            tx,
            utx,
            urrx,
            // @todo these should all come in some config.
            initial_difficulty,
            var_diff_config,
            connection_state,
        ));

        let stop_token = connection.get_stop_token();

        if upstream_config.enabled {
            let upstream = TcpStream::connect(upstream_config.url).await.unwrap();

            let (urh, uwh) = upstream.split();
            let mut upstream_buffer_stream = BufReader::new(urh);

            async_std::task::spawn(async move {
                match upstream_send_loop(urx, uwh).await {
                    //@todo not sure if we even want a info here, we need an ID tho.
                    Ok(_) => info!("Upstream Send Loop is closing for connection"),
                    Err(e) => warn!(
                        "Upstream Send loop is closed for connection: {}, Reason: {}",
                        1, e
                    ),
                }
            });

            async_std::task::spawn({
                let state = state.clone();
                let connection = connection.clone();
                let stop_token = stop_token.clone();

                async move {
                    loop {
                        // if connection.is_disconnected().await {
                        //     break;
                        // }

                        //Maybe have a wrap here or something and on Error instead of unwrap we
                        //break.
                        // let (method, values) = match connection.next_message().await {
                        //     Ok((method, values)) => (method, values),
                        //     Err(_) => {
                        //         break;
                        //     }
                        // };
                        // @todo here is our error on disconnecting miners.... If they get disconnected, BUT we
                        // never receive a message from them, then we will just sit on this loop indefinitely.
                        // So we need a way of actively telling this loop to disconnect.
                        // @todo it's possible this is super inefficient doing this on each message, but I
                        // can't really think of a better way to incorporate stop_token.
                        let next_message =
                            stop_token.stop_future(next_message(&mut upstream_buffer_stream));

                        let (method, values) = match next_message.await {
                            Some(result) => match result {
                                Ok((method, values)) => {
                                    if method == "result" {
                                        //@todo handle bad sends
                                        urtx.send(values).await;
                                        continue;
                                    }

                                    (method, values)
                                }
                                Err(_) => {
                                    break;
                                }
                            },
                            None => {
                                break;
                            }
                        };

                        upstream_router
                            .call(&method, values, state.clone(), connection.clone())
                            .await;
                    }
                }
            });
        }

        // let stop_token = connection.get_stop_token();

        //@todo use stop token inside this send loop.
        async_std::task::spawn(async move {
            match send_loop(rx, wh).await {
                //@todo not sure if we even want a info here, we need an ID tho.
                Ok(_) => info!("Send Loop is closing for connection"),
                Err(e) => warn!("Send loop is closed for connection: {}, Reason: {}", 1, e),
            }
        });

        connection_list
            .add_miner(addr, connection.clone())
            .await
            .unwrap();

        info!("Accepting stream from: {}", addr);

        loop {
            if connection.is_disconnected().await {
                break;
            }

            //Maybe have a wrap here or something and on Error instead of unwrap we
            //break.
            // let (method, values) = match connection.next_message().await {
            //     Ok((method, values)) => (method, values),
            //     Err(_) => {
            //         break;
            //     }
            // };
            // @todo here is our error on disconnecting miners.... If they get disconnected, BUT we
            // never receive a message from them, then we will just sit on this loop indefinitely.
            // So we need a way of actively telling this loop to disconnect.
            // @todo it's possible this is super inefficient doing this on each message, but I
            // can't really think of a better way to incorporate stop_token.
            let next_message = stop_token.stop_future(next_message(&mut buffer_stream));

            let (method, values) = match next_message.await {
                Some(result) => match result {
                    Ok((method, values)) => (method, values),
                    Err(_) => {
                        break;
                    }
                },
                None => {
                    break;
                }
            };

            router
                .call(&method, values, state.clone(), connection.clone())
                .await;
        }

        //@todo we can kill state in connection now
        //Unused for now, but may be useful for logging or bans
        // let _result = connection.start(router.clone()).await;

        info!("Closing stream from: {}", addr);

        connection_list.remove_miner(addr).await.unwrap();

        if connection.needs_ban().await {
            ban_manager.add_ban(&addr).await;
        }
    } else {
        warn!(
            "Banned connection attempting to connect: {}. Connected closed",
            addr
        );
    }
}

pub async fn next_message(
    stream: &mut BufReader<ReadHalf<TcpStream>>,
) -> Result<(String, serde_json::map::Map<String, serde_json::Value>)> {
    //I don't actually think this has to loop here.
    loop {
        // let mut stream = self.read_half.lock().await;

        let mut buf = String::new();
        let num_bytes = stream.read_line(&mut buf).await?;

        if num_bytes == 0 {
            // self.shutdown().await?;
            return Err(Error::StreamClosed);
        }

        if !buf.is_empty() {
            //@smells
            buf = buf.trim().to_owned();

            debug!("Received Message: {}", &buf);

            if buf.is_empty() {
                continue;
            }

            let msg: Map<String, Value> = match serde_json::from_str(&buf) {
                Ok(msg) => msg,
                Err(_) => continue,
            };

            let method = if msg.contains_key("method") {
                match msg.get("method") {
                    Some(method) => method.as_str(),
                    //@todo need better stratum erroring here.
                    None => return Err(Error::MethodDoesntExist),
                }
            } else if msg.contains_key("messsage") {
                match msg.get("message") {
                    Some(method) => method.as_str(),
                    //@todo need better stratum erroring here.
                    None => return Err(Error::MethodDoesntExist),
                }
                //@todo not sure if I totally like how this is working - it might blow up currently
                //existing shit, but let's get it working then we can refine REFINE REFINE.
            } else if msg.contains_key("result") {
                Some("result")
            } else {
                // return Err(Error::MethodDoesntExist);
                Some("")
            };

            if let Some(method_string) = method {
                //Mark the sender as active as we received a message.
                //We only mark them as active if the message/method was valid
                // self.stats.lock().await.last_active = Utc::now().naive_utc();

                return Ok((method_string.to_owned(), msg));
            } else {
                //@todo improper format
                return Err(Error::MethodDoesntExist);
            }
        };
    }
}

pub async fn send_loop(
    mut rx: UnboundedReceiver<String>,
    mut rh: WriteHalf<TcpStream>,
) -> Result<()> {
    while let Some(msg) = rx.next().await {
        rh.write_all(&msg.as_bytes()).await?;
        //@todo the reason we write this here is that JSON RPC messages are ended with a newline.
        //This probably should be built into the rpc library, but it works here for now.
        //Don't move this unless websockets ALSO require the newline, then we can move it back into
        //the Connection.send function.
        rh.write_all(b"\n").await?;
    }

    Ok(())
}

pub async fn upstream_send_loop(
    mut rx: UnboundedReceiver<String>,
    mut rh: WriteHalf<TcpStream>,
) -> Result<()> {
    while let Some(msg) = rx.next().await {
        rh.write_all(&msg.as_bytes()).await?;
        rh.write_all(b"\n").await?;
    }

    Ok(())
}
