nautalid 0.1.0

//! WebTransport over HTTP/3 (QUIC) — CONNECT with `:protocol webtransport` on a configurable path.
//!
//! | Variable | Role |
//! |----------|------|
//! | **`NAUTALID_WEBTRANSPORT_PATH`** | Demo/echo session path (default **`/wt`**) |
//! | **`NAUTALID_WEBTRANSPORT`** | Set **`1`** to enable WebTransport on HTTP/3 (default: **off**) |
//! | **`NAUTALID_WEBTRANSPORT_REQUIRE_MTLS`** | Demo session requires mTLS (default: inbound TLS client-auth policy) |
//! | **`NAUTALID_WEBTRANSPORT_DEMO_MAX_BIDI`** | Max concurrent demo bidi streams per session (default **16**) |
//! | **`NAUTALID_WEBTRANSPORT_DEMO_MAX_ECHO_BYTES`** | Max bytes echoed per demo bidi stream (default **1 MiB**) |
//! | **`NAUTALID_WEBTRANSPORT_DEMO_MAX_DATAGRAM_BYTES`** | Max demo datagram payload (default **1200**) |
//! | **`NAUTALID_WEBTRANSPORT_DEMO_FRAME_TIMEOUT_SECS`** | Demo bidi read/write timeout (default **15**) |
//! | **`NAUTALID_WEBTRANSPORT_AETHER`** | Route AetherDB wire frames over WT bidi streams |
//! | **`NAUTALID_WEBTRANSPORT_AETHER_PATH`** | Aether session path(s), comma-separated (default **`/wt/aether`**) |
//! | **`NAUTALID_WEBTRANSPORT_LISTEN`** | Optional dedicated UDP listen for WT-only HTTP/3 |

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

use bytes::Bytes;
use h3::ext::Protocol;
use h3::server::{Connection, RequestStream};
use h3_quinn::Connection as QuinnH3Conn;
use h3_webtransport::server::{AcceptedBi, WebTransportSession};
use http::{Method, Request, Response, StatusCode};
use tokio::io::AsyncWriteExt;
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use tracing::{debug, info, warn};

type H3ServerConn = Connection<QuinnH3Conn, Bytes>;
pub(crate) type H3RequestStream = RequestStream<h3_quinn::BidiStream<Bytes>, Bytes>;
pub(crate) type WtSession = WebTransportSession<QuinnH3Conn, Bytes>;

const DEFAULT_DEMO_MAX_BIDI: usize = 16;
const DEFAULT_DEMO_MAX_ECHO_BYTES: usize = 1024 * 1024;
const DEFAULT_DEMO_MAX_DATAGRAM_BYTES: usize = 1200;
const DEFAULT_DEMO_FRAME_TIMEOUT_SECS: u64 = 15;
const DEMO_READ_CHUNK: usize = 4096;

/// Demo session peer identity from QUIC connect time.
#[derive(Debug, Clone)]
pub struct WtDemoSessionAuth {
    pub peer_cert: crate::tls::PeerClientCert,
}

impl WtDemoSessionAuth {
    #[must_use]
    pub fn from_peer_cert(peer_cert: crate::tls::PeerClientCert) -> Self {
        Self { peer_cert }
    }

    #[must_use]
    pub fn peer_authenticated(&self) -> bool {
        !self.peer_cert.is_empty()
    }
}

#[derive(Clone)]
struct WtDemoLimits {
    bidi: Arc<Semaphore>,
    max_echo_bytes: usize,
    max_datagram_bytes: usize,
    frame_timeout: Duration,
    require_mtls: bool,
}

impl WtDemoLimits {
    fn from_env() -> Self {
        Self {
            bidi: Arc::new(Semaphore::new(demo_max_bidi_from_env())),
            max_echo_bytes: demo_max_echo_bytes_from_env(),
            max_datagram_bytes: demo_max_datagram_bytes_from_env(),
            frame_timeout: Duration::from_secs(demo_frame_timeout_secs_from_env()),
            require_mtls: demo_requires_mtls_from_env(),
        }
    }
}

/// WebTransport session mount path from env (default **`/wt`**).
#[must_use]
pub fn path_from_env() -> String {
    std::env::var("NAUTALID_WEBTRANSPORT_PATH")
        .ok()
        .map(|raw| raw.trim().to_string())
        .filter(|path| !path.is_empty() && path.starts_with('/'))
        .unwrap_or_else(|| "/wt".to_string())
}

