rusty_common/

zero_alloc_message.rs

//! Zero-allocation message processing for WebSocket exchanges
//!
//! This module provides concrete implementations showing how to process
//! WebSocket messages from exchanges with zero allocations using borrowed values.

use crate::SmartString;
use crate::error::{CommonError, Result};
use crate::memory::hft_pools::{HftBufferHandle, OrderHandle, TradeHandle, global_hft_pools};
use crate::zerocopy::{BorrowedValueExt, WebSocketJsonZeroCopy};
use log::warn;
use rust_decimal::Decimal;
use simd_json::BorrowedValue;
use smallvec::SmallVec;

/// Zero-allocation order update processing
pub struct ZeroAllocOrderUpdate<'a> {
    /// The parsed JSON value (borrowed)
    value: BorrowedValue<'a>,
}

impl<'a> ZeroAllocOrderUpdate<'a> {
    /// Parse an order update message with zero allocations
    pub fn parse(message_bytes: &'a mut [u8]) -> Result<Self> {
        let value = WebSocketJsonZeroCopy::parse_to_borrowed(message_bytes)?;
        Ok(Self { value })
    }

    /// Get the message type without allocation
    #[inline]
    pub fn message_type(&self) -> Option<&str> {
        self.value.get_str("type")
    }

    /// Get the order ID without allocation
    #[inline]
    pub fn order_id(&self) -> Option<&str> {
        self.value.get_str("order_id")
    }

    /// Get the client order ID without allocation
    #[inline]
    pub fn client_order_id(&self) -> Option<&str> {
        self.value
            .get_str("client_order_id")
            .or_else(|| self.value.get_str("client_oid"))
            .or_else(|| self.value.get_str("clOrdId"))
    }

    /// Get the symbol without allocation
    #[inline]
    pub fn symbol(&self) -> Option<&str> {
        self.value
            .get_str("symbol")
            .or_else(|| self.value.get_str("product_id"))
            .or_else(|| self.value.get_str("instrument"))
    }

    /// Get the price as string without allocation
    #[inline]
    pub fn price_str(&self) -> Option<&str> {
        self.value.get_str("price")
    }

    /// Get the quantity as string without allocation
    #[inline]
    pub fn quantity_str(&self) -> Option<&str> {
        self.value
            .get_str("quantity")
            .or_else(|| self.value.get_str("size"))
            .or_else(|| self.value.get_str("amount"))
    }

    /// Get the order status without allocation
    #[inline]
    pub fn status(&self) -> Option<&str> {
        self.value
            .get_str("status")
            .or_else(|| self.value.get_str("order_status"))
    }

    /// Get the side without allocation
    #[inline]
    pub fn side(&self) -> Option<&str> {
        self.value.get_str("side")
    }

    /// Get the timestamp as u64 without allocation
    #[inline]
    pub fn timestamp(&self) -> Option<u64> {
        self.value
            .get_u64("timestamp")
            .or_else(|| self.value.get_u64("time"))
    }

    /// Convert to owned values only when necessary
    pub fn to_owned_values(&self) -> OrderUpdateOwned {
        OrderUpdateOwned {
            order_id: self.order_id().map(SmartString::from),
            client_order_id: self.client_order_id().map(SmartString::from),
            symbol: self.symbol().map(SmartString::from),
            price: self.price_str().and_then(|s| {
                Decimal::from_str_exact(s)
                    .map_err(|e| {
                        warn!("Failed to parse price '{s}': {e}");
                        e
                    })
                    .ok()
            }),
            quantity: self.quantity_str().and_then(|s| {
                Decimal::from_str_exact(s)
                    .map_err(|e| {
                        warn!("Failed to parse quantity '{s}': {e}");
                        e
                    })
                    .ok()
            }),
            status: self.status().map(SmartString::from),
            side: self.side().map(SmartString::from),
            timestamp: self.timestamp(),
        }
    }
}

