flatbed 0.0.4

//! Core-NATS request-reply responders registered by `#[nats_route]`.
//!
//! A `#[nats_route]` handler is the subject-transport sibling of `#[route]`:
//! it takes the same [`Request`](crate::Request), returns the same
//! [`Response`](crate::Response) / [`FlatbedRouteError`](crate::FlatbedRouteError),
//! and is discovered through `inventory` without any re-export. Each handler
//! becomes a worker that subscribes to its subject and answers every message
//! carrying a reply subject.
//!
//! # Wire contract
//!
//! The request's `content-type` header selects the encoding for both the
//! decoded request body and the reply; a request with no recognized
//! `content-type` is read as FlatBuffers, so an empty payload sent without
//! headers is a decode failure rather than an empty body. Every reply carries
//! `content-type` and `x-request-id`. A handler failure is answered with an
//! error reply carrying `x-error-code`, `x-error-message`, and
//! `x-error-status` (the numeric HTTP status the handler chose), so a
//! requester's timeout never means the handler rejected the request. An
//! undecodable payload and a panicking handler are answered the same way; a
//! handler that never completes is the one case that produces no reply.
//!
//! A successful reply carries no status: [`Response::status`](crate::Response)
//! is not put on the wire, and the presence of `x-error-code` is what
//! distinguishes a rejection from a success.

use std::any::Any;
use std::collections::HashMap;
use std::sync::Arc;

use futures::{FutureExt, StreamExt};
use tracing::{error, info, warn};

use crate::nats::HasNatsClient;
use crate::{
    FlatbedRouteError, FlatbedWorkerError, HeaderMap, HeaderName, HeaderValue, Response,
    ToFlatBuffer,
};

/// Reply header naming the request this reply answers.
pub const REQUEST_ID_HEADER: &str = "x-request-id";
/// Reply header carrying the handler's error code (error replies only).
pub const ERROR_CODE_HEADER: &str = "x-error-code";
/// Reply header carrying the handler's error message (error replies only).
pub const ERROR_MESSAGE_HEADER: &str = "x-error-message";
/// Reply header carrying the handler's numeric HTTP status (error replies only).
pub const ERROR_STATUS_HEADER: &str = "x-error-status";

pub(crate) const CONTENT_TYPE_HEADER: &str = "content-type";
const JSON_CONTENT_TYPE: &str = "application/json";
const FLATBUFFER_CONTENT_TYPE: &str = "application/x-flatbuffers";

/// Payload encoding for a request and the reply it produces.
///
/// Selected from the request's `content-type` header. A request with no
/// `content-type`, or one naming neither JSON nor FlatBuffers, is read as
/// [`FlatBuffer`](Self::FlatBuffer) — the framework's native wire format.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum NatsEncoding {
    /// `application/json`.
    Json,
    /// `application/x-flatbuffers`.
    FlatBuffer,
}

impl NatsEncoding {
    /// Select the encoding named by a request's `content-type` header.
    #[must_use]
    pub fn from_content_type(content_type: Option<&str>) -> Self {
        match content_type {
            Some(ct) if ct.contains(JSON_CONTENT_TYPE) => Self::Json,
            _ => Self::FlatBuffer,
        }
    }

    /// The `content-type` a reply in this encoding announces.
    #[must_use]
    pub fn content_type(self) -> &'static str {
        match self {
            Self::Json => JSON_CONTENT_TYPE,
            Self::FlatBuffer => FLATBUFFER_CONTENT_TYPE,
        }
    }
}

/// Everything a `#[nats_route]` handler wrapper needs from the received
/// message besides its payload.
#[derive(Debug)]
pub struct NatsRequestParts {
    /// The concrete subject the message arrived on (wildcards resolved).
    pub subject: String,
    /// Message headers, with names lowercased to HTTP conventions.
    pub headers: HeaderMap,
    /// Values captured by the subject pattern's `{token}` segments.
    pub params: HashMap<String, String>,
    /// The requester's `x-request-id`, or a generated one.
    pub request_id: String,
    /// Encoding selected from the request's `content-type`.
    pub encoding: NatsEncoding,
}

