use std::net::SocketAddr;
use std::path::PathBuf;
use std::time::Duration;
use std::{future::Future, pin::Pin, sync::Arc};

use anyhow::{Context, Result};
use http_body_util::{BodyExt, Full};
use hyper::body::Incoming as IncomingBody;
use hyper::server::conn::http1;
use hyper::Method;
use hyper::{body::Bytes, Response};
use hyper::{service::Service, Request};
use hyper_util::rt::TokioIo;
use serde_json::Value;
use tokio::io::AsyncWriteExt;
use tokio::{net::TcpStream, sync::Mutex};
use tokio_rustls::TlsAcceptor;

use crate::config::Config;
use crate::doslimit::{ConnectionLimits, GlobalLimits};
use crate::errors::rpc_invalid_request;
use crate::query::Query;
use crate::rpc::daemon::dummycom::DummyCom;
use crate::rpc::daemon::ssl_utils::create_tls_acceptor;
use crate::rpc::daemon::to_rpc_error;
use crate::rpc::rpcstats::RpcStats;
use crate::signal::NetworkNotifier;
use crate::thread::{run_with_timeout, spawn_task};
use crate::util::Channel;

use super::connection::Connection;
use super::server::ConnectionId;
use crate::query::scripthash_subscriptions::ScripthashSubscriptions;

fn parse_post_request(body: Bytes) -> Result<(String, Value, Vec<Value>)> {
    let req_json: Value = serde_json::from_slice(&body).context("invalid JSON format")?;

    let empty_params = json!([]);
    let dummy_id = json!(0);

    let method = req_json
        .get("method")
        .context("missing method")?
        .as_str()
        .context("method not a string")?;
    let params = req_json.get("params").unwrap_or(&empty_params);

    // ID is optional, as we serve only one request. But honor it if passed.
    let id = req_json.get("id").unwrap_or(&dummy_id);

    let params = match params.as_array() {
        Some(p) => p,
        None => bail!("Expected 'params' to be an array"),
    };

    Ok((method.to_string(), id.to_owned(), params.to_vec()))
}

#[derive(Clone)]
struct RestReqService {
    connection: Arc<Mutex<Connection>>,
}

impl Service<Request<IncomingBody>> for RestReqService {
    type Response = Response<Full<Bytes>>;
    type Error = hyper::Error;
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

    fn call(&self, req: Request<IncomingBody>) -> Self::Future {
        let conn = self.connection.clone();

        let error_id = json!(-1);

        let to_response = |json_resp: Value| -> hyper::Result<Response<Full<Bytes>>> {
            let mut response = Response::new(Full::new(Bytes::from(json_resp.to_string())));
            response.headers_mut().insert(
                hyper::header::CONTENT_TYPE,
                hyper::header::HeaderValue::from_static("application/json"),
            );
            Ok(response)
        };

        let res = Box::pin(async move {
            if req.method() != Method::POST {
                return to_response(to_rpc_error(
                    rpc_invalid_request("Request method not supported (use POST)".into()),
                    &error_id,
                ));
            }

            let body = req.into_body().collect().await?.to_bytes();
            let (method, id, params) = match parse_post_request(body) {
                Ok(v) => v,
                Err(e) => {
                    return to_response(to_rpc_error(rpc_invalid_request(e.to_string()), &error_id))
                }
            };
            to_response(conn.lock().await.single_req(&method, &params, &id).await)
        });

        Box::pin(res)
    }
}

async fn run_rest_request(stream: TcpStream, conn: Connection) {
    let io = TokioIo::new(stream);
    let service = RestReqService {
        connection: Arc::new(Mutex::new(conn)),
    };

    if let Err(err) = http1::Builder::new().serve_connection(io, service).await {
        warn!("Failed to serve connection: {:?}", err);
    }
}

async fn run_https_request(stream: TcpStream, conn: Connection, ssl_acceptor: TlsAcceptor) {
    let service = RestReqService {
        connection: Arc::new(Mutex::new(conn)),
    };

    // Accept the TLS connection first
    let tls_stream = match ssl_acceptor.accept(stream).await {
        Ok(tls_stream) => tls_stream,
        Err(err) => {
            warn!("Failed to accept HTTPS connection: {:?}", err);
            return;
        }
    };

    let io = TokioIo::new(tls_stream);
    if let Err(err) = http1::Builder::new().serve_connection(io, service).await {
        warn!("Failed to serve HTTPS connection: {:?}", err);
    }
}

