shove 0.12.1

//! Simple Binary Encoding (SBE) codec — zero-copy.
//!
//! Enable via the `sbe` cargo feature (no extra dependencies). Works with the
//! flyweight encoders/decoders generated by
//! [`sbe-tool`](https://github.com/aeron-io/simple-binary-encoding)'s Rust
//! backend, but is generator-agnostic: anything that produces a standard SBE
//! frame (8-byte message header + root block + groups/var-data) rides through
//! unchanged.
//!
//! SBE has no owned "decoded struct" — generated codecs read and write fields
//! directly on a byte buffer. The topic message type is therefore
//! [`SbeFrame<T>`]: an owned, header-validated wire frame. Publishing passes
//! the frame's buffer through without re-encoding, and consuming hands the
//! received buffer to the handler without copying (on backends that own the
//! delivery buffer), so the handler wraps its generated decoder directly over
//! [`SbeFrame::as_bytes`].
//!
//! ```ignore
//! // `Order` is a type tag binding the topic to one SBE message type.
//! struct Order;
//! impl SbeMessage for Order {
//!     const SCHEMA_ID: u16 = order_codec::SBE_SCHEMA_ID;
//!     const TEMPLATE_ID: u16 = order_codec::SBE_TEMPLATE_ID;
//! }
//!
//! define_topic!(OrdersTopic, SbeFrame<Order>, orders_topology, codec = SbeCodec);
//!
//! // Publish: encode once with the generated flyweight (header included),
//! // then wrap. `SbeFrame::new` validates the header and takes ownership.
//! let frame = SbeFrame::<Order>::new(encoded)?;
//! broker.publisher().publish::<OrdersTopic>(&frame).await?;
//!
//! // Consume: wrap the generated decoder over the received frame.
//! let decoder = OrderDecoder::default().header(
//!     MessageHeaderDecoder::default().wrap(ReadBuf::new(frame.as_bytes()), 0),
//!     0,
//! );
//! ```
//!
//! Wire versioning follows SBE extension semantics: a frame whose header
//! carries a different `version` or `blockLength` than the consumer was built
//! against is accepted, and the acting values are exposed via
//! [`SbeFrame::header`] for the generated decoder to interpret. Only
//! `schemaId` and `templateId` are validated, so a frame from the wrong topic
//! or message type fails decode instead of being misread.

use std::marker::PhantomData;

use bytes::Bytes;

use crate::codec::Codec;
use crate::error::{Result, ShoveError};

/// Byte order of the SBE schema (`byteOrder` attribute in the schema XML).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SbeByteOrder {
    /// `byteOrder="littleEndian"` — the SBE default.
    LittleEndian,
    /// `byteOrder="bigEndian"`.
    BigEndian,
}

/// Identifies one SBE message type on a topic.
///
/// Implement on a zero-sized type tag next to the generated codec module and
/// copy the constants the generator emits (`SBE_SCHEMA_ID`,
/// `SBE_TEMPLATE_ID`). [`SbeFrame`] uses them to validate incoming frames.
pub trait SbeMessage: 'static {
    /// The schema's `id` attribute. Must match `SBE_SCHEMA_ID` in the
    /// generated code.
    const SCHEMA_ID: u16;

    /// The message's `template id`. Must match `SBE_TEMPLATE_ID` in the
    /// generated code.
    const TEMPLATE_ID: u16;

    /// The schema's `byteOrder`. Defaults to little-endian, the SBE default.
    const BYTE_ORDER: SbeByteOrder = SbeByteOrder::LittleEndian;
}

/// The standard 8-byte SBE message header composite.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SbeHeader {
    /// Length of the message's root block, as written by the producer.
    pub block_length: u16,
    /// Template id of the message.
    pub template_id: u16,
    /// Schema id the message was encoded with.
    pub schema_id: u16,
    /// Schema version the message was encoded with (the acting version).
    pub version: u16,
}

impl SbeHeader {
    /// Encoded size of the standard header on the wire.
    pub const ENCODED_LENGTH: usize = 8;