/// A reply published back to the requester's reply subject.
#[derive(Debug)]
pub struct NatsReply {
    /// Encoded reply body.
    pub payload: Vec<u8>,
    /// Reply headers, published as NATS headers.
    pub headers: HeaderMap,
}

/// Handler wrapper generated by `#[nats_route]`.
///
/// Returns a [`NatsReply`] unconditionally: request-decode failures and
/// handler errors are already encoded as error replies by the time the
/// future resolves.
pub type NatsHandlerFn =
    fn(
        NatsRequestParts,
        Vec<u8>,
        Arc<dyn Any + Send + Sync>,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = NatsReply> + Send>>;

/// A subject responder registered by `#[nats_route]`.
#[derive(Clone, Copy)]
pub struct NatsRouteInfo {
    /// The declared subject pattern, `{token}` segments included.
    pub subject: &'static str,
    /// The subject actually subscribed to, with `{token}` replaced by `*`.
    pub wire_subject: &'static str,
    /// Each `{token}` param's name paired with its subject-token index.
    pub params: &'static [(&'static str, usize)],
    /// Queue group to subscribe under; `None` subscribes without one, so
    /// every replica answers every request.
    pub queue: Option<&'static str>,
    /// Request body type name.
    pub request_type: &'static str,
    /// Response body type name.
    pub response_type: &'static str,
    /// The macro-generated dispatch wrapper.
    pub handler: NatsHandlerFn,
}

impl std::fmt::Debug for NatsRouteInfo {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("NatsRouteInfo")
            .field("subject", &self.subject)
            .field("wire_subject", &self.wire_subject)
            .field("params", &self.params)
            .field("queue", &self.queue)
            .field("request_type", &self.request_type)
            .field("response_type", &self.response_type)
            .field("handler", &"<async_function>")
            .finish()
    }
}

inventory::collect!(NatsRouteInfo);

/// Returns every subject responder registered via `#[nats_route]`.
#[must_use]
pub fn get_nats_routes() -> Vec<&'static NatsRouteInfo> {
    inventory::iter::<NatsRouteInfo>.into_iter().collect()
}

/// Two responders whose subjects can both match one message.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NatsRouteConflict {
    /// The first responder's declared pattern.
    pub first_subject: String,
    /// The second responder's declared pattern.
    pub second_subject: String,
}

impl std::fmt::Display for NatsRouteConflict {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "subjects '{}' and '{}' overlap: a message matching both is answered twice",
            self.first_subject, self.second_subject
        )
    }
}

impl std::error::Error for NatsRouteConflict {}

/// Rejects two responders whose subjects can both match one message.
///
/// A message matching two subscriptions is dispatched twice and answered
/// twice on the same reply subject, so the requester takes whichever reply
/// arrives first and the other lands on a closed inbox. Patterns differing
/// only in `{token}` names overlap, and so does a `{token}` segment against
/// a literal one in the same position.
///
/// # Errors
///
/// Returns [`NatsRouteConflict`] naming the first overlapping pair found.
pub fn validate_nats_routes() -> Result<(), NatsRouteConflict> {
    match find_conflict(get_nats_routes()) {
        Some(conflict) => Err(conflict),
        None => Ok(()),
    }
}

/// A NATS `*` matches exactly one token, so two subjects can both match a
/// message only when they have the same token count and every position is
/// either equal or wildcarded on one side.
fn subjects_overlap(first: &str, second: &str) -> bool {
    let first: Vec<&str> = first.split('.').collect();
    let second: Vec<&str> = second.split('.').collect();
    first.len() == second.len()
        && first
            .iter()
            .zip(&second)
            .all(|(a, b)| *a == "*" || *b == "*" || a == b)
}