/// Whether WebTransport is enabled (`NAUTALID_WEBTRANSPORT=1`).
#[must_use]
pub fn enabled_from_env() -> bool {
    crate::surfaces::env_enabled("NAUTALID_WEBTRANSPORT")
}

static WT_FORCE_ENABLED: AtomicBool = AtomicBool::new(false);

/// Enable WebTransport for crate embedding (in addition to env).
pub fn set_enabled(enabled: bool) {
    WT_FORCE_ENABLED.store(enabled, Ordering::Relaxed);
}

/// WebTransport active via env or [`set_enabled`].
#[must_use]
pub fn is_enabled() -> bool {
    enabled_from_env() || WT_FORCE_ENABLED.load(Ordering::Relaxed)
}

/// Build an HTTP/3 server [`h3::server::Builder`] with WebTransport settings applied.
pub fn h3_server_builder() -> h3::server::Builder {
    let mut builder = h3::server::builder();
    if is_enabled() {
        builder
            .enable_webtransport(true)
            .enable_extended_connect(true)
            .enable_datagram(true)
            .max_webtransport_sessions(100);
    }
    builder
}

/// True when the request is an HTTP/3 CONNECT for WebTransport.
#[must_use]
pub fn is_connect(request: &Request<()>) -> bool {
    let protocol = request.extensions().get::<Protocol>();
    matches!(
        (request.method(), protocol),
        (&Method::CONNECT, Some(p)) if p == &Protocol::WEB_TRANSPORT
    )
}

/// True when the request targets the configured WebTransport path.
#[must_use]
pub fn matches_path(request: &Request<()>, path: &str) -> bool {
    request.uri().path() == path
}

/// Whether a demo WT session may be established for the presented peer certificate.
#[must_use]
pub fn demo_peer_allowed(peer_cert: &crate::tls::PeerClientCert) -> bool {
    !demo_requires_mtls_from_env() || !peer_cert.is_empty()
}

/// Accept a WebTransport session (transfers ownership of the HTTP/3 connection).
pub async fn accept_session(
    request: Request<()>,
    stream: H3RequestStream,
    conn: H3ServerConn,
) -> Result<WtSession, h3::error::StreamError> {
    WebTransportSession::accept(request, stream, conn).await
}

/// Reject a CONNECT on the wrong path without starting a session.
pub async fn reject_connect(
    mut stream: H3RequestStream,
    status: StatusCode,
) -> Result<(), h3::error::StreamError> {
    let response = Response::builder()
        .status(status)
        .body(())
        .expect("response");
    stream.send_response(response).await?;
    stream.finish().await?;
    Ok(())
}

/// Brief exponential backoff before rejecting a bidi stream when at capacity.
pub(crate) async fn acquire_bidi_permit(
    sem: Arc<Semaphore>,
) -> Option<OwnedSemaphorePermit> {
    if let Ok(permit) = sem.clone().try_acquire_owned() {
        return Some(permit);
    }
    for delay_ms in [5_u64, 10, 20, 40, 80] {
        tokio::time::sleep(Duration::from_millis(delay_ms)).await;
        if let Ok(permit) = sem.clone().try_acquire_owned() {
            return Some(permit);
        }
    }
    None
}

