//! QUIC server accept loop.

use std::{
    collections::{HashMap, VecDeque},
    future::Future,
    net::{IpAddr, SocketAddr},
    panic::AssertUnwindSafe,
    sync::{Arc, Mutex},
    time::Duration,
};

use futures::FutureExt as _;
use s2n_quic::{
    Connection,
    Server,
    application,
    provider::endpoint_limits,
    stream::BidirectionalStream,
};
use tokio::{
    sync::{OwnedSemaphorePermit, Semaphore, oneshot},
    task::JoinSet,
};
use tokio_util::sync::CancellationToken;

use crate::{
    config::QUIC_HANDSHAKE_TIMEOUT,
    context::PeerCtx,
    library::prime_library_manifests,
    quic_runtime::{quic_congestion_controller, quic_server_limits, tracked_quic_io},
    scoped_blocking::scoped_blocking,
    services::{
        advertise::{close_mdns_advertiser, monitor_mdns_events, start_mdns_advertiser},
        stream::handle_peer_stream,
    },
    tls,
};

/// Limits unauthenticated handshake memory before QUIC admission completes.
const MAX_INFLIGHT_HANDSHAKES: usize = 64;
/// One unverified datagram source may start only this many handshakes during a
/// full handshake-timeout window, before application connection leases exist.
const MAX_INFLIGHT_HANDSHAKE_ATTEMPTS_PER_ORIGIN: usize = 8;
/// Bounds the unauthenticated source-rate ledger itself under spoofed-IP churn.
const MAX_TRACKED_HANDSHAKE_ORIGINS: usize = 256;
/// Limits established connection scopes owned by the application accept loop.
const MAX_ESTABLISHED_CONNECTIONS: usize = 64;
/// Mirrors the transport stream limit and bounds application stream futures.
const MAX_CONTROL_STREAM_TASKS: usize = 32;
/// Server-wide cap acquired before any length-delimited decoder is allocated.
const MAX_GLOBAL_CONTROL_STREAM_TASKS: usize = 16;
/// Long-lived transfers move from the decoder pool to this smaller pool so
/// saturated bulk egress cannot consume every control-plane permit.
const MAX_GLOBAL_BULK_TRANSFER_TASKS: usize = 48;
/// Native archive extraction is substantially more expensive than ordinary
/// catalog egress, so keep it behind a separate small global budget.
const MAX_GLOBAL_STREAM_INSTALL_TASKS: usize = 2;
/// One observed host may retain only this many established connection scopes.
const MAX_ESTABLISHED_CONNECTIONS_PER_ORIGIN: usize = 8;
/// Bounds decoded and undecoded control work across every connection from one IP.
const MAX_CONTROL_STREAM_TASKS_PER_ORIGIN: usize = 8;
/// Preserves bulk capacity for other LAN origins while allowing two full
/// four-stream ordinary download windows from one host.
const MAX_BULK_TRANSFER_TASKS_PER_ORIGIN: usize = 8;
/// A single origin cannot occupy both native Stream Install provider slots.
const MAX_STREAM_INSTALL_TASKS_PER_ORIGIN: usize = 1;
/// Application idle bound for an established connection with no active
/// request streams. This is independent of transport keepalive traffic.
const CONNECTION_NO_STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(10);

#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub(super) struct ObservedOrigin(IpAddr);

impl ObservedOrigin {
    fn from_remote_addr(remote_addr: Option<SocketAddr>) -> Option<Self> {
        remote_addr.map(|addr| Self(canonical_origin_ip(addr.ip())))
    }

    pub(super) const fn ip(self) -> IpAddr {
        self.0
    }
}