    /// Parse a header from the start of `bytes`.
    pub fn read(
        bytes: &[u8],
        byte_order: SbeByteOrder,
    ) -> std::result::Result<Self, SbeCodecError> {
        let read_u16 = |offset: usize| -> Option<u16> {
            let raw: [u8; 2] = bytes.get(offset..offset + 2)?.try_into().ok()?;
            Some(match byte_order {
                SbeByteOrder::LittleEndian => u16::from_le_bytes(raw),
                SbeByteOrder::BigEndian => u16::from_be_bytes(raw),
            })
        };
        let too_short = || SbeCodecError::TooShort { len: bytes.len() };
        Ok(Self {
            block_length: read_u16(0).ok_or_else(too_short)?,
            template_id: read_u16(2).ok_or_else(too_short)?,
            schema_id: read_u16(4).ok_or_else(too_short)?,
            version: read_u16(6).ok_or_else(too_short)?,
        })
    }

    /// Encode the header to its 8-byte wire form.
    pub fn to_bytes(self, byte_order: SbeByteOrder) -> [u8; Self::ENCODED_LENGTH] {
        let word = |v: u16| match byte_order {
            SbeByteOrder::LittleEndian => v.to_le_bytes(),
            SbeByteOrder::BigEndian => v.to_be_bytes(),
        };
        let mut out = [0u8; Self::ENCODED_LENGTH];
        let (bl, ti, si, ve) = (
            word(self.block_length),
            word(self.template_id),
            word(self.schema_id),
            word(self.version),
        );
        out[..2].copy_from_slice(&bl);
        out[2..4].copy_from_slice(&ti);
        out[4..6].copy_from_slice(&si);
        out[6..8].copy_from_slice(&ve);
        out
    }
}

/// Validation and framing failures for SBE frames.
#[derive(Debug, thiserror::Error)]
pub enum SbeCodecError {
    /// The buffer is shorter than the 8-byte SBE message header.
    #[error(
        "SBE frame too short: {len} bytes, need at least {}",
        SbeHeader::ENCODED_LENGTH
    )]
    TooShort {
        /// Length of the rejected buffer.
        len: usize,
    },

    /// The header's schema id does not match the topic's message type.
    #[error("SBE schema id mismatch: expected {expected}, got {actual}")]
    SchemaIdMismatch {
        /// Schema id declared by [`SbeMessage::SCHEMA_ID`].
        expected: u16,
        /// Schema id found on the wire.
        actual: u16,
    },

    /// The header's template id does not match the topic's message type.
    #[error("SBE template id mismatch: expected {expected}, got {actual}")]
    TemplateIdMismatch {
        /// Template id declared by [`SbeMessage::TEMPLATE_ID`].
        expected: u16,
        /// Template id found on the wire.
        actual: u16,
    },

    /// The buffer ends before the root block promised by the header's
    /// `blockLength`, so flyweight field reads would run past the buffer.
    #[error("SBE root block truncated: frame is {len} bytes, header promises {needed}")]
    RootBlockTruncated {
        /// Header length plus the acting `blockLength`.
        needed: usize,
        /// Length of the rejected buffer.
        len: usize,
    },

    /// SBE frames are binary and refuse string-based transports (SNS/SQS,
    /// Redis Streams) outright, even when a frame happens to be valid UTF-8.
    #[error("SBE frames cannot ride string-based transports")]
    StringTransportUnsupported,
}

impl SbeCodecError {
    fn into_shove(self) -> ShoveError {
        ShoveError::Codec {
            codec: SBE_CODEC_NAME,
            source: Box::new(self),
        }
    }
}

const SBE_CODEC_NAME: &str = "sbe";

/// An owned, header-validated SBE wire frame for message type `T`.
///
/// Holds the complete frame (header + body) in a reference-counted
/// [`Bytes`] buffer, so cloning and publishing never copy the payload.
/// Construction is the single validation point: every `SbeFrame<T>` in the
/// program is known to carry `T`'s schema and template ids.
pub struct SbeFrame<T: SbeMessage> {
    bytes: Bytes,
    header: SbeHeader,
    _message: PhantomData<fn() -> T>,
}

impl<T: SbeMessage> SbeFrame<T> {
    /// Take ownership of a fully encoded frame (header included), validating
    /// length, schema id, and template id against `T`.
    ///
    /// `Vec<u8>` and `Bytes` convert without copying.
    pub fn new(bytes: impl Into<Bytes>) -> std::result::Result<Self, SbeCodecError> {
        let bytes = bytes.into();
        let header = SbeHeader::read(&bytes, T::BYTE_ORDER)?;
        if header.schema_id != T::SCHEMA_ID {
            return Err(SbeCodecError::SchemaIdMismatch {
                expected: T::SCHEMA_ID,
                actual: header.schema_id,
            });
        }
        if header.template_id != T::TEMPLATE_ID {
            return Err(SbeCodecError::TemplateIdMismatch {
                expected: T::TEMPLATE_ID,
                actual: header.template_id,
            });
        }
        let needed = SbeHeader::ENCODED_LENGTH + usize::from(header.block_length);
        if bytes.len() < needed {
            return Err(SbeCodecError::RootBlockTruncated {
                needed,
                len: bytes.len(),
            });
        }
        Ok(Self {
            bytes,
            header,
            _message: PhantomData,
        })
    }

