// SPDX-License-Identifier: BUSL-1.1
// Copyright (c) 2026 vertexclique
// Licensed under the Business Source License 1.1.
// Change Date: 10 years after this version's release. Change License: Apache-2.0.

//! HTTP/3 (QUIC + h3) listener bolted onto the existing `DaemonHttp`
//! request/reply envelope path. Each accepted QUIC connection runs
//! an `h3::server::Connection`; each request stream is dispatched as
//! a `DaemonEvent` and the reply is written back through h3.
//!
//! TLS-1.3 is mandatory (no plaintext QUIC). The caller supplies a
//! PEM cert + key. ALPN advertises `h3` only - clients that don't
//! speak h3 see a TLS handshake reject, matching Node's behaviour.
//!
//! Off by default - the `http3` cargo feature gates the entire
//! coordinator. Without the feature, the host import returns
//! `LISTEN_ERR_NO_DAEMON` and the polyfill surfaces a typed error.

#![cfg(feature = "http3")]

use crate::daemon_http::{
    DaemonEvent, DaemonHttp, LISTEN_ERR_ADDR_IN_USE, LISTEN_ERR_IO, LISTEN_ERR_NO_DAEMON,
    PortClaim, ReplyEnvelope, ServerId,
};
use bytes::{Buf, Bytes};
use std::cell::RefCell;
use std::net::ToSocketAddrs;
use std::sync::Arc;

thread_local! {
    /// Stashed cert+key for the next `host_http3_listen` call on this
    /// thread. Set by `host_http3_listen_set_cert`; consumed by
    /// `host_http3_listen`. Two-step pattern keeps the wasm trampoline
    /// signature minimal - five mixed-i32-and-u32 args triggered a
    /// wasmtime trampoline anomaly we couldn't isolate.
    static PENDING_CERT: RefCell<Option<(String, String)>> = const { RefCell::new(None) };
}

pub fn stash_pending_cert(cert: String, key: String) {
    PENDING_CERT.with(|slot| {
        *slot.borrow_mut() = Some((cert, key));
    });
}

pub fn take_pending_cert() -> Option<(String, String)> {
    PENDING_CERT.with(|slot| slot.borrow_mut().take())
}

/// Bind a QUIC endpoint on `127.0.0.1:port` and spawn the h3 accept
/// loop. Events generated by incoming H3 requests dispatch under the
/// caller-supplied `server_id` - this is how QUIC piggybacks on a
/// JS-side handler chain that's already been wired by the
/// corresponding HTTP listener (`http.createServer().listen(port)`
/// happens *first* in the polyfill; H3 reuses the same server_id so
/// the handlers fire for both wire protocols).
///
/// Returns `server_id` on success (echoed back so JS gets a positive
/// number it can store), or a negative `LISTEN_ERR_*` on failure.
/// UDP bind collisions return `LISTEN_ERR_ADDR_IN_USE`.
pub fn bind_h3_listener(
    daemon: Arc<DaemonHttp>,
    server_id: ServerId,
    port: u16,
    cert_pem: &str,
    key_pem: &str,
) -> i32 {
    let Some(runtime) = daemon.runtime_handle() else {
        return LISTEN_ERR_NO_DAEMON;
    };
    if server_id <= 0 {
        // The JS side must hand us a real server_id allocated by
        // `__host_http_listen`. Without it, daemon-event dispatch has
        // nowhere to route the requests.
        return LISTEN_ERR_IO;
    }

    let server_config = match build_server_config(cert_pem, key_pem) {
        Ok(c) => c,
        Err(_) => return LISTEN_ERR_IO,
    };

    // UDP arbitration is independent from TCP - the HTTP listener
    // claimed the TCP socket on the same port; we claim the UDP one.
    // Multi-shard mode: only shard 0 binds the UDP socket; the
    // followers see the existing claim and skip the bind (the spawned
    // tokio task is shared via the daemon Arc).
    match daemon.try_claim_h3_port(port) {
        PortClaim::Owner(_) => { /* fall through to bind */ }
        PortClaim::Follower(_) => return server_id,
        PortClaim::Busy => return LISTEN_ERR_ADDR_IN_USE,
    }

    let bind_addr: std::net::SocketAddr = match format!("127.0.0.1:{port}").parse() {
        Ok(a) => a,
        Err(_) => {
            daemon.release_h3_port(port);
            return LISTEN_ERR_IO;
        }
    };

    let _enter = runtime.enter();
    // When the process is a forked cluster worker, switch to a
    // SO_REUSEPORT-bound UDP socket so sibling H3 listeners on the
    // same port co-bind successfully.
    let endpoint = if crate::daemon_cluster::cluster_mode_enabled() {
        let udp = match crate::daemon_cluster::build_udp_socket(bind_addr) {
            Ok(s) => s,
            Err(crate::daemon_cluster::ClusterBindError::AddrInUse) => {
                daemon.release_h3_port(port);
                return LISTEN_ERR_ADDR_IN_USE;
            }
            Err(crate::daemon_cluster::ClusterBindError::Io(_)) => {
                daemon.release_h3_port(port);
                return LISTEN_ERR_IO;
            }
        };
        let runtime_handle = quinn::default_runtime().expect("tokio runtime present");
        let endpoint_config = quinn::EndpointConfig::default();
        match quinn::Endpoint::new(endpoint_config, Some(server_config), udp, runtime_handle) {
            Ok(e) => e,
            Err(_) => {
                daemon.release_h3_port(port);
                return LISTEN_ERR_IO;
            }
        }
    } else {
        match quinn::Endpoint::server(server_config, bind_addr) {
            Ok(e) => e,
            Err(e) if e.kind() == std::io::ErrorKind::AddrInUse => {
                daemon.release_h3_port(port);
                return LISTEN_ERR_ADDR_IN_USE;
            }
            Err(_) => {
                daemon.release_h3_port(port);
                return LISTEN_ERR_IO;
            }
        }
    };

    let coord = Arc::clone(&daemon);
    runtime.spawn(serve_h3_endpoint(endpoint, server_id, coord));
    server_id
}