fn canonical_origin_ip(ip: IpAddr) -> IpAddr {
    match ip {
        IpAddr::V4(ip) => IpAddr::V4(ip),
        IpAddr::V6(ip) => ip.to_ipv4_mapped().map_or(IpAddr::V6(ip), IpAddr::V4),
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum OriginAdmissionClass {
    Connection,
    Control,
    Bulk,
    StreamInstall,
}

impl OriginAdmissionClass {
    const fn limit(self) -> usize {
        match self {
            Self::Connection => MAX_ESTABLISHED_CONNECTIONS_PER_ORIGIN,
            Self::Control => MAX_CONTROL_STREAM_TASKS_PER_ORIGIN,
            Self::Bulk => MAX_BULK_TRANSFER_TASKS_PER_ORIGIN,
            Self::StreamInstall => MAX_STREAM_INSTALL_TASKS_PER_ORIGIN,
        }
    }
}

#[derive(Default)]
struct OriginUsage {
    connections: usize,
    controls: usize,
    bulk: usize,
    stream_installs: usize,
}

impl OriginUsage {
    const fn count(&self, class: OriginAdmissionClass) -> usize {
        match class {
            OriginAdmissionClass::Connection => self.connections,
            OriginAdmissionClass::Control => self.controls,
            OriginAdmissionClass::Bulk => self.bulk,
            OriginAdmissionClass::StreamInstall => self.stream_installs,
        }
    }

    fn increment(&mut self, class: OriginAdmissionClass) {
        let count = self.count_mut(class);
        *count += 1;
    }

    fn decrement(&mut self, class: OriginAdmissionClass) -> bool {
        let count = self.count_mut(class);
        if *count == 0 {
            return false;
        }
        *count -= 1;
        true
    }

    const fn is_empty(&self) -> bool {
        self.connections == 0 && self.controls == 0 && self.bulk == 0 && self.stream_installs == 0
    }

    fn count_mut(&mut self, class: OriginAdmissionClass) -> &mut usize {
        match class {
            OriginAdmissionClass::Connection => &mut self.connections,
            OriginAdmissionClass::Control => &mut self.controls,
            OriginAdmissionClass::Bulk => &mut self.bulk,
            OriginAdmissionClass::StreamInstall => &mut self.stream_installs,
        }
    }
}

#[derive(Default)]
struct OriginAdmission {
    active: Mutex<HashMap<ObservedOrigin, OriginUsage>>,
}

impl OriginAdmission {
    fn try_acquire(
        self: &Arc<Self>,
        origin: ObservedOrigin,
        class: OriginAdmissionClass,
    ) -> Option<OriginLease> {
        let mut active = self
            .active
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let usage = active.entry(origin).or_default();
        if usage.count(class) >= class.limit() {
            return None;
        }
        usage.increment(class);
        drop(active);

        Some(OriginLease {
            admission: Arc::clone(self),
            origin,
            class,
        })
    }

    fn release(&self, origin: ObservedOrigin, class: OriginAdmissionClass) {
        let mut active = self
            .active
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let Some(usage) = active.get_mut(&origin) else {
            log::error!(
                "Origin admission lease for {:?} had no active entry",
                origin.ip()
            );
            return;
        };
        if !usage.decrement(class) {
            log::error!(
                "Origin admission lease for {:?} had no active {class:?} count",
                origin.ip()
            );
            return;
        }
        if usage.is_empty() {
            active.remove(&origin);
        }
    }

    #[cfg(test)]
    fn active_origin_count(&self) -> usize {
        self.active
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .len()
    }
}

struct OriginLease {
    admission: Arc<OriginAdmission>,
    origin: ObservedOrigin,
    class: OriginAdmissionClass,
}

impl Drop for OriginLease {
    fn drop(&mut self) {
        self.admission.release(self.origin, self.class);
    }
}

pub(super) struct ControlAdmission {
    _origin: OriginLease,
    _global: OwnedSemaphorePermit,
}

pub(super) struct BulkAdmission {
    _origin: OriginLease,
    _global: OwnedSemaphorePermit,
}

pub(super) struct StreamInstallAdmission {
    _origin: OriginLease,
    global: Option<OwnedSemaphorePermit>,
}

impl StreamInstallAdmission {
    pub(super) fn take_global_permit(&mut self) -> OwnedSemaphorePermit {
        self.global
            .take()
            .expect("Stream Install admission must own its global permit")
    }
}

pub(super) struct ServerAdmission {
    origins: Arc<OriginAdmission>,
    control: Arc<Semaphore>,
    bulk: Arc<Semaphore>,
    stream_install: Arc<Semaphore>,
}

impl ServerAdmission {
    fn new() -> Self {
        Self {
            origins: Arc::new(OriginAdmission::default()),
            control: Arc::new(Semaphore::new(MAX_GLOBAL_CONTROL_STREAM_TASKS)),
            bulk: Arc::new(Semaphore::new(MAX_GLOBAL_BULK_TRANSFER_TASKS)),
            stream_install: Arc::new(Semaphore::new(MAX_GLOBAL_STREAM_INSTALL_TASKS)),
        }
    }

    fn try_acquire_connection(&self, origin: ObservedOrigin) -> Option<OriginLease> {
        self.origins
            .try_acquire(origin, OriginAdmissionClass::Connection)
    }

    pub(super) fn try_acquire_control(&self, origin: ObservedOrigin) -> Option<ControlAdmission> {
        let origin = self
            .origins
            .try_acquire(origin, OriginAdmissionClass::Control)?;
        let global = Arc::clone(&self.control).try_acquire_owned().ok()?;
        Some(ControlAdmission {
            _origin: origin,
            _global: global,
        })
    }

    pub(super) fn try_acquire_bulk(&self, origin: ObservedOrigin) -> Option<BulkAdmission> {
        let origin = self
            .origins
            .try_acquire(origin, OriginAdmissionClass::Bulk)?;
        let global = Arc::clone(&self.bulk).try_acquire_owned().ok()?;
        Some(BulkAdmission {
            _origin: origin,
            _global: global,
        })
    }

    pub(super) fn try_acquire_stream_install(
        &self,
        origin: ObservedOrigin,
    ) -> Option<StreamInstallAdmission> {
        let origin = self
            .origins
            .try_acquire(origin, OriginAdmissionClass::StreamInstall)?;
        let global = Arc::clone(&self.stream_install).try_acquire_owned().ok()?;
        Some(StreamInstallAdmission {
            _origin: origin,
            global: Some(global),
        })
    }
}

struct BoundedEndpointLimits {
    inner: endpoint_limits::Default,
    handshakes: HandshakeOriginLimiter,
}

#[derive(Default)]
struct HandshakeOriginLimiter {
    attempts: HashMap<IpAddr, VecDeque<Duration>>,
}

impl HandshakeOriginLimiter {
    fn admit(&mut self, origin: IpAddr, now: Duration) -> bool {
        for attempts in self.attempts.values_mut() {
            while attempts
                .front()
                .is_some_and(|started| now.saturating_sub(*started) >= QUIC_HANDSHAKE_TIMEOUT)
            {
                attempts.pop_front();
            }
        }
        self.attempts.retain(|_, attempts| !attempts.is_empty());

        if !self.attempts.contains_key(&origin)
            && self.attempts.len() >= MAX_TRACKED_HANDSHAKE_ORIGINS
        {
            return false;
        }
        let attempts = self.attempts.entry(origin).or_default();
        if attempts.len() >= MAX_INFLIGHT_HANDSHAKE_ATTEMPTS_PER_ORIGIN {
            return false;
        }
        attempts.push_back(now);
        true
    }
}

impl endpoint_limits::Limiter for BoundedEndpointLimits {
    fn on_connection_attempt(
        &mut self,
        info: &endpoint_limits::ConnectionAttempt<'_>,
    ) -> endpoint_limits::Outcome {
        if !endpoint_connection_capacity_available(info.connection_count) {
            return endpoint_limits::Outcome::close();
        }
        let default_outcome =
            endpoint_limits::Limiter::on_connection_attempt(&mut self.inner, info);
        if default_outcome != endpoint_limits::Outcome::allow() {
            return default_outcome;
        }
        let remote_addr = SocketAddr::from(&info.remote_address);
        let origin = canonical_origin_ip(remote_addr.ip());
        if !self
            .handshakes
            .admit(origin, info.timestamp.duration_since_start())
        {
            // A valid Retry token bypasses s2n's connection-attempt callback on
            // the next Initial, so origin exhaustion must not mint one.
            return endpoint_limits::Outcome::drop();
        }
        default_outcome
    }
}

const fn endpoint_connection_capacity_available(connection_count: usize) -> bool {
    connection_count < MAX_ESTABLISHED_CONNECTIONS
}

fn bounded_endpoint_limits() -> eyre::Result<BoundedEndpointLimits> {
    Ok(BoundedEndpointLimits {
        inner: endpoint_limits::Default::builder()
            .with_inflight_handshake_limit(MAX_INFLIGHT_HANDSHAKES)?
            .build()?,
        handshakes: HandshakeOriginLimiter::default(),
    })
}

/// Runs the QUIC server and mDNS advertiser.
pub async fn run_server_component(
    addr: SocketAddr,
    ctx: PeerCtx,
    ready: oneshot::Sender<SocketAddr>,
) -> eyre::Result<()> {
    // Manifest bodies may be disk-backed on their first access. Resolve only
    // the currently publishable local set before opening the public endpoint;
    // this bounds retained manifest memory by actual local availability.
    let publication = {
        let library = ctx.local_library.read().await;
        library.publication(ctx.catalog.catalog())
    };
    let catalog = Arc::clone(&ctx.catalog);
    scoped_blocking(move || prime_library_manifests(&publication.game_ids, &catalog))?;

    let (io, endpoint_control) = tracked_quic_io(addr)?;

    let server = Server::builder()
        .with_tls(tls::server_provider(&ctx.peer_identity)?)?
        .with_io(io)?
        .with_endpoint_limits(bounded_endpoint_limits()?)?
        .with_limits(quic_server_limits()?)?
        .with_congestion_controller(quic_congestion_controller())?
        .start()?;
    let endpoint_task = endpoint_control.take_started()?;

    run_body_with_cleanup(
        run_server_body(server, ctx, ready),
        endpoint_task.shutdown_and_join(),
    )
    .await
}

async fn run_server_body(
    mut server: Server,
    ctx: PeerCtx,
    ready: oneshot::Sender<SocketAddr>,
) -> eyre::Result<()> {
    let server_addr = server.local_addr()?;
    log::info!("Peer server listening on {server_addr}");

    let mdns_advertiser = start_mdns_advertiser(&ctx, server_addr).await?;
    let mdns_monitor = mdns_advertiser.monitor.clone();
    let server_children_shutdown = ctx.shutdown.child_token();
    let mut mdns_tasks = JoinSet::new();
    let mdns_shutdown = server_children_shutdown.clone();
    mdns_tasks.spawn(async move {
        monitor_mdns_events(mdns_monitor, mdns_shutdown).await;
        Ok(())
    });
    let mut connection_tasks = JoinSet::new();
    let admission = Arc::new(ServerAdmission::new());
    let ready_addr =
        (*ctx.local_peer_addr.read().await).unwrap_or_else(|| direct_connect_addr(server_addr));

    ready
        .send(ready_addr)
        .map_err(|_| eyre::eyre!("network manager stopped before server readiness"))?;

    let server_result = match AssertUnwindSafe(async {
        loop {
            tokio::select! {
                biased;
                () = ctx.shutdown.cancelled() => break Ok(()),
                result = mdns_tasks.join_next(), if !mdns_tasks.is_empty() => {
                    log_joined_child_result(
                        "mDNS monitor",
                        result.expect("non-empty child set"),
                    );
                    log::warn!("mDNS monitor ended while the QUIC server is still running");
                }
                result = connection_tasks.join_next(), if !connection_tasks.is_empty() => {
                    log_joined_child_result(
                        "peer connection",
                        result.expect("non-empty child set"),
                    );
                }
                connection = server.accept() => {
                    let Some(connection) = connection else {
                        break Err(eyre::eyre!("QUIC server accept loop ended unexpectedly"));
                    };

                    let remote_addr = match connection.remote_addr() {
                        Ok(remote_addr) => remote_addr,
                        Err(error) => {
                            log::warn!("Closing peer connection without an observed remote address: {error}");
                            connection.close(application::Error::UNKNOWN);
                            continue;
                        }
                    };
                    let Some(origin) = ObservedOrigin::from_remote_addr(Some(remote_addr)) else {
                        log::warn!("Closing peer connection without an observed origin");
                        connection.close(application::Error::UNKNOWN);
                        continue;
                    };

                    if !has_child_capacity(connection_tasks.len(), MAX_ESTABLISHED_CONNECTIONS) {
                        log::warn!(
                            "Closing excess peer connection from {remote_addr} at application limit {MAX_ESTABLISHED_CONNECTIONS}",
                        );
                        connection.close(application::Error::UNKNOWN);
                        continue;
                    }
                    let Some(connection_admission) = admission.try_acquire_connection(origin) else {
                        log::warn!(
                            "Closing excess peer connection from {remote_addr} at per-origin limit {MAX_ESTABLISHED_CONNECTIONS_PER_ORIGIN}"
                        );
                        connection.close(application::Error::UNKNOWN);
                        continue;
                    };

                    connection_tasks.spawn(handle_peer_connection(
                        connection,
                        remote_addr,
                        origin,
                        connection_admission,
                        ctx.clone(),
                        server_children_shutdown.clone(),
                        Arc::clone(&admission),
                    ));
                }
            }
        }
    })
    .catch_unwind()
    .await
    {
        Ok(result) => result,
        Err(payload) => Err(eyre::eyre!(
            "QUIC server loop panicked: {}",
            panic_payload_to_string(payload.as_ref())
        )),
    };

    // Stop future work before dropping the accept owner. Connection children
    // cooperatively observe this token and close their stream scopes before
    // they return.
    server_children_shutdown.cancel();
    drop(server);
    drain_joined_child_tasks(&mut connection_tasks, "peer connection").await;
    drain_joined_child_tasks(&mut mdns_tasks, "mDNS monitor").await;
    let mdns_close_result = close_mdns_advertiser(mdns_advertiser);
    combine_server_results(server_result, mdns_close_result, Ok(()))
}

async fn run_body_with_cleanup<Body, Cleanup>(body: Body, cleanup: Cleanup) -> eyre::Result<()>
where
    Body: Future<Output = eyre::Result<()>>,
    Cleanup: Future<Output = eyre::Result<()>>,
{
    let body_result = match AssertUnwindSafe(body).catch_unwind().await {
        Ok(result) => result,
        Err(payload) => Err(eyre::eyre!(
            "QUIC server body panicked: {}",
            panic_payload_to_string(payload.as_ref())
        )),
    };

    // This is the unconditional endpoint-owner epilogue. Even an unexpected
    // panic during setup, serving, or descendant cleanup cannot skip the join.
    let endpoint_result = cleanup.await;
    combine_server_results(body_result, Ok(()), endpoint_result)
}

fn combine_server_results(
    server: eyre::Result<()>,
    mdns: eyre::Result<()>,
    endpoint: eyre::Result<()>,
) -> eyre::Result<()> {
    let mut errors = Vec::new();
    if let Err(error) = server {
        errors.push(format!("QUIC server failed: {error:#}"));
    }
    if let Err(error) = mdns {
        errors.push(format!("mDNS advertiser shutdown failed: {error:#}"));
    }
    if let Err(error) = endpoint {
        errors.push(format!("QUIC endpoint shutdown failed: {error:#}"));
    }

    if errors.is_empty() {
        Ok(())
    } else {
        Err(eyre::eyre!(errors.join("; ")))
    }
}

fn panic_payload_to_string(payload: &(dyn std::any::Any + Send)) -> String {
    if let Some(message) = payload.downcast_ref::<&'static str>() {
        return (*message).to_string();
    }
    if let Some(message) = payload.downcast_ref::<String>() {
        return message.clone();
    }
    "unknown panic payload".to_string()
}

fn direct_connect_addr(server_addr: SocketAddr) -> SocketAddr {
    if server_addr.ip().is_unspecified() {
        return SocketAddr::from(([127, 0, 0, 1], server_addr.port()));
    }
    server_addr
}

async fn handle_peer_connection(
    mut connection: Connection,
    remote_addr: SocketAddr,
    origin: ObservedOrigin,
    _connection_admission: OriginLease,
    ctx: PeerCtx,
    server_shutdown: CancellationToken,
    admission: Arc<ServerAdmission>,
) -> eyre::Result<()> {
    log::info!("{remote_addr} peer connected");

    let connection_shutdown = server_shutdown.child_token();
    let mut stream_tasks = JoinSet::new();
    let connection_result = match AssertUnwindSafe(async {
        loop {
            tokio::select! {
                biased;
                () = connection_shutdown.cancelled() => break Ok(()),
                result = stream_tasks.join_next(), if !stream_tasks.is_empty() => {
                    log_joined_child_result(
                        &format!("{remote_addr} peer stream"),
                        result.expect("non-empty child set"),
                    );
                }
                () = tokio::time::sleep(CONNECTION_NO_STREAM_IDLE_TIMEOUT),
                    if stream_tasks.is_empty() => {
                    log::debug!(
                        "Closing idle peer connection from {remote_addr} after {CONNECTION_NO_STREAM_IDLE_TIMEOUT:?}"
                    );
                    break Ok(());
                }
                stream = connection.accept_bidirectional_stream() => {
                    match stream {
                        Ok(Some(mut stream)) => {
                            if !has_child_capacity(stream_tasks.len(), MAX_CONTROL_STREAM_TASKS) {
                                let _ = stream.stop_sending(application::Error::UNKNOWN);
                                let _ = stream.reset(application::Error::UNKNOWN);
                                continue;
                            }
                            let Some(control_admission) = admission.try_acquire_control(origin) else {
                                let _ = stream.stop_sending(application::Error::UNKNOWN);
                                let _ = stream.reset(application::Error::UNKNOWN);
                                continue;
                            };
                            let stream_ctx = ctx.clone();
                            let stream_shutdown = connection_shutdown.child_token();
                            stream_tasks.spawn(handle_admitted_peer_stream(
                                stream,
                                stream_ctx,
                                remote_addr,
                                origin,
                                stream_shutdown,
                                control_admission,
                                Arc::clone(&admission),
                            ));
                        }
                        Ok(None) => break Ok(()),
                        Err(error) => break Err(error.into()),
                    }
                }
            }
        }
    })
    .catch_unwind()
    .await
    {
        Ok(result) => result,
        Err(payload) => Err(eyre::eyre!(
            "{remote_addr} peer connection scope panicked: {}",
            panic_payload_to_string(payload.as_ref())
        )),
    };

    // Cancel the connection-local scope before closing its QUIC owner. Every
    // accepted stream and outbound transfer derives from this token, so child
    // futures can settle without waiting for process-wide shutdown. Closing
    // the connection also wakes any transport operation already in progress.
    connection_shutdown.cancel();
    connection.close(0u32.into());
    drop(connection);
    let stream_label = format!("{remote_addr} peer stream");
    drain_joined_child_tasks(&mut stream_tasks, &stream_label).await;

    log::info!("{remote_addr} peer disconnected");
    connection_result
}

async fn handle_admitted_peer_stream(
    stream: BidirectionalStream,
    ctx: PeerCtx,
    remote_addr: SocketAddr,
    origin: ObservedOrigin,
    stream_shutdown: CancellationToken,
    control_admission: ControlAdmission,
    admission: Arc<ServerAdmission>,
) -> eyre::Result<()> {
    handle_peer_stream(
        stream,
        ctx,
        remote_addr,
        origin,
        stream_shutdown,
        control_admission,
        admission,
    )
    .await
}

const fn has_child_capacity(active: usize, limit: usize) -> bool {
    active < limit
}

fn log_child_result(label: &str, result: eyre::Result<()>) {
    if let Err(error) = result {
        log::error!("{label} error: {error}");
    }
}

fn log_joined_child_result(label: &str, result: Result<eyre::Result<()>, tokio::task::JoinError>) {
    match result {
        Ok(result) => log_child_result(label, result),
        Err(error) => log::error!("{label} task failed: {error}"),
    }
}

async fn drain_joined_child_tasks(children: &mut JoinSet<eyre::Result<()>>, label: &str) {
    while let Some(result) = children.join_next().await {
        log_joined_child_result(label, result);
    }
}

#[cfg(test)]
mod tests {
    use std::{
        net::{IpAddr, Ipv6Addr, SocketAddr},
        panic::{AssertUnwindSafe, catch_unwind},
        sync::{
            Arc,
            atomic::{AtomicUsize, Ordering},
        },
        time::Duration,
    };

    use tokio::{
        sync::{Semaphore, mpsc},
        task::JoinSet,
    };
    use tokio_util::sync::CancellationToken;

    use super::{
        CONNECTION_NO_STREAM_IDLE_TIMEOUT,
        HandshakeOriginLimiter,
        MAX_BULK_TRANSFER_TASKS_PER_ORIGIN,
        MAX_CONTROL_STREAM_TASKS,
        MAX_CONTROL_STREAM_TASKS_PER_ORIGIN,
        MAX_ESTABLISHED_CONNECTIONS,
        MAX_ESTABLISHED_CONNECTIONS_PER_ORIGIN,
        MAX_GLOBAL_BULK_TRANSFER_TASKS,
        MAX_GLOBAL_CONTROL_STREAM_TASKS,
        MAX_GLOBAL_STREAM_INSTALL_TASKS,
        MAX_INFLIGHT_HANDSHAKE_ATTEMPTS_PER_ORIGIN,
        MAX_INFLIGHT_HANDSHAKES,
        MAX_STREAM_INSTALL_TASKS_PER_ORIGIN,
        MAX_TRACKED_HANDSHAKE_ORIGINS,
        ObservedOrigin,
        OriginAdmission,
        OriginAdmissionClass,
        ServerAdmission,
        bounded_endpoint_limits,
        drain_joined_child_tasks,
        endpoint_connection_capacity_available,
        has_child_capacity,
        run_body_with_cleanup,
    };

    #[test]
    fn unauthenticated_runtime_bounds_are_explicit_and_closed_at_capacity() {
        assert_eq!(MAX_INFLIGHT_HANDSHAKES, 64);
        assert_eq!(MAX_INFLIGHT_HANDSHAKE_ATTEMPTS_PER_ORIGIN, 8);
        assert_eq!(MAX_ESTABLISHED_CONNECTIONS, 64);
        assert_eq!(MAX_CONTROL_STREAM_TASKS, 32);
        assert_eq!(MAX_GLOBAL_CONTROL_STREAM_TASKS, 16);
        assert_eq!(MAX_GLOBAL_BULK_TRANSFER_TASKS, 48);
        assert_eq!(MAX_GLOBAL_STREAM_INSTALL_TASKS, 2);
        assert_eq!(MAX_ESTABLISHED_CONNECTIONS_PER_ORIGIN, 8);
        assert_eq!(MAX_CONTROL_STREAM_TASKS_PER_ORIGIN, 8);
        assert_eq!(MAX_BULK_TRANSFER_TASKS_PER_ORIGIN, 8);
        assert_eq!(MAX_STREAM_INSTALL_TASKS_PER_ORIGIN, 1);
        assert_eq!(
            u64::try_from(MAX_CONTROL_STREAM_TASKS).expect("stream task bound fits u64"),
            crate::quic_runtime::MAX_OPEN_BIDIRECTIONAL_STREAMS,
        );
        assert!(has_child_capacity(63, MAX_ESTABLISHED_CONNECTIONS));
        assert!(!has_child_capacity(64, MAX_ESTABLISHED_CONNECTIONS));
        assert!(!has_child_capacity(32, MAX_CONTROL_STREAM_TASKS));
    }

    #[test]
    fn handshake_attempt_window_preserves_capacity_for_other_origins() {
        let mut limiter = HandshakeOriginLimiter::default();
        let now = Duration::from_secs(1);
        let flood = IpAddr::from([192, 0, 2, 10]);
        let healthy = IpAddr::from([192, 0, 2, 11]);

        for _ in 0..MAX_INFLIGHT_HANDSHAKE_ATTEMPTS_PER_ORIGIN {
            assert!(limiter.admit(flood, now));
        }
        assert!(!limiter.admit(flood, now));
        assert!(limiter.admit(healthy, now));
        assert!(limiter.admit(
            flood,
            now + crate::config::QUIC_HANDSHAKE_TIMEOUT
        ));
    }

    #[test]
    fn handshake_origin_ledger_rejects_spoofed_source_churn_at_capacity() {
        let mut limiter = HandshakeOriginLimiter::default();
        let now = Duration::from_secs(1);
        for index in 0..MAX_TRACKED_HANDSHAKE_ORIGINS {
            let origin = IpAddr::V6(Ipv6Addr::from(
                u128::try_from(index + 1).expect("test origin index should fit u128"),
            ));
            assert!(limiter.admit(origin, now));
        }
        assert!(!limiter.admit(IpAddr::V6(Ipv6Addr::from(1_000_u128)), now));
        assert_eq!(limiter.attempts.len(), MAX_TRACKED_HANDSHAKE_ORIGINS);
    }

    #[test]
    fn endpoint_limiter_rejects_before_the_internal_accept_queue_can_exceed_the_cap() {
        bounded_endpoint_limits().expect("endpoint limiter should build");
        assert!(endpoint_connection_capacity_available(
            MAX_ESTABLISHED_CONNECTIONS - 1
        ));
        assert!(!endpoint_connection_capacity_available(
            MAX_ESTABLISHED_CONNECTIONS
        ));
    }

    #[test]
    fn global_control_permit_saturates_and_releases_across_scopes() {
        let permits = Arc::new(Semaphore::new(MAX_GLOBAL_CONTROL_STREAM_TASKS));
        let held = (0..MAX_GLOBAL_CONTROL_STREAM_TASKS)
            .map(|_| {
                Arc::clone(&permits)
                    .try_acquire_owned()
                    .expect("configured global permit should be available")
            })
            .collect::<Vec<_>>();
        assert!(Arc::clone(&permits).try_acquire_owned().is_err());

        drop(held);
        assert!(Arc::clone(&permits).try_acquire_owned().is_ok());
    }

    #[test]
    fn saturated_bulk_pool_preserves_control_plane_permits() {
        let control = Arc::new(Semaphore::new(MAX_GLOBAL_CONTROL_STREAM_TASKS));
        let bulk = Arc::new(Semaphore::new(MAX_GLOBAL_BULK_TRANSFER_TASKS));
        let held_bulk = (0..MAX_GLOBAL_BULK_TRANSFER_TASKS)
            .map(|_| {
                Arc::clone(&bulk)
                    .try_acquire_owned()
                    .expect("configured bulk permit should be available")
            })
            .collect::<Vec<_>>();
        assert!(Arc::clone(&bulk).try_acquire_owned().is_err());
        assert!(
            Arc::clone(&control).try_acquire_owned().is_ok(),
            "bulk saturation must not consume control-plane admission"
        );
        drop(held_bulk);
    }

    fn origin(ip: [u8; 4], port: u16) -> ObservedOrigin {
        ObservedOrigin::from_remote_addr(Some(SocketAddr::from((ip, port))))
            .expect("test address should identify an origin")
    }

    #[test]
    fn missing_origin_is_rejected_and_port_rotation_cannot_escape_connection_limit() {
        assert!(ObservedOrigin::from_remote_addr(None).is_none());

        let admission = Arc::new(OriginAdmission::default());
        let first_ip = [192, 168, 1, 20];
        let held = (0..MAX_ESTABLISHED_CONNECTIONS_PER_ORIGIN)
            .map(|index| {
                admission
                    .try_acquire(
                        origin(
                            first_ip,
                            12_000 + u16::try_from(index).expect("test index should fit u16"),
                        ),
                        OriginAdmissionClass::Connection,
                    )
                    .expect("connection below the per-origin limit should be admitted")
            })
            .collect::<Vec<_>>();
        assert!(
            admission
                .try_acquire(origin(first_ip, 13_000), OriginAdmissionClass::Connection,)
                .is_none(),
            "a new source port must not create a new origin budget"
        );

        let distinct = admission
            .try_acquire(
                origin([192, 168, 1, 21], 12_000),
                OriginAdmissionClass::Connection,
            )
            .expect("a distinct observed IP should keep its own capacity");
        assert_eq!(admission.active_origin_count(), 2);

        drop(held);
        assert_eq!(admission.active_origin_count(), 1);
        drop(distinct);
        assert_eq!(admission.active_origin_count(), 0);
    }

    #[test]
    fn origin_leases_release_on_normal_drop_and_panic_unwind() {
        let admission = Arc::new(OriginAdmission::default());
        let test_origin = origin([10, 0, 0, 7], 12_000);
        {
            let _lease = admission
                .try_acquire(test_origin, OriginAdmissionClass::StreamInstall)
                .expect("first provider should be admitted");
            assert_eq!(admission.active_origin_count(), 1);
        }
        assert_eq!(admission.active_origin_count(), 0);

        let unwind_admission = Arc::clone(&admission);
        let unwind = catch_unwind(AssertUnwindSafe(move || {
            let _lease = unwind_admission
                .try_acquire(test_origin, OriginAdmissionClass::StreamInstall)
                .expect("provider should be admitted before the injected panic");
            panic!("injected origin lease panic");
        }));
        assert!(unwind.is_err());
        assert_eq!(admission.active_origin_count(), 0);
        assert!(
            admission
                .try_acquire(test_origin, OriginAdmissionClass::StreamInstall)
                .is_some(),
            "panic unwinding must restore the origin capacity"
        );
    }

    #[test]
    fn origin_resource_classes_have_independent_active_limits() {
        let admission = Arc::new(OriginAdmission::default());
        let test_origin = origin([10, 0, 0, 8], 12_000);
        let connections = (0..MAX_ESTABLISHED_CONNECTIONS_PER_ORIGIN)
            .map(|_| {
                admission
                    .try_acquire(test_origin, OriginAdmissionClass::Connection)
                    .expect("connection class should retain independent capacity")
            })
            .collect::<Vec<_>>();
        let controls = (0..MAX_CONTROL_STREAM_TASKS_PER_ORIGIN)
            .map(|_| {
                admission
                    .try_acquire(test_origin, OriginAdmissionClass::Control)
                    .expect("control class should retain independent capacity")
            })
            .collect::<Vec<_>>();
        let bulk = (0..MAX_BULK_TRANSFER_TASKS_PER_ORIGIN)
            .map(|_| {
                admission
                    .try_acquire(test_origin, OriginAdmissionClass::Bulk)
                    .expect("bulk class should retain independent capacity")
            })
            .collect::<Vec<_>>();
        let stream_install = admission
            .try_acquire(test_origin, OriginAdmissionClass::StreamInstall)
            .expect("Stream Install class should retain independent capacity");

        for class in [
            OriginAdmissionClass::Connection,
            OriginAdmissionClass::Control,
            OriginAdmissionClass::Bulk,
            OriginAdmissionClass::StreamInstall,
        ] {
            assert!(
                admission.try_acquire(test_origin, class).is_none(),
                "{class:?} should close at its own active limit"
            );
        }

        drop(connections);
        drop(controls);
        drop(bulk);
        drop(stream_install);
        assert_eq!(admission.active_origin_count(), 0);
    }

    #[test]
    fn saturated_stream_install_gate_preserves_ordinary_bulk_capacity() {
        let admission = ServerAdmission::new();
        let first = admission
            .try_acquire_stream_install(origin([10, 0, 0, 10], 12_000))
            .expect("first global provider slot should be available");
        let second = admission
            .try_acquire_stream_install(origin([10, 0, 0, 11], 12_000))
            .expect("second global provider slot should be available");
        assert!(
            admission
                .try_acquire_stream_install(origin([10, 0, 0, 12], 12_000))
                .is_none(),
            "the existing two-provider global gate must remain authoritative"
        );
        assert!(
            admission
                .try_acquire_bulk(origin([10, 0, 0, 12], 12_001))
                .is_some(),
            "provider saturation must not consume ordinary bulk capacity"
        );

        drop(first);
        drop(second);
    }

    #[tokio::test]
    async fn draining_waits_for_every_childs_natural_cleanup() {
        let owner_closed = CancellationToken::new();
        let cleanup_release = CancellationToken::new();
        let completed = Arc::new(AtomicUsize::new(0));
        let (started_tx, mut started_rx) = mpsc::unbounded_channel();
        let mut children = JoinSet::new();

        for child_id in 0..2 {
            let owner_closed = owner_closed.clone();
            let cleanup_release = cleanup_release.clone();
            let completed = completed.clone();
            let started_tx = started_tx.clone();
            children.spawn(async move {
                started_tx
                    .send(child_id)
                    .expect("test observer should remain available");
                owner_closed.cancelled().await;
                cleanup_release.cancelled().await;
                completed.fetch_add(1, Ordering::SeqCst);
                Ok(())
            });
        }
        drop(started_tx);

        let drain_task = tokio::spawn(async move {
            let mut children = children;
            drain_joined_child_tasks(&mut children, "synthetic child").await;
        });
        for _ in 0..2 {
            tokio::time::timeout(Duration::from_secs(1), started_rx.recv())
                .await
                .expect("child should start")
                .expect("start channel should remain open");
        }

        owner_closed.cancel();
        tokio::task::yield_now().await;
        assert_eq!(completed.load(Ordering::SeqCst), 0);
        assert!(!drain_task.is_finished(), "drain must await child cleanup");

        cleanup_release.cancel();
        tokio::time::timeout(Duration::from_secs(1), drain_task)
            .await
            .expect("drain should finish after cleanup is released")
            .expect("drain task should not panic");
        assert_eq!(completed.load(Ordering::SeqCst), 2);
    }

    #[tokio::test]
    async fn one_panicking_child_does_not_skip_a_siblings_cleanup() {
        let cleanup_release = CancellationToken::new();
        let completed = Arc::new(AtomicUsize::new(0));
        let mut children = JoinSet::new();
        children.spawn(async {
            panic!("injected child panic");
            #[allow(unreachable_code)]
            Ok(())
        });
        let child_release = cleanup_release.clone();
        let child_completed = completed.clone();
        children.spawn(async move {
            child_release.cancelled().await;
            child_completed.fetch_add(1, Ordering::SeqCst);
            Ok(())
        });

        let drain_task = tokio::spawn(async move {
            drain_joined_child_tasks(&mut children, "synthetic joined child").await;
        });
        tokio::task::yield_now().await;
        assert!(!drain_task.is_finished());
        cleanup_release.cancel();
        tokio::time::timeout(Duration::from_secs(1), drain_task)
            .await
            .expect("drain should await the surviving child")
            .expect("drain task should not panic");
        assert_eq!(completed.load(Ordering::SeqCst), 1);
    }

    #[tokio::test(start_paused = true)]
    async fn an_established_connection_without_streams_has_a_finite_application_idle_bound() {
        let started = tokio::time::Instant::now();
        tokio::time::sleep(CONNECTION_NO_STREAM_IDLE_TIMEOUT).await;
        assert_eq!(started.elapsed(), CONNECTION_NO_STREAM_IDLE_TIMEOUT);
    }

    #[tokio::test]
    async fn server_body_panic_still_awaits_owner_cleanup() {
        let cleanup_finished = Arc::new(AtomicUsize::new(0));
        let cleanup_probe = cleanup_finished.clone();

        let result = run_body_with_cleanup(
            async {
                panic!("injected server body panic");
                #[allow(unreachable_code)]
                Ok(())
            },
            async move {
                tokio::task::yield_now().await;
                cleanup_probe.store(1, Ordering::SeqCst);
                Ok(())
            },
        )
        .await;

        assert!(result.is_err());
        assert_eq!(cleanup_finished.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn nested_scopes_publish_completion_from_children_outward() {
        let global_shutdown = CancellationToken::new();
        let server_shutdown = global_shutdown.child_token();
        let stream_cleanup_release = CancellationToken::new();
        let (started_tx, mut started_rx) = mpsc::unbounded_channel();
        let (order_tx, mut order_rx) = mpsc::unbounded_channel();
        let mut connections = JoinSet::new();

        let connection_shutdown = server_shutdown.child_token();
        let connection_stream_cleanup_release = stream_cleanup_release.clone();
        let connection_order_tx = order_tx.clone();
        connections.spawn(async move {
            let mut streams = JoinSet::new();
            for label in ["stream-a", "stream-b"] {
                let shutdown = connection_shutdown.child_token();
                let cleanup_release = connection_stream_cleanup_release.clone();
                let started_tx = started_tx.clone();
                let order_tx = connection_order_tx.clone();
                streams.spawn(async move {
                    started_tx
                        .send(label)
                        .expect("test observer should remain available");
                    shutdown.cancelled().await;
                    cleanup_release.cancelled().await;
                    order_tx
                        .send(label)
                        .expect("order observer should remain available");
                    Ok(())
                });
            }

            connection_shutdown.cancelled().await;
            drain_joined_child_tasks(&mut streams, "synthetic stream").await;
            connection_order_tx
                .send("peer-disconnected")
                .expect("order observer should remain available");
            Ok(())
        });

        let hierarchy_task = tokio::spawn(async move {
            server_shutdown.cancel();
            drain_joined_child_tasks(&mut connections, "synthetic connection").await;
            order_tx
                .send("server-return")
                .expect("order observer should remain available");
        });
        for _ in 0..2 {
            tokio::time::timeout(Duration::from_secs(1), started_rx.recv())
                .await
                .expect("stream should start")
                .expect("start channel should remain open");
        }
        assert!(!hierarchy_task.is_finished());

        stream_cleanup_release.cancel();
        tokio::time::timeout(Duration::from_secs(1), hierarchy_task)
            .await
            .expect("hierarchy should drain")
            .expect("hierarchy task should not panic");

        let mut order = Vec::new();
        while let Ok(event) = order_rx.try_recv() {
            order.push(event);
        }
        let disconnected = order
            .iter()
            .position(|event| *event == "peer-disconnected")
            .expect("disconnect should be published");
        let server_return = order
            .iter()
            .position(|event| *event == "server-return")
            .expect("server return should be published");
        assert!(
            order[..disconnected]
                .iter()
                .all(|event| event.starts_with("stream-"))
        );
        assert_eq!(disconnected, 2);
        assert_eq!(server_return, 3);
        assert!(
            !global_shutdown.is_cancelled(),
            "server-local shutdown must not cancel its runtime parent"
        );
    }
}