frame-host 0.3.0

//! Same-origin proxy routes for the embedded bus's health listener.
//!
//! The bus health endpoint (`/health`, `/ready`, `/metrics`) binds its own
//! address and sends no CORS headers, so the console page — served from
//! frame-host's origin — cannot fetch it directly (introspection map §4.3).
//! Frame-host already owns the embedded component's real bound health address
//! (`EmbeddedLiminal::health_addr`), so these routes forward:
//!
//! - `GET /frame/bus/health`  → bus `GET /health`
//! - `GET /frame/bus/ready`   → bus `GET /ready`
//! - `GET /frame/bus/metrics` → bus `GET /metrics`
//!
//! **Compatibility window:** the pre-rename routes `/frame/liminal/*` answer
//! as aliases of the `/frame/bus/*` routes above (same handlers), and every
//! hit on a legacy route logs a loud `tracing::warn!` deprecation. The
//! aliases are removed after one window.
//!
//! The upstream status code, `Content-Type`, and body pass through verbatim —
//! a 503 from `/ready` during the drain window stays a 503 in the page —
//! with ONE exception: the metrics body is re-vocabularied by
//! [`bus_metrics_body`], which emits every upstream `liminal_*` family under
//! the frame-public `bus_*` name and keeps the original `liminal_*` lines as
//! deprecated duplicates for the same window. CORS is never opened on the bus
//! itself; same-origin is the whole point.
//!
//! Failure is typed and BOUNDED: an unreachable, hung, or malformed upstream
//! answers as a `502` JSON body naming the exact failure
//! (`BUS_UNREACHABLE` / `BUS_BAD_RESPONSE`) — never a hang, never a
//! silent empty 200. The bus's health server speaks hand-rolled HTTP/1.1 with
//! `Connection: close` + `Content-Length` on every response
//! (`liminal-server/src/health/endpoint.rs`), so the forwarder reads to EOF
//! under a deadline and validates the declared length.

use std::net::SocketAddr;
use std::time::Duration;

use axum::body::Body;
use axum::http::{HeaderValue, StatusCode, header};
use axum::response::Response;
use serde::Serialize;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;

/// Console-facing proxy route for bus liveness (`/health`).
pub const BUS_HEALTH_ROUTE: &str = "/frame/bus/health";
/// Console-facing proxy route for bus readiness (`/ready`).
pub const BUS_READY_ROUTE: &str = "/frame/bus/ready";
/// Console-facing proxy route for bus metrics (`/metrics`).
pub const BUS_METRICS_ROUTE: &str = "/frame/bus/metrics";

/// DEPRECATED alias of [`BUS_HEALTH_ROUTE`] (compatibility window only).
pub const LEGACY_LIMINAL_HEALTH_ROUTE: &str = "/frame/liminal/health";
/// DEPRECATED alias of [`BUS_READY_ROUTE`] (compatibility window only).
pub const LEGACY_LIMINAL_READY_ROUTE: &str = "/frame/liminal/ready";
/// DEPRECATED alias of [`BUS_METRICS_ROUTE`] (compatibility window only).
pub const LEGACY_LIMINAL_METRICS_ROUTE: &str = "/frame/liminal/metrics";

/// How long the proxy waits to open a TCP connection to the health listener.
/// Mirrors the liveness monitor's probe bound (`serve.rs`): past this, the
/// listener is not answering and the route must fail typed, not hang.
pub const PROXY_CONNECT_TIMEOUT: Duration = Duration::from_secs(2);

/// How long the proxy waits for the complete upstream response after
/// connecting. The health endpoint answers small bodies immediately from
/// memory; a response that cannot complete inside this window is a hung
/// upstream and must surface as a typed 502, never a hung page fetch.
pub const PROXY_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5);

/// Upstream responses are health JSON or a Prometheus text page — small by
/// construction. A response larger than this is not the liminal health
/// listener speaking and is refused as malformed rather than buffered without
/// bound.
pub const PROXY_MAX_RESPONSE_BYTES: usize = 4 * 1024 * 1024;