#[allow(clippy::too_many_arguments)]
pub async fn start_http_server(
    conn_acceptor: Channel<Option<(TcpStream, SocketAddr)>>,
    global_limits: Arc<GlobalLimits>,
    config: Arc<Config>,
    query: Arc<Query>,
    stats: Arc<RpcStats>,
    connection_limits: ConnectionLimits,
    network_notifier: Arc<NetworkNotifier>,
    idle_timeout: u64,
    subscription_manager: Arc<ScripthashSubscriptions>,
) -> Result<()> {
    while let Some(conn) = conn_acceptor.receiver().await.recv().await {
        let (mut stream, addr) = match conn {
            Some(c) => c,
            None => break,
        };

        let global_limits = global_limits.clone();

        let mut connections = match global_limits.inc_connection(&addr.ip()).await {
            Err(e) => {
                trace!("[{}] dropping http peer - {}", addr, e);
                let _ = stream.shutdown().await;
                continue;
            }
            Ok(n) => n,
        };

        if connections != (0, 0) {
            debug!(
                "[{} http] connected peer ({:?} out of {:?} connection slots used)",
                addr,
                connections,
                global_limits.connection_limits(),
            );
        }

        let dummycon = DummyCom {};
        let conn_id = ConnectionId::new();
        let subscription_manager_cpy = subscription_manager.clone();

        let conn = Connection::new(
            config.clone(),
            query.clone(),
            addr,
            stats.clone(),
            connection_limits,
            Box::new(dummycon),
            network_notifier.clone(),
            conn_id,
            subscription_manager_cpy,
        );

        spawn_task("http_peer", async move {
            let res = run_with_timeout(
                Duration::from_secs(idle_timeout),
                run_rest_request(stream, conn),
            )
            .await;
            if res.is_err() {
                debug!(
                    "[{} http] disconnected due to timeout ({}s)",
                    addr, idle_timeout
                );
            }

            match global_limits.dec_connection(&addr.ip()).await {
                Ok(n) => connections = n,
                Err(e) => error!("{}", e),
            };
            debug!(
                "[{} http] disconnected peer ({:?} out of {:?} connection slots used)",
                addr,
                connections,
                global_limits.connection_limits(),
            );

            Ok(())
        });
    }

    info!("HTTP connections no longer accepted");
    Ok(())
}

#[allow(clippy::too_many_arguments)]
pub async fn start_https_server(
    conn_acceptor: Channel<Option<(TcpStream, SocketAddr)>>,
    cert_file: PathBuf,
    key_file: PathBuf,
    global_limits: Arc<GlobalLimits>,
    config: Arc<Config>,
    query: Arc<Query>,
    stats: Arc<RpcStats>,
    connection_limits: ConnectionLimits,
    network_notifier: Arc<NetworkNotifier>,
    idle_timeout: u64,
    subscription_manager: Arc<ScripthashSubscriptions>,
) -> Result<()> {
    let ssl_acceptor = create_tls_acceptor(&cert_file, &key_file).await?;

    while let Some(conn) = conn_acceptor.receiver().await.recv().await {
        let (mut stream, addr) = match conn {
            Some(c) => c,
            None => break,
        };

        let global_limits = global_limits.clone();

        let mut connections = match global_limits.inc_connection(&addr.ip()).await {
            Err(e) => {
                trace!("[{}] dropping https peer - {}", addr, e);
                let _ = stream.shutdown().await;
                continue;
            }
            Ok(n) => n,
        };

        if connections != (0, 0) {
            debug!(
                "[{} https] connected peer ({:?} out of {:?} connection slots used)",
                addr,
                connections,
                global_limits.connection_limits(),
            );
        }

        let dummycon = DummyCom {};
        let conn_id = ConnectionId::new();
        let subscription_manager_cpy = subscription_manager.clone();

        let conn = Connection::new(
            config.clone(),
            query.clone(),
            addr,
            stats.clone(),
            connection_limits,
            Box::new(dummycon),
            network_notifier.clone(),
            conn_id,
            subscription_manager_cpy,
        );

        let ssl_acceptor = ssl_acceptor.clone();

        spawn_task("https_peer", async move {
            let res = run_with_timeout(
                Duration::from_secs(idle_timeout),
                run_https_request(stream, conn, ssl_acceptor),
            )
            .await;
            if res.is_err() {
                debug!(
                    "[{} https] disconnected due to timeout ({}s)",
                    addr, idle_timeout
                );
            }

            match global_limits.dec_connection(&addr.ip()).await {
                Ok(n) => connections = n,
                Err(e) => error!("{}", e),
            };
            debug!(
                "[{} https] disconnected peer ({:?} out of {:?} connection slots used)",
                addr,
                connections,
                global_limits.connection_limits(),
            );

            Ok(())
        });
    }

    info!("HTTPS connections no longer accepted");
    Ok(())
}
