autumn-web 0.9.0

//! The AEO response layer.
//!
//! - `Accept: text/markdown` on a `GET` for an HTML page returns Markdown.
//! - Every negotiable HTML response carries `Vary: Accept`, so a cache keeps
//!   the two forms apart.
//! - The homepage (`/`) carries the agent `Link` header.
//!
//! The layer sits inside response compression, so it reads plain bytes.

use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};

use axum::body::{Body, Bytes};
use axum::http::header::{
    CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LENGTH, CONTENT_RANGE, CONTENT_TYPE, ETAG,
    IF_MODIFIED_SINCE, IF_NONE_MATCH, LAST_MODIFIED, LINK, VARY,
};
use axum::http::{HeaderMap, HeaderValue, Method, Request, StatusCode};
use axum::response::Response;
use futures::StreamExt as _;
use http_body_util::BodyExt as _;

/// Settings the layer needs, built once per router.
#[derive(Debug, Clone, Default)]
pub struct NegotiateConfig {
    /// Convert HTML to Markdown on request.
    pub markdown: bool,
    /// Largest HTML body to convert.
    pub max_bytes: usize,
    /// The agent `Link` header of each home page: `/`, or `/{locale}` for
    /// each locale when the root is locale-prefixed.
    pub home_links: Vec<(String, HeaderValue)>,
    /// `Content-Signal` header for Markdown responses.
    pub content_signal: Option<HeaderValue>,
    /// `WWW-Authenticate` value for a `401` (RFC 9728 `resource_metadata`).
    pub resource_metadata: Option<HeaderValue>,
    /// Build the `401` hint from the request `Host` (OAuth is set, `[seo]
    /// base_url` is not), as the metadata document itself is.
    pub resource_metadata_from_host: bool,
}

impl NegotiateConfig {
    /// The index of `path` in [`Self::home_links`], when it is a home page.
    fn home(&self, path: &str) -> Option<usize> {
        self.home_links.iter().position(|(home, _)| home == path)
    }
}

/// Tower layer for [`NegotiateConfig`].
#[derive(Debug, Clone)]
pub struct NegotiateLayer {
    config: Arc<NegotiateConfig>,
}

impl NegotiateLayer {
    /// Build the layer.
    #[must_use]
    pub fn new(config: NegotiateConfig) -> Self {
        Self {
            config: Arc::new(config),
        }
    }
}

impl<S> tower::Layer<S> for NegotiateLayer {
    type Service = NegotiateService<S>;

    fn layer(&self, inner: S) -> Self::Service {
        NegotiateService {
            inner,
            config: Arc::clone(&self.config),
        }
    }
}

/// Service made by [`NegotiateLayer`].
#[derive(Debug, Clone)]
pub struct NegotiateService<S> {
    inner: S,
    config: Arc<NegotiateConfig>,
}

impl<S> tower::Service<Request<Body>> for NegotiateService<S>
where
    S: tower::Service<Request<Body>, Response = Response>,
    S::Future: Send + 'static,
{
    type Response = Response;
    type Error = S::Error;
    type Future = NegotiateFuture<S::Future>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, mut req: Request<Body>) -> Self::Future {
        let is_get = req.method() == Method::GET;
        let readable = is_get || req.method() == Method::HEAD;
        let flags = Flags {
            home: if readable {
                self.config.home(req.uri().path())
            } else {
                None
            },
            readable,
            is_head: !is_get && readable,
        };
        // The `Accept` to rank the response by, kept when Markdown may win.
        let accept = (self.config.markdown && readable)
            .then(|| accept_values(req.headers()))
            .filter(|accept| markdown_possible(accept));
        // A Markdown request's validators are checked here, against the
        // representation sent (Markdown, or the response as it is), not by
        // the handler against its HTML.
        let validators = if accept.is_some() {
            Validators::take(req.headers_mut())
        } else {
            None
        };
        // The host and scheme the client used, for a `401` hint built from
        // them: a trusted proxy's forwarded ones, else `Host`.
        let host = self
            .config
            .resource_metadata_from_host
            .then(|| super::documents::request_authority(&req));
        let page = accept.is_some().then(|| {
            req.uri()
                .path_and_query()
                .map_or_else(|| req.uri().path().to_owned(), ToString::to_string)
        });
        // The future holds only the inner future. It does not clone the
        // service. It boxes only a page that it converts.
        NegotiateFuture {
            inner: self.inner.call(req),
            config: Arc::clone(&self.config),
            flags,
            validators,
            host,
            page,
            accept,
            convert: None,
        }
    }
}

#[derive(Debug, Clone, Copy)]
#[allow(clippy::struct_excessive_bools)] // request facts, read once
struct Flags {
    /// The request's entry in `home_links`.
    home: Option<usize>,
    readable: bool,
    is_head: bool,
}

/// The charset a `charset` value names, by its common labels (the WHATWG
/// Encoding Standard's for UTF-8; IANA's for US-ASCII). `None` for any
/// other charset.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Charset {
    Utf8,
    Ascii,
}

impl Charset {
    fn label(value: &str) -> Option<Self> {
        let value = value.trim().trim_matches('"');
        if ["utf-8", "utf8", "unicode-1-1-utf-8"]
            .iter()
            .any(|l| value.eq_ignore_ascii_case(l))
        {
            Some(Self::Utf8)
        } else if ["us-ascii", "ascii", "iso646-us", "ansi_x3.4-1968"]
            .iter()
            .any(|l| value.eq_ignore_ascii_case(l))
        {
            Some(Self::Ascii)
        } else {
            None
        }
    }
}

type ConvertFuture = Pin<Box<dyn Future<Output = Response> + Send>>;

/// The conditional headers of a Markdown request.
#[derive(Debug)]
struct Validators {
    if_none_match: Vec<HeaderValue>,
    if_modified_since: Option<HeaderValue>,
}

