unb_core/

envelope.rs

use bytes::{BufMut, Bytes, BytesMut};
use serde::{Deserialize, Serialize};
use serde_json::Value;

use crate::{CoreError, TargetPath};

pub const PROTOCOL_VERSION: u16 = 1;
pub const DEFAULT_HOPS: u8 = 8;

const CONTROL_TARGET: &str = "*";
const MAX_HEADERS: usize = 64;
const MAX_HEAD_BYTES: usize = 64 * 1024;

pub const UNB_VERSION: &str = "unb-version";
pub const UNB_KIND: &str = "unb-kind";
pub const UNB_ID: &str = "unb-id";
pub const UNB_CORR: &str = "unb-corr";
pub const UNB_SEQ: &str = "unb-seq";
pub const UNB_HOPS: &str = "unb-hops";
pub const UNB_PATH: &str = "unb-path";
pub const UNB_CODE: &str = "unb-code";
pub const UNB_BODY: &str = "unb-body";

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Kind {
    Hello,
    Welcome,
    Request,
    Response,
    Subscribe,
    Event,
    Channel,
    Cancel,
    Error,
    Ping,
    Pong,
    Discover,
    Identify,
    IdentityAccepted,
    RouteSnapshot,
    RouteDelta,
    RouteAck,
}

impl Kind {
    pub fn token(self) -> &'static str {
        match self {
            Kind::Hello => "hello",
            Kind::Welcome => "welcome",
            Kind::Request => "request",
            Kind::Response => "response",
            Kind::Subscribe => "subscribe",
            Kind::Event => "event",
            Kind::Channel => "channel",
            Kind::Cancel => "cancel",
            Kind::Error => "error",
            Kind::Ping => "ping",
            Kind::Pong => "pong",
            Kind::Discover => "discover",
            Kind::Identify => "identify",
            Kind::IdentityAccepted => "identity_accepted",
            Kind::RouteSnapshot => "route_snapshot",
            Kind::RouteDelta => "route_delta",
            Kind::RouteAck => "route_ack",
        }
    }

    pub fn from_token(token: &str) -> Option<Kind> {
        Some(match token {
            "hello" => Kind::Hello,
            "welcome" => Kind::Welcome,
            "request" => Kind::Request,
            "response" => Kind::Response,
            "subscribe" => Kind::Subscribe,
            "event" => Kind::Event,
            "channel" => Kind::Channel,
            "cancel" => Kind::Cancel,
            "error" => Kind::Error,
            "ping" => Kind::Ping,
            "pong" => Kind::Pong,
            "discover" => Kind::Discover,
            "identify" => Kind::Identify,
            "identity_accepted" => Kind::IdentityAccepted,
            "route_snapshot" => Kind::RouteSnapshot,
            "route_delta" => Kind::RouteDelta,
            "route_ack" => Kind::RouteAck,
            _ => return None,
        })
    }

    pub fn is_application_request(self) -> bool {
        matches!(
            self,
            Kind::Request | Kind::Subscribe | Kind::Channel | Kind::Discover
        )
    }

    pub fn is_application_response(self) -> bool {
        matches!(self, Kind::Response | Kind::Event | Kind::Error)
    }
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Envelope {
    pub v: u16,
    pub id: String,
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub target: String,
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub subject: String,
    pub kind: Kind,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub corr: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub seq: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub hops: Option<u8>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub body_token: Option<String>,
    #[serde(skip)]
    pub payload: Bytes,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub path: Vec<String>,
    #[serde(flatten)]
    pub headers: serde_json::Map<String, Value>,
}

impl Envelope {
    pub fn decode(frame: Bytes) -> Result<Envelope, CoreError> {
        let (mut envelope, head_len) = Envelope::decode_head(&frame)?;
        envelope.payload = frame.slice(head_len..);
        Ok(envelope)
    }

    pub fn decode_head(frame: &[u8]) -> Result<(Envelope, usize), CoreError> {
        if frame.len() > MAX_HEAD_BYTES && !Envelope::head_within_bounds(frame) {
            return Err(CoreError::Malformed(
                "header section exceeds the maximum head size".into(),
            ));
        }
        let mut envelope = Envelope {
            v: PROTOCOL_VERSION,
            id: String::new(),
            target: String::new(),
            subject: String::new(),
            kind: Kind::Request,
            corr: None,
            seq: None,
            hops: None,
            body_token: None,
            payload: Bytes::new(),
            path: Vec::new(),
            headers: serde_json::Map::new(),
        };
        let mut storage = [httparse::EMPTY_HEADER; MAX_HEADERS];
        let is_status = frame.starts_with(b"HTTP/");
        let (head_len, status, target) = if is_status {
            let mut response = httparse::Response::new(&mut storage);
            let head_len = match response.parse(frame).map_err(Envelope::parse_error)? {
                httparse::Status::Complete(head_len) => head_len,
                httparse::Status::Partial => {
                    return Err(CoreError::Malformed("frame head is incomplete".into()))
                }
            };
            let code = response
                .code
                .ok_or_else(|| CoreError::Malformed("status line carries no status".into()))?;
            (head_len, Some(code), None)
        } else {
            let mut request = httparse::Request::new(&mut storage);
            let head_len = match request.parse(frame).map_err(Envelope::parse_error)? {
                httparse::Status::Complete(head_len) => head_len,
                httparse::Status::Partial => {
                    return Err(CoreError::Malformed("frame head is incomplete".into()))
                }
            };
            if request.method != Some("POST") {
                return Err(CoreError::Malformed(
                    "application and control frames use POST".into(),
                ));
            }
            let target = request
                .path
                .ok_or_else(|| CoreError::Malformed("request line carries no target".into()))?;
            (head_len, None, Some(target.to_string()))
        };
        let az_kind = envelope.absorb_wire_headers(&storage, is_status)?;
        if let Some(code) = status {
            if az_kind.is_some() {
                return Err(CoreError::Malformed(
                    "status-line frames carry no unb-kind".into(),
                ));
            }
            envelope.kind = match code {
                200 if envelope.seq.is_some() => Kind::Event,
                200 => Kind::Response,
                400..=599 => Kind::Error,
                other => {
                    return Err(CoreError::Malformed(format!(
                        "status {other} maps to no frame kind"
                    )));
                }
            };
        } else {
            let target = target.expect("request path present");
            if target.contains(['?', '#']) {
                return Err(CoreError::Malformed(
                    "targets must not contain uri query or fragment delimiters".into(),
                ));
            }
            if target == CONTROL_TARGET {
                let control = az_kind
                    .ok_or_else(|| CoreError::Malformed("control frames carry unb-kind".into()))?;
                if control.is_application_request() || control.is_application_response() {
                    return Err(CoreError::Malformed(format!(
                        "{control:?} is not a control kind"
                    )));
                }
                envelope.kind = control;
            } else if target.starts_with('/') {
                let kind = az_kind.unwrap_or(Kind::Request);
                if !kind.is_application_request() {
                    return Err(CoreError::Malformed(format!(
                        "{kind:?} is not an application request kind"
                    )));
                }
                let target_path = if kind == Kind::Discover {
                    TargetPath::parse_discovery(&target)?
                } else {
                    TargetPath::parse_application(&target)?
                };
                envelope.target = target_path.target().to_owned();
                envelope.subject = target_path.subject().to_owned();
                envelope.kind = kind;
            } else {
                return Err(CoreError::Malformed(
                    "request targets are origin-form and start with a slash".into(),
                ));
            }
        }
        Ok((envelope, head_len))
    }