/// The deadlines one forward operates under. Injected so tests exercise the
/// timeout paths quickly; the routes always pass the production consts.
#[derive(Clone, Copy, Debug)]
pub struct ProxyDeadlines {
    /// TCP connect bound.
    pub connect: Duration,
    /// Complete-response bound (after connect).
    pub response: Duration,
}

/// The production deadlines ([`PROXY_CONNECT_TIMEOUT`] / [`PROXY_RESPONSE_TIMEOUT`]).
pub const PRODUCTION_DEADLINES: ProxyDeadlines = ProxyDeadlines {
    connect: PROXY_CONNECT_TIMEOUT,
    response: PROXY_RESPONSE_TIMEOUT,
};

/// A typed forwarding failure. Every variant renders as a 502 JSON body; the
/// `code()` slug is the closed set the console's health strip switches on.
#[derive(Debug)]
pub enum ProxyFailure {
    /// The health listener could not be reached (refused / connect timeout).
    Unreachable {
        /// Exact connect failure.
        detail: String,
    },
    /// The listener answered, but not with a complete, well-formed HTTP/1.1
    /// response inside the deadline.
    BadResponse {
        /// Exact parse/deadline failure.
        detail: String,
    },
}

impl ProxyFailure {
    /// The closed failure-code slug carried in the 502 body.
    #[must_use]
    pub const fn code(&self) -> &'static str {
        match self {
            Self::Unreachable { .. } => "BUS_UNREACHABLE",
            Self::BadResponse { .. } => "BUS_BAD_RESPONSE",
        }
    }

    /// The exact failure detail carried in the 502 body.
    #[must_use]
    pub fn detail(&self) -> &str {
        match self {
            Self::Unreachable { detail } | Self::BadResponse { detail } => detail,
        }
    }
}

/// The 502 body shape the console's health strip parses on a proxy failure.
#[derive(Debug, Serialize)]
struct ProxyErrorBody<'a> {
    error: &'static str,
    upstream_path: &'a str,
    detail: &'a str,
}

/// One parsed upstream response: pass-through status, content type, and body.
#[derive(Debug)]
pub struct UpstreamResponse {
    /// Upstream HTTP status code (forwarded verbatim).
    pub status: u16,
    /// Upstream `Content-Type` header, when present.
    pub content_type: Option<String>,
    /// Upstream body bytes.
    pub body: Vec<u8>,
}

/// Forwards `GET {path}` to the bus health listener at `addr` and returns
/// the upstream response for verbatim pass-through.
///
/// # Errors
///
/// Returns a typed [`ProxyFailure`] when the listener is unreachable, hangs
/// past a deadline, or answers with a malformed / truncated HTTP response.
pub async fn forward_health_request(
    addr: SocketAddr,
    path: &str,
    deadlines: ProxyDeadlines,
) -> Result<UpstreamResponse, ProxyFailure> {
    let mut stream = tokio::time::timeout(deadlines.connect, TcpStream::connect(addr))
        .await
        .map_err(|_| ProxyFailure::Unreachable {
            detail: format!(
                "connecting to the bus health listener at {addr} timed out after {:?}",
                deadlines.connect
            ),
        })?
        .map_err(|error| ProxyFailure::Unreachable {
            detail: format!("connecting to the bus health listener at {addr} failed: {error}"),
        })?;

    let exchange = async {
        let request = format!("GET {path} HTTP/1.1\r\nHost: {addr}\r\nConnection: close\r\n\r\n");
        stream
            .write_all(request.as_bytes())
            .await
            .map_err(|error| ProxyFailure::BadResponse {
                detail: format!("writing the upstream request failed: {error}"),
            })?;
        // The health server answers `Connection: close` on every response, so
        // the complete response is exactly the bytes until EOF.
        let mut raw = Vec::new();
        let mut chunk = [0_u8; 8192];
        loop {
            let read =
                stream
                    .read(&mut chunk)
                    .await
                    .map_err(|error| ProxyFailure::BadResponse {
                        detail: format!("reading the upstream response failed: {error}"),
                    })?;
            if read == 0 {
                break;
            }
            raw.extend_from_slice(&chunk[..read]);
            if raw.len() > PROXY_MAX_RESPONSE_BYTES {
                return Err(ProxyFailure::BadResponse {
                    detail: format!(
                        "upstream response exceeded {PROXY_MAX_RESPONSE_BYTES} bytes; refusing to buffer further"
                    ),
                });
            }
        }
        Ok(raw)
    };
    let raw = tokio::time::timeout(deadlines.response, exchange)
        .await
        .map_err(|_| ProxyFailure::BadResponse {
            detail: format!(
                "the bus health listener did not complete a response within {:?}",
                deadlines.response
            ),
        })??;

    parse_http_response(&raw)
}