impl Validators {
    /// Remove the request's validators, if it has any.
    fn take(headers: &mut HeaderMap) -> Option<Box<Self>> {
        let if_none_match: Vec<HeaderValue> = match headers.entry(IF_NONE_MATCH) {
            axum::http::header::Entry::Occupied(e) => e.remove_entry_mult().1.collect(),
            axum::http::header::Entry::Vacant(_) => Vec::new(),
        };
        let if_modified_since = headers.remove(IF_MODIFIED_SINCE);
        (!if_none_match.is_empty() || if_modified_since.is_some()).then(|| {
            Box::new(Self {
                if_none_match,
                if_modified_since,
            })
        })
    }

    /// `res` as a `304` when these validators match it. A `206` carries the
    /// whole representation's validators, and a condition is evaluated
    /// before `Range` (RFC 9110 §13.2.2).
    fn apply(&self, res: Response) -> Response {
        if matches!(res.status(), StatusCode::OK | StatusCode::PARTIAL_CONTENT)
            && crate::etag::validators_match(
                &self.if_none_match,
                self.if_modified_since.as_ref(),
                res.headers(),
            )
        {
            crate::etag::not_modified_from(res.headers())
        } else {
            res
        }
    }
}

pin_project_lite::pin_project! {
    /// Future of [`NegotiateService`].
    pub struct NegotiateFuture<F> {
        #[pin]
        inner: F,
        config: Arc<NegotiateConfig>,
        flags: Flags,
        validators: Option<Box<Validators>>,
        host: Option<(Option<String>, Option<String>)>,
        // The request path and query, for a relative `<base href>`.
        page: Option<String>,
        // The request's `Accept`, when Markdown may win: the response's own
        // type decides.
        accept: Option<Vec<HeaderValue>>,
        convert: Option<ConvertFuture>,
    }
}

impl<F, E> Future for NegotiateFuture<F>
where
    F: Future<Output = Result<Response, E>>,
{
    type Output = Result<Response, E>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();
        if let Some(convert) = this.convert.as_mut() {
            return convert.as_mut().poll(cx).map(Ok);
        }
        let mut res = std::task::ready!(this.inner.poll(cx))?;
        let config = &**this.config;
        if res.extensions().get::<super::AeoDocument>().is_some() {
            res.headers_mut().remove(axum::http::header::SET_COOKIE);
        }
        if res.status() == StatusCode::UNAUTHORIZED {
            let hint = config.resource_metadata.clone().or_else(|| {
                config
                    .resource_metadata_from_host
                    .then(|| {
                        let (host, scheme) = this
                            .host
                            .as_ref()
                            .map_or((None, None), |(h, s)| (h.as_deref(), s.as_deref()));
                        host_resource_metadata(host, scheme)
                    })
                    .flatten()
            });
            if let Some(hint) = hint {
                add_resource_metadata(res.headers_mut(), &hint);
            }
        }
        if res.status().is_success()
            && let Some((_, link)) = this.flags.home.and_then(|i| config.home_links.get(i))
        {
            res.headers_mut().append(LINK, link.clone());
        }
        let validators = this.validators.take();
        let page = this.page.take();
        let checked = |res: Response| match &validators {
            Some(v) => v.apply(res),
            None => res,
        };
        if !(config.markdown && this.flags.readable && is_negotiable(&res)) {
            let mut res = checked(res);
            // A `304` has no type to show whether its `200` was a page this
            // layer negotiates, so it carries the `Vary: Accept` that `200`
            // may carry: a cache then keeps the HTML and Markdown apart.
            if config.markdown && this.flags.readable && res.status() == StatusCode::NOT_MODIFIED {
                add_vary_accept(res.headers_mut());
            }
            return Poll::Ready(Ok(res));
        }
        add_vary_accept(res.headers_mut());
        let content_type = res
            .headers()
            .get(CONTENT_TYPE)
            .and_then(|v| v.to_str().ok())
            .unwrap_or_default();
        if !this
            .accept
            .as_deref()
            .is_some_and(|accept| markdown_wins(accept, content_type))
        {
            // The validators, if taken, are checked against the HTML.
            return Poll::Ready(Ok(checked(res)));
        }
        // HEAD carries the headers a GET would get. A handler's HEAD body is
        // still here (the router drops it after this layer), so it converts
        // as GET does and the router sets the Markdown's `Content-Length`.
        // With the body already gone, only a known length shows that GET's
        // body fits: a stream of unknown length may outgrow the limit, and
        // GET then sends HTML.
        let body_here = {
            use http_body::Body as _;
            res.body().size_hint().exact().is_some_and(|n| n > 0)
        };
        if this.flags.is_head && !body_here {
            if known_to_fit(&res, config.max_bytes) {
                markdown_headers(res.headers_mut(), config, None);
            }
            return Poll::Ready(Ok(checked(res)));
        }
        let config = Arc::clone(this.config);
        let mut convert: ConvertFuture = Box::pin(async move {
            let res = to_markdown(res, &config, page.as_deref()).await;
            match validators {
                Some(v) => v.apply(res),
                None => res,
            }
        });
        let poll = convert.as_mut().poll(cx);
        *this.convert = Some(convert);
        poll.map(Ok)
    }
}

/// The `401` hint for a site with no `[seo] base_url`: the metadata URL at
/// the request's own origin.
fn host_resource_metadata(host: Option<&str>, scheme: Option<&str>) -> Option<HeaderValue> {
    let origin = super::documents::Origin::resolve_with_scheme(None, host, scheme);
    HeaderValue::from_str(&format!(
        "Bearer resource_metadata=\"{}\"",
        origin.url(super::documents::OAUTH_RESOURCE_PATH)
    ))
    .ok()
}