    /// The message header parsed at construction, including the acting
    /// `block_length` and `version` needed for cross-version decoding.
    pub fn header(&self) -> SbeHeader {
        self.header
    }

    /// The complete wire frame (header + body). Wrap generated decoders over
    /// this at offset 0, or over [`Self::body`] with explicit acting values.
    pub fn as_bytes(&self) -> &[u8] {
        &self.bytes
    }

    /// The message body after the 8-byte header.
    pub fn body(&self) -> &[u8] {
        self.bytes.get(SbeHeader::ENCODED_LENGTH..).unwrap_or(&[])
    }

    /// Consume the frame, returning the underlying buffer without copying.
    pub fn into_bytes(self) -> Bytes {
        self.bytes
    }
}

impl<T: SbeMessage> Clone for SbeFrame<T> {
    fn clone(&self) -> Self {
        Self {
            bytes: self.bytes.clone(),
            header: self.header,
            _message: PhantomData,
        }
    }
}

impl<T: SbeMessage> std::fmt::Debug for SbeFrame<T> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SbeFrame")
            .field("header", &self.header())
            .field("len", &self.bytes.len())
            .finish()
    }
}

impl<T: SbeMessage> PartialEq for SbeFrame<T> {
    fn eq(&self, other: &Self) -> bool {
        self.bytes == other.bytes
    }
}

impl<T: SbeMessage> Eq for SbeFrame<T> {}

impl<T: SbeMessage> AsRef<[u8]> for SbeFrame<T> {
    fn as_ref(&self) -> &[u8] {
        &self.bytes
    }
}

/// Zero-copy codec for [`SbeFrame`] topics.
///
/// The frame is already encoded, so `encode_bytes` is a reference-count
/// clone and `decode_owned` revalidates the header and keeps the received
/// buffer. The borrowed fallbacks (`encode`, `decode`) cost exactly one
/// buffer copy and are only hit on backends that cannot hand over ownership.
/// String-API backends (SNS/SQS, Redis Streams) are refused unconditionally
/// at publish time via `encode_to_string`.
pub struct SbeCodec;

impl<T: SbeMessage> Codec<SbeFrame<T>> for SbeCodec {
    const NAME: &'static str = SBE_CODEC_NAME;

    fn encode(value: &SbeFrame<T>) -> Result<Vec<u8>> {
        Ok(value.bytes.to_vec())
    }

    fn decode(bytes: &[u8]) -> Result<SbeFrame<T>> {
        Self::decode_owned(Bytes::copy_from_slice(bytes))
    }

    fn encode_bytes(value: &SbeFrame<T>) -> Result<Bytes> {
        Ok(value.bytes.clone())
    }

    // Unconditional: the default UTF-8 check would let frames that happen to
    // be UTF-8-valid ride string transports, making failures data-dependent.
    fn encode_to_string(_value: &SbeFrame<T>) -> Result<String> {
        Err(SbeCodecError::StringTransportUnsupported.into_shove())
    }