fn find_conflict<'a>(
    routes: impl IntoIterator<Item = &'a NatsRouteInfo>,
) -> Option<NatsRouteConflict> {
    let routes: Vec<&NatsRouteInfo> = routes.into_iter().collect();
    routes
        .iter()
        .enumerate()
        .flat_map(|(i, first)| routes[i + 1..].iter().map(move |second| (*first, *second)))
        .find(|(first, second)| subjects_overlap(first.wire_subject, second.wire_subject))
        .map(|(first, second)| NatsRouteConflict {
            first_subject: first.subject.to_string(),
            second_subject: second.subject.to_string(),
        })
}

/// Build a success reply carrying `response`'s body in `encoding`.
///
/// A [`Response::raw`] body is published verbatim under its own
/// content type; every other body goes through the encoding's serializer.
/// The handler's own response headers ride along, except that the reply's
/// `content-type` and `x-request-id` always describe what was actually
/// published, and the error headers are dropped — their presence is what a
/// requester reads a reply as a rejection by, so a success must not carry
/// them whatever the handler set.
#[must_use]
pub fn reply_ok<T: ToFlatBuffer>(
    encoding: NatsEncoding,
    request_id: &str,
    mut response: Response<T>,
) -> NatsReply {
    let (payload, content_type) = match response.take_raw() {
        Some((bytes, raw_content_type)) => (bytes, raw_content_type.into_owned()),
        None if encoding == NatsEncoding::Json => match serde_json::to_vec(&response.body) {
            Ok(bytes) => (bytes, encoding.content_type().to_string()),
            Err(e) => {
                return reply_err(
                    encoding,
                    request_id,
                    &FlatbedRouteError::internal(format!("JSON serialization error: {e}"))
                        .code("SERIALIZATION_ERROR"),
                )
            }
        },
        None => (
            response.body.to_flatbuffer(),
            encoding.content_type().to_string(),
        ),
    };

    let mut headers = copy_headers(&response.headers);
    for name in [ERROR_CODE_HEADER, ERROR_MESSAGE_HEADER, ERROR_STATUS_HEADER] {
        headers.remove(name);
    }
    set_header(&mut headers, CONTENT_TYPE_HEADER, &content_type);
    set_header(&mut headers, REQUEST_ID_HEADER, request_id);

    NatsReply { payload, headers }
}

#[derive(serde::Serialize)]
struct ErrorBody<'a, D: serde::Serialize> {
    code: &'a str,
    message: &'a str,
    #[serde(skip_serializing_if = "Option::is_none")]
    details: &'a Option<D>,
}

/// Build an error reply from a handler error.
///
/// The error code, message, and numeric status ride in headers under both
/// encodings so a requester can tell a rejection from a success without
/// decoding the payload. In JSON the payload repeats them as an object; in
/// FlatBuffers the payload is the encoded details, or empty when the error
/// carries none. The error's own headers ride along, but never displace the
/// framework's, so the discriminating headers always describe the error the
/// handler actually returned.
#[must_use]
pub fn reply_err<D: ToFlatBuffer>(
    encoding: NatsEncoding,
    request_id: &str,
    err: &FlatbedRouteError<D>,
) -> NatsReply {
    let code = if err.code.is_empty() {
        "ERROR"
    } else {
        &err.code
    };

    let payload = match encoding {
        NatsEncoding::Json => serde_json::to_vec(&ErrorBody {
            code,
            message: &err.message,
            details: &err.details,
        })
        .unwrap_or_default(),
        NatsEncoding::FlatBuffer => err
            .details
            .as_ref()
            .map(ToFlatBuffer::to_flatbuffer)
            .unwrap_or_default(),
    };

    let mut headers = copy_headers(&err.headers);
    set_header(&mut headers, CONTENT_TYPE_HEADER, encoding.content_type());
    set_header(&mut headers, REQUEST_ID_HEADER, request_id);
    set_header(&mut headers, ERROR_CODE_HEADER, code);
    set_header(&mut headers, ERROR_MESSAGE_HEADER, &err.message);
    set_header(&mut headers, ERROR_STATUS_HEADER, err.status.as_str());

    NatsReply { payload, headers }
}