/// Point a `401` at the RFC 9728 metadata. With no challenge, `hint` (a
/// `Bearer resource_metadata=...` challenge) is the challenge. A `Bearer`
/// challenge of the handler gets the `resource_metadata` parameter, and
/// keeps its own. Any other scheme is left alone.
fn add_resource_metadata(headers: &mut HeaderMap, hint: &HeaderValue) {
    use axum::http::header::WWW_AUTHENTICATE;

    if !headers.contains_key(WWW_AUTHENTICATE) {
        headers.insert(WWW_AUTHENTICATE, hint.clone());
        return;
    }
    let Some(param) = hint.to_str().ok().and_then(|h| h.strip_prefix("Bearer ")) else {
        return;
    };
    let merged: Vec<HeaderValue> = headers
        .get_all(WWW_AUTHENTICATE)
        .iter()
        .map(|v| {
            v.to_str()
                .ok()
                .and_then(|s| with_bearer_param(s, param))
                .and_then(|s| HeaderValue::from_str(&s).ok())
                .unwrap_or_else(|| v.clone())
        })
        .collect();
    headers.remove(WWW_AUTHENTICATE);
    for v in merged {
        headers.append(WWW_AUTHENTICATE, v);
    }
}

/// `challenges` with `param` added to its `Bearer` challenge, or `None` when
/// it has no `Bearer` challenge or already names `resource_metadata`.
fn with_bearer_param(challenges: &str, param: &str) -> Option<String> {
    if challenges
        .to_ascii_lowercase()
        .contains("resource_metadata")
    {
        return None;
    }
    let items: Vec<&str> = split_unquoted(challenges, ',')
        .into_iter()
        .map(str::trim)
        .collect();
    // An item opens a challenge when its first token is not followed by `=`.
    let opens = |item: &str| {
        let token_end = item
            .find(|c: char| c == '=' || c.is_whitespace())
            .unwrap_or(item.len());
        !item[token_end..].trim_start().starts_with('=')
    };
    let first = items.iter().position(|item| {
        opens(item)
            && item
                .split_whitespace()
                .next()
                .is_some_and(|s| s.eq_ignore_ascii_case("Bearer"))
    })?;
    let end = items[first + 1..]
        .iter()
        .position(|item| opens(item))
        .map_or(items.len(), |n| first + 1 + n);
    let mut out: Vec<String> = items.iter().map(|s| (*s).to_owned()).collect();
    if out[first].trim().eq_ignore_ascii_case("Bearer") {
        out[first] = format!("Bearer {param}");
    } else {
        out.insert(end, param.to_owned());
    }
    Some(out.join(", "))
}

/// Rewrite the HTML `ETag` to its Markdown form, `W/"<tag>-md"`.
fn markdown_etag(headers: &mut HeaderMap) {
    if let Some(etag) = headers.get(ETAG).and_then(|v| v.to_str().ok()) {
        let opaque = etag.trim_start_matches("W/").trim_matches('"');
        match HeaderValue::from_str(&format!("W/\"{opaque}-md\"")) {
            Ok(v) => {
                headers.insert(ETAG, v);
            }
            Err(_) => {
                headers.remove(ETAG);
            }
        }
    }
}

/// `true` when `Accept` may prefer Markdown to a page's HTML: it does for
/// UTF-8 or US-ASCII HTML, or it weighs HTML by a parameter only the page's
/// own type can show. The response's type then decides.
#[must_use]
pub fn prefers_markdown(headers: &HeaderMap) -> bool {
    markdown_possible(&accept_values(headers))
}

/// The request's `Accept` values.
fn accept_values(headers: &HeaderMap) -> Vec<HeaderValue> {
    headers
        .get_all(axum::http::header::ACCEPT)
        .iter()
        .cloned()
        .collect()
}

/// See [`prefers_markdown`].
fn markdown_possible(accept: &[HeaderValue]) -> bool {
    markdown_wins(accept, "text/html")
        || markdown_wins(accept, "text/html; charset=us-ascii")
        || (ranges(accept).any(|(media, params, _)| {
            !media.eq_ignore_ascii_case("text/markdown")
                && params
                    .iter()
                    .any(|(k, _)| !k.eq_ignore_ascii_case("charset"))
        }) && weights(accept, "text/html").0 > 0.0)
}

/// `true` when `Accept` names `text/markdown` with a weight at least as
/// high as the weight of the HTML of type `html_type`. Each weight comes
/// from the most specific range that covers the representation.
fn markdown_wins(accept: &[HeaderValue], html_type: &str) -> bool {
    let (markdown, html) = weights(accept, html_type);
    // Markdown must be named: `*/*` alone still gets HTML.
    markdown > 0.0 && markdown >= html
}

/// The media ranges of `accept`: type, parameters before `q`, and `q`.
fn ranges(accept: &[HeaderValue]) -> impl Iterator<Item = (&str, Vec<(&str, &str)>, f32)> {
    accept
        .iter()
        .filter_map(|v| v.to_str().ok())
        .flat_map(|v| split_unquoted(v, ','))
        .map(|range| {
            let parts = split_unquoted(range, ';');
            let media = parts.first().copied().unwrap_or("").trim();
            let q = accept_weight(parts.iter().skip(1).copied());
            let params = parts
                .into_iter()
                .skip(1)
                .map(str::trim)
                .filter(|p| !p.is_empty())
                .take_while(|p| {
                    !p.split_once('=')
                        .is_some_and(|(k, _)| k.trim().eq_ignore_ascii_case("q"))
                })
                .map(|p| {
                    p.split_once('=')
                        .map_or((p, ""), |(k, v)| (k.trim(), v.trim().trim_matches('"')))
                })
                .collect();
            (media, params, q)
        })
}

/// `s` split on `sep` outside quoted strings (RFC 9110 §5.6.4), so a
/// quoted parameter value may hold a `,` or `;`.
fn split_unquoted(s: &str, sep: char) -> Vec<&str> {
    let mut items = Vec::new();
    let (mut start, mut quoted, mut escaped) = (0, false, false);
    for (i, c) in s.char_indices() {
        match c {
            _ if escaped => escaped = false,
            '\\' if quoted => escaped = true,
            '"' => quoted = !quoted,
            c if c == sep && !quoted => {
                items.push(&s[start..i]);
                start = i + c.len_utf8();
            }
            _ => {}
        }
    }
    items.push(&s[start..]);
    items
}

