boatramp_server/

serve_pipeline.rs

//! The request-serving pipeline: host-based routing (`serve_by_host`), the
//! by-name admin route, preview and ACME domain-challenge serving, access
//! control (auth gate, basic-auth, rate-limit), request-context building, and
//! the resolve -> entry -> static/handler/proxy dispatch that streams the
//! chosen response. Pulls the shared response helpers and backends in via
//! `use super::*`.

use super::*;
use boatramp_core::project::ProjectRef;

/// A request's network identity, threaded into the serving pipeline for
/// access control: the socket peer plus the shared rate limiter.
struct Visitor<'a> {
    peer: IpAddr,
    limiter: &'a dyn RateLimitStore,
}

/// Serve under the explicit by-name admin/testing route `/_sites/<site>/...`.
/// The catch-all captures `<site>` or `<site>/<path...>`. Accepts any method so a
/// proxy rewrite can forward non-`GET` requests. This route is not host-routed and
/// does not serve a root-mounted site — for that, use host routing (see the
/// addressing docs).
pub(super) async fn serve_sites(
    State(deploy): State<DeployStore>,
    Extension(limiter): Extension<Arc<dyn RateLimitStore>>,
    Extension(handlers): Extension<Arc<HandlerRuntime>>,
    ConnectInfo(peer): ConnectInfo<SocketAddr>,
    request: Request,
) -> Response {
    let raw = request.uri().path();
    let rest = raw
        .strip_prefix("/_sites/")
        .unwrap_or("")
        .trim_start_matches('/');
    let (site, path) = rest.split_once('/').unwrap_or((rest, ""));
    if site.is_empty() {
        return not_found();
    }
    let (site, request_path) = (site.to_string(), format!("/{path}"));
    let visitor = Visitor {
        peer: peer.ip(),
        limiter: limiter.as_ref(),
    };
    // The explicit `/_sites/<name>/` admin/testing route is not host-routed, so
    // transport/canonical redirects don't apply. It addresses sites by bare name,
    // a default-project admin convenience.
    serve_request(
        &deploy,
        ProjectRef::DEFAULT.as_str(),
        &site,
        &request_path,
        request,
        &visitor,
        &handlers,
        false,
    )
    .await
}

/// The root-key-signed bootstrap-TLS identity attestation (base64url
/// `COSE_Sign1`), carried as an extension for [`serve_bootstrap_identity`].
#[derive(Clone)]
pub(super) struct BootstrapAttestation(pub(super) Option<String>);

/// `GET /.well-known/boatramp-bootstrap-identity` — serve the root-key attestation
/// of this node's `--tls rpk` control-plane TLS key. Public + unauthenticated: a
/// signed statement that reveals nothing (the TLS public key is already presented
/// in the handshake). A client pinning only the root key verifies it (root
/// signature + validity), extracts the attested TLS key, and pins it. `404` when
/// no attestation is set (not `--tls rpk`, or a verify-only node with no issuer).
pub(super) async fn serve_bootstrap_identity(
    Extension(att): Extension<BootstrapAttestation>,
) -> Response {
    match att.0 {
        Some(a) => (
            StatusCode::OK,
            [(header::CONTENT_TYPE, "application/octet-stream")],
            a,
        )
            .into_response(),
        None => not_found(),
    }
}

/// Serve a pending **HTTP domain-ownership challenge** from the edge, *before*
/// host routing — the fix for the verify-before-attach chicken-and-egg. A host
/// pointed at this server but not yet attached to any site (so it would
/// otherwise fall through to `default_site` and 404 its own challenge) fetches
/// its token here, letting `domain verify` succeed with no prior deploy. Returns
/// the token for a matching `(Host, token)` pending HTTP challenge, else `404`.
///
/// Gated by the `domain_verify_self_serve` posture knob (on by default). It only
/// ever echoes back a random token to the very host that owns the pending
/// challenge, so it leaks nothing and needs no auth (like an ACME http-01
/// challenge). Mounted on both the main router and the `:80` redirect router, so
/// the plain-HTTP probe is answered directly instead of 308-redirected to an
/// HTTPS endpoint that may have no cert yet.
pub(super) async fn serve_domain_challenge(
    State(deploy): State<DeployStore>,
    Extension(posture): Extension<boatramp_core::security::SecurityPosture>,
    Path(token): Path<String>,
    headers: HeaderMap,
) -> Response {
    if !posture.domain_verify_self_serve {
        return not_found();
    }
    let Some(host) = headers
        .get(header::HOST)
        .and_then(|value| value.to_str().ok())
        .map(strip_port)
    else {
        return not_found();
    };
    match deploy
        .find_pending_http_challenge(host, &token, now_unix())
        .await
    {
        Ok(Some(v)) => (
            StatusCode::OK,
            [(header::CONTENT_TYPE, "text/plain; charset=utf-8")],
            v.token,
        )
            .into_response(),
        Ok(None) => not_found(),
        Err(err) => deploy_error_response(err),
    }
}

/// The serve fast path (hot-path bypass): the dependencies the axum router's Extension
/// layers hold, cloned so [`RouterHandler`](crate::http_serve) can call
/// [`serve_by_host_inner`] directly for a plain site GET/HEAD — skipping the axum
/// Router + middleware future-composition tax (~15–20% of per-core CPU, profiled) — while
/// every security check still runs inside `serve_by_host_inner`. Non-eligible requests
/// fall through to the full router unchanged.
// An opaque handle held by the serve loops (`RouterHandler`); `pub` because the public
// `router_with_fast` returns it, but its fields are `pub(crate)` so it stays opaque to
// external crates — they only pass it back into a serve loop.
#[derive(Clone)]
pub struct FastServe {
    pub(crate) deploy: DeployStore,
    pub(crate) limiter: Arc<dyn RateLimitStore>,
    pub(crate) handlers: Arc<HandlerRuntime>,
    pub(crate) daemon: Arc<DaemonRuntime>,
    pub(crate) implicit: ImplicitRouting,
    pub(crate) preview_auth: Auth,
    /// The operator security posture and listener TLS flag the router carries as
    /// per-request extensions (`Extension(posture)` / `Extension(served_over_tls)`). The
    /// site path reads them from the request — the SSRF gate on a gateway-upstream proxy
    /// ([`proxy::request_posture`](crate::proxy)) and the scheme/HSTS derivation
    /// ([`serve_request`]) — so the bypass must re-insert them to stay byte-identical to
    /// the router (a permissive fleet would otherwise wrongly refuse a private upstream,
    /// and a TLS listener would derive `http`).
    pub(crate) posture: boatramp_core::security::SecurityPosture,
    pub(crate) served_over_tls: bool,
    /// When serving a TLS listener paired with an HTTP/3 endpoint, the `Alt-Svc` value the
    /// router's [`advertise_http3`](crate::advertise_http3) layer would add — re-applied by
    /// [`dispatch`](Self::dispatch) so a bypassed response still advertises h3. `None` on
    /// the plaintext path (no h3) and on TLS without http3.
    pub(crate) alt_svc: Option<axum::http::HeaderValue>,
}

