use std::sync::Arc;

#[cfg(feature = "use-async-std")]
use async_std::{
    channel::{unbounded as unbounded_channel, Receiver, Sender},
    io::{Read, Write},
    sync::Mutex,
    task::{spawn, JoinHandle},
};
#[cfg(feature = "use-smol")]
use smol::{
    channel::{unbounded as unbounded_channel, Receiver, Sender},
    io::{AsyncRead as Read, AsyncWrite as Write},
    lock::Mutex,
    spawn, Task as JoinHandle,
};
use tendermint_proto::abci::{Request, Response};
#[cfg(feature = "use-tokio")]
use tokio::{
    io::{AsyncRead as Read, AsyncWrite as Write},
    spawn,
    sync::{
        mpsc::{unbounded_channel, UnboundedReceiver as Receiver, UnboundedSender as Sender},
        Mutex,
    },
    task::JoinHandle,
};
use tracing::{debug, error, info, instrument};

use crate::{
    async_api::{Consensus, Info, Mempool, Snapshot},
    handler::*,
    state::ConsensusStateValidator,
    utils::{StreamReader, StreamWriter},
};

macro_rules! spawn {
    ($expr: expr) => {
        cfg_if::cfg_if! {
            if #[cfg(any(feature = "use-async-std", feature = "use-tokio"))] {
                spawn($expr)
            } else if #[cfg(feature = "use-smol")] {
                spawn($expr)
                    .detach()
            } else {
                unreachable!()
            }
        }
    };
}

#[instrument(skip(stream_reader, stream_writer, consensus))]
pub fn spawn_consensus_task<R, W, C>(
    stream_reader: StreamReader<R>,
    stream_writer: StreamWriter<W>,
    peer_addr: String,
    consensus: Arc<C>,
    validator: Arc<Mutex<ConsensusStateValidator>>,
) where
    R: Read + Unpin + Send + 'static,
    W: Write + Unpin + Send + 'static,
    C: Consensus + 'static,
{
    info!(message = "Spawning consensus task");

    spawn!(async move {
        consensus_task(
            stream_reader,
            stream_writer,
            peer_addr,
            consensus.as_ref(),
            validator,
        )
        .await
    });
}

#[instrument(skip(stream_reader, stream_writer, consensus))]
async fn consensus_task<R, W, C>(
    mut stream_reader: StreamReader<R>,
    mut stream_writer: StreamWriter<W>,
    peer_addr: String,
    consensus: &C,
    validator: Arc<Mutex<ConsensusStateValidator>>,
) where
    R: Read + Unpin,
    W: Write + Unpin,
    C: Consensus,
{
    while let Ok(request) = stream_reader.read().await {
        match request {
            None => debug!(message = "Received empty request"),
            Some(request) => {
                let request: Request = request;

                let response = match request.value {
                    None => {
                        debug!(message = "Received empty value in request", ?request);
                        Response::default()
                    }
                    Some(request_value) => {
                        handle_consensus_request(consensus, validator.clone(), request_value).await
                    }
                };

                if let Err(err) = stream_writer.write(response).await {
                    error!(message = "Error while writing to stream", %err);
                }
            }
        }
    }
}

#[instrument(skip(stream_reader, stream_writer, mempool))]
pub fn spawn_mempool_task<R, W, M>(
    stream_reader: StreamReader<R>,
    stream_writer: StreamWriter<W>,
    peer_addr: String,
    mempool: Arc<M>,
) where
    R: Read + Unpin + Send + 'static,
    W: Write + Unpin + Send + 'static,
    M: Mempool + 'static,
{
    info!(message = "Spawning mempool tasks");

    let (handle_sender, handle_receiver) = unbounded_channel();
    let peer_addr_clone = peer_addr.clone();

    spawn!(
        async move { mempool_writer_task(stream_writer, peer_addr_clone, handle_receiver).await }
    );

    spawn!(
        async move { mempool_reader_task(stream_reader, peer_addr, mempool, handle_sender).await }
    );
}

#[instrument(skip(stream_writer, handle_receiver))]
async fn mempool_writer_task<W>(
    mut stream_writer: StreamWriter<W>,
    peer_addr: String,
    handle_receiver: Receiver<JoinHandle<Response>>,
) where
    W: Write + Unpin,
{
    cfg_if::cfg_if! {
        if #[cfg(any(feature = "use-async-std", feature = "use-smol"))] {
            while let Ok(handle) = handle_receiver.recv().await {
                let response = handle.await;

                if let Err(err) = stream_writer.write(response).await {
                    error!(message = "Error while writing to stream", %err);
                }
            }
        } else if #[cfg(feature = "use-tokio")] {
            let mut handle_receiver = handle_receiver;

            while let Some(handle) = handle_receiver.recv().await {
                let response = handle.await;

                match response {
                    Ok(response) => {
                        if let Err(err) = stream_writer.write(response).await {
                            error!(message = "Error while writing to stream", %err);
                        }
                    }
                    Err(err) => error!(message = "Mempool request execution not completed", ?err),
                }
            }
        } else {
            unreachable!()
        }
    }
}