/// Drive a WebTransport demo session until the client closes or an error occurs.
pub async fn drive_session(session: WtSession, auth: WtDemoSessionAuth) {
    let session_id = session.session_id();
    let limits = WtDemoLimits::from_env();
    if limits.require_mtls && !auth.peer_authenticated() {
        warn!(?session_id, "demo webtransport session rejected: mTLS required");
        return;
    }
    info!(?session_id, peer_authenticated = auth.peer_authenticated(), "webtransport session established");
    let mut reader = session.datagram_reader();
    let mut sender = session.datagram_sender();
    loop {
        tokio::select! {
            bi = session.accept_bi() => {
                match bi {
                    Ok(Some(AcceptedBi::BidiStream(_sid, mut stream))) => {
                        let limits = limits.clone();
                        tokio::spawn(async move {
                            let Some(_permit) = acquire_bidi_permit(limits.bidi.clone()).await else {
                                warn!(?session_id, "demo webtransport bidi stream limit reached");
                                let _ = stream.shutdown().await;
                                return;
                            };
                            echo_bidi(stream, limits).await;
                        });
                    }
                    Ok(Some(AcceptedBi::Request(req, mut stream))) => {
                        debug!(method = %req.method(), uri = %req.uri(), "webtransport nested request");
                        if let Err(err) = serve_nested_request(req, &mut stream, auth.clone()).await {
                            warn!(%err, "webtransport nested request failed");
                        }
                    }
                    Ok(None) => {
                        debug!(?session_id, "webtransport session closed");
                        break;
                    }
                    Err(err) => {
                        warn!(?session_id, %err, "webtransport session error");
                        break;
                    }
                }
            }
            dg = reader.read_datagram() => {
                match dg {
                    Ok(datagram) => {
                        let payload = datagram.into_payload();
                        if payload.len() > limits.max_datagram_bytes {
                            debug!(
                                len = payload.len(),
                                max = limits.max_datagram_bytes,
                                "demo webtransport datagram dropped: too large"
                            );
                            continue;
                        }
                        if sender.send_datagram(payload).is_err() {
                            break;
                        }
                    }
                    Err(err) => {
                        debug!(%err, "webtransport datagram read ended");
                        break;
                    }
                }
            }
        }
    }
}

async fn serve_nested_request(
    req: Request<()>,
    stream: &mut H3RequestStream,
    auth: WtDemoSessionAuth,
) -> Result<(), h3::error::StreamError> {
    let (status, body) = demo_nested_http_response(auth, req.method().as_str(), req.uri().path());
    let response = Response::builder()
        .status(status)
        .header("content-type", "application/json")
        .body(())
        .expect("response");
    stream.send_response(response).await?;
    stream.send_data(Bytes::from(body.to_string())).await?;
    stream.finish().await
}

/// JSON + status for nested HTTP requests inside a demo WT session (testable policy surface).
#[must_use]
pub fn demo_nested_http_response(
    auth: WtDemoSessionAuth,
    method: &str,
    path: &str,
) -> (StatusCode, serde_json::Value) {
    let limits = WtDemoLimits::from_env();
    let status = if limits.require_mtls && !auth.peer_authenticated() {
        StatusCode::FORBIDDEN
    } else {
        StatusCode::OK
    };
    let body = if status == StatusCode::FORBIDDEN {
        serde_json::json!({ "error": "mTLS client certificate required" })
    } else {
        serde_json::json!({
            "method": method,
            "path": path,
        })
    };
    (status, body)
}

async fn echo_bidi(
    mut stream: h3_webtransport::stream::BidiStream<h3_quinn::BidiStream<Bytes>, Bytes>,
    limits: WtDemoLimits,
) {
    let mut buf = [0u8; DEMO_READ_CHUNK];
    let mut echoed = 0usize;
    loop {
        if echoed >= limits.max_echo_bytes {
            debug!(
                max = limits.max_echo_bytes,
                "demo webtransport bidi echo byte limit reached"
            );
            break;
        }
        let n = match tokio::time::timeout(limits.frame_timeout, tokio::io::AsyncReadExt::read(&mut stream, &mut buf)).await {
            Ok(Ok(0)) => break,
            Ok(Ok(n)) => n,
            Ok(Err(err)) => {
                debug!(%err, "webtransport bidi read ended");
                break;
            }
            Err(_) => {
                debug!("webtransport bidi read timed out");
                break;
            }
        };
        let chunk = n.min(limits.max_echo_bytes.saturating_sub(echoed));
        if chunk == 0 {
            break;
        }
        if tokio::time::timeout(
            limits.frame_timeout,
            tokio::io::AsyncWriteExt::write_all(&mut stream, &buf[..chunk]),
        )
        .await
        .is_err()
        {
            debug!("webtransport bidi write timed out");
            break;
        }
        echoed += chunk;
    }
    let _ = tokio::io::AsyncWriteExt::shutdown(&mut stream).await;
}

fn demo_requires_mtls_from_env() -> bool {
    match std::env::var("NAUTALID_WEBTRANSPORT_REQUIRE_MTLS")
        .ok()
        .map(|v| v.trim().to_ascii_lowercase())
        .as_deref()
    {
        Some("0") | Some("false") | Some("no") | Some("off") => false,
        Some("1") | Some("true") | Some("yes") | Some("on") => true,
        _ => crate::tls::client_auth_required_from_env(),
    }
}