fn build_server_config(
    cert_pem: &str,
    key_pem: &str,
) -> Result<quinn::ServerConfig, Box<dyn std::error::Error + Send + Sync>> {
    use std::io::Cursor;
    let mut cert_rdr = Cursor::new(cert_pem.as_bytes());
    let cert_chain: Vec<rustls::pki_types::CertificateDer<'static>> =
        rustls_pemfile::certs(&mut cert_rdr).collect::<Result<Vec<_>, _>>()?;
    if cert_chain.is_empty() {
        return Err("no certificates in PEM".into());
    }
    let mut key_rdr = Cursor::new(key_pem.as_bytes());
    let key = rustls_pemfile::private_key(&mut key_rdr)?.ok_or("no private key in PEM")?;

    let mut tls_config = rustls::ServerConfig::builder_with_provider(Arc::new(
        rustls::crypto::ring::default_provider(),
    ))
    .with_protocol_versions(&[&rustls::version::TLS13])?
    .with_no_client_auth()
    .with_single_cert(cert_chain, key)?;
    // RFC 9114 §3.1: ALPN identifier is "h3".
    tls_config.alpn_protocols = vec![b"h3".to_vec()];

    let quic_tls = quinn::crypto::rustls::QuicServerConfig::try_from(tls_config)
        .map_err(|e| format!("rustls→quic config: {e}"))?;
    Ok(quinn::ServerConfig::with_crypto(Arc::new(quic_tls)))
}

async fn serve_h3_endpoint(
    endpoint: quinn::Endpoint,
    server_id: ServerId,
    daemon: Arc<DaemonHttp>,
) {
    while let Some(incoming) = endpoint.accept().await {
        let daemon = Arc::clone(&daemon);
        tokio::spawn(async move {
            let conn = match incoming.await {
                Ok(c) => c,
                Err(_) => return,
            };
            // Wrap the quinn connection in h3-quinn's adapter and run
            // the h3 server handshake.
            let h3_conn = h3_quinn::Connection::new(conn);
            let mut h3_server = match h3::server::Connection::new(h3_conn).await {
                Ok(s) => s,
                Err(_) => return,
            };
            // Accept request streams in a loop until the client closes.
            while let Ok(Some(resolver)) = h3_server.accept().await {
                let daemon = Arc::clone(&daemon);
                tokio::spawn(handle_h3_request(resolver, server_id, daemon));
            }
        });
    }
}