fn copy_headers(source: &HeaderMap) -> HeaderMap {
    let mut headers = HeaderMap::new();
    for (name, value) in source.iter() {
        headers.append(name.clone(), value.clone());
    }
    headers
}

/// `http` rejects a header value containing a control character, so values
/// are flattened to stay carryable — an error message quoting a multi-line
/// payload is one.
pub(crate) fn set_header(headers: &mut HeaderMap, name: &str, value: &str) {
    let flattened: String = value
        .chars()
        .map(|c| if c.is_control() { ' ' } else { c })
        .collect();
    let (Ok(name), Ok(value)) = (
        HeaderName::try_from(name),
        HeaderValue::try_from(flattened.as_str()),
    ) else {
        return;
    };
    headers.insert(name, value);
}

/// NATS header names are case-preserving byte strings; `http::HeaderName`
/// parsing lowercases them, so a `X-Request-Id` sent by a requester is read
/// back under its lowercase name.
pub(crate) fn from_nats_headers(source: Option<&async_nats::HeaderMap>) -> HeaderMap {
    let mut headers = HeaderMap::new();
    let Some(source) = source else {
        return headers;
    };
    for (name, values) in source.iter() {
        let Ok(name) = HeaderName::try_from(AsRef::<str>::as_ref(name)) else {
            continue;
        };
        for value in values {
            let Ok(value) = HeaderValue::try_from(value.as_str()) else {
                continue;
            };
            headers.append(name.clone(), value);
        }
    }
    headers
}

pub(crate) fn to_nats_headers(source: &HeaderMap) -> async_nats::HeaderMap {
    let mut headers = async_nats::HeaderMap::new();
    for (name, value) in source.iter() {
        let Ok(value) = value.to_str() else {
            continue;
        };
        headers.append(name.as_str(), value);
    }
    headers
}

pub(crate) fn header_value<'a>(headers: &'a HeaderMap, name: &str) -> Option<&'a str> {
    headers.get(name).and_then(|value| value.to_str().ok())
}