    fn decode_owned(bytes: Bytes) -> Result<SbeFrame<T>> {
        SbeFrame::new(bytes).map_err(SbeCodecError::into_shove)
    }
}

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

    // Hand-rolled stand-in for sbe-tool output: <order> with two u64 fields
    // (price, quantity), schema id 42, template id 7, version 1.
    struct Order;
    impl SbeMessage for Order {
        const SCHEMA_ID: u16 = 42;
        const TEMPLATE_ID: u16 = 7;
    }

    struct BigEndianOrder;
    impl SbeMessage for BigEndianOrder {
        const SCHEMA_ID: u16 = 42;
        const TEMPLATE_ID: u16 = 7;
        const BYTE_ORDER: SbeByteOrder = SbeByteOrder::BigEndian;
    }

    fn encode_order(price: u64, quantity: u64) -> Vec<u8> {
        let header = SbeHeader {
            block_length: 16,
            template_id: 7,
            schema_id: 42,
            version: 1,
        };
        let mut frame = header.to_bytes(SbeByteOrder::LittleEndian).to_vec();
        frame.extend_from_slice(&price.to_le_bytes());
        frame.extend_from_slice(&quantity.to_le_bytes());
        frame
    }

    fn decode_order(body: &[u8]) -> (u64, u64) {
        let field = |offset: usize| {
            body.get(offset..offset + 8)
                .and_then(|b| b.try_into().ok())
                .map(u64::from_le_bytes)
                .unwrap()
        };
        (field(0), field(8))
    }

    #[test]
    fn header_round_trips_little_endian() {
        let header = SbeHeader {
            block_length: 16,
            template_id: 7,
            schema_id: 42,
            version: 3,
        };
        let bytes = header.to_bytes(SbeByteOrder::LittleEndian);
        assert_eq!(bytes[..2], [16, 0]);
        let back = SbeHeader::read(&bytes, SbeByteOrder::LittleEndian).unwrap();
        assert_eq!(back, header);
    }

    #[test]
    fn header_round_trips_big_endian() {
        let header = SbeHeader {
            block_length: 16,
            template_id: 7,
            schema_id: 42,
            version: 3,
        };
        let bytes = header.to_bytes(SbeByteOrder::BigEndian);
        assert_eq!(bytes[..2], [0, 16]);
        let back = SbeHeader::read(&bytes, SbeByteOrder::BigEndian).unwrap();
        assert_eq!(back, header);
    }

    #[test]
    fn frame_rejects_too_short_buffer() {
        let err = SbeFrame::<Order>::new(vec![1, 2, 3]).unwrap_err();
        match err {
            SbeCodecError::TooShort { len } => assert_eq!(len, 3),
            other => panic!("expected TooShort, got {other:?}"),
        }
    }

    #[test]
    fn frame_rejects_schema_id_mismatch() {
        let mut frame = encode_order(1, 2);
        frame[4..6].copy_from_slice(&99u16.to_le_bytes());
        let err = SbeFrame::<Order>::new(frame).unwrap_err();
        match err {
            SbeCodecError::SchemaIdMismatch { expected, actual } => {
                assert_eq!(expected, 42);
                assert_eq!(actual, 99);
            }
            other => panic!("expected SchemaIdMismatch, got {other:?}"),
        }
    }

    #[test]
    fn frame_rejects_template_id_mismatch() {
        let mut frame = encode_order(1, 2);
        frame[2..4].copy_from_slice(&8u16.to_le_bytes());
        let err = SbeFrame::<Order>::new(frame).unwrap_err();
        match err {
            SbeCodecError::TemplateIdMismatch { expected, actual } => {
                assert_eq!(expected, 7);
                assert_eq!(actual, 8);
            }
            other => panic!("expected TemplateIdMismatch, got {other:?}"),
        }
    }

    #[test]
    fn frame_exposes_header_body_and_full_frame() {
        let encoded = encode_order(1000, 5);
        let frame = SbeFrame::<Order>::new(encoded.clone()).unwrap();

        let header = frame.header();
        assert_eq!(header.block_length, 16);
        assert_eq!(header.template_id, 7);
        assert_eq!(header.schema_id, 42);
        assert_eq!(header.version, 1);

        assert_eq!(frame.as_bytes(), &encoded[..]);
        assert_eq!(frame.body(), &encoded[8..]);
        assert_eq!(decode_order(frame.body()), (1000, 5));
    }

    #[test]
    fn codec_round_trips_through_flyweight() {
        let frame = SbeFrame::<Order>::new(encode_order(250_000, 12)).unwrap();
        let wire = <SbeCodec as Codec<SbeFrame<Order>>>::encode(&frame).unwrap();
        let received = <SbeCodec as Codec<SbeFrame<Order>>>::decode(&wire).unwrap();
        assert_eq!(received, frame);
        assert_eq!(decode_order(received.body()), (250_000, 12));
    }

    #[test]
    fn encode_bytes_is_zero_copy() {
        let frame = SbeFrame::<Order>::new(encode_order(1, 2)).unwrap();
        let encoded = <SbeCodec as Codec<SbeFrame<Order>>>::encode_bytes(&frame).unwrap();
        assert_eq!(encoded.as_ptr(), frame.as_bytes().as_ptr());
    }

    #[test]
    fn decode_owned_is_zero_copy() {
        let bytes = Bytes::from(encode_order(1, 2));
        let ptr = bytes.as_ptr();
        let frame = <SbeCodec as Codec<SbeFrame<Order>>>::decode_owned(bytes).unwrap();
        assert_eq!(frame.as_bytes().as_ptr(), ptr);
    }

    #[test]
    fn frame_construction_from_vec_is_zero_copy() {
        let encoded = encode_order(1, 2);
        let ptr = encoded.as_ptr();
        let frame = SbeFrame::<Order>::new(encoded).unwrap();
        assert_eq!(frame.as_bytes().as_ptr(), ptr);
        let clone = frame.clone();
        assert_eq!(clone.as_bytes().as_ptr(), ptr);
    }

    #[test]
    fn newer_wire_version_and_block_length_pass_through() {
        let header = SbeHeader {
            block_length: 24,
            template_id: 7,
            schema_id: 42,
            version: 9,
        };
        let mut frame = header.to_bytes(SbeByteOrder::LittleEndian).to_vec();
        frame.extend_from_slice(&[0u8; 24]);

        let decoded = <SbeCodec as Codec<SbeFrame<Order>>>::decode(&frame).unwrap();
        assert_eq!(decoded.header().version, 9);
        assert_eq!(decoded.header().block_length, 24);
    }

    #[test]
    fn big_endian_message_uses_schema_byte_order() {
        let header = SbeHeader {
            block_length: 16,
            template_id: 7,
            schema_id: 42,
            version: 1,
        };
        let mut frame = header.to_bytes(SbeByteOrder::BigEndian).to_vec();
        frame.extend_from_slice(&[0u8; 16]);
        let decoded = SbeFrame::<BigEndianOrder>::new(frame.clone()).unwrap();
        assert_eq!(decoded.header().schema_id, 42);

        // The same bytes read as little-endian carry swapped ids and fail.
        assert!(SbeFrame::<Order>::new(frame).is_err());
    }

    #[test]
    fn codec_name_is_sbe() {
        assert_eq!(<SbeCodec as Codec<SbeFrame<Order>>>::NAME, "sbe");
    }

    #[test]
    fn decode_failure_surfaces_as_shove_codec_error() {
        let err = <SbeCodec as Codec<SbeFrame<Order>>>::decode(&[0xFF; 4]).unwrap_err();
        match err {
            ShoveError::Codec { codec, .. } => assert_eq!(codec, "sbe"),
            other => panic!("expected Codec variant, got {other:?}"),
        }
    }

    #[test]
    fn encode_to_string_rejects_even_utf8_valid_frames() {
        // Every byte of this frame is < 0x80, so it passes UTF-8 validation;
        // the codec must still refuse string-based transports outright.
        let frame = SbeFrame::<Order>::new(encode_order(65, 66)).unwrap();
        assert!(String::from_utf8(frame.as_bytes().to_vec()).is_ok());

        let err = <SbeCodec as Codec<SbeFrame<Order>>>::encode_to_string(&frame).unwrap_err();
        match err {
            ShoveError::Codec { codec, source } => {
                assert_eq!(codec, "sbe");
                assert!(matches!(
                    source.downcast_ref::<SbeCodecError>(),
                    Some(SbeCodecError::StringTransportUnsupported)
                ));
            }
            other => panic!("expected Codec variant, got {other:?}"),
        }
    }

    #[test]
    fn frame_rejects_root_block_shorter_than_block_length() {
        let header = SbeHeader {
            block_length: 16,
            template_id: 7,
            schema_id: 42,
            version: 1,
        };
        let mut frame = header.to_bytes(SbeByteOrder::LittleEndian).to_vec();
        frame.extend_from_slice(&[0u8; 15]);

        let err = SbeFrame::<Order>::new(frame).unwrap_err();
        match err {
            SbeCodecError::RootBlockTruncated { needed, len } => {
                assert_eq!(needed, 24);
                assert_eq!(len, 23);
            }
            other => panic!("expected RootBlockTruncated, got {other:?}"),
        }
    }

    #[test]
    fn frame_accepts_root_block_exactly_block_length() {
        let header = SbeHeader {
            block_length: 16,
            template_id: 7,
            schema_id: 42,
            version: 1,
        };
        let mut frame = header.to_bytes(SbeByteOrder::LittleEndian).to_vec();
        frame.extend_from_slice(&[0u8; 16]);
        assert!(SbeFrame::<Order>::new(frame).is_ok());
    }
}