/// Owned version of order update (only allocate when necessary)
#[derive(Debug, Clone)]
pub struct OrderUpdateOwned {
    /// The order ID.
    pub order_id: Option<SmartString>,
    /// The client order ID.
    pub client_order_id: Option<SmartString>,
    /// The trading symbol.
    pub symbol: Option<SmartString>,
    /// The order price.
    pub price: Option<Decimal>,
    /// The order quantity.
    pub quantity: Option<Decimal>,
    /// The order status.
    pub status: Option<SmartString>,
    /// The order side.
    pub side: Option<SmartString>,
    /// The timestamp of the order.
    pub timestamp: Option<u64>,
}

/// Zero-allocation market data processing
pub struct ZeroAllocMarketData<'a> {
    /// The parsed JSON value (borrowed)
    value: BorrowedValue<'a>,
}

impl<'a> ZeroAllocMarketData<'a> {
    /// Parse a market data message with zero allocations
    pub fn parse(message_bytes: &'a mut [u8]) -> Result<Self> {
        let value = WebSocketJsonZeroCopy::parse_to_borrowed(message_bytes)?;
        Ok(Self { value })
    }

    /// Get the message type without allocation
    #[inline]
    pub fn message_type(&self) -> Option<&str> {
        self.value
            .get_str("type")
            .or_else(|| self.value.get_str("e")) // Binance uses "e" for event type
    }

    /// Get the symbol without allocation
    #[inline]
    pub fn symbol(&self) -> Option<&str> {
        self.value
            .get_str("symbol")
            .or_else(|| self.value.get_str("s")) // Binance uses "s" for symbol
            .or_else(|| self.value.get_str("product_id"))
    }

    /// Get the event time without allocation
    #[inline]
    pub fn event_time(&self) -> Option<u64> {
        self.value
            .get_u64("E") // Binance event time
            .or_else(|| self.value.get_u64("T")) // Trade time
            .or_else(|| self.value.get_u64("timestamp"))
    }

    /// Get bid price for ticker updates
    #[inline]
    pub fn bid_price_str(&self) -> Option<&str> {
        self.value
            .get_str("b") // Binance bid price
            .or_else(|| self.value.get_str("best_bid"))
    }

    /// Get ask price for ticker updates
    #[inline]
    pub fn ask_price_str(&self) -> Option<&str> {
        self.value
            .get_str("a") // Binance ask price
            .or_else(|| self.value.get_str("best_ask"))
    }

    /// Get last price for ticker updates
    #[inline]
    pub fn last_price_str(&self) -> Option<&str> {
        self.value
            .get_str("c") // Binance close/last price
            .or_else(|| self.value.get_str("p")) // Price
            .or_else(|| self.value.get_str("price"))
    }

    /// Get volume for ticker updates
    #[inline]
    pub fn volume_str(&self) -> Option<&str> {
        self.value
            .get_str("v") // Binance volume
            .or_else(|| self.value.get_str("q")) // Quote volume
            .or_else(|| self.value.get_str("volume"))
    }

    /// Check if this is a snapshot or update
    #[inline]
    pub fn is_snapshot(&self) -> bool {
        self.message_type()
            .map(|t| t == "snapshot" || t == "depthUpdate")
            .unwrap_or(false)
    }
}

/// Stack buffer size for small JSON messages (4KB on stack)
const MESSAGE_BUFFER_SIZE: usize = 4096;

/// Stack-allocated buffer for small messages
pub type MessageBuffer = SmallVec<[u8; MESSAGE_BUFFER_SIZE]>;

/// Zero-allocation WebSocket message processor with buffer reuse
pub struct ZeroAllocMessageProcessor {
    /// Reusable buffer for parsing - stays on stack for small messages
    buffer: MessageBuffer,
    /// Pool buffer handle for large messages
    pool_buffer: Option<HftBufferHandle>,
}

impl Default for ZeroAllocMessageProcessor {
    fn default() -> Self {
        Self::new()
    }
}

impl ZeroAllocMessageProcessor {
    /// Create a new zero-allocation message processor
    #[must_use]
    pub fn new() -> Self {
        Self {
            buffer: MessageBuffer::new(),
            pool_buffer: None,
        }
    }

