use std::rc::Rc;

use futures_concurrency::future::Race;
use hyper_util::{
    rt::TokioIo,
    server::{conn::auto::Builder, graceful::GracefulShutdown},
};
use nailbox::boxed_future_within;
use nailcancel::CancelToken;
use nailip::IdentifyPeer;
use nailrater::{LimitModes, NailRater};
use nailresponder::NailResponder;
use nailrt::HyperExecutor;
use nailservice::normalize::NormalizeUrls;
use nailstate::ServerState;
use nailtrace::TraceConnection;
use tokio::net::TcpListener;

async fn server_loop(
    listener: TcpListener,
    app: ServerState,
    graceful: &GracefulShutdown,
) -> color_eyre::Result<()> {
    let server = Builder::new(HyperExecutor::new());

    let is_traced = app.config.open_telemetry.logs
        || app.config.open_telemetry.traces
        || app.config.open_telemetry.record_details;

    let partial_service = Rc::new(NormalizeUrls::new(Rc::new(TraceConnection::new(
        is_traced,
        Rc::new(NailRater::new(
            LimitModes::from(&app.config.rate_limiting),
            app.spicy_payloads.get(),
            app.config.clone_inner(),
            Rc::new(NailResponder::new(app)),
        )),
    ))));

    loop {
        let (socket, remote_addr) = listener.accept().await?;

        let socket = TokioIo::new(socket);

        let hyper_service = IdentifyPeer::new(remote_addr, partial_service.clone());

        let conn = boxed_future_within(|| {
            let conn = server.serve_connection(socket, hyper_service);

            graceful.watch(conn.into_owned())
        });

        tokio::task::spawn_local(conn);
    }
}

async fn cancel_loop(token: CancelToken) -> color_eyre::Result<()> {
    token.cancelled().await;
    Ok(())
}

/// Serves a `hyper` service app, taking a [`CancelToken`] to do a graceful shutdown
/// loop.
pub async fn serve(app: ServerState, token: CancelToken) -> color_eyre::Result<()> {
    let listener = nailnet::get_tcp_socket(&app.config.server.socket_addr)?;
    let ip = listener.local_addr()?;

    tracing::info!(
        port = ip.port(),
        address = ip.ip().to_string(),
        "{} listening on {}",
        std::thread::current().name().unwrap(),
        ip
    );

    let graceful = GracefulShutdown::new();

    (server_loop(listener, app, &graceful), cancel_loop(token))
        .race()
        .await?;

    graceful.shutdown().await;

    Ok(())
}