zc2 0.0.28

//! Transport-agnostic request/response for the broker HTTP handlers, so the
//! same handler logic runs under tiny_http (phase-1 interim) and axum.
//!
//! Phase 1 of the async migration (ZAK-12). See
//! docs/superpowers/specs/2026-06-21-zc-async-migration-design.md.

use tiny_http::Method;

/// Maximum accepted request-body size for synchronous HTTP endpoints.
/// Credit/worker JSON payloads are tiny; a few MiB is generous headroom.
/// Shared by the body-cap check (`body_or_413` in server.rs) and the axum
/// adapter so an over-cap body produces the identical 413 regardless of
/// frontend.
pub const MAX_BODY_BYTES: usize = 4 * 1024 * 1024;

/// A request reduced to what the handlers actually need, independent of the
/// underlying HTTP server.
pub struct BrokerRequest {
    pub method: Method,
    /// Path only (no query string).
    pub path: String,
    /// Raw query string ("" when absent).
    pub query: String,
    pub headers: Vec<(String, String)>,
    pub body: Vec<u8>,
    pub peer_addr: Option<std::net::SocketAddr>,
}

impl BrokerRequest {
    /// Case-insensitive header lookup (first match), mirroring the old
    /// `get_header` helper.
    pub fn header(&self, name: &str) -> Option<String> {
        self.headers
            .iter()
            .find(|(k, _)| k.eq_ignore_ascii_case(name))
            .map(|(_, v)| v.clone())
    }

    /// True when the client is on a loopback address — or unknown, which only
    /// happens for in-process callers (mirrors the old `is_loopback_request`).
    pub fn is_loopback(&self) -> bool {
        match self.peer_addr {
            Some(a) => a.ip().is_loopback(),
            None => true,
        }
    }
}

/// A response reduced to status + headers + body, independent of the server.
pub struct BrokerResponse {
    pub status: u16,
    pub headers: Vec<(String, String)>,
    pub body: Vec<u8>,
}

impl BrokerResponse {
    /// A JSON response with the appropriate Content-Type.
    pub fn json_bytes(body: Vec<u8>, status: u16) -> Self {
        BrokerResponse {
            status,
            headers: vec![("Content-Type".to_string(), "application/json".to_string())],
            body,
        }
    }

    /// Append a header (builder style).
    pub fn with_header(mut self, k: &str, v: &str) -> Self {
        self.headers.push((k.to_string(), v.to_string()));
        self
    }
}

impl BrokerRequest {
    /// Build from a tiny_http request (reads the whole body; the body cap is
    /// enforced by the caller via `body_or_413`).
    pub fn from_tiny_http(req: &mut tiny_http::Request) -> std::io::Result<Self> {
        let method = req.method().clone();
        let url = req.url().to_string();
        let (path, query) = match url.split_once('?') {
            Some((p, q)) => (p.to_string(), q.to_string()),
            None => (url, String::new()),
        };
        let headers = req
            .headers()
            .iter()
            .map(|h| {
                (
                    h.field.as_str().as_str().to_string(),
                    h.value.as_str().to_string(),
                )
            })
            .collect();
        let peer_addr = req.remote_addr().copied();
        let mut body = Vec::new();
        req.as_reader().read_to_end(&mut body)?;
        Ok(BrokerRequest {
            method,
            path,
            query,
            headers,
            body,
            peer_addr,
        })
    }
}

impl BrokerResponse {
    /// Convert to a tiny_http response for the interim frontend. Headers whose
    /// name/value bytes are rejected by tiny_http are skipped (never panics).
    pub fn into_tiny_http(self) -> tiny_http::Response<std::io::Cursor<Vec<u8>>> {
        let mut r = tiny_http::Response::from_data(self.body).with_status_code(self.status);
        for (k, v) in self.headers {
            if let Ok(h) = tiny_http::Header::from_bytes(k.as_bytes(), v.as_bytes()) {
                r = r.with_header(h);
            }
        }
        r
    }
}

impl BrokerRequest {
    /// Build from an axum request. Reads the whole body (capped at
    /// `MAX_BODY_BYTES + 1` so the downstream `body_or_413` still sees an
    /// over-limit body and emits the same 413). The method is mapped onto the
    /// `tiny_http::Method` the handlers match on; the broker only ever uses
    /// GET/POST/DELETE, but the full standard set is mapped for completeness,
    /// with any unknown method falling back to `Get`.
    pub async fn from_axum(
        req: axum::extract::Request,
        peer_addr: Option<std::net::SocketAddr>,
    ) -> Self {
        let method = match *req.method() {
            axum::http::Method::GET => Method::Get,
            axum::http::Method::POST => Method::Post,
            axum::http::Method::DELETE => Method::Delete,
            axum::http::Method::PUT => Method::Put,
            axum::http::Method::HEAD => Method::Head,
            axum::http::Method::PATCH => Method::Patch,
            axum::http::Method::OPTIONS => Method::Options,
            axum::http::Method::CONNECT => Method::Connect,
            axum::http::Method::TRACE => Method::Trace,
            _ => Method::Get,
        };

        let uri = req.uri();
        let path = uri.path().to_string();
        let query = uri.query().unwrap_or("").to_string();

        let headers = req
            .headers()
            .iter()
            .map(|(k, v)| (k.as_str().to_string(), v.to_str().unwrap_or("").to_string()))
            .collect();

        // Read the body, allowing one byte beyond the cap so the over-limit
        // case is detectable. On any read error (including over-limit), synthesize
        // an over-cap body so `body_or_413` returns the identical 413.
        let body = match axum::body::to_bytes(req.into_body(), MAX_BODY_BYTES + 1).await {
            Ok(b) => b.to_vec(),
            Err(_) => vec![0u8; MAX_BODY_BYTES + 1],
        };

        BrokerRequest {
            method,
            path,
            query,
            headers,
            body,
            peer_addr,
        }
    }
}