    /// Process JSON message with zero heap allocations for small messages
    ///
    /// Uses SmallVec for messages under 4KB (stack allocation),
    /// falls back to pooled buffers for larger messages.
    /// Note: This borrows from the internal buffer, so cannot call other methods
    /// on the processor while the returned value is alive.
    #[inline]
    fn process_json_bytes_internal(&mut self, data: &[u8]) -> Result<BorrowedValue<'_>> {
        // Clear previous buffer content
        self.buffer.clear();

        // For small messages, use stack-allocated SmallVec
        if data.len() <= MESSAGE_BUFFER_SIZE {
            self.buffer.extend_from_slice(data);
            // Parse in-place using borrowed parsing
            WebSocketJsonZeroCopy::parse_to_borrowed(&mut self.buffer)
        } else {
            // For large messages, use pooled buffer
            if self.pool_buffer.is_none() {
                self.pool_buffer = Some(global_hft_pools().acquire_buffer());
            }

            let pool_buf = self.pool_buffer.as_mut().unwrap();
            pool_buf.clear();
            pool_buf.extend_from_slice(data);

            WebSocketJsonZeroCopy::parse_to_borrowed(pool_buf.as_mut_slice())
        }
    }

    /// Parse order message with pooled object recycling
    ///
    /// Returns a pooled order handle that automatically returns to pool on drop.
    #[inline]
    pub fn parse_order_to_pooled(&mut self, data: &[u8]) -> Result<OrderHandle> {
        let json_value = self.process_json_bytes_internal(data)?;
        let mut order = global_hft_pools().acquire_order();

        // Parse order fields using borrowed JSON values (no string allocations)
        if let Some(order_id) = json_value.get_str("order_id") {
            order.order_id = SmartString::from(order_id);
        }

        if let Some(client_order_id) = json_value
            .get_str("client_order_id")
            .or_else(|| json_value.get_str("client_oid"))
            .or_else(|| json_value.get_str("clOrdId"))
        {
            order.client_order_id = SmartString::from(client_order_id);
        }

        if let Some(symbol) = json_value
            .get_str("symbol")
            .or_else(|| json_value.get_str("product_id"))
            .or_else(|| json_value.get_str("instrument"))
        {
            order.symbol = SmartString::from(symbol);
        }

        if let Some(side) = json_value.get_str("side") {
            order.side = SmartString::from(side);
        }

        if let Some(order_type) = json_value
            .get_str("type")
            .or_else(|| json_value.get_str("order_type"))
        {
            order.order_type = SmartString::from(order_type);
        }

        if let Some(status) = json_value
            .get_str("status")
            .or_else(|| json_value.get_str("order_status"))
        {
            order.status = SmartString::from(status);
        }

        // Parse decimal fields safely - return error if critical fields are invalid
        if let Some(price_str) = json_value.get_str("price") {
            match Decimal::from_str_exact(price_str) {
                Ok(price) => order.price = price,
                Err(e) => {
                    warn!("Failed to parse order price '{price_str}': {e}");
                    return Err(CommonError::Parse(
                        format!("Invalid price format: {price_str}").into(),
                    ));
                }
            }
        }

        if let Some(quantity_str) = json_value
            .get_str("quantity")
            .or_else(|| json_value.get_str("size"))
            .or_else(|| json_value.get_str("amount"))
        {
            match Decimal::from_str_exact(quantity_str) {
                Ok(quantity) => order.quantity = quantity,
                Err(e) => {
                    warn!("Failed to parse order quantity '{quantity_str}': {e}");
                    return Err(CommonError::Parse(
                        format!("Invalid quantity format: {quantity_str}").into(),
                    ));
                }
            }
        }

        if let Some(filled_str) = json_value
            .get_str("filled_quantity")
            .or_else(|| json_value.get_str("filled_size"))
            .or_else(|| json_value.get_str("executed_qty"))
        {
            match Decimal::from_str_exact(filled_str) {
                Ok(filled) => order.filled_quantity = filled,
                Err(e) => warn!("Failed to parse order filled quantity '{filled_str}': {e}"),
            }
        }

        // Parse timestamps
        if let Some(ts) = json_value
            .get_u64("timestamp")
            .or_else(|| json_value.get_u64("time"))
        {
            order.timestamp = ts;
        }

        if let Some(ex_ts) = json_value
            .get_u64("exchange_timestamp")
            .or_else(|| json_value.get_u64("exchange_time"))
            .or_else(|| json_value.get_u64("T"))
        {
            // Binance timestamp
            order.exchange_timestamp = ex_ts;
        }

        // Calculate remaining quantity
        order.remaining_quantity = order.quantity - order.filled_quantity;

        Ok(order)
    }

    /// Parse trade message with pooled object recycling
    ///
    /// Returns a pooled trade handle that automatically returns to pool on drop.
    #[inline]
    pub fn parse_trade_to_pooled(&mut self, data: &[u8]) -> Result<TradeHandle> {
        let json_value = self.process_json_bytes_internal(data)?;
        let mut trade = global_hft_pools().acquire_trade();

        // Parse trade fields using borrowed JSON values (no string allocations)
        if let Some(trade_id) = json_value
            .get_str("trade_id")
            .or_else(|| json_value.get_str("t"))
        {
            // Binance trade ID
            trade.trade_id = SmartString::from(trade_id);
        }

        if let Some(order_id) = json_value
            .get_str("order_id")
            .or_else(|| json_value.get_str("o"))
        {
            trade.order_id = SmartString::from(order_id);
        }

        if let Some(symbol) = json_value
            .get_str("symbol")
            .or_else(|| json_value.get_str("s")) // Binance symbol
            .or_else(|| json_value.get_str("product_id"))
        {
            trade.symbol = SmartString::from(symbol);
        }

        if let Some(side) = json_value
            .get_str("side")
            .or_else(|| json_value.get_str("S"))
        {
            // Binance side
            trade.side = SmartString::from(side);
        }

        // Parse decimal fields safely - return error if critical fields are invalid
        if let Some(price_str) = json_value
            .get_str("price")
            .or_else(|| json_value.get_str("p"))
        {
            // Binance price
            match Decimal::from_str_exact(price_str) {
                Ok(price) => trade.price = price,
                Err(e) => {
                    warn!("Failed to parse trade price '{price_str}': {e}");
                    return Err(CommonError::Parse(
                        format!("Invalid price format: {price_str}").into(),
                    ));
                }
            }
        }

        if let Some(quantity_str) = json_value
            .get_str("quantity")
            .or_else(|| json_value.get_str("q")) // Binance quantity
            .or_else(|| json_value.get_str("size"))
        {
            match Decimal::from_str_exact(quantity_str) {
                Ok(quantity) => trade.quantity = quantity,
                Err(e) => {
                    warn!("Failed to parse trade quantity '{quantity_str}': {e}");
                    return Err(CommonError::Parse(
                        format!("Invalid quantity format: {quantity_str}").into(),
                    ));
                }
            }
        }

        // Parse timestamps
        if let Some(ts) = json_value
            .get_u64("timestamp")
            .or_else(|| json_value.get_u64("time"))
            .or_else(|| json_value.get_u64("T"))
        {
            // Binance trade time
            trade.timestamp = ts;
        }

        if let Some(ex_ts) = json_value
            .get_u64("exchange_timestamp")
            .or_else(|| json_value.get_u64("E"))
        {
            // Binance event time
            trade.exchange_timestamp = ex_ts;
        }

        // Parse maker flag
        if let Some(is_maker) = json_value
            .get_bool("is_maker")
            .or_else(|| json_value.get_bool("m"))
        {
            // Binance maker flag
            trade.is_maker = is_maker;
        }

        Ok(trade)
    }

    /// Reset the processor for reuse
    ///
    /// Clears internal buffers while keeping allocated capacity.
    #[inline]
    pub fn reset(&mut self) {
        self.buffer.clear();
        if let Some(ref mut pool_buf) = self.pool_buffer {
            pool_buf.clear();
        }
    }

    /// Get buffer statistics for monitoring
    #[inline]
    pub fn buffer_stats(&self) -> BufferStats {
        BufferStats {
            stack_buffer_len: self.buffer.len(),
            stack_buffer_capacity: self.buffer.capacity(),
            stack_buffer_spilled: self.buffer.spilled(),
            pool_buffer_active: self.pool_buffer.is_some(),
            pool_buffer_len: self.pool_buffer.as_ref().map(|b| b.len()).unwrap_or(0),
        }
    }

    /// Process a message and determine its type without allocation
    pub fn identify_message_type(&mut self, message_bytes: &[u8]) -> Result<MessageType> {
        let value = self.process_json_bytes_internal(message_bytes)?;

        // Check common type fields
        if let Some(msg_type) = value.get_str("type") {
            Ok(match msg_type {
                "heartbeat" => MessageType::Heartbeat,
                "error" => MessageType::Error,
                "order" | "execution" | "fill" => MessageType::OrderUpdate,
                "ticker" | "trade" | "quote" => MessageType::MarketData,
                "snapshot" | "l2update" | "depthUpdate" => MessageType::OrderBook,
                _ => MessageType::Unknown,
            })
        } else if let Some(event_type) = value.get_str("e") {
            // Binance format
            Ok(match event_type {
                "executionReport" => MessageType::OrderUpdate,
                "24hrTicker" | "trade" => MessageType::MarketData,
                "depthUpdate" => MessageType::OrderBook,
                _ => MessageType::Unknown,
            })
        } else {
            Ok(MessageType::Unknown)
        }
    }

    /// Extract error message without allocation
    #[inline]
    pub fn extract_error_message(&mut self, data: &[u8]) -> Result<Option<SmartString>> {
        let value = self.process_json_bytes_internal(data)?;
        Ok(value
            .get_str("message")
            .or_else(|| value.get_str("msg"))
            .or_else(|| value.get_str("error"))
            .map(SmartString::from))
    }

    /// Extract request ID without allocation
    #[inline]
    pub fn extract_request_id(&mut self, data: &[u8]) -> Result<Option<SmartString>> {
        let value = self.process_json_bytes_internal(data)?;
        Ok(value
            .get_str("id")
            .or_else(|| value.get_str("request_id"))
            .or_else(|| value.get_str("req_id"))
            .map(SmartString::from))
    }

    /// Extract string field without allocation using borrowed parsing
    #[inline]
    pub fn extract_string_field(
        &mut self,
        data: &[u8],
        field: &str,
    ) -> Result<Option<SmartString>> {
        let value = self.process_json_bytes_internal(data)?;
        Ok(value.get_str(field).map(SmartString::from))
    }

    /// Extract decimal field without string allocation
    #[inline]
    pub fn extract_decimal_field(&mut self, data: &[u8], field: &str) -> Result<Option<Decimal>> {
        let value = self.process_json_bytes_internal(data)?;

        if let Some(str_val) = value.get_str(field) {
            match Decimal::from_str_exact(str_val) {
                Ok(decimal) => return Ok(Some(decimal)),
                Err(e) => {
                    warn!("Failed to parse decimal field '{field}' with value '{str_val}': {e}");
                    return Err(e.into());
                }
            }
        } else if let Some(num_val) = value.get_f64(field) {
            match Decimal::try_from(num_val) {
                Ok(decimal) => return Ok(Some(decimal)),
                Err(e) => {
                    warn!(
                        "Failed to convert f64 field '{field}' with value '{num_val}' to Decimal: {e}"
                    );
                    return Err(e.into());
                }
            }
        }

        Ok(None)
    }

    /// Extract u64 field without allocation
    #[inline]
    pub fn extract_u64_field(&mut self, data: &[u8], field: &str) -> Result<Option<u64>> {
        let value = self.process_json_bytes_internal(data)?;
        Ok(value.get_u64(field))
    }

    /// Extract boolean field without allocation
    #[inline]
    pub fn extract_bool_field(&mut self, data: &[u8], field: &str) -> Result<Option<bool>> {
        let value = self.process_json_bytes_internal(data)?;
        Ok(value.get_bool(field))
    }
}