fn demo_max_bidi_from_env() -> usize {
    std::env::var("NAUTALID_WEBTRANSPORT_DEMO_MAX_BIDI")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(DEFAULT_DEMO_MAX_BIDI)
        .max(1)
}

fn demo_max_echo_bytes_from_env() -> usize {
    std::env::var("NAUTALID_WEBTRANSPORT_DEMO_MAX_ECHO_BYTES")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(DEFAULT_DEMO_MAX_ECHO_BYTES)
        .max(1)
}

fn demo_max_datagram_bytes_from_env() -> usize {
    std::env::var("NAUTALID_WEBTRANSPORT_DEMO_MAX_DATAGRAM_BYTES")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(DEFAULT_DEMO_MAX_DATAGRAM_BYTES)
        .max(64)
}

fn demo_frame_timeout_secs_from_env() -> u64 {
    std::env::var("NAUTALID_WEBTRANSPORT_DEMO_FRAME_TIMEOUT_SECS")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(DEFAULT_DEMO_FRAME_TIMEOUT_SECS)
        .max(1)
}

#[cfg(test)]
mod tests {
    use super::*;
    use http::Uri;

    #[test]
    fn webtransport_disabled_by_default() {
        unsafe {
            std::env::remove_var("NAUTALID_WEBTRANSPORT");
        }
        assert!(!enabled_from_env());
    }

    #[test]
    fn default_path_is_wt() {
        unsafe {
            std::env::remove_var("NAUTALID_WEBTRANSPORT_PATH");
        }
        assert_eq!(path_from_env(), "/wt");
    }

    #[test]
    fn detects_webtransport_connect() {
        let mut req = Request::builder()
            .method(Method::CONNECT)
            .uri("/wt")
            .body(())
            .unwrap();
        req.extensions_mut().insert(Protocol::WEB_TRANSPORT);
        assert!(is_connect(&req));
    }

    #[test]
    fn rejects_non_connect() {
        let req = Request::builder()
            .method(Method::GET)
            .uri(Uri::from_static("/wt"))
            .body(())
            .unwrap();
        assert!(!is_connect(&req));
    }

    #[test]
    fn demo_limits_read_env() {
        unsafe {
            std::env::set_var("NAUTALID_WEBTRANSPORT_DEMO_MAX_BIDI", "4");
            std::env::set_var("NAUTALID_WEBTRANSPORT_DEMO_MAX_ECHO_BYTES", "8192");
            std::env::set_var("NAUTALID_WEBTRANSPORT_DEMO_MAX_DATAGRAM_BYTES", "512");
            std::env::set_var("NAUTALID_WEBTRANSPORT_DEMO_FRAME_TIMEOUT_SECS", "9");
        }
        let limits = WtDemoLimits::from_env();
        assert_eq!(limits.bidi.available_permits(), 4);
        assert_eq!(limits.max_echo_bytes, 8192);
        assert_eq!(limits.max_datagram_bytes, 512);
        assert_eq!(limits.frame_timeout, Duration::from_secs(9));
        unsafe {
            std::env::remove_var("NAUTALID_WEBTRANSPORT_DEMO_MAX_BIDI");
            std::env::remove_var("NAUTALID_WEBTRANSPORT_DEMO_MAX_ECHO_BYTES");
            std::env::remove_var("NAUTALID_WEBTRANSPORT_DEMO_MAX_DATAGRAM_BYTES");
            std::env::remove_var("NAUTALID_WEBTRANSPORT_DEMO_FRAME_TIMEOUT_SECS");
        }
    }

    #[test]
    fn demo_nested_http_response_requires_mtls_when_configured() {
        unsafe {
            std::env::set_var("NAUTALID_WEBTRANSPORT_REQUIRE_MTLS", "1");
        }
        let (status, body) = demo_nested_http_response(
            WtDemoSessionAuth::from_peer_cert(crate::tls::PeerClientCert(std::sync::Arc::new(
                Vec::new(),
            ))),
            "GET",
            "/nested",
        );
        assert_eq!(status, StatusCode::FORBIDDEN);
        assert!(body.get("method").is_none());
        unsafe {
            std::env::remove_var("NAUTALID_WEBTRANSPORT_REQUIRE_MTLS");
        }
    }
}