async fn handle_h3_request(
    resolver: h3::server::RequestResolver<h3_quinn::Connection, Bytes>,
    server_id: ServerId,
    daemon: Arc<DaemonHttp>,
) {
    let (req, mut stream) = match resolver.resolve_request().await {
        Ok(pair) => pair,
        Err(_) => return,
    };
    let (parts, _body) = req.into_parts();

    let method = parts.method.to_string();
    let url = parts
        .uri
        .path_and_query()
        .map(|pq| pq.to_string())
        .unwrap_or_else(|| "/".into());
    let headers: Vec<(String, String)> = parts
        .headers
        .iter()
        .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))
        .collect();

    // Pull the request body. h3 streams body frames through `recv_data`.
    const MAX_BODY: usize = 16 * 1024 * 1024;
    let mut body_bytes = Vec::new();
    while let Ok(Some(mut chunk)) = stream.recv_data().await {
        let to_copy = chunk.remaining();
        if body_bytes.len() + to_copy > MAX_BODY {
            // 413 - but in h3 we have to respond on the same stream.
            send_simple_response(&mut stream, 413, b"burn: body too large").await;
            return;
        }
        // Bytes::Buf - copy into our Vec via the chunk's data accessor.
        let bs = chunk.copy_to_bytes(to_copy);
        body_bytes.extend_from_slice(&bs);
    }

    let (reply_tx, reply_rx) = kovan_channel::bounded::<ReplyEnvelope>(1);
    let req_id = daemon.register_pending(reply_tx);
    let event = DaemonEvent {
        server_id,
        req_id,
        method,
        url,
        headers,
        body: body_bytes,
    };
    let Some(event_tx) = daemon.event_sender() else {
        send_simple_response(&mut stream, 500, b"burn: daemon not running\n").await;
        return;
    };
    event_tx.send_async(event).await;
    let reply = reply_rx.recv_async().await;

    let mut resp_builder = hyper::Response::builder().status(reply.status);
    for (name, value) in &reply.headers {
        resp_builder = resp_builder.header(name, value);
    }
    let resp = match resp_builder.body(()) {
        Ok(r) => r,
        Err(_) => {
            send_simple_response(&mut stream, 500, b"burn: response build failed").await;
            return;
        }
    };
    if stream.send_response(resp).await.is_err() {
        return;
    }
    if !reply.body.is_empty() {
        let _ = stream.send_data(Bytes::from(reply.body)).await;
    }
    let _ = stream.finish().await;
}

/// Outbound HTTP/3 request - synchronous wrapper around the quinn
/// client plus h3 client. Used by `quic.connect(...).request(...)`
/// from JS. Returns JSON `{"status":N,"headers":{...},"body_b64":"..."}`
/// on success or `__HOST_ERR__:<msg>` on failure. The runtime is
/// the same daemon tokio runtime - `block_on` on a sync host
/// import, matching the shape of `host_http_request`.
pub fn h3_request_sync(
    daemon: &Arc<DaemonHttp>,
    url: &str,
    method: &str,
    body: &[u8],
) -> Result<String, String> {
    let runtime = daemon
        .runtime_handle()
        .ok_or_else(|| "no daemon runtime".to_string())?;
    let url_owned = url.to_string();
    let method_owned = method.to_string();
    let body_owned = body.to_vec();
    runtime.block_on(async move { h3_request_inner(&url_owned, &method_owned, &body_owned).await })
}