    pub fn encode(&self) -> Bytes {
        use std::fmt::Write;
        let mut head = String::with_capacity(64 + self.headers.len() * 32);
        if self.kind.is_application_response() {
            let (status, reason) = match self.kind {
                Kind::Error => match self.error_code() {
                    Some(code) => (code.status().as_u16(), code.token()),
                    None => (500, "INTERNAL"),
                },
                _ => (200, "OK"),
            };
            write!(head, "HTTP/1.1 {status} {reason}\r\n")
                .expect("writing to a string cannot fail");
        } else if self.kind.is_application_request() {
            let target_path = if self.kind == Kind::Discover {
                TargetPath::discovery(self.target.clone())
            } else {
                TargetPath::application(self.target.clone(), self.subject.clone())
            }
            .expect("application envelopes carry a valid target path");
            write!(head, "POST {target_path} HTTP/1.1\r\n")
                .expect("writing to a string cannot fail");
        } else {
            write!(head, "POST {CONTROL_TARGET} HTTP/1.1\r\n")
                .expect("writing to a string cannot fail");
        }
        let line = |head: &mut String, name: &str, value: &str| {
            debug_assert!(
                !value.contains(['\r', '\n']),
                "header values are single-line strings"
            );
            head.push_str(name);
            head.push_str(": ");
            head.push_str(value);
            head.push_str("\r\n");
        };
        line(&mut head, UNB_VERSION, &self.v.to_string());
        if !self.id.is_empty() {
            line(&mut head, UNB_ID, &self.id);
        }
        if let Some(corr) = &self.corr {
            line(&mut head, UNB_CORR, corr);
        }
        if let Some(seq) = self.seq {
            line(&mut head, UNB_SEQ, &seq.to_string());
        }
        if let Some(hops) = self.hops {
            line(&mut head, UNB_HOPS, &hops.to_string());
        }
        if let Some(token) = &self.body_token {
            line(&mut head, UNB_BODY, token);
        }
        if !self.path.is_empty() {
            line(&mut head, UNB_PATH, &self.path.join(","));
        }
        if !self.kind.is_application_response() && self.kind != Kind::Request {
            line(&mut head, UNB_KIND, self.kind.token());
        }
        if self.kind == Kind::Error {
            if let Some(code) = self.payload_json()["code"].as_str() {
                line(&mut head, UNB_CODE, code);
            }
        }
        for (name, value) in &self.headers {
            let value = value
                .as_str()
                .expect("application headers carry string values on the wire");
            line(&mut head, name, value);
        }
        head.push_str("\r\n");
        let mut frame = BytesMut::with_capacity(head.len() + self.payload.len());
        frame.put_slice(head.as_bytes());
        frame.put_slice(&self.payload);
        frame.freeze()
    }

    fn head_within_bounds(frame: &[u8]) -> bool {
        frame
            .windows(4)
            .take(MAX_HEAD_BYTES)
            .position(|window| window == b"\r\n\r\n")
            .is_some_and(|position| position + 4 <= MAX_HEAD_BYTES)
    }

    fn parse_error(error: httparse::Error) -> CoreError {
        CoreError::Malformed(error.to_string())
    }

    fn absorb_wire_headers(
        &mut self,
        headers: &[httparse::Header],
        is_status: bool,
    ) -> Result<Option<Kind>, CoreError> {
        let mut az_kind = None;
        let mut az_version_seen = false;
        for header in headers {
            if header.name.is_empty() {
                continue;
            }
            let name = header.name.to_ascii_lowercase();
            let value = std::str::from_utf8(header.value)
                .map_err(|source| CoreError::Malformed(source.to_string()))?
                .trim_start_matches(' ');
            let duplicate = || CoreError::Malformed(format!("{name}: duplicate header"));
            match name.as_str() {
                "content-length" | "transfer-encoding" => {
                    return Err(CoreError::Malformed(format!(
                        "{name}: message-carrier frames carry no explicit body framing"
                    )));
                }
                UNB_VERSION => {
                    if az_version_seen {
                        return Err(duplicate());
                    }
                    az_version_seen = true;
                    self.v = value
                        .parse()
                        .map_err(|_| CoreError::Malformed(format!("{UNB_VERSION}: {value:?}")))?;
                }
                UNB_ID => {
                    if !self.id.is_empty() {
                        return Err(duplicate());
                    }
                    self.id = value.to_string();
                }
                UNB_CORR => {
                    if self.corr.is_some() {
                        return Err(duplicate());
                    }
                    self.corr = Some(value.to_string());
                }
                UNB_SEQ => {
                    if self.seq.is_some() {
                        return Err(duplicate());
                    }
                    self.seq = Some(
                        value
                            .parse()
                            .map_err(|_| CoreError::Malformed(format!("{UNB_SEQ}: {value:?}")))?,
                    );
                }
                UNB_HOPS => {
                    if self.hops.is_some() {
                        return Err(duplicate());
                    }
                    self.hops = Some(
                        value
                            .parse()
                            .map_err(|_| CoreError::Malformed(format!("{UNB_HOPS}: {value:?}")))?,
                    );
                }
                UNB_PATH => {
                    if !self.path.is_empty() {
                        return Err(duplicate());
                    }
                    self.path = value.split(',').map(str::to_string).collect();
                }
                UNB_BODY => {
                    if self.body_token.is_some() {
                        return Err(duplicate());
                    }
                    if value.is_empty() || value.len() > 256 {
                        return Err(CoreError::Malformed(format!(
                            "{UNB_BODY}: token length out of bounds"
                        )));
                    }
                    self.body_token = Some(value.to_string());
                }
                UNB_KIND => {
                    if az_kind.is_some() {
                        return Err(duplicate());
                    }
                    az_kind = Some(Kind::from_token(value).ok_or_else(|| {
                        CoreError::Malformed(format!("{UNB_KIND}: {value:?} is not a frame kind"))
                    })?);
                }
                UNB_CODE if is_status => {}
                other if other.starts_with("unb-") => {
                    return Err(CoreError::Malformed(format!(
                        "{other}: unknown reserved header"
                    )));
                }
                _ => {
                    if self
                        .headers
                        .insert(name.clone(), Value::String(value.to_string()))
                        .is_some()
                    {
                        return Err(duplicate());
                    }
                }
            }
        }
        Ok(az_kind)
    }

    fn error_code(&self) -> Option<crate::ErrorCode> {
        serde_json::from_value(self.payload_json()["code"].clone()).ok()
    }

    #[inline]
    pub fn payload_json(&self) -> Value {
        if self.payload.is_empty() {
            return Value::Null;
        }
        serde_json::from_slice(&self.payload).unwrap_or(Value::Null)
    }

    #[inline]
    pub fn encode_payload(value: &Value) -> Bytes {
        if value.is_null() {
            return Bytes::new();
        }
        serde_json::to_vec(value)
            .expect("payload is plain json data")
            .into()
    }

    #[inline]
    pub fn parse_payload<T: serde::de::DeserializeOwned>(&self) -> Result<T, CoreError> {
        serde_json::from_slice(&self.payload)
            .map_err(|source| CoreError::Malformed(source.to_string()))
    }

    pub fn to_request(&self) -> Result<http::Request<Bytes>, CoreError> {
        if !self.kind.is_application_request() {
            return Err(CoreError::Malformed(format!(
                "{:?} is not an application request kind",
                self.kind
            )));
        }
        let target_path = if self.kind == Kind::Discover {
            TargetPath::discovery(self.target.clone())?
        } else {
            TargetPath::application(self.target.clone(), self.subject.clone())?
        };
        let mut request = http::Request::builder()
            .method(http::Method::POST)
            .uri(target_path.to_string())
            .body(self.payload.clone())
            .map_err(|source| CoreError::Malformed(source.to_string()))?;
        if self.kind != Kind::Request {
            request
                .headers_mut()
                .insert(UNB_KIND, http::HeaderValue::from_static(self.kind.token()));
        }
        self.project_reserved(request.headers_mut())?;
        Self::project_custom(request.headers_mut(), &self.headers)?;
        Ok(request)
    }