impl FastServe {
    /// Advertise a paired HTTP/3 listener on `port`: an eligible bypassed response gains the
    /// same `Alt-Svc` header the router's [`advertise_http3`](crate::advertise_http3) layer
    /// adds. Call it (with the same port) exactly where `advertise_http3` wraps the router, so
    /// the two paths stay byte-identical. TLS + http3 only.
    #[cfg(feature = "http3")]
    pub fn advertise_http3(mut self, port: u16) -> Self {
        self.alt_svc = axum::http::HeaderValue::from_str(&crate::http3::alt_svc_value(port)).ok();
        self
    }

    /// Whether `request` may take the fast path: a plain site GET/HEAD the router would
    /// route to its `serve_by_host` fallback anyway. **Conservative** — anything the
    /// router matches explicitly (any `/api`, `/_*`, `/healthz`, `/readyz`, `/mcp`,
    /// `/.well-known`) or that the console owns falls through to the full router.
    /// Excluding too much only forgoes the speedup; excluding too little is the exact
    /// bug the differential test guards against.
    pub(crate) fn eligible(&self, request: &Request) -> bool {
        if !matches!(
            *request.method(),
            axum::http::Method::GET | axum::http::Method::HEAD
        ) {
            return false;
        }
        let path = request.uri().path();
        let reserved = path == "/api"
            || path.starts_with("/api/")
            || path.starts_with("/_") // /_deploy /_sites /_webhooks (path-form; host-form previews still serve)
            || path.starts_with("/.well-known/")
            || path == "/healthz"
            || path == "/readyz"
            || path == "/mcp"
            || path.starts_with("/mcp/");
        if reserved {
            return false;
        }
        // The console owns a runtime-configured host+path — never serve it as site
        // content. This is the *same* predicate the console middleware uses (host from the
        // `Host` header only, exactly as it resolves it), so the two can't disagree.
        #[cfg(feature = "console")]
        {
            let effective = self.daemon.effective();
            let host = request
                .headers()
                .get(header::HOST)
                .and_then(|v| v.to_str().ok())
                .map(crate::strip_port)
                .unwrap_or("");
            if crate::console::would_intercept(&effective, host, path) {
                return false;
            }
        }
        true
    }

    /// Serve an already-[`eligible`](Self::eligible) `request` directly, applying the same
    /// request-id + access-log/metrics guarantees the `access_log` middleware would (via the
    /// shared [`assign_request_id`](crate::assign_request_id) /
    /// [`AccessLogCtx`](crate::AccessLogCtx)). Auth, rate-limiting, host-routing, previews,
    /// and the DV gate all run inside [`serve_by_host_inner`], so the bypass can never skip
    /// them. The caller MUST have checked [`eligible`](Self::eligible) first — that gate is
    /// what keeps the bypass from stealing a request the router would match explicitly.
    pub(crate) async fn dispatch(&self, mut request: Request, peer: SocketAddr) -> Response {
        debug_assert!(
            self.eligible(&request),
            "FastServe::dispatch called on an ineligible request — caller must gate on eligible()"
        );
        // A HEAD is served like its GET but with the body dropped. The router path relies on
        // axum stripping the HEAD body before the response reaches the codec; the h1 codec
        // also suppresses it, but the h2 codec does not — so the bypass strips it here to be
        // correct (and byte-identical to the router) over every protocol. Captured before the
        // request is consumed below.
        let is_head = *request.method() == axum::http::Method::HEAD;
        let request_id = crate::assign_request_id(&mut request);
        // Re-insert the per-request extensions the router's tower layers would add and the
        // site path reads: the security posture (the gateway-upstream SSRF gate) and the
        // listener's TLS flag (scheme + HSTS). `RequestId` (WASM handlers) and `ConnectInfo`
        // (peer) are already present — from `assign_request_id` above and `RouterHandler`.
        {
            let ext = request.extensions_mut();
            ext.insert(self.posture);
            ext.insert(crate::ServedOverTls(self.served_over_tls));
        }
        let log = crate::AccessLogCtx::capture(&request, request_id);
        let response = serve_by_host_inner(
            self.deploy.clone(),
            self.limiter.clone(),
            self.handlers.clone(),
            self.daemon.clone(),
            self.implicit,
            self.preview_auth.clone(),
            peer,
            request,
        )
        .await;
        // Drop the body for HEAD (headers stand in for the GET response), matching the
        // router; the codec recomputes framing from the now-empty body identically.
        let mut response = if is_head {
            let (parts, _body) = response.into_parts();
            Response::from_parts(parts, axum::body::Body::empty())
        } else {
            response
        };
        // Advertise a paired h3 listener, matching the outermost `advertise_http3` router
        // layer on a TLS+http3 listener (`None`, so a no-op, on every other path).
        if let Some(alt_svc) = &self.alt_svc {
            response
                .headers_mut()
                .insert(axum::http::header::ALT_SVC, alt_svc.clone());
        }
        match log {
            Some(ctx) => ctx.finish(response),
            None => response,
        }
    }
}

/// axum extractor wrapper over [`serve_by_host_inner`] — the router's site fallback.
/// The plain-arg inner is called directly by the serve hot-path bypass (stage 2), so the
/// two paths share one serving implementation.
#[allow(clippy::too_many_arguments)] // axum extractors, not a real parameter list
pub(super) async fn serve_by_host(
    State(deploy): State<DeployStore>,
    Extension(limiter): Extension<Arc<dyn RateLimitStore>>,
    Extension(handlers): Extension<Arc<HandlerRuntime>>,
    Extension(daemon): Extension<Arc<DaemonRuntime>>,
    Extension(implicit): Extension<ImplicitRouting>,
    Extension(preview_auth): Extension<Auth>,
    ConnectInfo(peer): ConnectInfo<SocketAddr>,
    request: Request,
) -> Response {
    serve_by_host_inner(
        deploy,
        limiter,
        handlers,
        daemon,
        implicit,
        preview_auth,
        peer,
        request,
    )
    .await
}