impl axum::response::IntoResponse for BrokerResponse {
    fn into_response(self) -> axum::response::Response {
        use axum::http::{HeaderName, HeaderValue, StatusCode};

        let status = StatusCode::from_u16(self.status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
        let mut response = axum::response::Response::new(axum::body::Body::from(self.body));
        *response.status_mut() = status;
        let headers = response.headers_mut();
        for (k, v) in self.headers {
            // Skip headers whose name/value bytes are rejected (mirrors the
            // `into_tiny_http` behavior of never panicking on bad bytes).
            if let (Ok(name), Ok(val)) = (
                HeaderName::from_bytes(k.as_bytes()),
                HeaderValue::from_str(&v),
            ) {
                headers.append(name, val);
            }
        }
        response
    }
}

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

    fn req(headers: Vec<(&str, &str)>, peer: Option<&str>) -> BrokerRequest {
        BrokerRequest {
            method: Method::Get,
            path: "/x".into(),
            query: String::new(),
            headers: headers
                .into_iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect(),
            body: vec![],
            peer_addr: peer.map(|s| s.parse().unwrap()),
        }
    }

    #[test]
    fn header_is_case_insensitive() {
        let r = req(vec![("X-Api-Key", "abc")], None);
        assert_eq!(r.header("x-api-key").as_deref(), Some("abc"));
        assert_eq!(r.header("X-Api-Key").as_deref(), Some("abc"));
        assert_eq!(r.header("missing"), None);
    }

    #[test]
    fn is_loopback_matches_127_and_unknown() {
        assert!(req(vec![], Some("127.0.0.1:5000")).is_loopback());
        assert!(req(vec![], None).is_loopback()); // unknown == loopback (in-process)
        assert!(!req(vec![], Some("10.0.0.5:5000")).is_loopback());
    }

    #[test]
    fn json_bytes_sets_content_type() {
        let r = BrokerResponse::json_bytes(b"{}".to_vec(), 200);
        assert_eq!(r.status, 200);
        assert!(
            r.headers
                .iter()
                .any(|(k, v)| k.eq_ignore_ascii_case("content-type")
                    && v.contains("application/json"))
        );
    }

    #[test]
    fn with_header_appends() {
        let r = BrokerResponse::json_bytes(vec![], 200).with_header("X-Test", "1");
        assert!(r.headers.iter().any(|(k, v)| k == "X-Test" && v == "1"));
    }

    #[test]
    fn into_tiny_http_preserves_status_and_headers() {
        let resp = BrokerResponse::json_bytes(b"{\"ok\":true}".to_vec(), 201)
            .with_header("X-Zakuro-Cost", "0.50");
        let th = resp.into_tiny_http();
        assert_eq!(th.status_code().0, 201);
        let headers: Vec<(String, String)> = th
            .headers()
            .iter()
            .map(|h| {
                (
                    h.field.as_str().as_str().to_string(),
                    h.value.as_str().to_string(),
                )
            })
            .collect();
        assert!(
            headers
                .iter()
                .any(|(k, v)| k.eq_ignore_ascii_case("content-type")
                    && v.contains("application/json"))
        );
        assert!(headers
            .iter()
            .any(|(k, v)| k == "X-Zakuro-Cost" && v == "0.50"));
    }

    #[test]
    fn into_tiny_http_skips_invalid_headers() {
        // A header value with bytes tiny_http rejects (non-ASCII >127) must be
        // dropped rather than panic; valid headers still pass through.
        let resp = BrokerResponse {
            status: 200,
            headers: vec![
                ("X-Good".to_string(), "ok".to_string()),
                ("X-Bad".to_string(), "wörker\u{00ff}".to_string()),
            ],
            body: Vec::new(),
        };
        let th = resp.into_tiny_http();
        let headers: Vec<String> = th
            .headers()
            .iter()
            .map(|h| h.field.as_str().as_str().to_string())
            .collect();
        assert!(headers.iter().any(|k| k == "X-Good"));
        assert!(!headers.iter().any(|k| k == "X-Bad"));
    }

    #[test]
    fn response_round_trips_through_tiny_http() {
        // Build a response, convert to tiny_http, and confirm body bytes survive.
        let body = b"hello-bytes".to_vec();
        let resp = BrokerResponse {
            status: 200,
            headers: vec![(
                "Content-Type".to_string(),
                "application/octet-stream".to_string(),
            )],
            body: body.clone(),
        };
        let th = resp.into_tiny_http();
        // tiny_http exposes the data length; verify it matches the original body.
        assert_eq!(th.data_length(), Some(body.len()));
    }
}