liminal-rs 0.16.3

//! A worker's closed refusal vocabulary and its canonical wire record.

use core::fmt;
use serde_json::{Value, json};

use super::{Frame, ProtocolError};

/// A refused reply carries a canonical record instead of application bytes.
pub const PUSH_REPLY_REFUSED_FLAG: u8 = 0x01;

/// Wire-configured UTF-8 byte ceiling, keeping refusal allocation bounded while
/// allowing a useful explanation. Peers must share this protocol value.
pub const WORKER_REFUSAL_WORDS_MAX: usize = 1024;

/// Worst canonical record: every words byte requires a six-byte JSON escape,
/// plus the fixed keys, delimiters and longest code. Checked before parsing so
/// a worker cannot force an unbounded JSON allocation.
const RECORD_BYTES_MAX: usize =
    6 * WORKER_REFUSAL_WORDS_MAX + "{\"code\":\"unavailable\",\"words\":\"\"}".len();

/// The worker's reason, distinct from a route's admission refusal.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum WorkerRefusalCode {
    /// The worker cannot admit more work now.
    Busy,
    /// The request cannot be served as supplied.
    Invalid,
    /// A dependency or the worker is unavailable.
    Unavailable,
    /// The worker encountered an internal failure.
    Internal,
}

impl WorkerRefusalCode {
    /// The complete vocabulary, in wire-code order.
    pub const ALL: [Self; 4] = [Self::Busy, Self::Invalid, Self::Unavailable, Self::Internal];

    /// The name carried in the canonical worker record.
    #[must_use]
    pub const fn name(self) -> &'static str {
        match self {
            Self::Busy => "busy",
            Self::Invalid => "invalid",
            Self::Unavailable => "unavailable",
            Self::Internal => "internal",
        }
    }

    /// The caller's stable named refusal.
    #[must_use]
    pub const fn caller_name(self) -> &'static str {
        match self {
            Self::Busy => "worker_busy",
            Self::Invalid => "worker_invalid",
            Self::Unavailable => "worker_unavailable",
            Self::Internal => "worker_internal",
        }
    }

    /// Numeric representation on the existing conversation-error wire. The
    /// words remain entirely application supplied; classification uses this
    /// registry, never a prefix guessed from those words.
    #[must_use]
    pub const fn reason_code(self) -> u16 {
        match self {
            Self::Busy => 0x0401,
            Self::Invalid => 0x0402,
            Self::Unavailable => 0x0403,
            Self::Internal => 0x0404,
        }
    }

    /// Recognises only the four allocated conversation-error reasons.
    #[must_use]
    pub fn from_reason_code(reason: u16) -> Option<Self> {
        Self::ALL
            .into_iter()
            .find(|code| code.reason_code() == reason)
    }

    fn parse(name: &str) -> Result<Self, ProtocolError> {
        Self::ALL
            .into_iter()
            .find(|code| code.name() == name)
            .ok_or_else(|| ProtocolError::codec("worker_refusal_unknown_code"))
    }
}

/// A validated refusal. Words are data and are never included in `Debug`.
#[derive(Clone, PartialEq, Eq)]
pub struct WorkerRefusal {
    code: WorkerRefusalCode,
    words: String,
}

impl WorkerRefusal {
    /// Validates the closed code and the byte bound before retaining words.
    ///
    /// # Errors
    /// An unknown code or words exceeding the configured wire ceiling.
    pub fn new(code: &str, words: impl AsRef<str>) -> Result<Self, ProtocolError> {
        let code = WorkerRefusalCode::parse(code)?;
        let words = words.as_ref();
        if words.len() > WORKER_REFUSAL_WORDS_MAX {
            return Err(ProtocolError::codec(format!(
                "worker_refusal_words_too_long: words exceed {WORKER_REFUSAL_WORDS_MAX} UTF-8 bytes"
            )));
        }
        Ok(Self {
            code,
            words: words.to_owned(),
        })
    }

    /// The validated worker reason.
    #[must_use]
    pub const fn code(&self) -> WorkerRefusalCode {
        self.code
    }

    /// The worker's words, unchanged.
    #[must_use]
    pub fn words(&self) -> &str {
        &self.words
    }

    /// Transfers the words to the caller without another allocation.
    #[must_use]
    pub fn into_words(self) -> String {
        self.words
    }

    /// Compact JSON through a sorted value, the console's canonical form.
    ///
    /// # Errors
    /// The canonical JSON encoder failed; its input is never quoted.
    pub fn canonical_bytes(&self) -> Result<Vec<u8>, ProtocolError> {
        serde_json::to_vec(&json!({"code": self.code.name(), "words": self.words}))
            .map_err(|_| ProtocolError::codec("worker_refusal_encode_failed"))
    }

