//! RFC 9114 transport adapter: Quinn (QUIC/TLS 1.3) + h3 (HTTP/3/QPACK).
//! Every request uses the SAME router, ACLs, auth, cache and response policies.
//! No full-body buffering or 0-RTT replay. RFC 9220 WebSocket Extended CONNECT is supported; WebTransport is not.
use crate::{
    metrics::Metrics,
    runtime::App,
    server::{LocalAddr, PeerAddr, ServerShutdown, Transport},
    vhost::Listener,
};
use anyhow::{ensure, Context as _, Result};
use axum::{body::Body, Router};
use bytes::{Buf, Bytes};
use h3::error::Code;
use http::{HeaderMap, HeaderValue, Method, Version};
use http_body::{Body as HttpBody, Frame, SizeHint};
use http_body_util::BodyExt;
use serde::{Deserialize, Serialize};
use std::{
    io,
    net::SocketAddr,
    pin::Pin,
    sync::{atomic::Ordering, Arc},
    task::{Context, Poll},
    time::Duration,
};
use tokio::{
    sync::{OwnedSemaphorePermit, Semaphore},
    task::JoinSet,
};
use tower::ServiceExt;
include!("http3/settings.rs");
pub fn crypto(
    tls: &Arc<rustls::ServerConfig>,
    policy: &crate::tls::ProtocolPolicy,
) -> Result<Arc<quinn::crypto::rustls::QuicServerConfig>> {
    ensure!(
        policy.allows_http3(),
        "HTTP/3 requires TLS 1.3 and a TLS 1.3 cipher suite"
    );
    let mut provider = rustls::crypto::ring::default_provider();
    if !policy.cipher_suites.is_empty() {
        provider.cipher_suites.retain(|suite| {
            let name = format!("{:?}", suite.suite());
            policy
                .cipher_suites
                .iter()
                .any(|selected| selected == &name)
        });
    }
    let mut quic = rustls::ServerConfig::builder_with_provider(Arc::new(provider))
        .with_protocol_versions(&[&rustls::version::TLS13])?
        .with_no_client_auth()
        .with_cert_resolver(tls.cert_resolver.clone());
    quic.alpn_protocols = vec![b"h3".to_vec()];
    quic.max_early_data_size = 0;
    Ok(Arc::new(quinn::crypto::rustls::QuicServerConfig::try_from(
        quic,
    )?))
}
pub struct Bound {
    endpoint: quinn::Endpoint,
    server_config: quinn::ServerConfig,
    address: SocketAddr,
    settings: Settings,
}
pub fn bind(listener: Listener, state: &App) -> Result<Bound> {
    let settings = listener
        .http3
        .context("HTTP/3 is disabled on this listener")?;
    settings.validate()?;
    ensure!(listener.tls, "HTTP/3 listener requires TLS");
    let crypto = state
        .config
        .load()
        .quic_tls
        .clone()
        .context("HTTP/3 certificate unavailable")?;
    let mut server_config = quinn::ServerConfig::with_crypto(crypto);
    server_config
        .transport_config(Arc::new(settings.transport()?))
        .migration(false)
        .max_incoming(settings.max_concurrent_connections as usize)
        .incoming_buffer_size(64 * 1024)
        .incoming_buffer_size_total(1024 * 1024);
    let endpoint = quinn::Endpoint::server(server_config.clone(), listener.address)
        .with_context(|| format!("cannot bind HTTP/3 UDP {}", listener.address))?;
    Ok(Bound {
        endpoint,
        server_config,
        address: listener.address,
        settings,
    })
}
/// Derive the advertised endpoint from ACTUALLY bound sockets, never saved intent
/// or a backend Alt-Svc header. A pending H3 enable is not advertised before restart.
pub fn advertise(
    headers: &mut HeaderMap,
    local: Option<SocketAddr>,
    tls: bool,
    active: &[Listener],
) {
    headers.remove("alt-svc");
    if !tls {
        return;
    }
    let Some(local) = local else {
        return;
    };
    if let Some(settings) = active
        .iter()
        .find(|binding| binding.matches(local, true))
        .and_then(|binding| binding.http3)
    {
        let value = format!("h3=\":{}\"; ma={}", local.port(), settings.alt_svc_max_age);
        if let Ok(value) = HeaderValue::from_str(&value) {
            headers.insert("alt-svc", value);
        }
    }
}
pub async fn serve(bound: Bound, router: Router, state: Arc<App>, grace: Duration) -> Result<()> {
    let slots = Arc::new(Semaphore::new(
        bound.settings.max_concurrent_connections as usize,
    ));
    let mut tasks = JoinSet::new();
    loop {
        tokio::select! {
            biased;
            _=state.shutdown.cancelled()=>break,
            _=tasks.join_next(),if !tasks.is_empty()=>{},
            incoming=bound.endpoint.accept()=>{
                let Some(incoming)=incoming else {break;};
                let peer=incoming.remote_address();
                if !incoming.remote_address_validated() {let _=incoming.retry();continue;}
                let Ok(permit)=slots.clone().try_acquire_owned() else {
                    state.metrics.http3_rejected_connections.fetch_add(1,Ordering::Relaxed);
                    state.metrics.rejected_connections.fetch_add(1,Ordering::Relaxed);
                    state.logs.push_protocol_error(
                        "http3_connection_limit_rejected", "HTTP/3 connection concurrency limit reached",
                        peer, bound.address, "quic", "3", 0, None,
                    );
                    incoming.refuse();continue;
                };
                let mut config=bound.server_config.clone();
                let Some(crypto)=state.config.load().quic_tls.clone() else {incoming.refuse();continue;};
                // Certificate changes apply to new handshakes without changing
                // active transport limits or regenerating the Retry token key.
                config.crypto=crypto;
                let connecting=match incoming.accept_with(Arc::new(config)) {
                    Ok(c)=>c,
                    Err(error)=>{
                        state.logs.push_protocol_error(
                            "http3_accept_error", error, peer, bound.address, "quic", "3", 0, None,
                        );
                        continue;
                    }
                };
                let router=router.clone();let state=state.clone();let address=bound.address;
                let limit=bound.settings.max_concurrent_bidistream as usize;
                tasks.spawn(async move {
                    let connection=tokio::select! {
                        _=state.shutdown.cancelled()=>return,
                        result=tokio::time::timeout(Duration::from_secs(10),connecting)=>match result {
                            Ok(Ok(connection))=>connection,
                            Ok(Err(error))=>{
                                state.metrics.tls_handshake_errors.fetch_add(1,Ordering::Relaxed);
                                state.logs.push_protocol_error(
                                    "http3_handshake_error", error, peer, address, "quic", "3", 0, None,
                                );
                                return;
                            },
                            Err(_)=>{
                                state.metrics.tls_handshake_errors.fetch_add(1,Ordering::Relaxed);
                                state.logs.push_protocol_error(
                                    "http3_handshake_timeout", "HTTP/3 QUIC/TLS handshake timeout",
                                    peer, address, "quic", "3", 0, None,
                                );
                                return;
                            },
                        }
                    };
                    // A wildcard without destination metadata must fail closed.
                    let local=match connection.local_ip() {
                        Some(ip)=>SocketAddr::new(ip,address.port()),
                        None if !address.ip().is_unspecified()=>address,
                        None=>{connection.close(0x102u32.into(),b"local destination unavailable");return;},
                    };
                    let tls_sni=connection_sni(&connection);
                    state.metrics.connections.fetch_add(1,Ordering::Relaxed);
                    state.metrics.active_connections.fetch_add(1,Ordering::Relaxed);
                    state.metrics.http3_connections.fetch_add(1,Ordering::Relaxed);
                    state.metrics.http3_active_connections.fetch_add(1,Ordering::Relaxed);
                    state.metrics.tls_connections.fetch_add(1,Ordering::Relaxed);
                    let _lease=Lease {_permit:permit,metrics:state.metrics.clone()};
                    if let Err(error)=connection_loop(connection,local,router,state.clone(),limit,grace,tls_sni.clone()).await {
                        log_protocol_error(
                            &state, "http3_connection_error", &error, peer, local, 0, None, None, tls_sni,
                        );
                        tracing::debug!(%error,"HTTP/3 connection ended");
                    }
                });
            }
        }
    }
    // Connection tasks send GOAWAY and drain; only close the endpoint after the
    // grace period, not immediately after a response FIN (which can lose data).
    let _ = tokio::time::timeout(grace, async { while tasks.join_next().await.is_some() {} }).await;
    tasks.abort_all();
    while tasks.join_next().await.is_some() {}
    bound.endpoint.close(0x100u32.into(), b"shutdown");
    let _ = tokio::time::timeout(Duration::from_secs(1), bound.endpoint.wait_idle()).await;
    Ok(())
}
async fn connection_loop(
    connection: quinn::Connection,
    local: SocketAddr,
    router: Router,
    state: Arc<App>,
    limit: usize,
    grace: Duration,
    tls_sni: Option<String>,
) -> Result<()> {
    let mut builder = h3::server::builder();
    builder.max_field_section_size(32 * 1024);
    // RFC 9220 remains an explicit experimental opt-in. Avoid advertising
    // Extended CONNECT on listeners that only use classic HTTP/1.1 WebSockets.
    builder.enable_extended_connect(state.config.load().websocket_rfc9220_on(local));
    let mut h3 = builder
        .build(h3::quinn::Connection::new(connection.clone()))
        .await?;
    let mut requests = JoinSet::new();
    let mut draining = false;
    let deadline = tokio::time::sleep(Duration::from_secs(86400));
    tokio::pin!(deadline);
    loop {
        tokio::select! {
            biased;
            _=state.shutdown.cancelled(),if !draining=>{
                draining=true;deadline.as_mut().reset(tokio::time::Instant::now()+grace);
                h3.shutdown(0).await?;
            },
            _=&mut deadline,if draining=>break,
            _=requests.join_next(),if !requests.is_empty()=>{},
            next=h3.accept()=>match next {
                Ok(Some(resolver))=>{
                    // Completed JoinSet entries are drained before accepting more.
                    if requests.len()>=limit {
                        log_protocol_error(
                            &state, "http3_stream_limit_rejected", "HTTP/3 concurrent request stream limit reached",
                            connection.remote_address(), local, 0, None, None, tls_sni.clone(),
                        );
                        drop(resolver);continue;
                    }
                    let router=router.clone();let state=state.clone();let peer=connection.remote_address();let tls_sni=tls_sni.clone();
                    requests.spawn(async move {
                        let resolved=tokio::time::timeout(state.header_timeout,resolver.resolve_request()).await;
                        let (mut request,stream)=match resolved {
                            Ok(Ok(request))=>request,
                            Ok(Err(error))=>{
                                if client_cancelled_request(&error) {
                                    // Record the cancellation with status=0 (no HTTP response)
                                    // for iOS navigation investigations, never synthetic 400.
                                    log_protocol_error(
                                        &state, "http3_peer_cancelled_before_headers",
                                        format!("phase=request_headers stream_status=cancelled error={error}"),
                                        peer, local, 0, None, None, tls_sni.clone(),
                                    );
                                } else {
                                    log_protocol_error(
                                        &state, "http3_request_error", error, peer, local, 400, None, None, tls_sni.clone(),
                                    );
                                }
                                return;
                            },
                            Err(_)=>{
                                log_protocol_error(
                                    &state, "http3_header_timeout", "HTTP/3 request header timeout",
                                    peer, local, 408, None, None, tls_sni.clone(),
                                );
                                return;
                            },
                        };
                        *request.version_mut()=Version::HTTP_3;
                        let protocol=request.extensions().get::<h3::ext::Protocol>().map(|protocol|protocol.as_str().to_owned());
                        let extended_connect=request.method()==Method::CONNECT && protocol.is_some();
                        let websocket=extended_connect && protocol.as_deref()==Some("websocket");
                        let (mut send,receive)=stream.split();
                        let stream_id=format!("{:?}",send.send_id());
                        let diagnostic_method=request.method().as_str().to_owned();
                        // Do not record query strings or cookies in transport diagnostics.
                        let diagnostic_path=request.uri().path().chars().take(160).collect::<String>();
                        let (diagnostic_authority, diagnostic_host)=request_identity(&request);
                        // GET / is a full document navigation (the original logo).
                        // Emit only query-free, bounded navigation metadata.
                        let root_navigation=diagnostic_method=="GET" && diagnostic_path=="/";
                        if root_navigation { log_root_navigation(
                            "started", peer, local, &tls_sni, &stream_id, "request_headers", 0, 0, false,
                        ); }
                        tracing::debug!(
                            peer=%peer, stream_id=%stream_id,
                            method=%diagnostic_method, path=%diagnostic_path,
                            "HTTP/3 request headers accepted"
                        );
                        if invalid_headers(request.headers()) {
                            let (authority, host) = request_identity(&request);
                            log_protocol_error(
                                &state, "http3_invalid_headers", "HTTP/3 request contains forbidden connection-specific headers",
                                peer, local, 400, authority, host, tls_sni.clone(),
                            );
                            send.stop_stream(Code::H3_MESSAGE_ERROR);return;
                        }
                        let expected=request.headers().get("content-length").and_then(|v|v.to_str().ok()).and_then(|v|v.parse::<u64>().ok());
                        let mut tunnel_receive=None;
                        let mut request=if websocket {
                            // RFC 9220 DATA belongs to the WebSocket byte stream, not
                            // to an HTTP request body. Keep the receive half for the
                            // transport bridge after the router accepts the tunnel.
                            tunnel_receive=Some(receive);
                            request.map(|_|Body::empty())
                        } else if extended_connect {
                            // Unsupported Extended CONNECT protocols are answered by
                            // the router (501). They are not interpreted as HTTP bodies.
                            drop(receive);
                            request.map(|_|Body::empty())
                        } else {
                            request.map(|_|Body::new(ReceiveBody {stream:receive,expected,received:0,data_done:false,done:false}))
                        };
                        request.extensions_mut().insert(PeerAddr(peer));request.extensions_mut().insert(LocalAddr(local));
                        request.extensions_mut().insert(Transport {tls:true});
                        request.extensions_mut().insert(ServerShutdown(state.shutdown.clone()));
                        let response=match router.oneshot(request).await {Ok(response)=>response,Err(never)=>match never {}};
                        let (mut parts,mut body)=response.into_parts();
                        let tunnel=parts.extensions.remove::<crate::proxy::websocket::H3Tunnel>();
                        crate::proxy::headers::strip_hop_by_hop(&mut parts.headers);
                        parts.version=Version::HTTP_3;
                        let response=http::Response::from_parts(parts,());
                        let response_status=response.status().as_u16();
                        let mut response_phase="response_headers";
                        let mut sent_body_bytes=0u64;
                        let sent=async {
                            tokio::time::timeout(state.body_idle_timeout,send.send_response(response)).await??;
                            response_phase="response_body";
                            if let Some(tunnel)=tunnel {
                                let Some(mut receive)=tunnel_receive.take() else {
                                    send.stop_stream(Code::H3_INTERNAL_ERROR);
                                    return Err(anyhow::anyhow!("HTTP/3 tunnel response without a WebSocket Extended CONNECT request"));
                                };
                                drop(body);
                                if let Err(error)=crate::proxy::websocket::run_h3(state.clone(),&mut send,&mut receive,tunnel).await {
                                    // run_h3 owns RFC 9220 tunnel shutdown/reset semantics.
                                    // Do not overwrite its H3_REQUEST_CANCELLED with a
                                    // second generic H3_INTERNAL_ERROR here.
                                    tracing::debug!(%error,"HTTP/3 WebSocket tunnel closed");
                                }
                                return Ok::<(),anyhow::Error>(());
                            }
                            if let Some(mut receive)=tunnel_receive.take() {
                                // The WebSocket bootstrap was rejected before becoming
                                // a tunnel. Stop accepting request DATA after sending the
                                // ordinary HTTP error response.
                                receive.stop_sending(Code::H3_REQUEST_CANCELLED);
                            }
                            while let Some(frame)=body.frame().await {
                                let frame=frame?;
                                match frame.into_data() {
                                    Ok(data)=>{if !data.is_empty() {
                                        let data_len=data.len() as u64;
                                        tokio::time::timeout(state.body_idle_timeout,send.send_data(data)).await??;
                                        sent_body_bytes=sent_body_bytes.saturating_add(data_len);
                                    }},
                                    Err(frame)=>{if let Ok(mut trailers)=frame.into_trailers() {
                                        crate::proxy::headers::strip_hop_by_hop(&mut trailers);
                                        tokio::time::timeout(state.body_idle_timeout,send.send_trailers(trailers)).await??;
                                    }},
                                }
                            }
                            response_phase="response_finish";
                            tokio::time::timeout(state.body_idle_timeout,send.finish()).await??;
                            Ok::<(),anyhow::Error>(())
                        }.await;
                        if let Err(error)=sent {
                            let cancelled=error.downcast_ref::<h3::error::StreamError>()
                                .is_some_and(client_cancelled_request);
                            let kind=if cancelled { "http3_peer_cancelled_after_headers" }
                                else { "http3_response_write_error" };
                            log_protocol_error(
                                &state, kind,
                                format!("phase={response_phase} stream_id={stream_id} method={diagnostic_method} path={diagnostic_path} response_status={response_status} body_bytes_sent={sent_body_bytes} error={error}"),
                                peer, local, 0,
                                diagnostic_authority, diagnostic_host, tls_sni.clone(),
                            );
                            if root_navigation { log_root_navigation(
                                "interrupted", peer, local, &tls_sni, &stream_id, response_phase,
                                response_status, sent_body_bytes, cancelled,
                            ); }
                            if !cancelled {
                                // Never mark a truncated upstream response as success.
                                // Reset only this response stream, not sibling requests.
                                send.stop_stream(Code::H3_INTERNAL_ERROR);
                            }
                        } else if root_navigation {
                            log_root_navigation("complete", peer, local, &tls_sni, &stream_id,
                                "response_finish", response_status, sent_body_bytes, false);
                        } else {
                            log_h3_response_complete(peer, &stream_id, response_status, sent_body_bytes);
                        }
                    });
                },
                Ok(None)=>break,
                Err(error)=>{
                    // The parser/QUIC transport can stop accepting after a peer
                    // refresh while unrelated response streams are still alive.
                    // Abort_all here reset those streams (including a new GET /).
                    // Stop accepting and allow the bounded drain below instead.
                    log_protocol_error(
                        &state, "http3_accept_error", &error,
                        connection.remote_address(), local, 0, None, None, tls_sni.clone(),
                    );
                    tracing::debug!(%error, "HTTP/3 accept stopped; draining in-flight streams");
                    break;
                },
            }
        }
    }
    let _ = tokio::time::timeout(grace, async {
        while requests.join_next().await.is_some() {}
    })
    .await;
    requests.abort_all();
    while requests.join_next().await.is_some() {}
    // Retain the transport until the client closes or the bounded drain expires.
    if draining {
        let _ = tokio::time::timeout(grace, connection.closed()).await;
    }
    Ok(())
}
include!("http3/identity.rs");
include!("http3/diagnostics.rs");
include!("http3/body.rs");

#[cfg(test)]
include!("http3/tests.rs");