async fn h3_request_inner(url: &str, method: &str, body: &[u8]) -> Result<String, String> {
    use base64::Engine;
    use hyper::Uri;

    let uri: Uri = url.parse().map_err(|e| format!("bad url: {e}"))?;
    let host = uri.host().ok_or("url missing host")?.to_string();
    let port = uri.port_u16().unwrap_or(443);

    // Resolve the host via a sync getaddrinfo since quinn accepts a
    // SocketAddr. For loopback IPs this is fast; remote hosts go
    // through the OS resolver.
    let mut addrs_iter = (host.as_str(), port)
        .to_socket_addrs()
        .map_err(|e| format!("dns: {e}"))?;
    let server_addr = addrs_iter.next().ok_or("dns: no addresses")?;

    // Client TLS - accept any cert per Manifold::net (which already
    // gates the outbound). The client equivalent of
    // `webpki_roots` is the right default but tests with self-signed
    // certs won't validate; for now mirror what fetch does.
    let mut tls = rustls::ClientConfig::builder_with_provider(Arc::new(
        rustls::crypto::ring::default_provider(),
    ))
    .with_protocol_versions(&[&rustls::version::TLS13])
    .map_err(|e| format!("tls config: {e}"))?
    .dangerous()
    .with_custom_certificate_verifier(Arc::new(NoVerifyClient))
    .with_no_client_auth();
    tls.alpn_protocols = vec![b"h3".to_vec()];

    let quic_client = quinn::crypto::rustls::QuicClientConfig::try_from(tls)
        .map_err(|e| format!("quic client cfg: {e}"))?;
    let client_cfg = quinn::ClientConfig::new(Arc::new(quic_client));

    let mut endpoint = quinn::Endpoint::client("0.0.0.0:0".parse().unwrap())
        .map_err(|e| format!("client endpoint: {e}"))?;
    endpoint.set_default_client_config(client_cfg);

    let conn = endpoint
        .connect(server_addr, &host)
        .map_err(|e| format!("connect: {e}"))?
        .await
        .map_err(|e| format!("handshake: {e}"))?;
    let h3_conn = h3_quinn::Connection::new(conn);
    let (mut driver, mut send_request) = h3::client::new(h3_conn)
        .await
        .map_err(|e| format!("h3 client: {e}"))?;
    let drive_task = tokio::spawn(async move {
        let _ = std::future::poll_fn(|cx| driver.poll_close(cx)).await;
    });

    let req = hyper::Request::builder()
        .method(method)
        .uri(url)
        .body(())
        .map_err(|e| format!("req build: {e}"))?;
    let mut stream = send_request
        .send_request(req)
        .await
        .map_err(|e| format!("send: {e}"))?;
    if !body.is_empty() {
        stream
            .send_data(Bytes::copy_from_slice(body))
            .await
            .map_err(|e| format!("send body: {e}"))?;
    }
    stream.finish().await.map_err(|e| format!("finish: {e}"))?;

    let resp = stream
        .recv_response()
        .await
        .map_err(|e| format!("recv resp: {e}"))?;
    let status = resp.status().as_u16();
    let mut hmap = std::collections::BTreeMap::<String, String>::new();
    for (k, v) in resp.headers().iter() {
        hmap.insert(k.to_string(), v.to_str().unwrap_or("").to_string());
    }

    let mut body_bytes: Vec<u8> = Vec::new();
    while let Some(mut chunk) = stream
        .recv_data()
        .await
        .map_err(|e| format!("recv data: {e}"))?
    {
        let n = chunk.remaining();
        let bs = chunk.copy_to_bytes(n);
        body_bytes.extend_from_slice(&bs);
    }
    drive_task.abort();
    let body_b64 = base64::engine::general_purpose::STANDARD.encode(&body_bytes);

    Ok(serde_json::json!({
        "status": status,
        "headers": hmap,
        "body_b64": body_b64,
    })
    .to_string())
}

#[derive(Debug)]
struct NoVerifyClient;
impl rustls::client::danger::ServerCertVerifier for NoVerifyClient {
    fn verify_server_cert(
        &self,
        _: &rustls::pki_types::CertificateDer<'_>,
        _: &[rustls::pki_types::CertificateDer<'_>],
        _: &rustls::pki_types::ServerName<'_>,
        _: &[u8],
        _: rustls::pki_types::UnixTime,
    ) -> Result<rustls::client::danger::ServerCertVerified, rustls::Error> {
        Ok(rustls::client::danger::ServerCertVerified::assertion())
    }
    fn verify_tls12_signature(
        &self,
        _: &[u8],
        _: &rustls::pki_types::CertificateDer<'_>,
        _: &rustls::DigitallySignedStruct,
    ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
        Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
    }
    fn verify_tls13_signature(
        &self,
        _: &[u8],
        _: &rustls::pki_types::CertificateDer<'_>,
        _: &rustls::DigitallySignedStruct,
    ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
        Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
    }
    fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
        vec![
            rustls::SignatureScheme::ECDSA_NISTP256_SHA256,
            rustls::SignatureScheme::ECDSA_NISTP384_SHA384,
            rustls::SignatureScheme::ED25519,
            rustls::SignatureScheme::RSA_PSS_SHA256,
            rustls::SignatureScheme::RSA_PSS_SHA384,
            rustls::SignatureScheme::RSA_PSS_SHA512,
        ]
    }
}

async fn send_simple_response(
    stream: &mut h3::server::RequestStream<h3_quinn::BidiStream<Bytes>, Bytes>,
    status: u16,
    body: &[u8],
) {
    let resp = match hyper::Response::builder().status(status).body(()) {
        Ok(r) => r,
        Err(_) => return,
    };
    if stream.send_response(resp).await.is_err() {
        return;
    }
    let _ = stream.send_data(Bytes::copy_from_slice(body)).await;
    let _ = stream.finish().await;
}