/// Host-routed static + reverse-proxy serving: resolves the site from the request host,
/// applies preview auth + rate-limiting (`Visitor`) + virtualhost routing, then serves.
/// Plain-arg form so both the axum router (via [`serve_by_host`]) and the hot-path bypass
/// call one implementation — the bypass can never diverge on the security-relevant checks
/// because they live here, not in the middleware. (Serve hot-path bypass, stage 2.)
#[allow(clippy::too_many_arguments)]
pub(crate) async fn serve_by_host_inner(
    deploy: DeployStore,
    limiter: Arc<dyn RateLimitStore>,
    handlers: Arc<HandlerRuntime>,
    daemon: Arc<DaemonRuntime>,
    implicit: ImplicitRouting,
    preview_auth: Auth,
    peer: SocketAddr,
    request: Request,
) -> Response {
    // Catch-all site + preview protection are read live from the daemon-config
    // runtime, so `config set default_site …` / `protect_previews …` take effect
    // without a restart.
    let effective = daemon.effective();
    let preview_policy = PreviewPolicy {
        protect: effective.protect_previews,
    };
    // Resolve the request host from the `Host` header (HTTP/1.1) or, when that is
    // absent, the URI authority. HTTP/2 sends the host as the `:authority`
    // pseudo-header, which hyper places in the request URI rather than a `Host`
    // header — so without this fallback every H2 request has no host and 404s.
    let Some(host) = request
        .headers()
        .get(header::HOST)
        .and_then(|value| value.to_str().ok())
        .or_else(|| request.uri().host())
        .map(strip_port)
        .map(str::to_string)
    else {
        return not_found();
    };
    let request_path = request.uri().path().to_string();
    // Wildcard preview host form `<id>.deploy.<site-host>`: the deploy id rides
    // as a subdomain (an unguessable content-hash capability, like the path form
    // `<site-host>/_deploy/<id>/…`). The remaining host resolves the site, and
    // the deployment is served with a preview-scoped binding identity. Falls
    // through to normal virtualhost routing when the host isn't a preview host.
    if let Some((id_prefix, site_host)) = parse_deploy_host(&host) {
        if let Some(blocked) =
            preview_auth_gate(preview_policy, &preview_auth, request.headers()).await
        {
            return blocked;
        }
        return serve_host_preview(
            &deploy,
            &handlers,
            peer.ip(),
            &request_path,
            request,
            id_prefix,
            site_host,
        )
        .await;
    }
    match deploy.resolve_site_by_host(&host).await {
        Ok(Some(owner)) => {
            let visitor = Visitor {
                peer: peer.ip(),
                limiter: limiter.as_ref(),
            };
            // Host-routed: transport/canonical redirects + HSTS apply.
            serve_request(
                &deploy,
                &owner.project,
                &owner.site,
                &request_path,
                request,
                &visitor,
                &handlers,
                true,
            )
            .await
        }
        // Unmatched host — no verified, attached virtualhost. Resolution order:
        //   (0) mandatory verification — a **non-local** host that isn't verified
        //       gets the "verification pending" page (421), never a fallback;
        //   (A) implicit first-label routing — `<site>.host` names a served site;
        //   the configured catch-all `default_site` (explicit operator intent).
        // (A) runs only when `implicit` is on (dev / single-tenant / a loopback
        // bind). There is deliberately no implicit *sole-site* auto-default: an
        // operator makes a site the catch-all explicitly with `default_site`.
        Ok(None) => {
            // (0) Strict gate (DV-2): a non-local public host that matched no
            // verified virtualhost is refused with the holding page — so
            // `default_site`/implicit never silently serve an unverified host.
            // Local hosts (localhost/*.localhost/*.local/IPs) and a fleet with the
            // gate off (`[security] require_domain_verification = false`, or an
            // admin `domain add --unverified` that attached the host above) pass.
            if effective.posture.require_domain_verification && !is_local_host(&host, implicit.0) {
                return verification_pending_page(&deploy, &host).await;
            }
            // (A) First host label naming a served site: `blog.localhost` → `blog`.
            if implicit.0 {
                let label = host.split('.').next().unwrap_or("");
                if !label.is_empty()
                    && matches!(
                        deploy.current_id(ProjectRef::DEFAULT, label).await,
                        Ok(Some(_))
                    )
                {
                    let visitor = Visitor {
                        peer: peer.ip(),
                        limiter: limiter.as_ref(),
                    };
                    // Implicit first-label routing is a default-project convenience.
                    return serve_request(
                        &deploy,
                        ProjectRef::DEFAULT.as_str(),
                        label,
                        &request_path,
                        request,
                        &visitor,
                        &handlers,
                        true,
                    )
                    .await;
                }
            }
            match effective.default_site.as_deref() {
                Some(site) => {
                    let visitor = Visitor {
                        peer: peer.ip(),
                        limiter: limiter.as_ref(),
                    };
                    // The operator's catch-all `default_site` is a default-project site.
                    serve_request(
                        &deploy,
                        ProjectRef::DEFAULT.as_str(),
                        site,
                        &request_path,
                        request,
                        &visitor,
                        &handlers,
                        true,
                    )
                    .await
                }
                None => not_found(),
            }
        }
        Err(err) => deploy_error_response(err),
    }
}

/// Serve a wildcard-host preview: resolve `id_prefix` to a full deployment id
/// and `site_host` to a site, then run the deployment with a **preview-scoped**
/// binding identity (like [`serve_preview`], but reached by subdomain). Handlers
/// run only when the host resolves to a real site; otherwise the preview serves
/// static content only. No visitor access control — the unguessable id is the
/// capability (consistent with the path-form preview).
#[allow(clippy::too_many_arguments)]
async fn serve_host_preview(
    deploy: &DeployStore,
    handlers: &HandlerRuntime,
    peer: IpAddr,
    request_path: &str,
    request: Request,
    id_prefix: &str,
    site_host: &str,
) -> Response {
    let id = match deploy.resolve_manifest_id(id_prefix).await {
        Ok(Some(id)) => id,
        Ok(None) => return not_found(),
        Err(err) => return deploy_error_response(err),
    };
    let owner = match deploy.resolve_site_by_host(site_host).await {
        Ok(owner) => owner,
        Err(err) => return deploy_error_response(err),
    };
    let project = owner.as_ref().map(|o| o.project.clone());
    let site = owner.map(|o| o.site);
    let site_config = match (&project, &site) {
        (Some(project), Some(site)) => {
            match deploy.get_site_config(ProjectRef::new(project), site).await {
                Ok(config) => config,
                Err(err) => return deploy_error_response(err),
            }
        }
        _ => None,
    };
    match deploy.get_manifest(&id).await {
        Ok(Some(manifest)) => {
            serve_resolved(
                deploy,
                &manifest,
                request_path,
                request,
                peer,
                project.as_deref(),
                site.as_deref(),
                site_config.as_ref(),
                handlers,
                Some(&id),
            )
            .await
        }
        Ok(None) => not_found(),
        Err(err) => deploy_error_response(err),
    }
}