/// The parameters of a representation's type. HTML with no `charset` is
/// UTF-8.
fn type_params(content_type: &str) -> Vec<(&str, &str)> {
    let mut params: Vec<(&str, &str)> = split_unquoted(content_type, ';')
        .into_iter()
        .skip(1)
        .filter_map(|p| p.split_once('='))
        .map(|(k, v)| (k.trim(), v.trim().trim_matches('"')))
        .collect();
    if !params
        .iter()
        .any(|(k, _)| k.eq_ignore_ascii_case("charset"))
    {
        params.push(("charset", "utf-8"));
    }
    params
}

/// The weights `accept` gives the Markdown sent and the HTML of type
/// `html_type`. RFC 9110 §12.5.1: the most specific range that matches
/// gives the quality. A range names a type (`text/html` over `text/*` over
/// `*/*`), then parameters: `text/markdown;variant=GFM` overrides
/// `text/markdown`.
fn weights(accept: &[HeaderValue], html_type: &str) -> (f32, f32) {
    const MARKDOWN: &[(&str, &str)] = &[("charset", "utf-8"), ("variant", "GFM")];
    let html_params = type_params(html_type);
    let mut markdown = ((0_u8, 0_usize), 0.0_f32);
    let mut html = ((0_u8, 0_usize), 0.0_f32);
    let take = |best: &mut ((u8, usize), f32), specificity: (u8, usize), q: f32| {
        if specificity > best.0 || (specificity == best.0 && q > best.1) {
            *best = (specificity, q);
        }
    };
    for (media, params, q) in ranges(accept) {
        let markdown_range = media.eq_ignore_ascii_case("text/markdown");
        let level = if markdown_range || media.eq_ignore_ascii_case("text/html") {
            3
        } else if media.eq_ignore_ascii_case("text/*") {
            2
        } else if media == "*/*" {
            1
        } else {
            continue;
        };
        // A range whose parameters the representation does not have (another
        // `charset`, `variant`, or `profile`) does not cover it.
        let sent = if markdown_range {
            MARKDOWN
        } else {
            &html_params[..]
        };
        if !params.iter().all(|&(k, v)| has_param(sent, k, v)) {
            continue;
        }
        let best = if markdown_range {
            &mut markdown
        } else {
            &mut html
        };
        take(best, (level, params.len()), q);
    }
    (markdown.1, html.1)
}

/// `true` when `sent` has parameter `key` with value `value`: a charset by
/// what its label names, `variant` ignoring case, any other by its value.
fn has_param(sent: &[(&str, &str)], key: &str, value: &str) -> bool {
    sent.iter().any(|&(k, v)| {
        k.eq_ignore_ascii_case(key)
            && if key.eq_ignore_ascii_case("charset") {
                match (Charset::label(value), Charset::label(v)) {
                    (Some(a), Some(b)) => a == b,
                    _ => value.eq_ignore_ascii_case(v),
                }
            } else if key.eq_ignore_ascii_case("variant") {
                value.eq_ignore_ascii_case(v)
            } else {
                value == v
            }
    })
}

/// The `q` weight of one media range (RFC 9110 §12.4.2). A missing `q` is
/// 1; a `q` that does not parse is 0.
fn accept_weight<'a>(params: impl Iterator<Item = &'a str>) -> f32 {
    for param in params {
        let Some((key, value)) = param.split_once('=') else {
            continue;
        };
        if key.trim().eq_ignore_ascii_case("q") {
            return value
                .trim()
                .parse::<f32>()
                .map_or(0.0, |q| q.clamp(0.0, 1.0));
        }
    }
    1.0
}

/// A whole `200` HTML document: not encoded, not a range, not a download,
/// and not signed. An RFC 9421 signature can cover the body's digest or its
/// type, and the Markdown would carry a signature that no longer verifies.
fn is_negotiable(res: &Response) -> bool {
    let headers = res.headers();
    res.status() == StatusCode::OK
        && !headers.contains_key("signature")
        && !headers.contains_key("signature-input")
        && headers
            .get(CONTENT_TYPE)
            .and_then(|v| v.to_str().ok())
            .is_some_and(utf8_html)
        && !headers.contains_key(CONTENT_ENCODING)
        && !headers.contains_key(CONTENT_RANGE)
        && !headers
            .get(CONTENT_DISPOSITION)
            .and_then(|v| v.to_str().ok())
            .is_some_and(|v| {
                v.trim_start()
                    .to_ascii_lowercase()
                    .starts_with("attachment")
            })
}

/// `text/html` whose bytes read as UTF-8: no charset, or `utf-8` or
/// `us-ascii`. The converter reads UTF-8, and the Markdown says UTF-8.
fn utf8_html(content_type: &str) -> bool {
    let mut parts = split_unquoted(content_type, ';').into_iter();
    let html = parts
        .next()
        .is_some_and(|m| m.trim().eq_ignore_ascii_case("text/html"));
    html && parts.all(|param| {
        let Some((name, value)) = param.split_once('=') else {
            return true;
        };
        !name.trim().eq_ignore_ascii_case("charset") || Charset::label(value).is_some()
    })
}

fn add_vary_accept(headers: &mut HeaderMap) {
    let present = headers.get_all(VARY).iter().any(|v| {
        v.to_str().is_ok_and(|v| {
            v.split(',')
                .any(|p| p.trim() == "*" || p.trim().eq_ignore_ascii_case("accept"))
        })
    });
    if !present {
        headers.append(VARY, HeaderValue::from_static("Accept"));
    }
}

/// The declared `Content-Length`.
fn content_length(res: &Response) -> Option<u64> {
    res.headers()
        .get(CONTENT_LENGTH)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.parse::<u64>().ok())
}

/// `true` when the body has an exact size that fits in `max`. A stream is not
/// collected, even one that declares a `Content-Length`: this layer runs
/// outside the request timeout and the admission limits, so a slow stream
/// would hold the request open. An exact size is not proof the bytes are in
/// memory, so [`collect_limited`] also gives up after [`COLLECT_WITHIN`].
fn fits(res: &Response, max: usize) -> bool {
    use http_body::Body as _;

    res.body()
        .size_hint()
        .exact()
        .is_some_and(|len| len <= max as u64)
}

