use std::convert::Infallible;
use std::sync::Arc;
use std::time::Duration;

use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper_util::rt::{TokioIo, TokioTimer};
use tokio::net::TcpListener;
use tokio::sync::watch;
use tokio::task::JoinHandle;

use super::GatewayHandle;
use crate::config::{GatewayBind, GatewayConfig};
use crate::error::GatewayError;
use crate::service::GatewayServiceInfo;

// Shorter than Nomad and Prometheus scrape timeouts
const HEADER_READ_TIMEOUT: Duration = Duration::from_secs(2);
// Keeps gateway stack use and parsing work bounded
const MAX_HEADERS: usize = 64;
// Hyper requires at least 8192 bytes for HTTP/1 buffers
const MAX_READ_BUFFER: usize = 16 * 1024;

/// Starts the telemetry gateway.
///
/// # Errors
/// [`GatewayError`] if config validation or listener binding fails.
pub async fn start<I>(config: GatewayConfig, input: I) -> Result<GatewayHandle, GatewayError>
where
    I: Into<GatewayServiceInfo>,
{
    config
        .validate()
        .map_err(|error| GatewayError::Config(error.to_string()))?;
    let input = input.into();
    let service = service_info(&input);
    if !config.enabled.is_enabled() {
        return Ok(GatewayHandle::new(
            &config, service, input, None, None, None,
        ));
    }
    let listener = TcpListener::bind(config.bind.socket_addr())
        .await
        .map_err(GatewayError::bind)?;
    let local_addr = GatewayBind::new(listener.local_addr().map_err(GatewayError::bind)?);
    let (shutdown_tx, shutdown_rx) = watch::channel(false);
    let handle = GatewayHandle::new(
        &config,
        service,
        input,
        Some(shutdown_tx),
        None,
        Some(local_addr),
    );
    let task = spawn(listener, shutdown_rx, handle.clone());
    handle.set_task(task);
    Ok(handle)
}

fn service_info(input: &GatewayServiceInfo) -> efdn_telem::canonical::service::ServiceInfo {
    efdn_telem::canonical::service::ServiceInfo::new(input.name().clone(), input.version().clone())
}

fn spawn(
    listener: TcpListener,
    shutdown: watch::Receiver<bool>,
    handle: GatewayHandle,
) -> JoinHandle<()> {
    tokio::spawn(async move {
        accept_loop(listener, shutdown, handle.inner()).await;
    })
}

async fn accept_loop(
    listener: TcpListener,
    mut shutdown: watch::Receiver<bool>,
    state: Arc<super::GatewayInner>,
) {
    loop {
        tokio::select! {
            _ = shutdown.changed() => break,
            accepted = listener.accept() => {
                if !serve_accepted(accepted, Arc::clone(&state), shutdown.clone()) {
                    tokio::time::sleep(Duration::from_millis(100)).await;
                }
            }
        }
    }
}

fn serve_accepted(
    accepted: Result<(tokio::net::TcpStream, std::net::SocketAddr), std::io::Error>,
    state: Arc<super::GatewayInner>,
    mut shutdown: watch::Receiver<bool>,
) -> bool {
    let Ok((stream, _)) = accepted else {
        return false;
    };
    let Ok(permit) = state.connection_limit.clone().try_acquire_owned() else {
        return true;
    };
    let tracker = state.connections.clone();
    let task_state = Arc::clone(&state);
    let task = tokio::spawn(async move {
        let _permit = permit;
        let io = TokioIo::new(stream);
        let service = service_fn(move |req| {
            let state = Arc::clone(&state);
            async move { Ok::<_, Infallible>(handle_with_timeout(req, state).await) }
        });
        let mut builder = http1::Builder::new();
        builder
            .timer(TokioTimer::new())
            .header_read_timeout(HEADER_READ_TIMEOUT)
            .max_headers(MAX_HEADERS)
            .max_buf_size(MAX_READ_BUFFER);
        let connection = builder.serve_connection(io, service);
        tokio::pin!(connection);
        tokio::select! {
            result = &mut connection => {
                let _ = result;
            }
            _ = shutdown.changed() => {
                connection.as_mut().graceful_shutdown();
                let _ = connection.await;
            }
        }
    });
    tracker.spawn(async move {
        if let Err(error) = task.await {
            task_state
                .connection_errors
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner())
                .push(error.to_string());
        }
    });
    true
}

async fn handle_with_timeout(
    req: hyper::Request<hyper::body::Incoming>,
    state: Arc<super::GatewayInner>,
) -> hyper::Response<http_body_util::Full<bytes::Bytes>> {
    let timeout = state.request_timeout.get();
    match tokio::time::timeout(timeout, super::routes::handle(req, state)).await {
        Ok(response) => response,
        Err(_) => super::routes::timeout(),
    }
}