/// Run the serving pipeline for a resolved `site` and request path: apply the
/// deploy config (redirects, rewrites/SPA, clean URLs, custom 404, headers,
/// cache) via [`route::resolve`], then HTTP correctness (conditional `304`,
/// `Range`/`206`, `ETag`).
#[allow(clippy::too_many_arguments)]
async fn serve_request(
    deploy: &DeployStore,
    project: &str,
    site: &str,
    request_path: &str,
    request: Request,
    visitor: &Visitor<'_>,
    handlers: &HandlerRuntime,
    host_routed: bool,
) -> Response {
    let project_ref = ProjectRef::new(project);
    // Load the site config once (for access policy + client-IP resolution). Cached
    // by content hash — the hot path skips the body read + JSON parse per request.
    let site_config = match deploy.get_site_config_cached(project_ref, site).await {
        Ok(config) => config,
        Err(err) => return deploy_error_response(err),
    };

    // Transport redirects + HSTS. The effective scheme
    // honors `X-Forwarded-Proto` **only from a configured trusted proxy**
    // — otherwise a direct HTTP client could forge `…: https` to
    // skip the HTTPS redirect. For an untrusted/direct peer the scheme is the
    // listener's own (TLS ⇒ `https`, else `http`). Host-routed traffic only.
    let listener_scheme = if request
        .extensions()
        .get::<ServedOverTls>()
        .map(|s| s.0)
        .unwrap_or(false)
    {
        "https"
    } else {
        "http"
    };
    let peer_trusted = site_config
        .as_ref()
        .map(|c| c.access.is_trusted_proxy(visitor.peer))
        .unwrap_or(false);
    let effective_scheme = if peer_trusted {
        request
            .headers()
            .get("x-forwarded-proto")
            .and_then(|v| v.to_str().ok())
            .unwrap_or(listener_scheme)
            .to_string()
    } else {
        listener_scheme.to_string()
    };
    // Captured before the request body is consumed, for on-the-fly compression.
    #[cfg(feature = "compression")]
    let accept_encoding = request
        .headers()
        .get(header::ACCEPT_ENCODING)
        .and_then(|v| v.to_str().ok())
        .map(str::to_string);
    // Site-tier security response headers, applied (host-routed only) after the
    // response is built: HSTS (HTTPS only), plus opt-in CSP / X-Frame-Options.
    let mut security_headers: Vec<(HeaderName, String)> = Vec::new();
    if host_routed {
        if let Some(cfg) = site_config.as_ref() {
            let host = request
                .headers()
                .get(header::HOST)
                .and_then(|v| v.to_str().ok())
                .map(strip_port)
                .unwrap_or("");
            let path_and_query = request
                .uri()
                .path_and_query()
                .map(axum::http::uri::PathAndQuery::as_str)
                .unwrap_or(request_path);
            if let Some(target) = boatramp_core::config::transport_redirect(
                &cfg.security,
                &cfg.domains,
                &effective_scheme,
                host,
                path_and_query,
            ) {
                return redirect_to(&target);
            }
            // HSTS only over HTTPS (it's meaningless / ignored over plain HTTP).
            if effective_scheme == "https" {
                if let Some(hsts) = cfg.security.hsts.as_ref() {
                    security_headers.push((
                        HeaderName::from_static("strict-transport-security"),
                        hsts.header_value(),
                    ));
                }
            }
            // CSP + X-Frame-Options apply on either scheme, when configured.
            if let Some(csp) = cfg.security.csp.as_deref() {
                security_headers.push((header::CONTENT_SECURITY_POLICY, csp.to_string()));
            }
            if let Some(frame) = cfg.security.frame_options.as_deref() {
                security_headers.push((header::X_FRAME_OPTIONS, frame.to_string()));
            }
        }
    }

    let access = site_config.as_ref().map(|c| &c.access);

    // Resolve the real client IP, honoring X-Forwarded-For only from a
    // configured trusted proxy.
    let trusted = access.map(|a| a.trusted_proxies.as_slice()).unwrap_or(&[]);
    let forwarded_for = request
        .headers()
        .get("x-forwarded-for")
        .and_then(|value| value.to_str().ok());
    let client_ip = boatramp_core::access::resolve_client_ip(visitor.peer, forwarded_for, trusted);

    // Visitor access control (WAF → IP rules → rate limit → basic auth) runs
    // before any content is read.
    if let Some(access) = access {
        if let Some(denied) = enforce_access(
            access,
            site,
            request.headers(),
            request_path,
            client_ip,
            visitor.limiter,
        )
        .await
        {
            return denied;
        }
    }

    let manifest = match deploy.current_manifest(project_ref, site).await {
        Ok(Some(manifest)) => manifest,
        Ok(None) => return not_found(),
        Err(err) => return deploy_error_response(err),
    };
    let mut response = serve_resolved(
        deploy,
        &manifest,
        request_path,
        request,
        client_ip,
        Some(project),
        Some(site),
        site_config.as_deref(),
        handlers,
        None,
    )
    .await;
    // Site-tier security headers (HSTS / CSP / X-Frame-Options), computed above.
    for (name, value) in security_headers {
        if let Ok(value) = HeaderValue::from_str(&value) {
            response.headers_mut().insert(name, value);
        }
    }
    // On-the-fly compression (opt-in per site; covers dynamic + variant-less
    // static responses). A no-op without the `compression` feature.
    #[cfg(feature = "compression")]
    let response = match site_config.as_ref() {
        Some(cfg) if cfg.compression.enabled => maybe_compress(
            response,
            accept_encoding.as_deref(),
            cfg.compression.min_size,
        ),
        _ => response,
    };
    response
}

/// When previews are protected, require a valid control-plane token. Returns
/// `Some(401)` to block, `None` to allow. "Any valid token" (no scope needed).
async fn preview_auth_gate(
    policy: PreviewPolicy,
    auth: &Auth,
    headers: &HeaderMap,
) -> Option<Response> {
    if !policy.protect {
        return None;
    }
    let bearer = headers
        .get(header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.strip_prefix("Bearer "));
    let ok = match bearer {
        Some(token) => auth.verify_bearer(token).await,
        None => false,
    };
    (!ok).then(|| {
        (
            StatusCode::UNAUTHORIZED,
            "preview requires a valid bearer token\n",
        )
            .into_response()
    })
}

/// A `301 Moved Permanently` to `target` (transport/canonical redirects).
fn redirect_to(target: &str) -> Response {
    match HeaderValue::from_str(target) {
        Ok(location) => (
            StatusCode::MOVED_PERMANENTLY,
            [(header::LOCATION, location)],
        )
            .into_response(),
        Err(_) => (StatusCode::INTERNAL_SERVER_ERROR, "bad redirect target\n").into_response(),
    }
}

/// A standalone router for a plain `:80` listener that permanently redirects
/// every request to its HTTPS equivalent. Bound alongside the HTTPS listener so
/// plain-HTTP visitors are upgraded even when boatramp terminates TLS itself.
///
/// It does serve one thing directly rather than redirecting: the HTTP
/// domain-ownership challenge (`/.well-known/boatramp-domain-verification/…`).
/// That probe arrives on plain `:80` for a host that may have no cert yet, so a
/// 308 to HTTPS would bounce it to an endpoint that can't answer — the token
/// must be served here. (ACME's own challenges use ALPN-01/DNS-01, so there is
/// no `/.well-known/acme-challenge` to serve.)
pub fn http_redirect_router(
    deploy: DeployStore,
    posture: boatramp_core::security::SecurityPosture,
) -> Router {
    Router::new()
        .route(
            "/.well-known/boatramp-domain-verification/{token}",
            get(serve_domain_challenge),
        )
        .fallback(redirect_http_to_https)
        .with_state(deploy)
        .layer(Extension(posture))
}