#[instrument(skip(stream_reader, mempool, handle_sender))]
async fn mempool_reader_task<R, M>(
    mut stream_reader: StreamReader<R>,
    peer_addr: String,
    mempool: Arc<M>,
    handle_sender: Sender<JoinHandle<Response>>,
) where
    R: Read + Unpin,
    M: Mempool + 'static,
{
    while let Ok(request) = stream_reader.read().await {
        match request {
            None => {
                debug!(message = "Received empty request")
            }
            Some(request) => {
                let peer_addr = peer_addr.clone();
                let mempool = mempool.clone();

                let handle = spawn(async move {
                    let request: Request = request;

                    match request.value {
                        None => {
                            debug!(
                                message = "Received empty value in request",
                                ?peer_addr,
                                ?request
                            );
                            Response::default()
                        }
                        Some(request_value) => {
                            handle_mempool_request(mempool.as_ref(), request_value).await
                        }
                    }
                });

                cfg_if::cfg_if! {
                    if #[cfg(any(feature = "use-async-std", feature = "use-smol"))] {
                        handle_sender
                            .send(handle)
                            .await
                            .expect("Channel receiver dropped");
                    } else if #[cfg(feature = "use-tokio")] {
                        handle_sender
                            .send(handle)
                            .expect("Channel receiver dropped");
                    } else {
                        unreachable!()
                    }
                }
            }
        }
    }
}

#[instrument(skip(stream_reader, stream_writer, info))]
pub fn spawn_info_task<R, W, I>(
    stream_reader: StreamReader<R>,
    stream_writer: StreamWriter<W>,
    peer_addr: String,
    info: Arc<I>,
    validator: Arc<Mutex<ConsensusStateValidator>>,
) where
    R: Read + Unpin + Send + 'static,
    W: Write + Unpin + Send + 'static,
    I: Info + 'static,
{
    info!(message = "Spawning info task");

    spawn!(async move {
        info_task(
            stream_reader,
            stream_writer,
            peer_addr,
            info.as_ref(),
            validator,
        )
        .await
    });
}

#[instrument(skip(stream_reader, stream_writer, info))]
async fn info_task<R, W, I>(
    mut stream_reader: StreamReader<R>,
    mut stream_writer: StreamWriter<W>,
    peer_addr: String,
    info: &I,
    validator: Arc<Mutex<ConsensusStateValidator>>,
) where
    R: Read + Unpin,
    W: Write + Unpin,
    I: Info,
{
    while let Ok(request) = stream_reader.read().await {
        match request {
            None => debug!(message = "Received empty request"),
            Some(request) => {
                let request: Request = request;

                let response = match request.value {
                    None => {
                        debug!(message = "Received empty value in request", ?request);
                        Response::default()
                    }
                    Some(request_value) => {
                        handle_info_request(info, validator.clone(), request_value).await
                    }
                };

                if let Err(err) = stream_writer.write(response).await {
                    error!(message = "Error while writing to stream", %err);
                }
            }
        }
    }
}

#[instrument(skip(stream_reader, stream_writer, snapshot))]
pub fn spawn_snapshot_task<R, W, S>(
    stream_reader: StreamReader<R>,
    stream_writer: StreamWriter<W>,
    peer_addr: String,
    snapshot: Arc<S>,
) where
    R: Read + Unpin + Send + 'static,
    W: Write + Unpin + Send + 'static,
    S: Snapshot + 'static,
{
    info!(message = "Spawning snapshot task");

    spawn!(async move {
        snapshot_task(stream_reader, stream_writer, peer_addr, snapshot.as_ref()).await
    });
}

#[instrument(skip(stream_reader, stream_writer, snapshot))]
async fn snapshot_task<R, W, S>(
    mut stream_reader: StreamReader<R>,
    mut stream_writer: StreamWriter<W>,
    peer_addr: String,
    snapshot: &S,
) where
    R: Read + Unpin,
    W: Write + Unpin,
    S: Snapshot,
{
    while let Ok(request) = stream_reader.read().await {
        match request {
            None => debug!(message = "Received empty request"),
            Some(request) => {
                let request: Request = request;

                let response = match request.value {
                    None => {
                        debug!(message = "Received empty value in request", ?request);
                        Response::default()
                    }
                    Some(request_value) => handle_snapshot_request(snapshot, request_value).await,
                };

                if let Err(err) = stream_writer.write(response).await {
                    error!(message = "Error while writing to stream", %err);
                }
            }
        }
    }
}