/// Parses a complete raw HTTP/1.1 response (status line + headers + body) as
/// the bus's health server writes it.
fn parse_http_response(raw: &[u8]) -> Result<UpstreamResponse, ProxyFailure> {
    let header_end = find_header_end(raw).ok_or_else(|| ProxyFailure::BadResponse {
        detail: "upstream response has no header terminator (connection closed early?)".to_owned(),
    })?;
    let head = std::str::from_utf8(&raw[..header_end]).map_err(|_| ProxyFailure::BadResponse {
        detail: "upstream response headers are not valid UTF-8".to_owned(),
    })?;
    let body = raw[header_end + 4..].to_vec();

    let mut lines = head.split("\r\n");
    let status_line = lines.next().ok_or_else(|| ProxyFailure::BadResponse {
        detail: "upstream response is empty".to_owned(),
    })?;
    let status = parse_status_line(status_line)?;

    let mut content_type = None;
    let mut content_length: Option<usize> = None;
    for line in lines {
        let Some((name, value)) = line.split_once(':') else {
            continue;
        };
        let name = name.trim().to_ascii_lowercase();
        let value = value.trim();
        if name == "content-type" {
            content_type = Some(value.to_owned());
        } else if name == "content-length" {
            content_length = Some(value.parse().map_err(|_| ProxyFailure::BadResponse {
                detail: format!("upstream Content-Length {value:?} is not a length"),
            })?);
        }
    }

    // The health server declares Content-Length on every response; a mismatch
    // means the body was truncated or over-read and must not pass through.
    if let Some(declared) = content_length
        && declared != body.len()
    {
        return Err(ProxyFailure::BadResponse {
            detail: format!(
                "upstream declared Content-Length {declared} but sent {} body byte(s)",
                body.len()
            ),
        });
    }

    Ok(UpstreamResponse {
        status,
        content_type,
        body,
    })
}

/// Parses `HTTP/1.1 <code> <reason>` into the status code.
fn parse_status_line(line: &str) -> Result<u16, ProxyFailure> {
    let mut parts = line.split_whitespace();
    let version = parts.next().unwrap_or_default();
    if !version.starts_with("HTTP/1.") {
        return Err(ProxyFailure::BadResponse {
            detail: format!("upstream status line {line:?} is not HTTP/1.x"),
        });
    }
    let code = parts.next().ok_or_else(|| ProxyFailure::BadResponse {
        detail: format!("upstream status line {line:?} has no status code"),
    })?;
    code.parse().map_err(|_| ProxyFailure::BadResponse {
        detail: format!("upstream status code {code:?} is not a number"),
    })
}

/// Locates the `\r\n\r\n` header/body boundary.
fn find_header_end(raw: &[u8]) -> Option<usize> {
    raw.windows(4).position(|window| window == b"\r\n\r\n")
}