/// 308-redirect any request to `https://<host><path-and-query>` (308 preserves
/// the method/body, unlike 301).
async fn redirect_http_to_https(req: Request) -> Response {
    let host = req
        .headers()
        .get(header::HOST)
        .and_then(|v| v.to_str().ok())
        .map(strip_port)
        .unwrap_or("");
    if host.is_empty() {
        return (StatusCode::BAD_REQUEST, "missing Host header\n").into_response();
    }
    let path_and_query = req
        .uri()
        .path_and_query()
        .map(axum::http::uri::PathAndQuery::as_str)
        .unwrap_or("/");
    match HeaderValue::from_str(&format!("https://{host}{path_and_query}")) {
        Ok(location) => (
            StatusCode::PERMANENT_REDIRECT,
            [(header::LOCATION, location)],
        )
            .into_response(),
        Err(_) => (StatusCode::BAD_REQUEST, "invalid host\n").into_response(),
    }
}

/// Evaluate a site's [`AccessConfig`] against an already-resolved `client_ip`.
/// Returns `Some(response)` to short-circuit (403/429/401), or `None` to allow.
/// Order: WAF → IP rules → rate limit → basic auth. `async` because the
/// cluster-wide rate-limit store does a KV round-trip.
async fn enforce_access(
    access: &AccessConfig,
    site: &str,
    req_headers: &HeaderMap,
    path: &str,
    client_ip: IpAddr,
    limiter: &dyn RateLimitStore,
) -> Option<Response> {
    if !access.is_enforced() {
        return None;
    }
    // WAF (user-agent rules + anomaly scoring) is the outermost filter: a blocked
    // request shouldn't reach rate limiting or auth.
    if access.waf.is_enabled() {
        let header_str = |name| req_headers.get(name).and_then(|v| v.to_str().ok());
        let waf_req = boatramp_core::waf::WafRequest {
            user_agent: header_str(header::USER_AGENT),
            accept: header_str(header::ACCEPT),
            path,
        };
        if let boatramp_core::waf::WafVerdict::Block(reason) =
            boatramp_core::waf::evaluate(&access.waf, &waf_req)
        {
            tracing::debug!(%client_ip, site, %reason, "request blocked by WAF");
            return Some((StatusCode::FORBIDDEN, "forbidden\n").into_response());
        }
    }
    if !access.ip.allows(client_ip) {
        tracing::debug!(%client_ip, site, "request blocked by IP rules");
        return Some((StatusCode::FORBIDDEN, "forbidden\n").into_response());
    }
    if let Some(limit) = &access.rate_limit {
        if !limiter.check(site, client_ip, limit).await {
            return Some(too_many_requests());
        }
    }
    if let Some(basic) = &access.basic_auth {
        if !verify_basic_auth(basic, req_headers) {
            return Some(basic_auth_challenge(basic));
        }
    }
    None
}

/// Verify an HTTP `Authorization: Basic` header against the site credentials.
fn verify_basic_auth(basic: &BasicAuth, req_headers: &HeaderMap) -> bool {
    use base64::Engine;
    let Some(encoded) = req_headers
        .get(header::AUTHORIZATION)
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.strip_prefix("Basic "))
    else {
        return false;
    };
    let Ok(decoded) = base64::engine::general_purpose::STANDARD.decode(encoded.trim()) else {
        return false;
    };
    let Ok(text) = String::from_utf8(decoded) else {
        return false;
    };
    match text.split_once(':') {
        Some((user, pass)) => basic.verify(user, pass),
        None => false,
    }
}

/// `401` with a `WWW-Authenticate: Basic` challenge.
fn basic_auth_challenge(basic: &BasicAuth) -> Response {
    let realm = basic.realm.replace(['"', '\\'], "");
    let mut headers = HeaderMap::new();
    if let Ok(value) = HeaderValue::from_str(&format!("Basic realm=\"{realm}\", charset=\"UTF-8\""))
    {
        headers.insert(header::WWW_AUTHENTICATE, value);
    }
    (
        StatusCode::UNAUTHORIZED,
        headers,
        "authentication required\n",
    )
        .into_response()
}

/// `429 Too Many Requests` with a `Retry-After`.
fn too_many_requests() -> Response {
    let mut headers = HeaderMap::new();
    headers.insert(header::RETRY_AFTER, HeaderValue::from_static("1"));
    (
        StatusCode::TOO_MANY_REQUESTS,
        headers,
        "rate limit exceeded\n",
    )
        .into_response()
}

/// Serve an immutable deployment by id under `/_deploy/<id>/...`. Like
/// [`serve_sites`], a single catch-all captures `<id>` or `<id>/<path...>`, so
/// `/_deploy/<id>`, `/_deploy/<id>/`, and `/_deploy/<id>/about` all route here.
pub(super) async fn serve_preview(
    State(deploy): State<DeployStore>,
    Extension(handlers): Extension<Arc<HandlerRuntime>>,
    Extension(daemon): Extension<Arc<DaemonRuntime>>,
    Extension(preview_auth): Extension<Auth>,
    ConnectInfo(peer): ConnectInfo<SocketAddr>,
    request: Request,
) -> Response {
    let preview_policy = PreviewPolicy {
        protect: daemon.effective().protect_previews,
    };
    if let Some(blocked) = preview_auth_gate(preview_policy, &preview_auth, request.headers()).await
    {
        return blocked;
    }
    let raw = request.uri().path();
    let rest = raw
        .strip_prefix("/_deploy/")
        .unwrap_or("")
        .trim_start_matches('/');
    let (id, path) = rest.split_once('/').unwrap_or((rest, ""));
    if id.is_empty() {
        return not_found();
    }
    let (id, request_path) = (id.to_string(), format!("/{path}"));
    // When the preview is reached via the *site's own hostname*
    // (`site.example.com/_deploy/<id>/…`), resolve that site so handlers can run
    // — with **preview-scoped** bindings (`Some(&id)` below) so they never touch
    // the live site's kv/blob/sql. Reached via any other host
    // (no site resolves), handlers stay off — the preview serves static only.
    let site = request
        .headers()
        .get(header::HOST)
        .and_then(|value| value.to_str().ok())
        .map(strip_port);
    let owner = match site {
        Some(host) => match deploy.resolve_site_by_host(host).await {
            Ok(owner) => owner,
            Err(err) => return deploy_error_response(err),
        },
        None => None,
    };
    let project = owner.as_ref().map(|o| o.project.clone());
    let site = owner.map(|o| o.site);
    let site_config = match (&project, &site) {
        (Some(project), Some(site)) => {
            match deploy.get_site_config(ProjectRef::new(project), site).await {
                Ok(config) => config,
                Err(err) => return deploy_error_response(err),
            }
        }
        _ => None,
    };
    match deploy.get_manifest(&id).await {
        Ok(Some(manifest)) => {
            serve_resolved(
                &deploy,
                &manifest,
                &request_path,
                request,
                peer.ip(),
                project.as_deref(),
                site.as_deref(),
                site_config.as_ref(),
                &handlers,
                Some(&id),
            )
            .await
        }
        Ok(None) => not_found(),
        Err(err) => deploy_error_response(err),
    }
}

