saddle-boundary 0.3.24

//! Fixed `profusegw` HTTP ingress codec and listener adapter.
//!
//! The adapter validates transport-owned identity and framing only. It does
//! not resolve `interfaceId`, dispatch business handlers, or own Runtime.

use serde::{Deserialize, Serialize};
use serde_json::Value;

pub const METHOD: &str = "POST";
pub const PATH: &str = "/saddle/v1/ingress/profusegw/invoke";
pub const MEDIA_TYPE: &str = "application/json";
pub const MAX_BODY_BYTES: usize = 1024 * 1024;
pub const MAX_ID_BYTES: usize = 256;

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProfuseGwTarget {
    pub app: String,
    #[serde(rename = "interfaceId")]
    pub interface_id: String,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProfuseGwUserInfo {
    #[serde(rename = "userId")]
    pub user_id: String,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct ProfuseGwTraceInfo {
    #[serde(default)]
    pub trace_id: String,
    pub rpc_id: String,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProfuseGwLdcInfo {
    pub zone: String,
    pub idc: String,
    pub env: String,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProfuseGwContext {
    #[serde(rename = "userInfo")]
    pub user_info: ProfuseGwUserInfo,
    #[serde(rename = "traceInfo")]
    pub trace_info: ProfuseGwTraceInfo,
    #[serde(rename = "ldcInfo")]
    pub ldc_info: ProfuseGwLdcInfo,
}

/// The exact JSON body forwarded by `profusegw`.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct IngressEnvelope {
    pub target: ProfuseGwTarget,
    #[serde(rename = "profuseGwContext")]
    pub profuse_gw_context: ProfuseGwContext,
    #[serde(rename = "requestData")]
    pub request_data: Value,
}

/// Transport identity supplied by the listener integration, never by
/// business JSON. All three values remain attached to the accepted request.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct IngressIdentity {
    pub request_id: String,
    pub call_id: String,
    pub deadline_unix_ms: i64,
}

impl IngressIdentity {
    pub fn new(
        request_id: impl Into<String>,
        call_id: impl Into<String>,
        deadline_unix_ms: i64,
    ) -> Result<Self, CodecError> {
        let identity = Self {
            request_id: request_id.into(),
            call_id: call_id.into(),
            deadline_unix_ms,
        };
        validate_identity(&identity)?;
        Ok(identity)
    }
}

/// A listener-owned, validated input. `interface_id` deliberately remains
/// unresolved so Transport cannot become business dispatch authority.
#[derive(Clone, Debug, PartialEq)]
pub struct AcceptedIngress {
    pub identity: IngressIdentity,
    pub interface_id: String,
    pub user_id: String,
    pub trace_id: String,
    pub rpc_id: String,
    pub zone: String,
    pub idc: String,
    pub env: String,
    pub request_data: Value,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct CodecError {
    pub http_status: u16,
    pub code: &'static str,
}

/// Fixed deployment application identity owned by the listener adapter.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ProfuseGwListenerAdapter {
    application: String,
}

impl ProfuseGwListenerAdapter {
    pub fn new(application: impl Into<String>) -> Result<Self, CodecError> {
        let application = application.into();
        validate_id(&application)?;
        Ok(Self { application })
    }

    pub fn application(&self) -> &str {
        &self.application
    }

    pub fn accept(
        &self,
        method: &str,
        path: &str,
        content_type: &str,
        identity: IngressIdentity,
        body: &[u8],
    ) -> Result<AcceptedIngress, CodecError> {
        self.accept_observed(method, path, content_type, identity, body, |_| {})
    }

    /// Borrow the original parser error before the stable protocol mapping.
    /// The observer neither replaces the response nor retains input authority.
    pub fn accept_observed(
        &self,
        method: &str,
        path: &str,
        content_type: &str,
        identity: IngressIdentity,
        body: &[u8],
        parser_error: impl FnOnce(&serde_json::Error),
    ) -> Result<AcceptedIngress, CodecError> {
        validate_identity(&identity)?;
        let envelope = decode_observed(method, path, content_type, body, parser_error)?;
        if envelope.target.app != self.application {
            return Err(error(404, "APPLICATION_NOT_FOUND"));
        }
        let trace_id = if envelope.profuse_gw_context.trace_info.trace_id.is_empty() {
            format!("saddle-{}", identity.request_id)
        } else {
            envelope.profuse_gw_context.trace_info.trace_id
        };
        Ok(AcceptedIngress {
            identity,
            interface_id: envelope.target.interface_id,
            user_id: envelope.profuse_gw_context.user_info.user_id,
            trace_id,
            rpc_id: envelope.profuse_gw_context.trace_info.rpc_id,
            zone: envelope.profuse_gw_context.ldc_info.zone,
            idc: envelope.profuse_gw_context.ldc_info.idc,
            env: envelope.profuse_gw_context.ldc_info.env,
            request_data: envelope.request_data,
        })
    }
}

/// Decodes the sole HTTP request shape. Any error occurs before a valid
/// protocol request exists and therefore uses a non-200 status.
pub fn decode(
    method: &str,
    path: &str,
    content_type: &str,
    body: &[u8],
) -> Result<IngressEnvelope, CodecError> {
    decode_observed(method, path, content_type, body, |_| {})
}

fn decode_observed(
    method: &str,
    path: &str,
    content_type: &str,
    body: &[u8],
    parser_error: impl FnOnce(&serde_json::Error),
) -> Result<IngressEnvelope, CodecError> {
    if method != METHOD {
        return Err(error(405, "METHOD_NOT_ALLOWED"));
    }
    if path != PATH {
        return Err(error(404, "PATH_NOT_FOUND"));
    }
    if content_type
        .split(';')
        .next()
        .map(str::trim)
        .filter(|value| value.eq_ignore_ascii_case(MEDIA_TYPE))
        .is_none()
    {
        return Err(error(415, "MEDIA_TYPE_NOT_SUPPORTED"));
    }
    if body.len() > MAX_BODY_BYTES {
        return Err(error(413, "BODY_TOO_LARGE"));
    }
    let envelope: IngressEnvelope = serde_json::from_slice(body).map_err(|original| {
        parser_error(&original);
        error(400, "INVALID_JSON_ENVELOPE")
    })?;
    validate(&envelope)?;
    Ok(envelope)
}

fn validate(envelope: &IngressEnvelope) -> Result<(), CodecError> {
    validate_id(&envelope.target.app)?;
    validate_id(&envelope.target.interface_id)?;
    validate_id(&envelope.profuse_gw_context.user_info.user_id)?;
    if !envelope.profuse_gw_context.trace_info.trace_id.is_empty() {
        validate_trace_id(&envelope.profuse_gw_context.trace_info.trace_id)?;
    }
    validate_id(&envelope.profuse_gw_context.trace_info.rpc_id)?;
    validate_id(&envelope.profuse_gw_context.ldc_info.zone)?;
    validate_id(&envelope.profuse_gw_context.ldc_info.idc)?;
    validate_id(&envelope.profuse_gw_context.ldc_info.env)?;
    if !envelope.request_data.is_object() {
        return Err(error(400, "INVALID_REQUEST_DATA"));
    }
    Ok(())
}

fn validate_identity(identity: &IngressIdentity) -> Result<(), CodecError> {
    validate_id(&identity.request_id)?;
    validate_id(&identity.call_id)?;
    if identity.deadline_unix_ms <= 0 {
        return Err(error(400, "INVALID_DEADLINE"));
    }
    Ok(())
}

fn validate_id(value: &str) -> Result<(), CodecError> {
    if value.trim().is_empty() || value.len() > MAX_ID_BYTES {
        return Err(error(400, "INVALID_IDENTITY"));
    }
    Ok(())
}

fn validate_trace_id(value: &str) -> Result<(), CodecError> {
    if value.is_empty() || value.len() > MAX_ID_BYTES || value.chars().any(char::is_control) {
        return Err(error(400, "INVALID_TRACE_ID"));
    }
    Ok(())
}

const fn error(http_status: u16, code: &'static str) -> CodecError {
    CodecError { http_status, code }
}

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

    const VALID: &[u8] = br#"{
      "target":{"app":"bill-service","interfaceId":"bill.query"},
      "profuseGwContext":{"userInfo":{"userId":"2088\u4e2d\u6587\u7528\u6237"},"traceInfo":{"traceId":"trace-1","rpcId":"0"},"ldcInfo":{"zone":"z1","idc":"i1","env":"test"}},
      "requestData":{"account":"A-1"}
    }"#;

    fn identity() -> IngressIdentity {
        IngressIdentity::new("request-1", "call-1", 1_800_000_000_000).unwrap()
    }

    #[test]
    fn listener_accepts_fixed_app_and_preserves_identity() {
        let accepted = ProfuseGwListenerAdapter::new("bill-service")
            .unwrap()
            .accept(METHOD, PATH, MEDIA_TYPE, identity(), VALID)
            .unwrap();
        assert_eq!(accepted.identity.request_id, "request-1");
        assert_eq!(accepted.identity.call_id, "call-1");
        assert_eq!(accepted.identity.deadline_unix_ms, 1_800_000_000_000);
        assert_eq!(accepted.interface_id, "bill.query");
        assert_eq!(accepted.user_id, "2088中文用户");
        assert_eq!(accepted.trace_id, "trace-1");
        assert_eq!(accepted.rpc_id, "0");
        assert_eq!(accepted.zone, "z1");
        assert_eq!(accepted.request_data["account"], "A-1");
    }

    #[test]
    fn listener_generates_trace_only_when_ingress_omits_it() {
        let body = String::from_utf8(VALID.to_vec())
            .unwrap()
            .replace("trace-1", "");
        let accepted = ProfuseGwListenerAdapter::new("bill-service")
            .unwrap()
            .accept(METHOD, PATH, MEDIA_TYPE, identity(), body.as_bytes())
            .unwrap();
        assert_eq!(accepted.trace_id, "saddle-request-1");
        assert_eq!(accepted.rpc_id, "0");
    }

    #[test]
    fn trace_id_is_opaque_bounded_and_control_free() {
        let opaque = String::from_utf8(VALID.to_vec())
            .unwrap()
            .replace("trace-1", "opaque/非HEX:值");
        let accepted = ProfuseGwListenerAdapter::new("bill-service")
            .unwrap()
            .accept(METHOD, PATH, MEDIA_TYPE, identity(), opaque.as_bytes())
            .unwrap();
        assert_eq!(accepted.trace_id, "opaque/非HEX:值");

        let control = String::from_utf8(VALID.to_vec())
            .unwrap()
            .replace("trace-1", "bad\\ttrace");
        assert_eq!(
            ProfuseGwListenerAdapter::new("bill-service")
                .unwrap()
                .accept(METHOD, PATH, MEDIA_TYPE, identity(), control.as_bytes())
                .unwrap_err()
                .code,
            "INVALID_TRACE_ID"
        );
        let oversized = String::from_utf8(VALID.to_vec())
            .unwrap()
            .replace("trace-1", &"x".repeat(MAX_ID_BYTES + 1));
        assert_eq!(
            ProfuseGwListenerAdapter::new("bill-service")
                .unwrap()
                .accept(METHOD, PATH, MEDIA_TYPE, identity(), oversized.as_bytes())
                .unwrap_err()
                .code,
            "INVALID_TRACE_ID"
        );
    }

    #[test]
    fn codec_rejects_old_shape_and_foreign_app() {
        let old = br#"{"protocol":"saddle-profusegw/1","request_id":"r"}"#;
        let foreign_app = String::from_utf8(VALID.to_vec())
            .unwrap()
            .replace("bill-service", "other-app");
        assert_eq!(
            decode(METHOD, PATH, MEDIA_TYPE, old)
                .unwrap_err()
                .http_status,
            400
        );
        assert_eq!(
            ProfuseGwListenerAdapter::new("bill-service")
                .unwrap()
                .accept(METHOD, PATH, MEDIA_TYPE, identity(), foreign_app.as_bytes())
                .unwrap_err()
                .code,
            "APPLICATION_NOT_FOUND"
        );
    }

    #[test]
    fn framing_and_identity_fail_before_acceptance() {
        for (method, path, media_type, expected) in [
            ("GET", PATH, MEDIA_TYPE, 405),
            (METHOD, "/business/route", MEDIA_TYPE, 404),
            (METHOD, PATH, "text/plain", 415),
        ] {
            assert_eq!(
                decode(method, path, media_type, VALID)
                    .unwrap_err()
                    .http_status,
                expected
            );
        }
        assert_eq!(
            IngressIdentity::new("", "call-1", 1).unwrap_err().code,
            "INVALID_IDENTITY"
        );
    }

    #[test]
    fn context_overlay_and_oversized_body_are_rejected() {
        let overlay = br#"{
          "target":{"app":"bill-service","interfaceId":"bill.query"},
          "profuseGwContext":{"userInfo":{"userId":"2088"},"headers":{}},
          "requestData":{}
        }"#;
        assert_eq!(
            decode(METHOD, PATH, MEDIA_TYPE, overlay)
                .unwrap_err()
                .http_status,
            400
        );
        assert_eq!(
            decode(METHOD, PATH, MEDIA_TYPE, &vec![b' '; MAX_BODY_BYTES + 1])
                .unwrap_err()
                .http_status,
            413
        );
    }
}