fn extract_params(params: &[(&'static str, usize)], subject: &str) -> HashMap<String, String> {
    if params.is_empty() {
        return HashMap::new();
    }
    let tokens: Vec<&str> = subject.split('.').collect();
    params
        .iter()
        .filter_map(|(name, index)| {
            tokens
                .get(*index)
                .map(|token| ((*name).to_string(), (*token).to_string()))
        })
        .collect()
}

async fn dispatch(
    route: NatsRouteInfo,
    message: &async_nats::Message,
    ctx: Arc<dyn Any + Send + Sync>,
) -> NatsReply {
    let headers = from_nats_headers(message.headers.as_ref());
    let encoding = NatsEncoding::from_content_type(header_value(&headers, CONTENT_TYPE_HEADER));
    let request_id = header_value(&headers, REQUEST_ID_HEADER)
        .map(str::to_string)
        .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());

    let parts = NatsRequestParts {
        subject: message.subject.to_string(),
        params: extract_params(route.params, &message.subject),
        headers,
        request_id: request_id.clone(),
        encoding,
    };

    let handled = (route.handler)(parts, message.payload.to_vec(), ctx);
    match std::panic::AssertUnwindSafe(handled).catch_unwind().await {
        Ok(reply) => reply,
        Err(_) => {
            error!(
                subject = %message.subject,
                "nats_route handler panicked; answering with an error reply"
            );
            reply_err(
                encoding,
                &request_id,
                &FlatbedRouteError::internal("handler panicked").code("PANIC"),
            )
        }
    }
}

/// Subscribe to a responder's subject and answer requests until the
/// subscription ends.
///
/// This is the runtime executor `#[nats_route]` registers as a worker: each
/// message is dispatched on its own task, so a slow handler never stalls the
/// subscription, and every message carrying a reply subject is answered.
/// Dispatch tasks are detached, so replies still in flight when the process
/// shuts down are not waited for.
///
/// # Errors
///
/// Returns [`FlatbedWorkerError`] if the service booted with a context other
/// than `C`, if the subscription cannot be established, or if the
/// subscription ends — a responder that stops answering its subject must
/// fail the worker rather than leave the process healthy and silent.
pub async fn run_nats_route<C>(
    ctx: Arc<dyn Any + Send + Sync>,
    route: NatsRouteInfo,
) -> Result<(), FlatbedWorkerError>
where
    C: HasNatsClient + Send + Sync + 'static,
{
    let ctx: Arc<C> = ctx.downcast::<C>().map_err(|_| {
        FlatbedWorkerError::new(
            "nats_route_context",
            format!(
                "[{}] context type mismatch: the service booted with a different context",
                route.subject
            ),
        )
    })?;

    let client = HasNatsClient::nats_client(&*ctx).clone();
    let subscribe = match route.queue {
        Some(queue) => {
            client
                .queue_subscribe(route.wire_subject, queue.to_string())
                .await
        }
        None => client.subscribe(route.wire_subject).await,
    };
    let mut subscription = subscribe.map_err(|e| {
        FlatbedWorkerError::new(
            "nats_route_subscribe",
            format!("[{}] subscribe failed: {}", route.subject, e),
        )
    })?;

    info!(
        subject = route.subject,
        wire_subject = route.wire_subject,
        queue = route.queue.unwrap_or("<none>"),
        "nats route subscribed"
    );

    while let Some(message) = subscription.next().await {
        let Some(reply_subject) = message.reply.clone() else {
            warn!(
                subject = %message.subject,
                "message has no reply subject; nothing to answer"
            );
            continue;
        };

        let ctx: Arc<dyn Any + Send + Sync> = ctx.clone();
        let client = client.clone();
        tokio::spawn(async move {
            let reply = dispatch(route, &message, ctx).await;
            let headers = to_nats_headers(&reply.headers);
            if let Err(e) = client
                .publish_with_headers(reply_subject, headers, reply.payload.into())
                .await
            {
                error!(subject = route.subject, error = %e, "reply publish failed");
            }
        });
    }

    Err(FlatbedWorkerError::new(
        "nats_route_subscription_ended",
        format!(
            "[{}] subscription ended; the subject is no longer answered",
            route.subject
        ),
    ))
}

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

    #[test]
    fn encoding_defaults_to_flatbuffers_for_absent_or_unknown_content_type() {
        assert_eq!(
            NatsEncoding::from_content_type(None),
            NatsEncoding::FlatBuffer
        );
        assert_eq!(
            NatsEncoding::from_content_type(Some("text/plain")),
            NatsEncoding::FlatBuffer
        );
        assert_eq!(
            NatsEncoding::from_content_type(Some("application/x-flatbuffers")),
            NatsEncoding::FlatBuffer
        );
    }

    #[test]
    fn encoding_selects_json_from_a_parameterized_content_type() {
        assert_eq!(
            NatsEncoding::from_content_type(Some("application/json; charset=utf-8")),
            NatsEncoding::Json
        );
        assert_eq!(NatsEncoding::Json.content_type(), "application/json");
        assert_eq!(
            NatsEncoding::FlatBuffer.content_type(),
            "application/x-flatbuffers"
        );
    }

    fn unreachable_handler(
        _: NatsRequestParts,
        _: Vec<u8>,
        _: Arc<dyn Any + Send + Sync>,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = NatsReply> + Send>> {
        unreachable!("the conflict scan never dispatches")
    }

    fn route(subject: &'static str, wire_subject: &'static str) -> NatsRouteInfo {
        NatsRouteInfo {
            subject,
            wire_subject,
            params: &[],
            queue: None,
            request_type: "Req",
            response_type: "Res",
            handler: unreachable_handler,
        }
    }

    #[test]
    fn distinct_wire_subjects_do_not_conflict() {
        let routes = [
            route("a.{id}.status", "a.*.status"),
            route("a.{id}.config", "a.*.config"),
            route("b.report", "b.report"),
        ];
        assert!(find_conflict(&routes).is_none());
    }

    /// Two patterns that differ only in their capture names subscribe to the
    /// same wire subject, so both answer every matching request.
    #[test]
    fn patterns_differing_only_in_capture_names_conflict() {
        let routes = [
            route("a.report", "a.report"),
            route("plonk.sat.{id}.status", "plonk.sat.*.status"),
            route("plonk.sat.{name}.status", "plonk.sat.*.status"),
        ];
        let conflict = find_conflict(&routes).expect("the colliding pair must be reported");

        assert_eq!(conflict.first_subject, "plonk.sat.{id}.status");
        assert_eq!(conflict.second_subject, "plonk.sat.{name}.status");
        assert_eq!(
            conflict.to_string(),
            "subjects 'plonk.sat.{id}.status' and 'plonk.sat.{name}.status' \
             overlap: a message matching both is answered twice",
        );
    }

    /// A wildcard and a literal in the same position both match one concrete
    /// subject, so the pair conflicts even though neither wire subject equals
    /// the other.
    #[test]
    fn a_wildcard_conflicts_with_a_literal_in_the_same_position() {
        let routes = [
            route("plonk.sat.{id}.status", "plonk.sat.*.status"),
            route("plonk.sat.alpha.status", "plonk.sat.alpha.status"),
        ];
        let conflict = find_conflict(&routes).expect("the overlapping pair must be reported");

        assert_eq!(conflict.first_subject, "plonk.sat.{id}.status");
        assert_eq!(conflict.second_subject, "plonk.sat.alpha.status");
    }

    /// A NATS `*` matches exactly one token, so subjects of different lengths
    /// can never both match the same message.
    #[test]
    fn a_wildcard_does_not_conflict_across_different_token_counts() {
        let routes = [
            route("plonk.{id}", "plonk.*"),
            route("plonk.{id}.status", "plonk.*.status"),
        ];
        assert!(find_conflict(&routes).is_none());
    }

    #[test]
    fn params_are_read_off_the_concrete_subject_by_token_index() {
        let params = extract_params(&[("id", 2), ("verb", 4)], "plonk.satellite.x07.call.status");
        assert_eq!(params.get("id").map(String::as_str), Some("x07"));
        assert_eq!(params.get("verb").map(String::as_str), Some("status"));
    }

    #[test]
    fn params_are_empty_for_a_pattern_without_tokens() {
        assert!(extract_params(&[], "plonk.ground.report").is_empty());
    }

    #[test]
    fn error_reply_carries_code_message_and_status_headers() {
        let err = FlatbedRouteError::not_found("no such satellite").code("NOT_FOUND");
        let reply = reply_err(NatsEncoding::FlatBuffer, "req-1", &err);

        assert_eq!(
            header_value(&reply.headers, ERROR_CODE_HEADER),
            Some("NOT_FOUND")
        );
        assert_eq!(
            header_value(&reply.headers, ERROR_MESSAGE_HEADER),
            Some("no such satellite")
        );
        assert_eq!(
            header_value(&reply.headers, ERROR_STATUS_HEADER),
            Some("404")
        );
        assert_eq!(
            header_value(&reply.headers, REQUEST_ID_HEADER),
            Some("req-1")
        );
        assert!(
            reply.payload.is_empty(),
            "a FlatBuffers error without details has an empty payload"
        );
    }

    #[test]
    fn error_reply_without_a_code_falls_back_to_a_generic_one() {
        let err = FlatbedRouteError::with_status(StatusCode::IM_A_TEAPOT, "brew failed");
        let reply = reply_err(NatsEncoding::Json, "req-2", &err);

        assert_eq!(
            header_value(&reply.headers, ERROR_CODE_HEADER),
            Some("ERROR")
        );
        assert_eq!(
            header_value(&reply.headers, ERROR_STATUS_HEADER),
            Some("418")
        );
        let body: serde_json::Value = serde_json::from_slice(&reply.payload).unwrap();
        assert_eq!(body["code"], "ERROR");
        assert_eq!(body["message"], "brew failed");
        assert!(
            body.get("details").is_none(),
            "an error without details omits the field"
        );
    }

    /// The error headers are what a requester reads to tell a rejection from a
    /// success, so a handler that sets them on its own error must not be able
    /// to misreport the error it actually returned.
    #[test]
    fn handler_headers_never_displace_the_frameworks_own() {
        let err = FlatbedRouteError::not_found("gone")
            .code("NOT_FOUND")
            .header("x-error-code", "OK")
            .header("content-type", "text/plain")
            .header("x-trace", "t-9");
        let reply = reply_err(NatsEncoding::Json, "req-4", &err);

        assert_eq!(
            header_value(&reply.headers, ERROR_CODE_HEADER),
            Some("NOT_FOUND")
        );
        assert_eq!(
            header_value(&reply.headers, CONTENT_TYPE_HEADER),
            Some("application/json")
        );
        assert_eq!(header_value(&reply.headers, "x-trace"), Some("t-9"));
    }

    /// A header value cannot carry a control character, so a multi-line error
    /// message reaches the requester flattened rather than not at all.
    #[test]
    fn a_multi_line_error_message_is_flattened_onto_the_header() {
        let err = FlatbedRouteError::bad_request("line one\nline two\r\tend").code("BAD");
        let reply = reply_err(NatsEncoding::FlatBuffer, "req-5", &err);

        assert_eq!(
            header_value(&reply.headers, ERROR_MESSAGE_HEADER),
            Some("line one line two  end")
        );
    }

    /// A requester reads a reply as a rejection by the presence of
    /// `x-error-code`, so a handler that puts one on a success must not be
    /// able to make its own answer look like a failure.
    #[test]
    fn a_success_reply_never_carries_the_error_headers() {
        let reply = reply_ok(
            NatsEncoding::Json,
            "req-6",
            Response::ok(()).header("x-error-code", "NOT_FOUND"),
        );

        assert_eq!(header_value(&reply.headers, ERROR_CODE_HEADER), None);
    }

    #[test]
    fn raw_response_bytes_are_published_verbatim_under_their_own_content_type() {
        let reply = reply_ok(
            NatsEncoding::Json,
            "req-3",
            Response::raw(b"id,count\n1,2\n".to_vec(), "text/csv"),
        );

        assert_eq!(reply.payload, b"id,count\n1,2\n");
        assert_eq!(
            header_value(&reply.headers, CONTENT_TYPE_HEADER),
            Some("text/csv")
        );
    }

    #[test]
    fn nats_headers_round_trip_through_the_http_header_map() {
        let mut source = async_nats::HeaderMap::new();
        source.append("Content-Type", "application/json");
        source.append("X-Request-Id", "abc");

        let converted = from_nats_headers(Some(&source));
        assert_eq!(
            header_value(&converted, CONTENT_TYPE_HEADER),
            Some("application/json"),
            "NATS header names are matched case-insensitively after conversion"
        );
        assert_eq!(header_value(&converted, REQUEST_ID_HEADER), Some("abc"));

        let back = to_nats_headers(&converted);
        assert_eq!(
            back.get("content-type")
                .map(async_nats::HeaderValue::as_str),
            Some("application/json")
        );
    }
}