/// Run the deploy-config routing pipeline against a resolved `manifest`, then
/// stream the chosen entry (or proxy). `client_ip` is the resolved visitor
/// address (for proxy `X-Forwarded-For`).
/// Build the [`RequestContext`](boatramp_core::predicate::RequestContext) a
/// conditional-routing `when` predicate reads from the live request. Only called
/// when the deployment actually has conditional rules, so the non-conditional hot
/// path never pays for it.
fn build_request_context(request: &Request) -> boatramp_core::predicate::RequestContext {
    use boatramp_core::predicate::RequestContext;
    let headers = request.headers();
    let mut hmap: std::collections::BTreeMap<String, String> = std::collections::BTreeMap::new();
    for (name, value) in headers {
        if let Ok(v) = value.to_str() {
            hmap.entry(name.as_str().to_ascii_lowercase())
                .and_modify(|e| {
                    e.push_str(", ");
                    e.push_str(v);
                })
                .or_insert_with(|| v.to_string());
        }
    }
    let host = headers
        .get(header::HOST)
        .and_then(|h| h.to_str().ok())
        .map(|h| h.split(':').next().unwrap_or(h).to_string())
        .unwrap_or_default();
    let cookies = headers
        .get(header::COOKIE)
        .and_then(|h| h.to_str().ok())
        .map(parse_cookie_header)
        .unwrap_or_default();
    let query = request
        .uri()
        .query()
        .map(parse_query_string)
        .unwrap_or_default();
    let accept_languages = headers
        .get(header::ACCEPT_LANGUAGE)
        .and_then(|h| h.to_str().ok())
        .map(RequestContext::parse_accept_language)
        .unwrap_or_default();
    RequestContext {
        method: request.method().as_str().to_ascii_uppercase(),
        host,
        headers: hmap,
        cookies,
        query,
        accept_languages,
    }
}

/// Parse a `Cookie` header into name→value pairs (first value wins).
pub(super) fn parse_cookie_header(raw: &str) -> std::collections::BTreeMap<String, String> {
    raw.split(';')
        .filter_map(|pair| pair.split_once('='))
        .map(|(k, v)| (k.trim().to_string(), v.trim().to_string()))
        .fold(std::collections::BTreeMap::new(), |mut m, (k, v)| {
            m.entry(k).or_insert(v);
            m
        })
}

/// Parse a URL query string into key→value pairs (first value wins), with
/// `application/x-www-form-urlencoded` decoding (`+` → space, `%XX` → byte) so a
/// condition compares against the real value.
pub(super) fn parse_query_string(raw: &str) -> std::collections::BTreeMap<String, String> {
    raw.split('&')
        .filter(|p| !p.is_empty())
        .map(|pair| match pair.split_once('=') {
            Some((k, v)) => (percent_decode(k), percent_decode(v)),
            None => (percent_decode(pair), String::new()),
        })
        .fold(std::collections::BTreeMap::new(), |mut m, (k, v)| {
            m.entry(k).or_insert(v);
            m
        })
}

/// Decode a `application/x-www-form-urlencoded` component: `+` → space, `%XX` →
/// the byte, everything else verbatim. Invalid `%` escapes are left as-is;
/// non-UTF-8 results are lossily replaced.
fn percent_decode(s: &str) -> String {
    let bytes = s.as_bytes();
    let mut out: Vec<u8> = Vec::with_capacity(bytes.len());
    let mut i = 0;
    while i < bytes.len() {
        match bytes[i] {
            b'+' => {
                out.push(b' ');
                i += 1;
            }
            b'%' if i + 2 < bytes.len() => {
                let hi = (bytes[i + 1] as char).to_digit(16);
                let lo = (bytes[i + 2] as char).to_digit(16);
                match (hi, lo) {
                    (Some(h), Some(l)) => {
                        out.push((h * 16 + l) as u8);
                        i += 3;
                    }
                    _ => {
                        out.push(b'%');
                        i += 1;
                    }
                }
            }
            b => {
                out.push(b);
                i += 1;
            }
        }
    }
    String::from_utf8_lossy(&out).into_owned()
}

/// Merge conditional-routing `Vary` header names into a response, so a per-visitor
/// (language/cookie/header) redirect or page is never shared across visitors by a
/// downstream cache. A no-op when `vary` is empty (the non-conditional case).
pub(super) fn apply_vary(mut response: Response, vary: &[String]) -> Response {
    if vary.is_empty() {
        return response;
    }
    let mut names: Vec<String> = response
        .headers()
        .get(header::VARY)
        .and_then(|v| v.to_str().ok())
        .map(|s| {
            s.split(',')
                .map(|p| p.trim().to_ascii_lowercase())
                .filter(|p| !p.is_empty())
                .collect()
        })
        .unwrap_or_default();
    for v in vary {
        if !names.iter().any(|n| n == v) {
            names.push(v.clone());
        }
    }
    if let Ok(hv) = HeaderValue::from_str(&names.join(", ")) {
        response.headers_mut().insert(header::VARY, hv);
    }
    response
}