/// `true` when a `HEAD` response's `GET` body is known to fit in `max`.
/// The `HEAD` body may already be empty, so an empty exact size says
/// nothing; a `Content-Length`, even `0`, or a non-empty exact size does.
fn known_to_fit(res: &Response, max: usize) -> bool {
    use http_body::Body as _;

    content_length(res)
        .or_else(|| res.body().size_hint().exact().filter(|&n| n > 0))
        .is_some_and(|len| len <= max as u64)
}

async fn to_markdown(res: Response, config: &NegotiateConfig, page: Option<&str>) -> Response {
    if !fits(&res, config.max_bytes) {
        return res;
    }
    let (mut parts, body) = res.into_parts();
    let bytes = match collect_limited(body, config.max_bytes, COLLECT_WITHIN).await {
        Ok(bytes) => bytes,
        Err(body) => return Response::from_parts(parts, body),
    };
    let markdown = if bytes.len() > BLOCKING_THRESHOLD {
        // A large page converts off the async worker threads.
        let html = bytes.clone();
        let page = page.map(ToOwned::to_owned);
        match crate::time::spawn_blocking(move || {
            super::markdown::html_to_markdown_at(&String::from_utf8_lossy(&html), page.as_deref())
        })
        .await
        {
            Ok(markdown) => markdown,
            Err(_) => return Response::from_parts(parts, Body::from(bytes)),
        }
    } else {
        super::markdown::html_to_markdown_at(&String::from_utf8_lossy(&bytes), page)
    };
    markdown_headers(
        &mut parts.headers,
        config,
        Some(super::markdown::estimate_tokens(&markdown)),
    );
    Response::from_parts(parts, Body::from(markdown))
}

/// How long Markdown conversion waits for an HTML body. A body in memory is
/// ready at once; one that is still being produced past this goes out as HTML.
const COLLECT_WITHIN: std::time::Duration = std::time::Duration::from_secs(2);

/// Pages larger than this convert on a blocking thread.
const BLOCKING_THRESHOLD: usize = 256 * 1024;

/// Turn HTML response headers into Markdown ones. `tokens` is `None` for
/// `HEAD`, which has no body to count.
fn markdown_headers(headers: &mut HeaderMap, config: &NegotiateConfig, tokens: Option<usize>) {
    headers.insert(
        CONTENT_TYPE,
        // RFC 7763: the variant is GitHub Flavored Markdown (its tables and
        // strikethrough), which a client can name in `Accept`.
        HeaderValue::from_static("text/markdown; charset=utf-8; variant=GFM"),
    );
    headers.remove(CONTENT_LENGTH);
    headers.remove(LAST_MODIFIED);
    // These describe the HTML bytes, not the Markdown.
    for name in ["content-digest", "repr-digest", "digest", "content-md5"] {
        headers.remove(name);
    }
    headers.remove(axum::http::header::ACCEPT_RANGES);
    // The HTML's trailers are dropped with its body; none follow the Markdown.
    headers.remove(axum::http::header::TRAILER);
    markdown_etag(headers);
    // Some CDNs ignore `Vary` on HTML. With no cache policy from the app,
    // keep the Markdown copy out of shared caches.
    if !headers.contains_key(axum::http::header::CACHE_CONTROL) {
        headers.insert(
            axum::http::header::CACHE_CONTROL,
            HeaderValue::from_static("private"),
        );
    }
    if let Some(tokens) = tokens {
        headers.insert("x-markdown-tokens", HeaderValue::from(tokens));
    }
    if let Some(signal) = &config.content_signal {
        headers.insert("content-signal", signal.clone());
    }
}

/// Read `body` up to `limit` bytes, waiting at most `within`. Over the limit,
/// or past the deadline, return a body that sends the bytes read, then the
/// remaining bytes.
async fn collect_limited(
    mut body: Body,
    limit: usize,
    within: std::time::Duration,
) -> Result<Bytes, Body> {
    let deadline = tokio::time::Instant::now() + within;
    let mut chunks: Vec<Bytes> = Vec::new();
    let mut total = 0usize;
    loop {
        // A body still producing at the deadline is sent on as it is.
        let Ok(frame) = tokio::time::timeout_at(deadline, body.frame()).await else {
            return Err(replay(chunks, body));
        };
        match frame {
            None => break,
            Some(Ok(frame)) => {
                let Ok(data) = frame.into_data() else {
                    continue;
                };
                total += data.len();
                chunks.push(data);
                if total > limit {
                    return Err(replay(chunks, body));
                }
            }
            Some(Err(err)) => {
                let head = futures::stream::iter(chunks.into_iter().map(Ok::<_, axum::Error>));
                return Err(Body::from_stream(
                    head.chain(futures::stream::once(async { Err(err) })),
                ));
            }
        }
    }
    let mut buf = Vec::with_capacity(total);
    for chunk in chunks {
        buf.extend_from_slice(&chunk);
    }
    Ok(Bytes::from(buf))
}

