// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: 2026 ndaal Gesellschaft für Sicherheit in der Informationstechnik mbH & Co KG, Cologne
// SPDX-FileCopyrightText: Author: Pierre Gronau <Pierre.Gronau@ndaal.eu>

//! The localhost listeners.
//!
//! * [`run_tls`] — HTTPS (HTTP/1.1 + HTTP/2) over TCP/TLS 1.3, with bounded
//!   concurrency, a TLS-handshake timeout and a slowloris header-read timeout
//!   (the ndaal OSCAL viewer accept loop).
//! * [`run_quic`] — HTTP/3 over QUIC/UDP (quinn + h3), the QUIC server reused
//!   from the ndaal CSAF-CRUD listener.
//!
//! Both share one self-signed TLS 1.3 certificate so the TCP and UDP listeners
//! present a single identity.

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

use hyper::service::service_fn;
use hyper_util::rt::{TokioExecutor, TokioIo, TokioTimer};
use rustls::pki_types::{CertificateDer, PrivateKeyDer};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use tokio_rustls::TlsAcceptor;

use crate::router::SharedRouter;
use crate::tls::build_tls13_server_config;

/// Upper bound on concurrently served TCP connections.
const MAX_CONNECTIONS: usize = 512;
/// How long a client may take to send its request head (slowloris defence).
const HEADER_READ_TIMEOUT: Duration = Duration::from_secs(30);
/// How long a client may take to complete the TLS handshake.
const TLS_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);

// ---------------------------------------------------------------------------
// HTTPS over TCP (HTTP/1.1 + HTTP/2)
// ---------------------------------------------------------------------------

/// Accept TLS connections until Ctrl-C, bounded by [`MAX_CONNECTIONS`].
pub async fn run_tls(listener: TcpListener, acceptor: TlsAcceptor, router: SharedRouter) {
    let limiter = Arc::new(Semaphore::new(MAX_CONNECTIONS));
    loop {
        tokio::select! {
            accepted = listener.accept() => {
                match accepted {
                    Ok((stream, _peer)) => admit(stream, &acceptor, &router, &limiter),
                    Err(error) => tracing::warn!(%error, "accept failed"),
                }
            },
            _ = tokio::signal::ctrl_c() => {
                tracing::info!("shutdown requested");
                return;
            },
        }
    }
}

/// Admit one connection when the limit allows; shed it otherwise.
fn admit(
    stream: TcpStream,
    acceptor: &TlsAcceptor,
    router: &SharedRouter,
    limiter: &Arc<Semaphore>,
) {
    match Arc::clone(limiter).try_acquire_owned() {
        Ok(permit) => spawn(stream, acceptor.clone(), router.clone(), permit),
        Err(_) => tracing::warn!(
            limit = MAX_CONNECTIONS,
            "connection limit reached — shedding connection"
        ),
    }
}

/// Serve one TLS connection on its own task; the permit is held for its life.
fn spawn(
    stream: TcpStream,
    acceptor: TlsAcceptor,
    router: SharedRouter,
    permit: OwnedSemaphorePermit,
) {
    tokio::spawn(async move {
        let _permit = permit;
        let handshake = tokio::time::timeout(TLS_HANDSHAKE_TIMEOUT, acceptor.accept(stream)).await;
        let tls_stream = match handshake {
            Ok(Ok(stream)) => stream,
            Ok(Err(error)) => {
                tracing::debug!(%error, "TLS handshake failed");
                return;
            }
            Err(_) => {
                tracing::debug!("TLS handshake timed out");
                return;
            }
        };
        let io = TokioIo::new(tls_stream);
        let service = service_fn(move |req| {
            let router = router.clone();
            async move { Ok::<_, Infallible>(router.handle(req).await) }
        });
        let mut builder = hyper_util::server::conn::auto::Builder::new(TokioExecutor::new());
        builder
            .http1()
            .timer(TokioTimer::new())
            .header_read_timeout(HEADER_READ_TIMEOUT);
        builder.http2().timer(TokioTimer::new());
        if let Err(error) = builder.serve_connection(io, service).await {
            tracing::debug!(%error, "connection closed with error");
        }
    });
}

// ---------------------------------------------------------------------------
// HTTP/3 over QUIC/UDP
// ---------------------------------------------------------------------------

/// Run the QUIC/HTTP-3 endpoint until the process exits.
///
/// # Errors
///
/// Returns an error if the QUIC endpoint cannot be created or bound.
pub async fn run_quic(
    addr: SocketAddr,
    router: SharedRouter,
    cert_chain: Vec<CertificateDer<'static>>,
    key_der: PrivateKeyDer<'static>,
) -> anyhow::Result<()> {
    let mut tls_config = build_tls13_server_config(cert_chain, key_der, vec![b"h3".to_vec()])?;
    tls_config.max_early_data_size = u32::MAX;

    let server_config = quinn::ServerConfig::with_crypto(Arc::new(
        quinn::crypto::rustls::QuicServerConfig::try_from(tls_config)?,
    ));
    let endpoint = quinn::Endpoint::server(server_config, addr)?;
    tracing::info!(%addr, "QUIC/HTTP3 endpoint listening (TLS 1.3)");

    while let Some(incoming) = endpoint.accept().await {
        let router = router.clone();
        tokio::spawn(async move {
            if let Err(error) = handle_quic_connection(incoming, router).await {
                tracing::debug!(%error, "QUIC connection error");
            }
        });
    }
    Ok(())
}

/// Serve one QUIC connection's HTTP/3 request streams through the router.
async fn handle_quic_connection(
    incoming: quinn::Incoming,
    router: SharedRouter,
) -> anyhow::Result<()> {
    let connection = incoming.await?;
    let mut h3_conn = h3::server::Connection::new(h3_quinn::Connection::new(connection)).await?;

    loop {
        match h3_conn.accept().await {
            Ok(Some(resolver)) => {
                let (req, mut stream) = resolver.resolve_request().await?;
                let router = router.clone();
                tokio::spawn(async move {
                    let (parts, _body) = req.into_parts();
                    let response = router.handle_parts(parts).await;
                    let resp_headers = http::Response::builder()
                        .status(response.status())
                        .body(())
                        .unwrap_or_default();
                    if let Err(error) = stream.send_response(resp_headers).await {
                        tracing::debug!(%error, "failed to send h3 response headers");
                        return;
                    }
                    let body = http_body_util::BodyExt::collect(response.into_body())
                        .await
                        .map(|b| b.to_bytes())
                        .unwrap_or_default();
                    if let Err(error) = stream.send_data(body).await {
                        tracing::debug!(%error, "failed to send h3 response body");
                    }
                    let _ = stream.finish().await;
                });
            }
            Ok(None) => break,
            Err(error) => {
                tracing::debug!(%error, "h3 accept error");
                break;
            }
        }
    }
    Ok(())
}