    /// Accepts only the exact canonical record, including its member order and
    /// spelling. Comparing the re-encoding also rejects duplicate members and
    /// alternate escapes without ever copying their values into an error.
    ///
    /// # Errors
    /// An oversized, malformed, noncanonical or out-of-vocabulary record.
    pub fn decode(payload: &[u8]) -> Result<Self, ProtocolError> {
        if payload.len() > RECORD_BYTES_MAX {
            return Err(ProtocolError::codec("worker_refusal_record_too_large"));
        }
        let value: Value = serde_json::from_slice(payload)
            .map_err(|_| ProtocolError::codec("worker_refusal_malformed_record"))?;
        let object = value
            .as_object()
            .filter(|object| object.len() == 2)
            .ok_or_else(|| ProtocolError::codec("worker_refusal_invalid_fields"))?;
        let code = object
            .get("code")
            .and_then(Value::as_str)
            .ok_or_else(|| ProtocolError::codec("worker_refusal_invalid_code"))?;
        let words = object
            .get("words")
            .and_then(Value::as_str)
            .ok_or_else(|| ProtocolError::codec("worker_refusal_invalid_words"))?;
        let refusal = Self::new(code, words)?;
        if refusal.canonical_bytes()? != payload {
            return Err(ProtocolError::codec("worker_refusal_noncanonical_record"));
        }
        Ok(refusal)
    }
}

impl fmt::Debug for WorkerRefusal {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("WorkerRefusal")
            .field("code", &self.code)
            .field("words_bytes", &self.words.len())
            .finish()
    }
}

// Debug is kept beside the refusal so its opaque bytes share the redaction boundary.
mod frame_debug {
    use super::Frame;
    use core::fmt;

    /// An opaque field: its length, never its bytes.
    struct ByteLength(usize);

    impl fmt::Debug for ByteLength {
        fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
            write!(formatter, "{} bytes", self.0)
        }
    }

    macro_rules! frame_debug_field {
        ($out:ident, envelope, $value:expr) => {
            $out.field("envelope_payload", &ByteLength($value.payload.len()));
        };
        ($out:ident, payload, $value:expr) => {
            $out.field("payload", &ByteLength($value.len()));
        };
        ($out:ident, sealed, $value:expr) => {
            $out.field("sealed", &ByteLength($value.len()));
        };
        ($out:ident, $field:ident, $value:expr) => {
            $out.field(stringify!($field), $value);
        };
    }

    macro_rules! frame_debug {
    ($($variant:ident { $($field:ident),* }),* $(,)?) => {
        impl fmt::Debug for Frame {
            fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
                match self {
                    Self::Connect {
                        flags,
                        min_version,
                        max_version,
                        auth_token,
                    } => formatter
                        .debug_struct("Connect")
                        .field("flags", flags)
                        .field("min_version", min_version)
                        .field("max_version", max_version)
                        .field("auth_token", &ByteLength(auth_token.len()))
                        .finish(),
                    Self::Push { flags, stream_id, correlation_id, payload }
                    | Self::PushReply { flags, stream_id, correlation_id, payload } => formatter
                        .debug_struct(if matches!(self, Self::Push { .. }) { "Push" } else { "PushReply" })
                        .field("flags", flags).field("stream_id", stream_id)
                        .field("correlation_id", correlation_id)
                        .field("payload", &ByteLength(payload.len())).finish(),
                    Self::ConversationError { flags, stream_id, conversation_id, reason_code, message } => formatter
                        .debug_struct("ConversationError").field("flags", flags)
                        .field("stream_id", stream_id).field("conversation_id", conversation_id)
                        .field("reason_code", reason_code)
                        .field("message_bytes", &message.as_ref().map(String::len)).finish(),
                    $(Self::$variant { $($field),* } => {
                        let mut out = formatter.debug_struct(stringify!($variant));
                        $(frame_debug_field!(out, $field, $field);)*
                        out.finish()
                    })*
                }
            }
        }
    };
}

    frame_debug! {
        ConnectAck { flags, selected_version, capabilities },
        ConnectError { flags, reason_code, message },
        Disconnect { flags },
        DisconnectReason { flags, reason },
        Subscribe { flags, stream_id, channel, accepted_schemas, max_in_flight },
        SubscribeAck { flags, stream_id, subscription_id, selected_schema },
        SubscribeError { flags, stream_id, reason_code, message },
        Unsubscribe { flags, stream_id, subscription_id },
        Publish { flags, stream_id, channel, envelope, idempotency_key },
        PublishAck { flags, stream_id, message_id, durable_position },
        PublishError { flags, stream_id, reason_code, message },
        ConversationOpen { flags, stream_id, conversation_id, subject },
        ConversationMessage { flags, stream_id, conversation_id, envelope },
        ConversationClose { flags, stream_id, conversation_id, reason_code, message },
        Accept { flags, stream_id, referenced_message_id },
        Defer { flags, stream_id, referenced_message_id, reason },
        Reject { flags, stream_id, referenced_message_id, reason },
        Ping { flags },
        Pong { flags },
        WorkerRegister { flags, registration },
        WorkerRegisterAck { flags, outcome },
        Deliver { flags, stream_id, delivery_seq, envelope, message },
        LinkOpen { flags, stream_id, offer },
        LinkOpenAck { flags, stream_id, outcome },
        LinkMessage { flags, stream_id, subject, sealed },
        Unknown { type_id, flags, stream_id, payload },
    }
}