/// The frames read so far, then the rest of `body`, trailers too: the
/// response stays HTML.
fn replay(chunks: Vec<Bytes>, body: Body) -> Body {
    let head = futures::stream::iter(
        chunks
            .into_iter()
            .map(|b| Ok::<_, axum::Error>(http_body::Frame::data(b))),
    );
    let rest = http_body_util::BodyStream::new(body);
    Body::new(http_body_util::StreamBody::new(head.chain(rest)))
}

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

    #[test]
    fn head_knows_the_length_from_content_length_only() {
        let res = |len: Option<&'static str>| {
            let mut res = Response::new(Body::empty());
            if let Some(len) = len {
                res.headers_mut()
                    .insert(CONTENT_LENGTH, HeaderValue::from_static(len));
            }
            res
        };
        assert!(known_to_fit(&res(Some("0")), 10), "an empty page fits");
        assert!(known_to_fit(&res(Some("10")), 10));
        assert!(!known_to_fit(&res(Some("11")), 10));
        assert!(
            !known_to_fit(&res(None), 10),
            "a stripped body says nothing"
        );
        assert!(known_to_fit(&Response::new(Body::from("<p>x</p>")), 10));
    }

    #[test]
    fn a_matching_validator_turns_a_partial_response_into_304() {
        let mut headers = HeaderMap::new();
        headers.insert(IF_NONE_MATCH, HeaderValue::from_static("\"h1\""));
        let validators = Validators::take(&mut headers).unwrap();
        let partial = |etag: &'static str| {
            let mut res = Response::new(Body::from("abc"));
            *res.status_mut() = StatusCode::PARTIAL_CONTENT;
            res.headers_mut()
                .insert(axum::http::header::ETAG, HeaderValue::from_static(etag));
            res
        };
        assert_eq!(validators.apply(partial("\"h1\"")).status(), 304);
        assert_eq!(validators.apply(partial("\"h2\"")).status(), 206);
    }

    #[test]
    fn only_utf8_html_is_converted() {
        assert!(utf8_html("text/html"));
        assert!(utf8_html("text/html; charset=utf-8"));
        assert!(utf8_html("Text/HTML; Charset=\"UTF-8\""));
        assert!(utf8_html("text/html; charset=us-ascii"));
        assert!(!utf8_html("text/html; charset=iso-8859-1"));
        assert!(!utf8_html("text/html;charset=windows-1252"));
        assert!(!utf8_html("application/json"));
    }

    /// Codex review (P2): a signed response keeps its body, so its
    /// signature still verifies.
    #[test]
    fn a_signed_response_is_not_converted() {
        let html = || {
            Response::builder()
                .header(CONTENT_TYPE, "text/html; charset=utf-8")
                .body(Body::from("<h1>Hi</h1>"))
                .unwrap()
        };
        assert!(is_negotiable(&html()));
        for name in ["signature", "signature-input"] {
            let mut res = html();
            res.headers_mut()
                .insert(name, HeaderValue::from_static("sig1=:YQ==:"));
            assert!(!is_negotiable(&res), "{name}");
        }
    }

    #[test]
    fn markdown_drops_the_html_trailer_declaration() {
        let mut headers = HeaderMap::new();
        headers.insert(
            axum::http::header::TRAILER,
            HeaderValue::from_static("content-digest"),
        );
        markdown_headers(&mut headers, &NegotiateConfig::default(), None);
        assert!(!headers.contains_key(axum::http::header::TRAILER));
    }

    fn accept(v: &str) -> HeaderMap {
        let mut h = HeaderMap::new();
        h.insert(
            axum::http::header::ACCEPT,
            HeaderValue::from_str(v).unwrap(),
        );
        h
    }

    /// The choice for UTF-8 HTML, the common page.
    fn utf8(headers: &HeaderMap) -> bool {
        markdown_wins(&accept_values(headers), "text/html")
    }

    #[test]
    fn accept_ranking() {
        assert!(utf8(&accept("text/markdown")));
        assert!(utf8(&accept("text/markdown, text/html;q=0.9")));
        assert!(utf8(&accept("text/html;q=0.5, text/markdown")));
        assert!(!utf8(&accept("text/html, text/markdown;q=0.5")));
        assert!(!utf8(&accept("text/markdown;q=0")));
        assert!(!utf8(&accept("*/*")));
        assert!(!utf8(&accept("text/markdown;q=0.5, */*")));
        assert!(!utf8(&accept("text/markdown;q=0.5, text/*")));
        assert!(utf8(&accept("text/markdown, */*;q=0.8")));
        // Media parameters restrict the match: this layer sends UTF-8 GFM.
        assert!(!utf8(&accept(
            "text/markdown; charset=iso-8859-1, text/html;q=0.5"
        )));
        assert!(!utf8(&accept("text/markdown;variant=CommonMark")));
        assert!(!utf8(&accept("text/markdown;level=1")));
        assert!(utf8(&accept(
            "text/markdown; charset=\"UTF-8\"; variant=gfm; q=0.9, text/html;q=0.5"
        )));
        assert!(utf8(&accept("text/markdown;")));
        // The most specific matching range gives the quality.
        assert!(!utf8(&accept(
            "text/markdown;variant=GFM;q=0, text/markdown;q=1, text/html;q=0.5"
        )));
        assert!(utf8(&accept(
            "text/markdown;q=0, text/markdown;variant=GFM;q=1, text/html;q=0.5"
        )));
        assert!(!utf8(&accept(
            "text/html;charset=utf-8;q=1, text/html;q=0, text/markdown;q=0.5"
        )));
        assert!(utf8(&accept(
            "text/html;charset=utf-8;q=0, text/html;q=1, text/markdown;q=0.5"
        )));
        // An HTML range for another charset does not cover the UTF-8 HTML.
        assert!(utf8(&accept(
            "text/html; charset=iso-8859-1, text/markdown;q=0.5"
        )));
        assert!(!utf8(&accept(
            "text/html; charset=utf-8, text/markdown;q=0.5"
        )));
        // The more specific range is skipped, so `*/*` decides.
        assert!(!utf8(&accept(
            "text/html;charset=latin1, */*, text/markdown;q=0.5"
        )));
        assert!(utf8(&accept("text/markdown;Q=0.9, text/html;q=0.5")));
        assert!(!utf8(&accept("text/markdown;q=abc")));
        assert!(!utf8(&HeaderMap::new()));
    }

    #[test]
    fn the_most_specific_range_sets_the_html_weight() {
        // RFC 9110 §12.5.1: `text/html;q=0` overrides `*/*`.
        assert!(utf8(&accept("text/html;q=0, text/markdown;q=0.5, */*;q=1")));
        assert!(utf8(&accept("text/*;q=0.2, text/markdown;q=0.5, */*")));
        assert!(!utf8(&accept(
            "text/html;q=0.9, text/markdown;q=0.5, */*;q=0.1"
        )));
    }

    /// Codex review (P2): an HTML range matches the charset the page
    /// declares, so the choice waits for the response.
    #[test]
    fn the_html_range_matches_the_page_charset() {
        let wins = |a: &str, html: &str| markdown_wins(&accept_values(&accept(a)), html);
        let possible = |a: &str| prefers_markdown(&accept(a));

        let ascii = "text/html;charset=us-ascii;q=1, text/markdown;q=0.5";
        assert!(!wins(ascii, "text/html; charset=us-ascii"));
        assert!(!wins(ascii, "text/html; charset=\"US-ASCII\""));
        assert!(wins(ascii, "text/html; charset=utf-8"));
        assert!(wins(ascii, "text/html"));
        assert!(possible(ascii));

        let utf8 = "text/html;charset=utf-8;q=1, text/markdown;q=0.5";
        assert!(!wins(utf8, "text/html; charset=utf-8"));
        assert!(!wins(utf8, "text/html"));
        assert!(wins(utf8, "text/html; charset=us-ascii"));

        assert!(!possible("text/html, text/markdown;q=0.5"));
    }

    /// Codex review (P2): equivalent labels name the same charset.
    #[test]
    fn charset_labels_are_matched_by_what_they_name() {
        let wins = |a: &str, html: &str| markdown_wins(&accept_values(&accept(a)), html);
        assert!(utf8(&accept("text/markdown;charset=utf8, text/html;q=0")));
        assert!(utf8(&accept(
            "text/markdown;charset=\"UTF8\", text/html;q=0.5"
        )));
        assert!(!wins(
            "text/html;charset=utf-8, text/markdown;q=0.5",
            "text/html; charset=utf8"
        ));
        let ascii = "text/html;charset=ascii, text/markdown;q=0.5";
        assert!(!wins(ascii, "text/html; charset=us-ascii"));
        assert!(wins(ascii, "text/html"));
        assert!(utf8_html("text/html; charset=ASCII"));
        assert!(!utf8_html("text/html; charset=latin1"));
    }

    /// Codex review (P2): an HTML range's parameters are compared with the
    /// ones the page's own type declares.
    #[test]
    fn html_range_parameters_match_the_page_type() {
        let wins = |a: &str, html: &str| markdown_wins(&accept_values(&accept(a)), html);
        let profiled = "text/html;profile=\"https://example.test/p\", text/markdown;q=0.5";
        assert!(!wins(
            profiled,
            "text/html; profile=\"https://example.test/p\""
        ));
        assert!(!wins(
            profiled,
            "text/html; charset=utf-8; profile=https://example.test/p"
        ));
        // A quoted value may hold the separators.
        let quoted = "text/html;profile=\"a,b;c\";q=1, text/markdown;q=0.5";
        assert!(!wins(quoted, "text/html; profile=\"a,b;c\""));
        assert!(wins(quoted, "text/html; profile=\"a\""));
        // Another profile, or none, is not covered: Markdown wins.
        assert!(wins(
            profiled,
            "text/html; profile=\"https://example.test/q\""
        ));
        assert!(wins(profiled, "text/html"));
        // Such a range keeps the choice open until the response is here.
        assert!(prefers_markdown(&accept(profiled)));
    }

    #[test]
    fn localized_home_pages_are_home() {
        let link = |s: &'static str| HeaderValue::from_static(s);
        let config = NegotiateConfig {
            home_links: vec![
                ("/en".to_owned(), link("<x>; rel=\"alternate\"")),
                ("/fr".to_owned(), link("<y>; rel=\"alternate\"")),
            ],
            ..NegotiateConfig::default()
        };
        assert_eq!(config.home("/fr"), Some(1));
        assert_eq!(config.home("/"), None);
        assert_eq!(config.home("/en/x"), None);
    }

    #[test]
    fn validators_are_checked_against_the_representation_sent() {
        let mut req = HeaderMap::new();
        req.insert(IF_NONE_MATCH, HeaderValue::from_static("W/\"v1-md\""));
        req.insert(IF_MODIFIED_SINCE, HeaderValue::from_static("x"));
        let v = Validators::take(&mut req).expect("validators");
        assert!(req.get(IF_NONE_MATCH).is_none() && req.get(IF_MODIFIED_SINCE).is_none());

        let md = |etag: &'static str| {
            let mut res = Response::new(Body::from("body"));
            res.headers_mut()
                .insert(ETAG, HeaderValue::from_static(etag));
            res
        };
        assert_eq!(
            v.apply(md("W/\"v1-md\"")).status(),
            StatusCode::NOT_MODIFIED
        );
        // The HTML tag is another representation: no 304.
        assert_eq!(v.apply(md("\"v1\"")).status(), StatusCode::OK);
        assert!(Validators::take(&mut HeaderMap::new()).is_none());
    }

    #[test]
    fn resource_metadata_joins_the_handlers_bearer_challenge() {
        let p = "resource_metadata=\"https://s/m\"";
        assert_eq!(
            with_bearer_param("Bearer realm=\"api\", error=\"invalid_token\"", p).as_deref(),
            Some(
                "Bearer realm=\"api\", error=\"invalid_token\", resource_metadata=\"https://s/m\""
            )
        );
        assert_eq!(
            with_bearer_param("Bearer", p).as_deref(),
            Some("Bearer resource_metadata=\"https://s/m\"")
        );
        // The parameter goes on the Bearer challenge, not the one after it.
        assert_eq!(
            with_bearer_param("Bearer realm=\"a, b\", Basic realm=\"x\"", p).as_deref(),
            Some("Bearer realm=\"a, b\", resource_metadata=\"https://s/m\", Basic realm=\"x\"")
        );
        assert_eq!(with_bearer_param("Basic realm=\"x\"", p), None);
        assert_eq!(
            with_bearer_param("Bearer resource_metadata=\"https://o\"", p),
            None
        );
    }

    #[test]
    fn conversion_drops_metadata_about_the_html_bytes() {
        let mut h = HeaderMap::new();
        for name in [
            "content-digest",
            "repr-digest",
            "digest",
            "content-md5",
            "accept-ranges",
        ] {
            h.insert(
                axum::http::HeaderName::from_static(name),
                HeaderValue::from_static("x"),
            );
        }
        let config = NegotiateConfig {
            markdown: true,
            max_bytes: 1024,
            home_links: Vec::new(),
            content_signal: None,
            resource_metadata: None,
            resource_metadata_from_host: false,
        };
        markdown_headers(&mut h, &config, Some(1));
        for name in [
            "content-digest",
            "repr-digest",
            "digest",
            "content-md5",
            "accept-ranges",
        ] {
            assert!(h.get(name).is_none(), "{name}");
        }
    }

    /// Codex review (P1): a stream of unknown length is never collected,
    /// so a slow one cannot hold the request open; it stays HTML.
    #[tokio::test]
    async fn a_stream_of_unknown_length_stays_html() {
        let config = NegotiateConfig {
            markdown: true,
            max_bytes: 1024,
            home_links: Vec::new(),
            content_signal: None,
            resource_metadata: None,
            resource_metadata_from_host: false,
        };
        let html = || Response::builder().header(CONTENT_TYPE, "text/html; charset=utf-8");
        let streamed = html()
            .body(Body::from_stream(futures::stream::pending::<
                Result<Bytes, std::io::Error>,
            >()))
            .unwrap();
        // A pending stream would hang `to_markdown` if it were collected.
        let res = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            to_markdown(streamed, &config, None),
        )
        .await
        .expect("not collected");
        assert!(
            res.headers()[CONTENT_TYPE]
                .to_str()
                .unwrap()
                .starts_with("text/html")
        );
        // Codex review (P1): a declared `Content-Length` does not make a
        // stream safe to wait for.
        let declared = html()
            .header(CONTENT_LENGTH, "11")
            .body(Body::from_stream(futures::stream::pending::<
                Result<Bytes, std::io::Error>,
            >()))
            .unwrap();
        let res = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            to_markdown(declared, &config, None),
        )
        .await
        .expect("not collected");
        assert!(
            res.headers()[CONTENT_TYPE]
                .to_str()
                .unwrap()
                .starts_with("text/html")
        );

        let known = html().body(Body::from("<h1>Hi</h1>")).unwrap();
        let res = to_markdown(known, &config, None).await;
        assert!(
            res.headers()[CONTENT_TYPE]
                .to_str()
                .unwrap()
                .starts_with("text/markdown")
        );
    }

    /// An exact size hint whose bytes are produced later: the first frame
    /// is ready, the rest never comes.
    struct ExactButLazy {
        sent: bool,
    }

    impl http_body::Body for ExactButLazy {
        type Data = Bytes;
        type Error = std::io::Error;

        fn poll_frame(
            mut self: std::pin::Pin<&mut Self>,
            _cx: &mut std::task::Context<'_>,
        ) -> std::task::Poll<Option<Result<http_body::Frame<Bytes>, Self::Error>>> {
            if self.sent {
                return std::task::Poll::Pending;
            }
            self.sent = true;
            std::task::Poll::Ready(Some(Ok(http_body::Frame::data(Bytes::from("<h1>")))))
        }

        fn size_hint(&self) -> http_body::SizeHint {
            http_body::SizeHint::with_exact(11)
        }
    }

    // Codex review (P1): an exact size hint is not a buffered body. The wait
    // is bounded, and the page goes out as HTML with the bytes already read.
    #[tokio::test(start_paused = true)]
    async fn an_exact_but_lazy_body_stays_html_after_the_deadline() {
        let config = NegotiateConfig {
            markdown: true,
            max_bytes: 1024,
            home_links: Vec::new(),
            content_signal: None,
            resource_metadata: None,
            resource_metadata_from_host: false,
        };
        let res = Response::builder()
            .header(CONTENT_TYPE, "text/html; charset=utf-8")
            .body(Body::new(ExactButLazy { sent: false }))
            .unwrap();
        let started = tokio::time::Instant::now();
        let res = to_markdown(res, &config, None).await;
        assert_eq!(started.elapsed(), COLLECT_WITHIN);
        assert!(
            res.headers()[CONTENT_TYPE]
                .to_str()
                .unwrap()
                .starts_with("text/html")
        );
        let mut body = res.into_body();
        let first = body.frame().await.unwrap().unwrap().into_data().unwrap();
        assert_eq!(first, Bytes::from("<h1>"));
    }

    #[tokio::test]
    async fn collect_limited_keeps_trailers_of_an_oversized_body() {
        let mut trailers = HeaderMap::new();
        trailers.insert("x-digest", HeaderValue::from_static("abc"));
        let frames = futures::stream::iter([
            Ok::<_, std::io::Error>(http_body::Frame::data(Bytes::from("aaaa"))),
            Ok(http_body::Frame::data(Bytes::from("bbbb"))),
            Ok(http_body::Frame::trailers(trailers)),
        ]);
        let body = Body::new(http_body_util::StreamBody::new(frames));
        let Err(body) = collect_limited(body, 6, COLLECT_WITHIN).await else {
            panic!("over the limit")
        };
        let collected = body.collect().await.unwrap();
        assert_eq!(
            collected.trailers().and_then(|t| t.get("x-digest")),
            Some(&HeaderValue::from_static("abc"))
        );
        assert_eq!(&collected.to_bytes()[..], b"aaaabbbb");
    }

    #[tokio::test]
    async fn collect_limited_replays_an_oversized_body() {
        let body = Body::from_stream(futures::stream::iter(
            ["aaaa", "bbbb", "cccc"].map(|s| Ok::<_, std::io::Error>(Bytes::from(s))),
        ));
        let Err(body) = collect_limited(body, 6, COLLECT_WITHIN).await else {
            panic!("over the limit")
        };
        let all = body.collect().await.unwrap().to_bytes();
        assert_eq!(&all[..], b"aaaabbbbcccc");
        let ok = collect_limited(Body::from("abc"), 6, COLLECT_WITHIN)
            .await
            .unwrap();
        assert_eq!(&ok[..], b"abc");
    }
}