#[allow(clippy::too_many_arguments)]
#[cfg_attr(not(feature = "handlers"), allow(unused_variables))]
async fn serve_resolved(
    deploy: &DeployStore,
    manifest: &Manifest,
    request_path: &str,
    request: Request,
    client_ip: IpAddr,
    // The tenant project the resolved site belongs to; `None` mirrors `site:
    // None` (a preview reached via a non-resolving host, handlers off).
    project: Option<&str>,
    site: Option<&str>,
    site_config: Option<&SiteConfig>,
    handlers: &HandlerRuntime,
    // `Some(deploy_id)` when serving a by-id preview, so handler bindings get a
    // preview-scoped identity; `None` for live serving.
    preview: Option<&str>,
) -> Response {
    // Evaluate conditional (`when`) routing against the request. The request
    // context is built only when the deploy has conditional rules, and `vary`
    // carries the request dimensions those conditions read (applied to the
    // response below so a per-language/-cookie outcome isn't wrongly cached).
    let ctx = if manifest.config.redirects.iter().any(|r| r.when.is_some())
        || manifest.config.rewrites.iter().any(|r| r.when.is_some())
    {
        build_request_context(&request)
    } else {
        boatramp_core::predicate::RequestContext::default()
    };
    let route::ResolveResult { outcome, vary } =
        route::resolve_ctx(&manifest.config, &manifest.files, request_path, &ctx);
    // Routing precedence: redirects win over handlers, which
    // win over rewrites/static. A redirect short-circuits below; otherwise a
    // matching handler is dispatched in preference to the file/rewrite outcome.
    #[cfg(feature = "handlers")]
    if !matches!(outcome, Outcome::Redirect { .. }) {
        if let Some(site) = site {
            if let Some(handler) = route::match_handler(
                &manifest.config.handlers,
                request.method().as_str(),
                request_path,
            ) {
                return apply_vary(
                    dispatch_handler(
                        handlers,
                        deploy,
                        manifest,
                        // Handler dispatch only runs when `site` is Some, and the
                        // project is threaded alongside it; default-project fallback
                        // keeps a by-id preview reached via a non-resolving host safe.
                        project.unwrap_or(ProjectRef::DEFAULT.as_str()),
                        site,
                        request_path,
                        site_config,
                        handler,
                        request,
                        client_ip,
                        preview,
                    )
                    .await,
                    &vary,
                );
            }
            // No handler matched: a GET to a configured SSE stream route fans out
            // its messaging topics. Streams are GET-only.
            if request.method() == Method::GET {
                if let Some(stream) = manifest
                    .config
                    .streams
                    .iter()
                    .find(|s| route_matches(&s.route, request_path))
                {
                    if let (Some(inner), Some(site_handlers)) = (
                        handlers.inner.as_ref(),
                        site_config
                            .and_then(|c| c.handlers.as_ref())
                            .filter(|h| h.enabled),
                    ) {
                        // A `websocket` stream upgraded by the client is served
                        // bidirectionally (WebSocket fan-out); otherwise it's SSE.
                        // serve_ws_stream does the RFC 6455 handshake + takes over the
                        // connection via boatramp-http's upgrade seam (the request body
                        // isn't held across an await — WS carries none).
                        if stream.websocket && is_upgrade_request(request.headers()) {
                            return apply_vary(
                                serve_ws_stream(
                                    inner,
                                    site,
                                    site_handlers,
                                    stream,
                                    request,
                                    client_ip,
                                    preview,
                                )
                                .await,
                                &vary,
                            );
                        }
                        // Pull the only field needed from the request as an owned
                        // value: `&Request` is not `Send` (the body isn't `Sync`),
                        // so it must not be held across the dispatch await.
                        let after = request
                            .headers()
                            .get("last-event-id")
                            .and_then(|value| value.to_str().ok())
                            .map(str::to_string);
                        return apply_vary(
                            serve_stream(
                                inner,
                                site,
                                site_handlers,
                                stream,
                                after,
                                client_ip,
                                preview,
                            )
                            .await,
                            &vary,
                        );
                    }
                    // A stream route on a site with handlers disabled / no runtime
                    // is not served (deny by default).
                    return apply_vary(not_found(), &vary);
                }
            }
        }
    }
    // Gateway: an operator-declared route forwards to a private
    // upstream. Independent of the handlers feature; runs after redirects/
    // handlers and **wins over static files** (the operator declared it). Access
    // control already ran up front; only declared upstreams reach private addrs.
    if !matches!(outcome, Outcome::Redirect { .. }) {
        if let Some(gw) = site_config
            .and_then(|c| c.gateway.as_ref())
            .filter(|g| g.is_enabled())
        {
            if let Some(route) = gw.match_route(request_path) {
                return apply_vary(
                    match gw.upstreams.get(&route.upstream) {
                        Some(upstream) => {
                            // A compute-backed upstream resolves its pool live from
                            // the workload's healthy replica endpoints. Record
                            // the request as activity so the reconcile loop
                            // keeps the workload warm / wakes it, and only sleeps it
                            // once genuinely idle.
                            let (compute_backends, compute_regions) = match &upstream.compute {
                                Some(workload) => {
                                    // Resolve the compute upstream against the site's OWN
                                    // project (not `default`), so a non-default tenant's
                                    // replica state is found — closing the project-blind
                                    // resolution that 502'd / never woke it.
                                    let compute_project =
                                        project.unwrap_or(ProjectRef::DEFAULT.as_str());
                                    gateway::record_activity(workload);
                                    let mut pool =
                                        compute_endpoints(deploy, compute_project, workload).await;
                                    // Wake-from-zero: no live replica but one
                                    // is parked → nudge the reconcile loop to restore it
                                    // and hold this request until it's serving. The cold
                                    // start is invisible to the client; only a genuine
                                    // restore failure (timeout) falls through to 502.
                                    if pool.is_empty()
                                        && has_parked_replica(deploy, compute_project, workload)
                                            .await
                                    {
                                        gateway::wake_reconcile();
                                        pool = await_warm(
                                            deploy,
                                            compute_project,
                                            workload,
                                            COMPUTE_WAKE_TIMEOUT,
                                        )
                                        .await;
                                    }
                                    // FA-8: for a nearest-region pool, tag each replica
                                    // endpoint with its node's region (from placement).
                                    let regions = if upstream.lb
                                        == boatramp_core::gateway::LbPolicy::Nearest
                                    {
                                        Some(
                                            compute_endpoint_regions(
                                                deploy,
                                                compute_project,
                                                workload,
                                            )
                                            .await,
                                        )
                                    } else {
                                        None
                                    };
                                    (Some(pool), regions)
                                }
                                None => (None, None),
                            };
                            dispatch_gateway(
                                request,
                                site.unwrap_or(""),
                                &route.upstream,
                                upstream,
                                request_path,
                                client_ip,
                                compute_backends,
                                compute_regions,
                            )
                            .await
                        }
                        None => (
                            StatusCode::BAD_GATEWAY,
                            "gateway route references an unknown upstream\n",
                        )
                            .into_response(),
                    },
                    &vary,
                );
            }
        }
    }
    // Plaintext (not TLS) gates the zero-copy `sendfile` static body. The listener
    // stamps `ServedOverTls`; absent (an unusual direct call) defaults to plaintext,
    // which is still correct — the codec only `sendfile`s a bare TCP socket and reads
    // the file through the (encrypting) write half otherwise.
    let plaintext = !request
        .extensions()
        .get::<ServedOverTls>()
        .map(|s| s.0)
        .unwrap_or(false);
    let response = match outcome {
        Outcome::Redirect { location, status } => redirect(status, &location),
        Outcome::Proxy { url } => proxy(request, &url, &manifest.config, client_ip).await,
        Outcome::File {
            path: served,
            entry,
        } => {
            // Static content answers only GET/HEAD; other methods are 405.
            if !matches!(*request.method(), Method::GET | Method::HEAD) {
                return apply_vary(method_not_allowed(), &vary);
            }
            serve_entry(
                deploy,
                &manifest.config,
                request_path,
                &served,
                &entry,
                request.headers(),
                StatusCode::OK,
                plaintext,
            )
            .await
        }
        Outcome::NotFound { error } => match error {
            Some((served, entry)) => {
                serve_entry(
                    deploy,
                    &manifest.config,
                    request_path,
                    &served,
                    &entry,
                    request.headers(),
                    StatusCode::NOT_FOUND,
                    plaintext,
                )
                .await
            }
            None => not_found(),
        },
    };
    apply_vary(response, &vary)
}

/// `405` for a non-`GET`/`HEAD` request to static content.
fn method_not_allowed() -> Response {
    let mut headers = HeaderMap::new();
    headers.insert(header::ALLOW, HeaderValue::from_static("GET, HEAD"));
    (
        StatusCode::METHOD_NOT_ALLOWED,
        headers,
        "method not allowed\n",
    )
        .into_response()
}

/// A large static blob to serve zero-copy via `sendfile` — attached as a response
/// extension (survives `into_parts`), the [`http_serve`](crate::http_serve) bridge
/// turns it into a [`boatramp_http::Body::File`] the h1 codec moves kernel-to-kernel.
/// Only produced for plaintext connections; TLS keeps the memory-mapped path.
#[derive(Clone)]
pub(crate) struct SendfileSource {
    // `Arc` so this can ride an `http::Extensions` (which requires `Clone`).
    pub(crate) file: std::sync::Arc<std::fs::File>,
    pub(crate) offset: u64,
    pub(crate) len: u64,
}