/// Re-vocabularies an upstream Prometheus metrics page into frame's public
/// metric names: every `liminal_*` family (its `# HELP` line, `# TYPE` line,
/// and sample lines) is emitted under the `bus_*` name FIRST, then the
/// original page follows verbatim so the `liminal_*` families remain as
/// deprecated duplicates for one compatibility window. Families without the
/// `liminal_` prefix are not duplicated (a duplicated identical family would
/// be malformed Prometheus text).
///
/// A non-UTF-8 upstream body cannot be renamed; it passes through unchanged
/// with a loud `tracing::warn!` (never silently altered, never dropped).
#[must_use]
pub fn bus_metrics_body(upstream: &[u8]) -> Vec<u8> {
    let Ok(text) = std::str::from_utf8(upstream) else {
        tracing::warn!(
            bytes = upstream.len(),
            "bus metrics upstream body is not UTF-8; serving it verbatim without the bus_* rename"
        );
        return upstream.to_vec();
    };
    let mut bus_section = String::new();
    for line in text.lines() {
        if let Some(renamed) = rename_metric_line(line) {
            bus_section.push_str(&renamed);
            bus_section.push('\n');
        }
    }
    if bus_section.is_empty() {
        return upstream.to_vec();
    }
    let mut body = bus_section;
    body.push_str(
        "# The liminal_* families below are DEPRECATED duplicates of the bus_* families above \
         (one compatibility window).\n",
    );
    body.push_str(text);
    if !text.ends_with('\n') {
        body.push('\n');
    }
    body.into_bytes()
}

/// Renames one metrics line from the `liminal_*` family vocabulary to
/// `bus_*`; returns `None` for lines that do not reference a `liminal_*`
/// family (they are not duplicated into the bus section).
fn rename_metric_line(line: &str) -> Option<String> {
    for prefix in ["# HELP liminal_", "# TYPE liminal_"] {
        if let Some(rest) = line.strip_prefix(prefix) {
            let renamed_prefix = prefix.replace("liminal_", "bus_");
            return Some(format!("{renamed_prefix}{rest}"));
        }
    }
    line.strip_prefix("liminal_")
        .map(|rest| format!("bus_{rest}"))
}