    pub fn to_local_request(&self) -> Result<http::Request<Bytes>, CoreError> {
        if !matches!(self.kind, Kind::Request | Kind::Subscribe | Kind::Channel) {
            return Err(CoreError::Malformed(format!(
                "{:?} is not a destination-local handler request kind",
                self.kind
            )));
        }
        let target_path = TargetPath::application(self.target.clone(), self.subject.clone())?;
        let mut request = http::Request::builder()
            .method(http::Method::POST)
            .uri(format!("/{}", target_path.subject().replace('.', "/")))
            .body(self.payload.clone())
            .map_err(|source| CoreError::Malformed(source.to_string()))?;
        if self.kind != Kind::Request {
            request
                .headers_mut()
                .insert(UNB_KIND, http::HeaderValue::from_static(self.kind.token()));
        }
        self.project_reserved(request.headers_mut())?;
        Self::project_custom(request.headers_mut(), &self.headers)?;
        Ok(request)
    }

    pub fn subject_of(uri: &http::Uri) -> String {
        let path = uri.path().trim_start_matches('/');
        if path.is_empty() {
            uri.authority()
                .map(|authority| authority.as_str().replace('/', "."))
                .unwrap_or_default()
        } else {
            path.replace('/', ".")
        }
    }

    pub fn ensure_headers_wire_safe(
        headers: &serde_json::Map<String, Value>,
    ) -> Result<(), CoreError> {
        Self::project_custom(&mut http::HeaderMap::new(), headers)
    }

    pub fn from_request(request: http::Request<Bytes>) -> Result<Envelope, CoreError> {
        if request.method() != http::Method::POST {
            return Err(CoreError::Malformed(format!(
                "application requests use POST, not {}",
                request.method()
            )));
        }
        if request.uri().query().is_some() {
            return Err(CoreError::Malformed(
                "application request targets must not carry a query string".into(),
            ));
        }
        let kind = match request.headers().get(UNB_KIND) {
            None => Kind::Request,
            Some(value) => {
                let value = value
                    .to_str()
                    .map_err(|source| CoreError::Malformed(format!("{UNB_KIND}: {source}")))?;
                match Kind::from_token(value) {
                    Some(kind) if kind.is_application_request() => kind,
                    _ => {
                        return Err(CoreError::Malformed(format!(
                            "{UNB_KIND}: {value:?} is not an application request kind"
                        )))
                    }
                }
            }
        };
        let (parts, payload) = request.into_parts();
        let target_path = if kind == Kind::Discover {
            TargetPath::parse_discovery(parts.uri.path())?
        } else {
            TargetPath::parse_application(parts.uri.path())?
        };
        let mut envelope = Envelope {
            v: PROTOCOL_VERSION,
            id: String::new(),
            target: target_path.target().to_owned(),
            subject: target_path.subject().to_owned(),
            kind,
            corr: None,
            seq: None,
            hops: None,
            body_token: None,
            payload,
            path: Vec::new(),
            headers: serde_json::Map::new(),
        };
        envelope.absorb_headers(&parts.headers, false)?;
        Ok(envelope)
    }

    pub fn to_response(&self) -> Result<http::Response<Bytes>, CoreError> {
        let status = match self.kind {
            Kind::Response | Kind::Event => http::StatusCode::OK,
            Kind::Error => self
                .error_code()
                .ok_or_else(|| {
                    CoreError::Malformed("error frame payload carries no known code".into())
                })?
                .status(),
            other => {
                return Err(CoreError::Malformed(format!(
                    "{other:?} is not an application response kind"
                )))
            }
        };
        if !self.subject.is_empty() {
            return Err(CoreError::Malformed(
                "response frames carry no subject".into(),
            ));
        }
        if self.kind == Kind::Response && self.seq.is_some() {
            return Err(CoreError::Malformed(
                "a unary response carries no seq; seq marks stream events".into(),
            ));
        }
        if self.kind == Kind::Event && self.seq.is_none() {
            return Err(CoreError::Malformed(
                "an event frame requires seq to remain distinguishable".into(),
            ));
        }
        let mut response = http::Response::builder()
            .status(status)
            .body(self.payload.clone())
            .map_err(|source| CoreError::Malformed(source.to_string()))?;
        self.project_reserved(response.headers_mut())?;
        if self.kind == Kind::Error {
            let code = self.payload_json()["code"].clone();
            if let Some(code) = code.as_str() {
                response.headers_mut().insert(
                    UNB_CODE,
                    http::HeaderValue::from_str(code)
                        .map_err(|source| CoreError::Malformed(source.to_string()))?,
                );
            }
        }
        Self::project_custom(response.headers_mut(), &self.headers)?;
        Ok(response)
    }

    pub fn from_response(response: http::Response<Bytes>) -> Result<Envelope, CoreError> {
        let status = response.status();
        let (parts, payload) = response.into_parts();
        let kind = if status == http::StatusCode::OK {
            if parts.headers.contains_key(UNB_SEQ) {
                Kind::Event
            } else {
                Kind::Response
            }
        } else if status.is_client_error() || status.is_server_error() {
            Kind::Error
        } else {
            return Err(CoreError::Malformed(format!(
                "status {status} maps to no application response kind"
            )));
        };
        let mut envelope = Envelope {
            v: PROTOCOL_VERSION,
            id: String::new(),
            target: String::new(),
            subject: String::new(),
            kind,
            corr: None,
            seq: None,
            hops: None,
            body_token: None,
            payload,
            path: Vec::new(),
            headers: serde_json::Map::new(),
        };
        envelope.absorb_headers(&parts.headers, true)?;
        Ok(envelope)
    }