/// Message type classification
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MessageType {
    /// A heartbeat message.
    Heartbeat,
    /// An error message.
    Error,
    /// An order update message.
    OrderUpdate,
    /// A market data message.
    MarketData,
    /// An order book message.
    OrderBook,
    /// An unknown message type.
    Unknown,
}

/// Buffer usage statistics for monitoring performance
#[derive(Debug, Clone, Copy)]
pub struct BufferStats {
    /// The length of the stack buffer.
    pub stack_buffer_len: usize,
    /// The capacity of the stack buffer.
    pub stack_buffer_capacity: usize,
    /// Whether the stack buffer has spilled to the heap.
    pub stack_buffer_spilled: bool,
    /// Whether a pool buffer is active.
    pub pool_buffer_active: bool,
    /// The length of the pool buffer.
    pub pool_buffer_len: usize,
}

// Thread-local zero-allocation message processor
thread_local! {
    static LOCAL_PROCESSOR: std::cell::RefCell<ZeroAllocMessageProcessor> =
        std::cell::RefCell::new(ZeroAllocMessageProcessor::new());
}

/// Access thread-local zero-allocation message processor
///
/// This avoids the overhead of creating new processors for each message.
pub fn with_zero_alloc_processor<F, R>(f: F) -> R
where
    F: FnOnce(&mut ZeroAllocMessageProcessor) -> R,
{
    LOCAL_PROCESSOR.with(|processor| f(&mut processor.borrow_mut()))
}