/// Renders an upstream response as the pass-through page response.
#[must_use]
pub fn passthrough_response(upstream: UpstreamResponse) -> Response {
    let status = StatusCode::from_u16(upstream.status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
    let mut response = Response::new(Body::from(upstream.body));
    *response.status_mut() = status;
    if let Some(content_type) = upstream.content_type
        && let Ok(value) = HeaderValue::from_str(&content_type)
    {
        response.headers_mut().insert(header::CONTENT_TYPE, value);
    }
    response
}

/// Renders a typed proxy failure as the 502 JSON page response.
#[must_use]
pub fn failure_response(path: &str, failure: &ProxyFailure) -> Response {
    let body = ProxyErrorBody {
        error: failure.code(),
        upstream_path: path,
        detail: failure.detail(),
    };
    let bytes = serde_json::to_vec(&body).unwrap_or_else(|_| {
        // ProxyErrorBody is three plain strings; serialization cannot fail in
        // practice, but a silent empty body is still forbidden.
        format!(
            "{{\"error\":\"{}\",\"upstream_path\":\"{path}\",\"detail\":\"(detail serialization failed)\"}}",
            failure.code()
        )
        .into_bytes()
    });
    let mut response = Response::new(Body::from(bytes));
    *response.status_mut() = StatusCode::BAD_GATEWAY;
    response.headers_mut().insert(
        header::CONTENT_TYPE,
        HeaderValue::from_static("application/json; charset=utf-8"),
    );
    response
}

#[cfg(test)]
mod tests {
    use super::{
        ProxyDeadlines, ProxyFailure, bus_metrics_body, forward_health_request,
        parse_http_response, parse_status_line,
    };
    use std::io::{Read, Write};
    use std::net::{SocketAddr, TcpListener};
    use std::time::Duration;

    type TestResult = Result<(), Box<dyn std::error::Error>>;

    /// Deadlines short enough to keep the timeout tests fast.
    const TEST_DEADLINES: ProxyDeadlines = ProxyDeadlines {
        connect: Duration::from_millis(500),
        response: Duration::from_millis(500),
    };

    /// A fake health listener answering one connection with canned bytes,
    /// written exactly as liminal's health server writes responses.
    fn fake_upstream(
        status: &str,
        content_type: Option<&str>,
        body: &str,
    ) -> Result<SocketAddr, Box<dyn std::error::Error>> {
        let listener = TcpListener::bind("127.0.0.1:0")?;
        let addr = listener.local_addr()?;
        let type_header = content_type
            .map(|content_type| format!("Content-Type: {content_type}\r\n"))
            .unwrap_or_default();
        let response = format!(
            "HTTP/1.1 {status}\r\nContent-Length: {}\r\nConnection: close\r\n{type_header}\r\n{body}",
            body.len()
        );
        std::thread::spawn(move || {
            if let Ok((mut stream, _)) = listener.accept() {
                let mut request = [0_u8; 2048];
                if stream.read(&mut request).is_ok() {
                    // A write failure fails the test through its read side.
                    let _outcome = stream.write_all(response.as_bytes());
                }
            }
        });
        Ok(addr)
    }

    #[tokio::test]
    async fn forwards_status_content_type_and_body_verbatim() -> TestResult {
        let body = r#"{"status":"healthy","message":null}"#;
        let addr = fake_upstream("200 OK", Some("application/json"), body)?;
        let upstream = forward_health_request(addr, "/health", TEST_DEADLINES)
            .await
            .map_err(|failure| format!("forward failed: {failure:?}"))?;
        assert_eq!(upstream.status, 200);
        assert_eq!(upstream.content_type.as_deref(), Some("application/json"));
        assert_eq!(upstream.body, body.as_bytes());
        Ok(())
    }

    #[tokio::test]
    async fn a_503_ready_response_passes_through_as_503() -> TestResult {
        let body = r#"{"ready":false,"unmet_conditions":["listener_bound"]}"#;
        let addr = fake_upstream("503 Service Unavailable", Some("application/json"), body)?;
        let upstream = forward_health_request(addr, "/ready", TEST_DEADLINES)
            .await
            .map_err(|failure| format!("forward failed: {failure:?}"))?;
        assert_eq!(upstream.status, 503);
        assert_eq!(upstream.body, body.as_bytes());
        Ok(())
    }

    #[tokio::test]
    async fn bus_down_is_a_typed_unreachable_error_not_a_hang() -> TestResult {
        // Bind then release to obtain an address nothing listens on.
        let released = TcpListener::bind("127.0.0.1:0")?;
        let dead_addr = released.local_addr()?;
        drop(released);
        let outcome = tokio::time::timeout(
            Duration::from_secs(5),
            forward_health_request(dead_addr, "/health", TEST_DEADLINES),
        )
        .await
        .map_err(|_| "the down path must be bounded, not a hang")?;
        let failure = outcome.err().ok_or("a down listener must fail typed")?;
        assert!(
            matches!(failure, ProxyFailure::Unreachable { .. }),
            "got {failure:?}"
        );
        assert_eq!(failure.code(), "BUS_UNREACHABLE");
        Ok(())
    }

    #[tokio::test]
    async fn a_hung_upstream_is_a_typed_bad_response_within_the_deadline() -> TestResult {
        // Accepts the connection, then never writes a byte.
        let listener = TcpListener::bind("127.0.0.1:0")?;
        let addr = listener.local_addr()?;
        let hold = std::thread::spawn(move || listener.accept().map(|(stream, _)| stream));
        let outcome = tokio::time::timeout(
            Duration::from_secs(5),
            forward_health_request(addr, "/health", TEST_DEADLINES),
        )
        .await
        .map_err(|_| "the hang path must be bounded, not a hang")?;
        let failure = outcome.err().ok_or("a hung listener must fail typed")?;
        assert!(
            matches!(failure, ProxyFailure::BadResponse { .. }),
            "got {failure:?}"
        );
        assert_eq!(failure.code(), "BUS_BAD_RESPONSE");
        drop(hold);
        Ok(())
    }

    #[tokio::test]
    async fn a_truncated_body_is_refused_not_passed_through() -> TestResult {
        let listener = TcpListener::bind("127.0.0.1:0")?;
        let addr = listener.local_addr()?;
        std::thread::spawn(move || {
            if let Ok((mut stream, _)) = listener.accept() {
                let mut request = [0_u8; 2048];
                if stream.read(&mut request).is_ok() {
                    // Declares 100 bytes, sends 4, closes.
                    let _outcome = stream.write_all(
                        b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\nConnection: close\r\n\r\nbody",
                    );
                }
            }
        });
        let failure = forward_health_request(addr, "/metrics", TEST_DEADLINES)
            .await
            .err()
            .ok_or("a truncated body must fail typed")?;
        match failure {
            ProxyFailure::BadResponse { ref detail } => {
                assert!(detail.contains("Content-Length"), "detail: {detail}");
            }
            ProxyFailure::Unreachable { .. } => {
                return Err(format!("expected BadResponse, got {failure:?}").into());
            }
        }
        Ok(())
    }

    /// The metrics rename emits every `liminal_*` family under `bus_*` first
    /// (HELP/TYPE/sample lines included) and keeps the original `liminal_*`
    /// lines as deprecated duplicates; non-liminal families are not duplicated.
    #[test]
    fn metrics_body_is_renamed_to_bus_families_with_legacy_duplicates()
    -> Result<(), Box<dyn std::error::Error>> {
        let upstream = "# HELP liminal_connections_active Active connections\n\
                        # TYPE liminal_connections_active gauge\n\
                        liminal_connections_active 2\n\
                        liminal_publishes_total 41\n\
                        other_family 7\n";
        let body = String::from_utf8(bus_metrics_body(upstream.as_bytes()))?;
        assert!(body.contains("# HELP bus_connections_active Active connections\n"));
        assert!(body.contains("# TYPE bus_connections_active gauge\n"));
        assert!(body.contains("bus_connections_active 2\n"));
        assert!(body.contains("bus_publishes_total 41\n"));
        // The legacy lines survive verbatim for the compatibility window.
        assert!(body.contains("liminal_connections_active 2\n"));
        assert!(body.contains("liminal_publishes_total 41\n"));
        // The new vocabulary is primary: it appears before the legacy lines.
        let bus_at = body.find("bus_connections_active 2").ok_or("bus line")?;
        let legacy_at = body
            .find("liminal_connections_active 2")
            .ok_or("legacy line")?;
        assert!(bus_at < legacy_at, "bus_* families must be emitted first");
        // A non-liminal family is present exactly once, never duplicated.
        assert_eq!(body.matches("other_family 7").count(), 1);
        Ok(())
    }

    /// A page with no `liminal_*` families passes through unchanged.
    #[test]
    fn metrics_body_without_liminal_families_is_untouched() {
        let upstream = b"# HELP other_family Something\nother_family 7\n";
        assert_eq!(bus_metrics_body(upstream), upstream.to_vec());
    }

    /// A labeled sample line is renamed too — the rename is prefix-based, not
    /// tied to the exactly-three unlabeled production families.
    #[test]
    fn labeled_liminal_samples_are_renamed() -> Result<(), Box<dyn std::error::Error>> {
        let upstream = "liminal_publishes_total{channel=\"a\"} 9\n";
        let body = String::from_utf8(bus_metrics_body(upstream.as_bytes()))?;
        assert!(body.contains("bus_publishes_total{channel=\"a\"} 9\n"));
        assert!(body.contains("liminal_publishes_total{channel=\"a\"} 9\n"));
        Ok(())
    }

    #[test]
    fn status_line_parsing_is_strict() {
        assert_eq!(parse_status_line("HTTP/1.1 200 OK").ok(), Some(200));
        assert_eq!(
            parse_status_line("HTTP/1.1 503 Service Unavailable").ok(),
            Some(503)
        );
        assert!(parse_status_line("SPDY/9 200 OK").is_err());
        assert!(parse_status_line("HTTP/1.1").is_err());
        assert!(parse_status_line("HTTP/1.1 abc OK").is_err());
    }

    #[test]
    fn response_parsing_requires_a_header_terminator() -> TestResult {
        assert!(parse_http_response(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n").is_err());
        let parsed = parse_http_response(
            b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok",
        )
        .map_err(|failure| format!("well-formed response must parse: {failure:?}"))?;
        assert_eq!(parsed.status, 200);
        assert_eq!(parsed.body, b"ok");
        Ok(())
    }
}