/// Smallest blob that `sendfile`s rather than serving from the in-memory small-blob
/// cache. There is a crossover: `sendfile` avoids the socket-buffer page-alloc/zero cost
/// (`clear_page`) but adds an `open` + the `sendfile` syscall dance per request, a losing
/// trade for a tiny cache-resident body at very high request rates. Measured on the EPYC
/// benchmark box, the knee is ~32 KiB (≤16 KiB the cache wins by ~10–18%; ≥64 KiB
/// `sendfile` wins by ~11% at 64 KiB, ~19% at 100 KiB, ~85% at 1 MiB). Default 64 KiB —
/// captures the clear medium+ wins while leaving small bodies and the marginal 32–64 KiB
/// range on the cache. Tune with `BOATRAMP_SENDFILE_MIN_KB` (floored at 4 KiB).
static SENDFILE_MIN_BYTES: std::sync::LazyLock<u64> = std::sync::LazyLock::new(|| {
    std::env::var("BOATRAMP_SENDFILE_MIN_KB")
        .ok()
        .and_then(|v| v.parse::<u64>().ok())
        .map_or(64 * 1024, |kb| kb.saturating_mul(1024).max(4 * 1024))
});

/// Kill-switch for the zero-copy `sendfile` static path (default on). `BOATRAMP_SENDFILE=0`
/// (or `false`) forces the memory-mapped body path — a safety valve for the hot path and
/// the knob for a same-binary A/B of the two.
static SENDFILE_ENABLED: std::sync::LazyLock<bool> = std::sync::LazyLock::new(|| {
    std::env::var("BOATRAMP_SENDFILE")
        .map(|v| v != "0" && !v.eq_ignore_ascii_case("false"))
        .unwrap_or(true)
});

/// Stream a resolved entry, applying conditional/range/headers. `base_status` is
/// `200` for a normal hit and `404` for a custom error document. `plaintext` is true
/// when the connection is not TLS — the gate for the zero-copy `sendfile` body.
#[allow(clippy::too_many_arguments)]
async fn serve_entry(
    deploy: &DeployStore,
    config: &DeployConfig,
    request_path: &str,
    served_path: &str,
    entry: &FileEntry,
    req_headers: &HeaderMap,
    base_status: StatusCode,
    plaintext: bool,
) -> Response {
    let is_range = base_status == StatusCode::OK && req_headers.contains_key(header::RANGE);

    // Content-encoding negotiation. Range requests are served from the identity
    // representation (Range over a compressed variant is intentionally avoided).
    let chosen = if is_range {
        None
    } else {
        negotiate_encoding(entry, req_headers)
    };
    let (blob_hash, blob_size, encoding) = match chosen {
        Some((enc, variant)) => (variant.hash.as_str(), variant.size, Some(enc)),
        None => (entry.hash.as_str(), entry.size, None),
    };
    // ETag is per-representation (identity vs br vs gzip differ in bytes).
    let etag = format!("\"{blob_hash}\"");

    // Conditional GET — content hash is a strong validator.
    if base_status == StatusCode::OK && if_none_match(req_headers, &etag) {
        let mut headers = response_headers(config, request_path, served_path, entry, &etag);
        set_content_encoding(&mut headers, encoding);
        return (StatusCode::NOT_MODIFIED, headers).into_response();
    }

    // Range request (identity only).
    if is_range {
        if let Some(spec) = req_headers
            .get(header::RANGE)
            .and_then(|value| value.to_str().ok())
        {
            match parse_ranges(spec, entry.size) {
                // A single range → `206` with `Content-Range`, streamed.
                Some(ranges) if ranges.len() == 1 => {
                    let (offset, len) = ranges[0];
                    let object = match deploy.open_blob_range(&entry.hash, offset, Some(len)).await
                    {
                        Ok(object) => object,
                        Err(err) => return deploy_error_response(err),
                    };
                    let mut headers =
                        response_headers(config, request_path, served_path, entry, &etag);
                    set_header(&mut headers, header::CONTENT_LENGTH, &len.to_string());
                    set_header(
                        &mut headers,
                        header::CONTENT_RANGE,
                        &format!("bytes {}-{}/{}", offset, offset + len - 1, entry.size),
                    );
                    return (
                        StatusCode::PARTIAL_CONTENT,
                        headers,
                        Body::from_stream(object.body),
                    )
                        .into_response();
                }
                // Several ranges → `206 multipart/byteranges`, streamed.
                Some(ranges) if ranges.len() <= MAX_RANGES => {
                    return multipart_byteranges(
                        deploy,
                        config,
                        request_path,
                        served_path,
                        entry,
                        &etag,
                        &ranges,
                    )
                    .await;
                }
                // Too many ranges: ignore `Range`, serve the full `200` body.
                Some(_) => {}
                // Malformed / wholly unsatisfiable → `416`.
                None => {
                    let mut headers = HeaderMap::new();
                    set_header(
                        &mut headers,
                        header::CONTENT_RANGE,
                        &format!("bytes */{}", entry.size),
                    );
                    return (StatusCode::RANGE_NOT_SATISFIABLE, headers).into_response();
                }
            }
        }
    }

    // Full body (identity or negotiated variant). Small blobs are served from the
    // in-memory body cache as one refcounted frame (no open/ReaderStream/disk read);
    // larger blobs stream straight from storage.
    let mut headers = response_headers(config, request_path, served_path, entry, &etag);
    set_header(&mut headers, header::CONTENT_LENGTH, &blob_size.to_string());
    set_content_encoding(&mut headers, encoding);

    // Zero-copy `sendfile` fast-path: a large static blob on a local backend served
    // over a plaintext connection is moved kernel-to-kernel by the h1 codec (no
    // userspace copy — what nginx/caddy do). Hand an empty body + the file source as an
    // extension; the `http_serve` bridge turns it into a `Body::File`. TLS, remote
    // backends, and small (cached) blobs fall through to the mapped/cached/stream path,
    // byte-identical to before. HEAD is unaffected (the codec suppresses the body).
    if *SENDFILE_ENABLED && plaintext && blob_size > *SENDFILE_MIN_BYTES {
        if let Some(file) = deploy.blob_file(blob_hash) {
            let mut resp = (base_status, headers, axum::body::Body::empty()).into_response();
            resp.extensions_mut().insert(SendfileSource {
                file: std::sync::Arc::new(file),
                offset: 0,
                len: blob_size,
            });
            return resp;
        }
    }

    match deploy.open_blob_cached(blob_hash, blob_size).await {
        // Cached (small) and Mapped (large, memory-mapped file) both serve one
        // borrowed frame as a content-length body.
        Ok(boatramp_core::deploy::BlobBody::Cached(bytes))
        | Ok(boatramp_core::deploy::BlobBody::Mapped(bytes)) => {
            (base_status, headers, Body::from(bytes)).into_response()
        }
        Ok(boatramp_core::deploy::BlobBody::Stream(object)) => {
            (base_status, headers, Body::from_stream(object.body)).into_response()
        }
        Err(err) => deploy_error_response(err),
    }
}