/// Convenience function for zero-allocation order parsing
#[inline]
pub fn parse_order_zero_alloc(data: &[u8]) -> Result<OrderHandle> {
    with_zero_alloc_processor(|processor| processor.parse_order_to_pooled(data))
}

/// Convenience function for zero-allocation trade parsing
#[inline]
pub fn parse_trade_zero_alloc(data: &[u8]) -> Result<TradeHandle> {
    with_zero_alloc_processor(|processor| processor.parse_trade_to_pooled(data))
}

/// Convenience function for zero-allocation JSON field extraction
#[inline]
pub fn extract_string_field_zero_alloc(data: &[u8], field: &str) -> Result<Option<SmartString>> {
    with_zero_alloc_processor(|processor| processor.extract_string_field(data, field))
}

/// Convenience function for zero-allocation message type identification
#[inline]
pub fn identify_message_type_zero_alloc(data: &[u8]) -> Result<MessageType> {
    with_zero_alloc_processor(|processor| processor.identify_message_type(data))
}

/// Example usage in WebSocket handler
pub async fn example_zero_alloc_handler(message: &mut [u8]) -> Result<()> {
    // Use thread-local processor for best performance
    with_zero_alloc_processor(|processor| {
        // Identify message type with zero allocations
        let msg_type = processor.identify_message_type(message)?;

        match msg_type {
            MessageType::OrderUpdate => {
                // Process order update with zero allocations and pooled objects
                let order_update = ZeroAllocOrderUpdate::parse(message)?;

                // Access fields without allocation
                if let Some(order_id) = order_update.order_id() {
                    // Use borrowed string directly
                    println!("Order ID: {order_id}");
                }

                // Use pooled objects for storage when needed
                if order_update.status() == Some("filled") {
                    // Note: Cannot use both order_update and processor simultaneously
                    // due to lifetime constraints. In practice, choose one approach.
                    // For example, drop order_update and parse with pooled object:
                    drop(order_update);
                    let _pooled_order = processor.parse_order_to_pooled(message)?;
                    // Process with pooled object...
                }
            }
            MessageType::MarketData => {
                // Process market data with zero allocations
                let market_data = ZeroAllocMarketData::parse(message)?;

                // Use borrowed values directly
                if let Some(symbol) = market_data.symbol()
                    && let Some(price) = market_data.last_price_str()
                {
                    // Process without allocation
                    println!("{symbol}: {price}");
                }

                // Or parse as trade with pooled object if needed
                if market_data.message_type() == Some("trade") {
                    // Note: Cannot use both market_data and processor simultaneously
                    // due to lifetime constraints. Drop market_data first:
                    drop(market_data);
                    let _pooled_trade = processor.parse_trade_to_pooled(message)?;
                    // Process with pooled object...
                }
            }
            _ => {
                // Handle other message types
            }
        }

        Ok(())
    })
}

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

    #[test]
    fn test_zero_alloc_order_update() {
        let mut json = r#"{"type":"order","order_id":"123","symbol":"BTC-USDT","price":"50000.5","quantity":"0.1","status":"new","side":"buy","timestamp":1234567890}"#.to_string();
        let bytes = unsafe { json.as_bytes_mut() };

        let update = ZeroAllocOrderUpdate::parse(bytes).unwrap();
        assert_eq!(update.message_type(), Some("order"));
        assert_eq!(update.order_id(), Some("123"));
        assert_eq!(update.symbol(), Some("BTC-USDT"));
        assert_eq!(update.price_str(), Some("50000.5"));
        assert_eq!(update.quantity_str(), Some("0.1"));
        assert_eq!(update.status(), Some("new"));
        assert_eq!(update.side(), Some("buy"));
        assert_eq!(update.timestamp(), Some(1234567890));
    }

    #[test]
    fn test_zero_alloc_market_data() {
        let mut json = r#"{"e":"24hrTicker","s":"BTCUSDT","b":"49999.0","a":"50001.0","c":"50000.0","v":"1000.5","E":1234567890}"#.to_string();
        let bytes = unsafe { json.as_bytes_mut() };

        let data = ZeroAllocMarketData::parse(bytes).unwrap();
        assert_eq!(data.message_type(), Some("24hrTicker"));
        assert_eq!(data.symbol(), Some("BTCUSDT"));
        assert_eq!(data.bid_price_str(), Some("49999.0"));
        assert_eq!(data.ask_price_str(), Some("50001.0"));
        assert_eq!(data.last_price_str(), Some("50000.0"));
        assert_eq!(data.volume_str(), Some("1000.5"));
        assert_eq!(data.event_time(), Some(1234567890));
    }

    #[test]
    fn test_pooled_order_parsing() {
        let order_json = r#"{
            "order_id": "ORDER123",
            "client_order_id": "CLIENT456",
            "symbol": "BTC-USDT",
            "side": "buy",
            "type": "limit",
            "status": "open",
            "price": "50000.00",
            "quantity": "0.001",
            "filled_quantity": "0.0005",
            "timestamp": 1640995200000,
            "exchange_timestamp": 1640995200100
        }"#;

        let mut processor = ZeroAllocMessageProcessor::new();
        let order = processor
            .parse_order_to_pooled(order_json.as_bytes())
            .unwrap();

        assert_eq!(order.order_id.as_str(), "ORDER123");
        assert_eq!(order.symbol.as_str(), "BTC-USDT");
        assert_eq!(order.side.as_str(), "buy");
        assert_eq!(order.price, Decimal::from_str_exact("50000.00").unwrap());
        assert_eq!(order.quantity, Decimal::from_str_exact("0.001").unwrap());
        assert_eq!(
            order.filled_quantity,
            Decimal::from_str_exact("0.0005").unwrap()
        );
        assert_eq!(
            order.remaining_quantity,
            Decimal::from_str_exact("0.0005").unwrap()
        );
        assert_eq!(order.timestamp, 1640995200000);
    }

    #[test]
    fn test_pooled_trade_parsing() {
        let trade_json = r#"{
            "trade_id": "TRADE789",
            "order_id": "ORDER123",
            "symbol": "ETH-USDT",
            "side": "sell",
            "price": "3000.50",
            "quantity": "0.1",
            "timestamp": 1640995200000,
            "exchange_timestamp": 1640995200050,
            "is_maker": true
        }"#;

        let mut processor = ZeroAllocMessageProcessor::new();
        let trade = processor
            .parse_trade_to_pooled(trade_json.as_bytes())
            .unwrap();

        assert_eq!(trade.trade_id.as_str(), "TRADE789");
        assert_eq!(trade.symbol.as_str(), "ETH-USDT");
        assert_eq!(trade.side.as_str(), "sell");
        assert_eq!(trade.price, Decimal::from_str_exact("3000.50").unwrap());
        assert_eq!(trade.quantity, Decimal::from_str_exact("0.1").unwrap());
        assert!(trade.is_maker);
    }

    #[test]
    fn test_small_message_stack_allocation() {
        let mut processor = ZeroAllocMessageProcessor::new();

        // Small JSON message should stay on stack
        let small_json = r#"{"id":1,"price":100.0}"#;
        {
            let _result = processor
                .process_json_bytes_internal(small_json.as_bytes())
                .unwrap();
            // _result drops here, releasing the borrow
        }

        let stats = processor.buffer_stats();
        assert!(!stats.stack_buffer_spilled);
        assert!(!stats.pool_buffer_active);
    }

    #[test]
    fn test_large_message_pool_allocation() {
        let mut processor = ZeroAllocMessageProcessor::new();

        // Large JSON message should use pool buffer
        let large_json = format!(r#"{{"data":"{}"}}"#, "x".repeat(5000));
        {
            let _result = processor
                .process_json_bytes_internal(large_json.as_bytes())
                .unwrap();
            // _result drops here, releasing the borrow
        }

        let stats = processor.buffer_stats();
        assert!(stats.pool_buffer_active);
    }

    #[test]
    fn test_thread_local_processor() {
        let order_json = r#"{
            "order_id": "TL123",
            "symbol": "BTC-USDT",
            "side": "buy",
            "price": "40000.00",
            "quantity": "0.01"
        }"#;

        let order = parse_order_zero_alloc(order_json.as_bytes()).unwrap();
        assert_eq!(order.order_id.as_str(), "TL123");
        assert_eq!(order.symbol.as_str(), "BTC-USDT");
    }

    #[test]
    fn test_buffer_reuse() {
        let mut processor = ZeroAllocMessageProcessor::new();

        // Process first message
        let json1 = r#"{"id":1,"value":"test1"}"#;
        {
            let _result1 = processor
                .process_json_bytes_internal(json1.as_bytes())
                .unwrap();
            // _result1 drops here
        }

        // Process second message - should reuse buffer
        let json2 = r#"{"id":2,"value":"test2"}"#;
        {
            let _result2 = processor
                .process_json_bytes_internal(json2.as_bytes())
                .unwrap();
            // _result2 drops here
        }

        // Buffer should maintain capacity from previous use
        let stats = processor.buffer_stats();
        assert!(stats.stack_buffer_capacity >= json1.len().max(json2.len()));
    }

    #[test]
    fn test_message_type_identification() {
        let test_cases = vec![
            (r#"{"type":"heartbeat"}"#, MessageType::Heartbeat),
            (
                r#"{"type":"error","message":"Invalid request"}"#,
                MessageType::Error,
            ),
            (
                r#"{"type":"order","order_id":"123"}"#,
                MessageType::OrderUpdate,
            ),
            (r#"{"e":"executionReport"}"#, MessageType::OrderUpdate),
            (r#"{"type":"ticker"}"#, MessageType::MarketData),
            (r#"{"type":"snapshot"}"#, MessageType::OrderBook),
            (r#"{"e":"depthUpdate"}"#, MessageType::OrderBook),
            (r#"{"unknown":"field"}"#, MessageType::Unknown),
        ];

        let mut processor = ZeroAllocMessageProcessor::new();
        for (json, expected) in test_cases {
            let msg_type = processor.identify_message_type(json.as_bytes()).unwrap();
            assert_eq!(msg_type, expected, "Failed for JSON: {json}");
        }
    }
}