    fn project_reserved(&self, headers: &mut http::HeaderMap) -> Result<(), CoreError> {
        let mut put = |name: &'static str, value: String| -> Result<(), CoreError> {
            let value = http::HeaderValue::from_str(&value)
                .map_err(|source| CoreError::Malformed(format!("{name}: {source}")))?;
            headers.insert(name, value);
            Ok(())
        };
        put(UNB_VERSION, self.v.to_string())?;
        if !self.id.is_empty() {
            put(UNB_ID, self.id.clone())?;
        }
        if let Some(corr) = &self.corr {
            put(UNB_CORR, corr.clone())?;
        }
        if let Some(seq) = self.seq {
            put(UNB_SEQ, seq.to_string())?;
        }
        if let Some(hops) = self.hops {
            put(UNB_HOPS, hops.to_string())?;
        }
        if !self.path.is_empty() {
            if self.path.iter().any(|hop| hop.contains(',')) {
                return Err(CoreError::Malformed(
                    "path elements must not contain commas".into(),
                ));
            }
            put(UNB_PATH, self.path.join(","))?;
        }
        Ok(())
    }

    fn project_custom(
        headers: &mut http::HeaderMap,
        custom: &serde_json::Map<String, Value>,
    ) -> Result<(), CoreError> {
        for (name, value) in custom {
            let name = http::HeaderName::try_from(name.as_str())
                .map_err(|source| CoreError::Malformed(source.to_string()))?;
            if name.as_str().starts_with("unb-") {
                return Err(CoreError::Malformed(format!(
                    "{name}: unb-* header names are reserved for framing metadata"
                )));
            }
            let Value::String(value) = value else {
                return Err(CoreError::Malformed(format!(
                    "{name}: application-frame header values must be strings"
                )));
            };
            let value = http::HeaderValue::from_str(value)
                .map_err(|source| CoreError::Malformed(format!("{name}: {source}")))?;
            headers.insert(name, value);
        }
        Ok(())
    }

    fn absorb_headers(
        &mut self,
        headers: &http::HeaderMap,
        response: bool,
    ) -> Result<(), CoreError> {
        for name in headers.keys() {
            let mut values = headers.get_all(name).iter();
            let value = values.next().expect("keys() yields present names");
            if values.next().is_some() {
                return Err(CoreError::Malformed(format!(
                    "{name}: duplicate header values are not representable"
                )));
            }
            let value = std::str::from_utf8(value.as_bytes())
                .map_err(|source| CoreError::Malformed(format!("{name}: {source}")))?;
            match name.as_str() {
                UNB_VERSION => {
                    self.v = value
                        .parse()
                        .map_err(|_| CoreError::Malformed(format!("{name}: {value:?}")))?;
                }
                UNB_ID => self.id = value.to_string(),
                UNB_CORR => self.corr = Some(value.to_string()),
                UNB_SEQ => {
                    self.seq = Some(
                        value
                            .parse()
                            .map_err(|_| CoreError::Malformed(format!("{name}: {value:?}")))?,
                    );
                }
                UNB_HOPS => {
                    self.hops = Some(
                        value
                            .parse()
                            .map_err(|_| CoreError::Malformed(format!("{name}: {value:?}")))?,
                    );
                }
                UNB_PATH => {
                    self.path = value.split(',').map(str::to_string).collect();
                }
                UNB_CODE if response => {}
                UNB_KIND if !response => {}
                other if other.starts_with("unb-") => {
                    return Err(CoreError::Malformed(format!(
                        "{other}: unknown reserved header"
                    )));
                }
                other => {
                    self.headers
                        .insert(other.to_string(), Value::String(value.to_string()));
                }
            }
        }
        Ok(())
    }
}

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

    fn hex_bytes(hex: &str) -> Bytes {
        let hex = hex.trim();
        (0..hex.len())
            .step_by(2)
            .map(|i| u8::from_str_radix(&hex[i..i + 2], 16).expect("hex fixture"))
            .collect::<Vec<u8>>()
            .into()
    }

    fn envelope(kind: Kind) -> Envelope {
        Envelope {
            v: PROTOCOL_VERSION,
            id: "f1".into(),
            target: if kind.is_application_request() {
                "node-a".into()
            } else {
                String::new()
            },
            subject: String::new(),
            kind,
            corr: None,
            seq: None,
            hops: None,
            body_token: None,
            payload: Bytes::new(),
            path: Vec::new(),
            headers: Default::default(),
        }
    }

    #[test]
    fn a_v1_request_line_frame_round_trips() {
        let mut headers = serde_json::Map::new();
        headers.insert("authorization".into(), json!("Bearer jwt-abc"));
        let request = Envelope {
            v: PROTOCOL_VERSION,
            id: "f2".into(),
            target: "node-b".into(),
            subject: "chess.move".into(),
            kind: Kind::Request,
            corr: Some("s1".into()),
            seq: None,
            hops: Some(DEFAULT_HOPS),
            body_token: None,
            payload: Envelope::encode_payload(&json!({"from": "e2", "to": "e4"})),
            path: vec!["node-a".into()],
            headers,
        };
        let frame = request.encode();
        let text = std::str::from_utf8(&frame).unwrap();
        assert_eq!(
            text,
            "POST /node-b/chess/move HTTP/1.1\r\nunb-version: 1\r\nunb-id: f2\r\nunb-corr: s1\r\nunb-hops: 8\r\n\
             unb-path: node-a\r\nauthorization: Bearer jwt-abc\r\n\r\n\
             {\"from\":\"e2\",\"to\":\"e4\"}"
        );
        assert_eq!(Envelope::decode(frame).unwrap(), request);
    }

    #[test]
    fn every_kind_takes_its_lane_on_the_wire() {
        for kind in [
            Kind::Hello,
            Kind::Welcome,
            Kind::Request,
            Kind::Response,
            Kind::Subscribe,
            Kind::Event,
            Kind::Channel,
            Kind::Cancel,
            Kind::Error,
            Kind::Ping,
            Kind::Pong,
            Kind::Discover,
            Kind::Identify,
            Kind::IdentityAccepted,
            Kind::RouteSnapshot,
            Kind::RouteDelta,
            Kind::RouteAck,
        ] {
            let mut envelope = envelope(kind);
            if kind.is_application_request() && kind != Kind::Discover {
                envelope.subject = "chess".into();
                envelope.corr = Some("s1".into());
            }
            if kind == Kind::Discover {
                envelope.corr = Some("s1".into());
            }
            if kind.is_application_response() {
                envelope.corr = Some("s1".into());
            }
            if kind == Kind::Event {
                envelope.seq = Some(7);
            }
            if kind == Kind::Error {
                envelope.payload =
                    Envelope::encode_payload(&json!({"code": "BUSY", "message": "full"}));
            }
            let frame = envelope.encode();
            let text = std::str::from_utf8(&frame).unwrap();
            if kind.is_application_response() {
                assert!(text.starts_with("HTTP/1.1 "), "{kind:?}: {text}");
            } else if kind == Kind::Discover {
                assert!(
                    text.starts_with("POST /node-a HTTP/1.1"),
                    "{kind:?}: {text}"
                );
            } else if kind.is_application_request() {
                assert!(
                    text.starts_with("POST /node-a/chess HTTP/1.1"),
                    "{kind:?}: {text}"
                );
            } else {
                assert!(text.starts_with("POST * HTTP/1.1"), "{kind:?}: {text}");
            }
            assert_eq!(Envelope::decode(frame).unwrap(), envelope, "{kind:?}");
        }
    }

    #[test]
    fn a_control_frame_routes_by_az_kind_over_the_asterisk_target() {
        let mut hello = envelope(Kind::Hello);
        hello.payload = Envelope::encode_payload(&json!({"versions": [1]}));
        assert_eq!(
            std::str::from_utf8(&hello.encode()).unwrap(),
            "POST * HTTP/1.1\r\nunb-version: 1\r\nunb-id: f1\r\nunb-kind: hello\r\n\r\n{\"versions\":[1]}"
        );
    }

    #[test]
    fn a_ping_frame_has_an_empty_body() {
        let ping = envelope(Kind::Ping);
        let frame = ping.encode();
        assert_eq!(
            std::str::from_utf8(&frame).unwrap(),
            "POST * HTTP/1.1\r\nunb-version: 1\r\nunb-id: f1\r\nunb-kind: ping\r\n\r\n"
        );
        assert_eq!(Envelope::decode(frame).unwrap(), ping);
    }

    #[test]
    fn an_error_status_line_names_the_error_code() {
        let mut error = envelope(Kind::Error);
        error.corr = Some("s1".into());
        error.payload =
            Envelope::encode_payload(&json!({"code": "BUSY", "message": "node at capacity"}));
        let frame = error.encode();
        let text = std::str::from_utf8(&frame).unwrap();
        assert!(text.starts_with("HTTP/1.1 503 BUSY\r\n"), "{text}");
        assert!(text.contains("\r\nunb-code: BUSY\r\n"), "{text}");
        assert_eq!(Envelope::decode(frame).unwrap(), error);
    }

    #[test]
    fn every_error_code_has_canonical_wire_status_reason_and_restoration() {
        for (code, status) in [
            (crate::ErrorCode::VersionMismatch, 505),
            (crate::ErrorCode::Protocol, 400),
            (crate::ErrorCode::UnknownSubject, 404),
            (crate::ErrorCode::UnresolvedAtPeer, 421),
            (crate::ErrorCode::PeerUnreachable, 502),
            (crate::ErrorCode::HopLimitExceeded, 508),
            (crate::ErrorCode::InvalidInput, 400),
            (crate::ErrorCode::Unauthorized, 401),
            (crate::ErrorCode::Conflict, 409),
            (crate::ErrorCode::Busy, 503),
            (crate::ErrorCode::Cancelled, 499),
            (crate::ErrorCode::Internal, 500),
        ] {
            let mut error = envelope(Kind::Error);
            error.corr = Some("s1".into());
            error.payload = Envelope::encode_payload(&json!({
                "code": code.token(),
                "message": "failure"
            }));
            let frame = error.encode();
            let text = std::str::from_utf8(&frame).unwrap();
            assert!(
                text.starts_with(&format!("HTTP/1.1 {status} {}\r\n", code.token())),
                "{code:?}: {text}"
            );
            assert!(
                text.contains(&format!("\r\nunb-code: {}\r\n", code.token())),
                "{code:?}: {text}"
            );
            let decoded = Envelope::decode(frame.clone()).unwrap();
            assert_eq!(decoded, error, "{code:?}");
            let restored = Envelope::from_response(decoded.to_response().unwrap()).unwrap();
            assert_eq!(restored.payload_json()["code"], code.token(), "{code:?}");
            assert_eq!(restored.encode(), frame, "{code:?}");
        }
    }

    #[test]
    fn an_unmapped_error_payload_still_frames_as_a_server_error() {
        let mut error = envelope(Kind::Error);
        error.corr = Some("s1".into());
        error.payload = Envelope::encode_payload(&json!({"note": "no code field"}));
        let frame = error.encode();
        let text = std::str::from_utf8(&frame).unwrap();
        assert!(text.starts_with("HTTP/1.1 500 INTERNAL\r\n"), "{text}");
        let decoded = Envelope::decode(frame).unwrap();
        assert_eq!(decoded.kind, Kind::Error);
    }

    #[test]
    fn decode_slices_the_body_zero_copy() {
        let mut request = envelope(Kind::Request);
        request.subject = "chess".into();
        request.corr = Some("s1".into());
        request.payload = Envelope::encode_payload(&json!({"from": "e2"}));
        let frame = request.encode();
        let body_start = frame.len() - request.payload.len();
        let decoded = Envelope::decode(frame.clone()).unwrap();
        assert_eq!(decoded.payload.as_ptr(), frame[body_start..].as_ptr());
    }

    #[test]
    fn dot_and_slash_subject_forms_are_equivalent() {
        for uri in ["/node-a/chess.move", "/node-a/chess/move"] {
            let request = http::Request::builder()
                .method("POST")
                .uri(uri)
                .body(Bytes::new())
                .unwrap();
            let envelope = Envelope::from_request(request).unwrap();
            assert_eq!(envelope.target, "node-a");
            assert_eq!(envelope.subject, "chess.move");
        }
    }

    #[test]
    fn a_malformed_head_is_a_typed_decode_error() {
        for frame in [
            &b"GARBAGE\r\n\r\n"[..],
            b"POST /x HTTP/1.1\r\nunb-id: f1",
            b"GET /x HTTP/1.1\r\n\r\n",
            b"POST x HTTP/1.1\r\n\r\n",
            b"POST /x?side=w HTTP/1.1\r\nunb-version: 1\r\n\r\n",
            b"POST /az/teleport HTTP/1.1\r\nunb-version: 1\r\n\r\n",
            b"POST /az HTTP/1.1\r\nunb-version: 1\r\n\r\n",
            b"POST /az/move HTTP/1.1\r\nunb-version: 1\r\n\r\n",
            b"POST /x HTTP/1.1\r\nunb-kind: response\r\n\r\n",
            b"POST /x HTTP/1.1\r\nunb-magic: 1\r\n\r\n",
            b"POST /x HTTP/1.1\r\nunb-corr: a\r\nunb-corr: b\r\n\r\n",
            b"POST /x HTTP/1.1\r\nactor: a\r\nactor: b\r\n\r\n",
            b"POST * HTTP/1.1\r\nunb-kind: request\r\n\r\n",
            b"POST * HTTP/1.1\r\n\r\n",
            b"POST /x HTTP/1.1\r\ncontent-length: 5\r\n\r\nhello",
            b"POST /x HTTP/1.1\r\ntransfer-encoding: chunked\r\n\r\n",
            b"HTTP/1.1 302 FOUND\r\n\r\n",
            b"HTTP/1.1 200 OK\r\nunb-kind: response\r\n\r\n",
        ] {
            let error = Envelope::decode(Bytes::copy_from_slice(frame)).unwrap_err();
            assert!(
                matches!(error, CoreError::Malformed(_)),
                "{:?}: {error}",
                std::str::from_utf8(frame)
            );
        }
    }

    #[test]
    fn decode_rejects_bare_carriage_returns_and_accepts_supported_line_endings() {
        for frame in [
            &b"POST /x\rHTTP/1.1\r\n\r\n"[..],
            &b"POST /x HTTP/1.1\r\nact\ror: a\r\n\r\n"[..],
            &b"POST /x HTTP/1.1\r\nactor: a\rb\r\n\r\n"[..],
        ] {
            assert!(matches!(
                Envelope::decode(Bytes::copy_from_slice(frame)),
                Err(CoreError::Malformed(_))
            ));
        }

        assert_eq!(
            Envelope::decode(Bytes::from_static(
                b"POST /node-a/x HTTP/1.1\r\nactor: a\r\n\r\n",
            ))
            .unwrap()
            .headers["actor"],
            "a"
        );
    }

    #[test]
    fn the_exact_head_bound_is_accepted_and_one_byte_over_is_rejected() {
        let prefix = "POST /node-a/x HTTP/1.1\r\nactor: ";
        let suffix = "\r\n\r\n";
        let value_len = MAX_HEAD_BYTES - prefix.len() - suffix.len();
        let exact = format!("{prefix}{}{suffix}", "x".repeat(value_len));
        assert_eq!(exact.len(), MAX_HEAD_BYTES);
        assert_eq!(
            Envelope::decode(Bytes::from(exact)).unwrap().headers["actor"],
            "x".repeat(value_len)
        );

        let text = format!("{prefix}{}{suffix}", "x".repeat(value_len + 1));
        assert_eq!(text.len(), MAX_HEAD_BYTES + 1);
        assert!(matches!(
            Envelope::decode(Bytes::from(text)),
            Err(CoreError::Malformed(_))
        ));
    }

    #[test]
    fn custom_headers_stay_headers_not_core_fields() {
        let frame = Bytes::from_static(
            b"POST /node-a/chess HTTP/1.1\r\nunb-corr: s1\r\nsubject: sneaky\r\n\r\n",
        );
        let decoded = Envelope::decode(frame).unwrap();
        assert_eq!(decoded.target, "node-a");
        assert_eq!(decoded.subject, "chess");
        assert_eq!(decoded.corr.as_deref(), Some("s1"));
        assert_eq!(decoded.headers["subject"], "sneaky");
    }

    #[test]
    fn a_non_json_payload_round_trips_verbatim() {
        let mut opaque = envelope(Kind::Request);
        opaque.subject = "files".into();
        opaque.corr = Some("s1".into());
        opaque.payload = Bytes::from_static(&[0x00, 0x01, 0xff, 0xfe, b'!', 0x80]);
        let decoded = Envelope::decode(opaque.encode()).unwrap();
        assert_eq!(decoded, opaque);
        assert_eq!(decoded.payload_json(), Value::Null);
    }

    #[test]
    fn a_custom_header_round_trips_verbatim() {
        let mut headers = serde_json::Map::new();
        headers.insert("actor".into(), json!("jwt-abc"));
        headers.insert("x-trace".into(), json!("span-7"));
        let mut request = envelope(Kind::Request);
        request.subject = "chess".into();
        request.corr = Some("s1".into());
        request.headers = headers;
        let decoded = Envelope::decode(request.encode()).unwrap();
        assert_eq!(decoded, request);
        assert_eq!(decoded.headers["actor"], "jwt-abc");
    }

    #[test]
    fn a_mixed_case_unb_header_is_rejected_as_reserved() {
        for name in ["Unb-Corr", "UNB-KIND", "uNb-hOpS"] {
            let mut headers = serde_json::Map::new();
            headers.insert(name.into(), json!("forged"));
            assert!(matches!(
                Envelope::ensure_headers_wire_safe(&headers),
                Err(CoreError::Malformed(message)) if message.contains("reserved")
            ));
            let mut request = envelope(Kind::Request);
            request.subject = "chess".into();
            request.corr = Some("s1".into());
            request.headers.insert(name.into(), json!("forged"));
            assert!(matches!(
                request.to_request(),
                Err(CoreError::Malformed(message)) if message.contains("reserved")
            ));
        }
    }

    #[test]
    fn an_application_envelope_round_trips_through_the_request_model() {
        let mut headers = serde_json::Map::new();
        headers.insert("authorization".into(), json!("Bearer jwt-abc"));
        let envelope = Envelope {
            v: PROTOCOL_VERSION,
            id: "f7".into(),
            target: "node-a".into(),
            subject: "chess.move".into(),
            kind: Kind::Request,
            corr: Some("s1".into()),
            seq: None,
            hops: Some(DEFAULT_HOPS),
            body_token: None,
            payload: Envelope::encode_payload(&json!({"from": "e2", "to": "e4"})),
            path: vec!["node-a".into()],
            headers,
        };
        let request = envelope.to_request().unwrap();
        assert_eq!(request.method(), http::Method::POST);
        assert_eq!(request.uri().path(), "/node-a/chess/move");
        assert_eq!(request.headers()["authorization"], "Bearer jwt-abc");
        assert_eq!(request.headers()[UNB_CORR], "s1");
        assert_eq!(request.headers()[UNB_ID], "f7");
        let back = Envelope::from_request(request).unwrap();
        assert_eq!(back.encode(), envelope.encode());
    }

    #[test]
    fn every_request_kind_rides_az_kind_over_post_and_back() {
        for (kind, marker) in [
            (Kind::Request, None),
            (Kind::Subscribe, Some("subscribe")),
            (Kind::Channel, Some("channel")),
            (Kind::Discover, Some("discover")),
        ] {
            let mut envelope = envelope(kind);
            if kind != Kind::Discover {
                envelope.subject = "chess".into();
            }
            envelope.corr = Some("s1".into());
            let request = envelope.to_request().unwrap();
            assert_eq!(request.method(), http::Method::POST);
            match marker {
                Some(marker) => assert_eq!(request.headers()[UNB_KIND], marker),
                None => assert!(!request.headers().contains_key(UNB_KIND)),
            }
            assert_eq!(Envelope::from_request(request).unwrap().kind, kind);
        }
    }

    #[test]
    fn request_conversion_rejects_what_the_model_forbids() {
        let mut base = envelope(Kind::Request);
        base.subject = "chess".into();
        base.corr = Some("s1".into());

        let mut control = base.clone();
        control.kind = Kind::Ping;
        assert!(matches!(control.to_request(), Err(CoreError::Malformed(_))));

        let mut nested = base.clone();
        nested.headers.insert("trace".into(), json!({"span": 7}));
        assert!(matches!(nested.to_request(), Err(CoreError::Malformed(_))));

        let mut shadowing = base.clone();
        shadowing.headers.insert(UNB_CORR.into(), json!("spoof"));
        assert!(matches!(
            shadowing.to_request(),
            Err(CoreError::Malformed(_))
        ));

        let mut queried = base.clone();
        queried.subject = "chess.move?side=white".into();
        assert!(matches!(queried.to_request(), Err(CoreError::Malformed(_))));

        let mut reserved = base.clone();
        reserved.target = "az".into();
        assert!(matches!(
            reserved.to_request(),
            Err(CoreError::Malformed(_))
        ));

        let wrong_method = http::Request::builder()
            .method("GET")
            .uri("/node-a/chess")
            .body(Bytes::new())
            .unwrap();
        assert!(matches!(
            Envelope::from_request(wrong_method),
            Err(CoreError::Malformed(_))
        ));

        let unknown_kind = http::Request::builder()
            .method("POST")
            .uri("/node-a/chess")
            .header(UNB_KIND, "teleport")
            .body(Bytes::new())
            .unwrap();
        assert!(matches!(
            Envelope::from_request(unknown_kind),
            Err(CoreError::Malformed(_))
        ));

        let queried = http::Request::builder()
            .method("POST")
            .uri("/node-a/chess/move?draft=1")
            .body(Bytes::new())
            .unwrap();
        assert!(matches!(
            Envelope::from_request(queried),
            Err(CoreError::Malformed(message)) if message.contains("query string")
        ));

        let reserved_subject = http::Request::builder()
            .method("POST")
            .uri("/az/hello")
            .body(Bytes::new())
            .unwrap();
        assert!(matches!(
            Envelope::from_request(reserved_subject),
            Err(CoreError::Malformed(_))
        ));
    }

    #[test]
    fn a_fresh_user_request_converts_with_wire_defaults() {
        let request = http::Request::builder()
            .method("POST")
            .uri("/node-a/chess/move")
            .header("authorization", "jwt-abc")
            .body(Bytes::from_static(b"{}"))
            .unwrap();
        let envelope = Envelope::from_request(request).unwrap();
        assert_eq!(envelope.v, PROTOCOL_VERSION);
        assert_eq!(envelope.kind, Kind::Request);
        assert_eq!(envelope.target, "node-a");
        assert_eq!(envelope.subject, "chess.move");
        assert!(envelope.id.is_empty());
        assert_eq!(envelope.corr, None);
        assert_eq!(envelope.headers["authorization"], "jwt-abc");
    }

    #[test]
    fn model_conversion_shares_the_payload_allocation() {
        let mut source = envelope(Kind::Request);
        source.subject = "chess.move".into();
        source.corr = Some("s1".into());
        source.payload = Envelope::encode_payload(&json!({"from": "e2"}));
        let request = source.to_request().unwrap();
        assert_eq!(request.body().as_ptr(), source.payload.as_ptr());
        let back = Envelope::from_request(request).unwrap();
        assert_eq!(back.payload.as_ptr(), source.payload.as_ptr());
    }

    #[test]
    fn response_event_and_error_envelopes_round_trip_through_the_response_model() {
        let mut response = envelope(Kind::Response);
        response.id = "f8".into();
        response.corr = Some("s1".into());
        response.payload = Envelope::encode_payload(&json!({"ok": true}));
        let converted = response.to_response().unwrap();
        assert_eq!(converted.status(), http::StatusCode::OK);
        let back = Envelope::from_response(converted).unwrap();
        assert_eq!(back.encode(), response.encode());

        let mut event = response.clone();
        event.kind = Kind::Event;
        event.seq = Some(42);
        let converted = event.to_response().unwrap();
        assert_eq!(converted.headers()[UNB_SEQ], "42");
        let back = Envelope::from_response(converted).unwrap();
        assert_eq!(back.kind, Kind::Event);
        assert_eq!(back.encode(), event.encode());

        let mut error = response.clone();
        error.kind = Kind::Error;
        error.payload =
            Envelope::encode_payload(&json!({"code": "BUSY", "message": "node at capacity"}));
        let converted = error.to_response().unwrap();
        assert_eq!(converted.status(), http::StatusCode::SERVICE_UNAVAILABLE);
        assert_eq!(converted.headers()[UNB_CODE], "BUSY");
        let back = Envelope::from_response(converted).unwrap();
        assert_eq!(back.encode(), error.encode());
    }

    #[test]
    fn reserved_fields_project_and_restore_across_legal_application_lanes() {
        for kind in [Kind::Request, Kind::Event, Kind::Error] {
            let mut source = envelope(kind);
            source.id = "f-reserved".into();
            source.subject = if kind == Kind::Request {
                "chess.move".into()
            } else {
                String::new()
            };
            source.corr = Some("s-reserved".into());
            source.seq = (kind == Kind::Event).then_some(17);
            source.hops = Some(6);
            source.path = vec!["leaf-a".into(), "hub-b".into(), "root-c".into()];
            source.headers.insert("x-trace".into(), json!("span-7"));
            source.payload = if kind == Kind::Error {
                Envelope::encode_payload(&json!({"code": "PROTOCOL", "message": "bad"}))
            } else {
                Bytes::from_static(b"opaque")
            };

            let restored = if kind == Kind::Request {
                let projected = source.to_request().unwrap();
                assert_eq!(projected.headers()[UNB_VERSION], "1");
                assert_eq!(projected.headers()[UNB_ID], "f-reserved");
                assert_eq!(projected.headers()[UNB_CORR], "s-reserved");
                assert_eq!(projected.headers()[UNB_HOPS], "6");
                assert_eq!(projected.headers()[UNB_PATH], "leaf-a,hub-b,root-c");
                assert_eq!(projected.headers()["x-trace"], "span-7");
                Envelope::from_request(projected).unwrap()
            } else {
                let projected = source.to_response().unwrap();
                assert_eq!(projected.headers()[UNB_VERSION], "1");
                assert_eq!(projected.headers()[UNB_ID], "f-reserved");
                assert_eq!(projected.headers()[UNB_CORR], "s-reserved");
                assert_eq!(projected.headers()[UNB_HOPS], "6");
                assert_eq!(projected.headers()[UNB_PATH], "leaf-a,hub-b,root-c");
                assert_eq!(projected.headers()["x-trace"], "span-7");
                if kind == Kind::Event {
                    assert_eq!(projected.headers()[UNB_SEQ], "17");
                } else {
                    assert_eq!(projected.headers()[UNB_CODE], "PROTOCOL");
                }
                Envelope::from_response(projected).unwrap()
            };
            assert_eq!(restored, source, "{kind:?}");
        }
    }

    #[test]
    fn comma_path_elements_are_rejected_on_every_legal_projection_lane() {
        for kind in [Kind::Request, Kind::Event, Kind::Error] {
            let mut source = envelope(kind);
            source.subject = if kind == Kind::Request {
                "chess".into()
            } else {
                String::new()
            };
            source.corr = Some("s1".into());
            source.seq = (kind == Kind::Event).then_some(1);
            source.path = vec!["leaf-a,forged-hop".into()];
            if kind == Kind::Error {
                source.payload = Envelope::encode_payload(&json!({"code": "INTERNAL"}));
            }
            let result = if kind == Kind::Request {
                source.to_request().map(|_| ())
            } else {
                source.to_response().map(|_| ())
            };
            assert!(matches!(result, Err(CoreError::Malformed(_))), "{kind:?}");
        }
    }

    #[test]
    fn error_status_collisions_restore_the_exact_code_from_az_code() {
        for code in ["INVALID_INPUT", "PROTOCOL"] {
            let mut error = envelope(Kind::Error);
            error.id = "f9".into();
            error.corr = Some("s1".into());
            error.payload = Envelope::encode_payload(&json!({"code": code, "message": "bad"}));
            let converted = error.to_response().unwrap();
            assert_eq!(converted.status(), http::StatusCode::BAD_REQUEST);
            assert_eq!(converted.headers()[UNB_CODE], code);
            let back = Envelope::from_response(converted).unwrap();
            assert_eq!(back.encode(), error.encode());
        }
    }

    #[test]
    fn header_validation_rejects_frame_splitting_input() {
        let clean = {
            let mut headers = serde_json::Map::new();
            headers.insert("x-trace".into(), json!("span-7"));
            headers
        };
        assert!(Envelope::ensure_headers_wire_safe(&clean).is_ok());

        let crlf_value = {
            let mut headers = serde_json::Map::new();
            headers.insert("x-trace".into(), json!("span\r\nunb-corr: forged"));
            headers
        };
        assert!(matches!(
            Envelope::ensure_headers_wire_safe(&crlf_value),
            Err(CoreError::Malformed(_))
        ));

        let crlf_name = {
            let mut headers = serde_json::Map::new();
            headers.insert("x\r\nInjected".into(), json!("v"));
            headers
        };
        assert!(matches!(
            Envelope::ensure_headers_wire_safe(&crlf_name),
            Err(CoreError::Malformed(_))
        ));

        let non_string = {
            let mut headers = serde_json::Map::new();
            headers.insert("x-trace".into(), json!({ "nested": true }));
            headers
        };
        assert!(matches!(
            Envelope::ensure_headers_wire_safe(&non_string),
            Err(CoreError::Malformed(_))
        ));
    }

    fn golden_envelopes() -> Vec<(&'static str, Envelope)> {
        let build = |id: &str, subject: &str, kind: Kind, corr: Option<&str>| Envelope {
            v: PROTOCOL_VERSION,
            id: id.into(),
            target: if kind.is_application_request() {
                "node-a".into()
            } else {
                String::new()
            },
            subject: subject.into(),
            kind,
            corr: corr.map(str::to_string),
            seq: None,
            hops: None,
            body_token: None,
            payload: Bytes::new(),
            path: Vec::new(),
            headers: Default::default(),
        };
        let mut request = build("f2", "chess", Kind::Request, Some("s1"));
        request.hops = Some(DEFAULT_HOPS);
        request.payload = Envelope::encode_payload(
            &json!({"action": "move", "input": {"from": "e2", "to": "e4"}}),
        );
        let mut event = build("f9", "", Kind::Event, Some("s1"));
        event.seq = Some(3);
        event.payload =
            Envelope::encode_payload(&json!({"as_of": 4711, "value": {"done": false, "id": "t1"}}));
        let mut response = build("f3", "", Kind::Response, Some("s1"));
        response.payload = Envelope::encode_payload(&json!({"ok": true}));
        let mut subscribe = build("f8", "todo.changes", Kind::Subscribe, Some("s2"));
        subscribe.hops = Some(DEFAULT_HOPS);
        subscribe.payload = Envelope::encode_payload(&json!({"after": 4711}));
        let mut error = build("f4", "", Kind::Error, Some("s1"));
        error.path = vec!["node-a".into(), "node-b".into()];
        error.payload = Envelope::encode_payload(&json!({
            "code": "UNKNOWN_SUBJECT",
            "message": "Unknown subject \"ches\". Did you mean \"chess\"?"
        }));
        let ping = build("f1", "", Kind::Ping, None);
        let mut identify = build("f1", "", Kind::Identify, None);
        identify.payload = Envelope::encode_payload(
            &json!({"node_id": "node-a", "instance_id": "node-a-1", "epoch": 1}),
        );
        let identity_accepted = build("f2", "", Kind::IdentityAccepted, None);
        let mut route_snapshot = build("f3", "", Kind::RouteSnapshot, None);
        route_snapshot.payload = Envelope::encode_payload(&json!({
            "generation": 1,
            "routes": [{
                "subject": "chess", "owner": "leaf-a", "owner_instance": "leaf-a-1",
                "owner_epoch": 1, "owner_revision": 0, "distance": 1,
                "path": ["leaf-a", "hub"]
            }]
        }));
        let mut route_delta = build("f4", "", Kind::RouteDelta, None);
        route_delta.payload = Envelope::encode_payload(&json!({
            "generation": 2,
            "upsert": [],
            "withdraw": [{
                "subject": "chess", "owner": "leaf-a", "owner_instance": "leaf-a-1",
                "owner_epoch": 1, "owner_revision": 0
            }]
        }));
        let mut route_ack = build("f5", "", Kind::RouteAck, None);
        route_ack.payload =
            Envelope::encode_payload(&json!({"generation": 2, "status": "applied"}));
        vec![
            ("request", request),
            ("response", response),
            ("subscribe", subscribe),
            ("event", event),
            ("error", error),
            ("ping", ping),
            ("identify", identify),
            ("identity_accepted", identity_accepted),
            ("route_snapshot", route_snapshot),
            ("route_delta", route_delta),
            ("route_ack", route_ack),
        ]
    }

    #[test]
    #[ignore = "regenerates the golden fixtures from the canonical encoder"]
    fn regenerate_golden_fixtures() {
        let root = concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../../conformance/fixtures/envelope"
        );
        for (name, envelope) in golden_envelopes() {
            let frame = envelope.encode();
            let hex: String = frame.iter().map(|byte| format!("{byte:02x}")).collect();
            std::fs::write(format!("{root}/{name}.frame.hex"), format!("{hex}\n")).unwrap();
            std::fs::write(
                format!("{root}/{name}.header.json"),
                format!("{}\n", serde_json::to_string(&envelope).unwrap()),
            )
            .unwrap();
        }
    }

    #[test]
    fn az_body_is_a_core_field_on_the_wire_and_reserved_for_users() {
        let mut headers = serde_json::Map::new();
        headers.insert(UNB_BODY.into(), Value::String("token".into()));
        assert!(matches!(
            Envelope::ensure_headers_wire_safe(&headers),
            Err(CoreError::Malformed(_))
        ));
        let frame = Bytes::from_static(
            b"POST /node-a/echo HTTP/1.1\r\nunb-id: a\r\nunb-body: token\r\n\r\n",
        );
        let decoded = Envelope::decode(frame).unwrap();
        assert_eq!(decoded.body_token.as_deref(), Some("token"));
        assert!(!decoded.headers.contains_key(UNB_BODY));
        let round = Envelope::decode(decoded.encode()).unwrap();
        assert_eq!(round.body_token.as_deref(), Some("token"));
        let oversized = format!(
            "POST /node-a/echo HTTP/1.1\r\nunb-id: a\r\nunb-body: {}\r\n\r\n",
            "x".repeat(257)
        );
        assert!(matches!(
            Envelope::decode(Bytes::from(oversized)),
            Err(CoreError::Malformed(_))
        ));
    }

    #[test]
    fn head_first_decode_matches_whole_message_decode() {
        for (name, envelope) in golden_envelopes() {
            let frame = envelope.encode();
            let whole = Envelope::decode(frame.clone()).unwrap();
            let (mut head, head_len) = Envelope::decode_head(&frame).unwrap();
            assert!(
                head.payload.is_empty(),
                "{name}: head decode carries no payload"
            );
            head.payload = frame.slice(head_len..);
            assert_eq!(head, whole, "{name}");
        }
    }

    #[test]
    fn golden_fixtures_round_trip_byte_identically() {
        for (frame_hex, header_json) in [
            (
                include_str!("../../../conformance/fixtures/envelope/request.frame.hex"),
                include_str!("../../../conformance/fixtures/envelope/request.header.json"),
            ),
            (
                include_str!("../../../conformance/fixtures/envelope/response.frame.hex"),
                include_str!("../../../conformance/fixtures/envelope/response.header.json"),
            ),
            (
                include_str!("../../../conformance/fixtures/envelope/subscribe.frame.hex"),
                include_str!("../../../conformance/fixtures/envelope/subscribe.header.json"),
            ),
            (
                include_str!("../../../conformance/fixtures/envelope/event.frame.hex"),
                include_str!("../../../conformance/fixtures/envelope/event.header.json"),
            ),
            (
                include_str!("../../../conformance/fixtures/envelope/error.frame.hex"),
                include_str!("../../../conformance/fixtures/envelope/error.header.json"),
            ),
            (
                include_str!("../../../conformance/fixtures/envelope/ping.frame.hex"),
                include_str!("../../../conformance/fixtures/envelope/ping.header.json"),
            ),
            (
                include_str!("../../../conformance/fixtures/envelope/identify.frame.hex"),
                include_str!("../../../conformance/fixtures/envelope/identify.header.json"),
            ),
            (
                include_str!("../../../conformance/fixtures/envelope/identity_accepted.frame.hex"),
                include_str!(
                    "../../../conformance/fixtures/envelope/identity_accepted.header.json"
                ),
            ),
            (
                include_str!("../../../conformance/fixtures/envelope/route_snapshot.frame.hex"),
                include_str!("../../../conformance/fixtures/envelope/route_snapshot.header.json"),
            ),
            (
                include_str!("../../../conformance/fixtures/envelope/route_delta.frame.hex"),
                include_str!("../../../conformance/fixtures/envelope/route_delta.header.json"),
            ),
            (
                include_str!("../../../conformance/fixtures/envelope/route_ack.frame.hex"),
                include_str!("../../../conformance/fixtures/envelope/route_ack.header.json"),
            ),
        ] {
            let frame = hex_bytes(frame_hex);
            let envelope = Envelope::decode(frame.clone()).unwrap();
            assert_eq!(
                serde_json::to_string(&envelope).unwrap(),
                header_json.trim(),
                "header fixture must be the canonical envelope encoding"
            );
            assert_eq!(
                envelope.encode(),
                frame,
                "fixture must be the canonical frame encoding"
            );
        }
    }

    #[test]
    fn application_golden_fixtures_survive_the_model_byte_identically() {
        let request_frame = hex_bytes(include_str!(
            "../../../conformance/fixtures/envelope/request.frame.hex"
        ));
        let envelope = Envelope::decode(request_frame.clone()).unwrap();
        let back = Envelope::from_request(envelope.to_request().unwrap()).unwrap();
        assert_eq!(back.encode(), request_frame);

        let subscribe_frame = hex_bytes(include_str!(
            "../../../conformance/fixtures/envelope/subscribe.frame.hex"
        ));
        let envelope = Envelope::decode(subscribe_frame.clone()).unwrap();
        let back = Envelope::from_request(envelope.to_request().unwrap()).unwrap();
        assert_eq!(back.encode(), subscribe_frame);

        for fixture in [
            include_str!("../../../conformance/fixtures/envelope/response.frame.hex"),
            include_str!("../../../conformance/fixtures/envelope/event.frame.hex"),
            include_str!("../../../conformance/fixtures/envelope/error.frame.hex"),
        ] {
            let frame = hex_bytes(fixture);
            let envelope = Envelope::decode(frame.clone()).unwrap();
            let back = Envelope::from_response(envelope.to_response().unwrap()).unwrap();
            assert_eq!(back.encode(), frame);
        }
    }
}