use bytes::{Buf, Bytes, BytesMut};
use tokio_util::codec::Decoder;

use crate::error::Error;
use crate::types::{ApiKey, GroupName, MessageId, NamespaceId, TopicName, TopicOffset};

use super::MessageCodec;
use super::frame::{
    ErrorCode, Frame, MessageType, ReconnectStatus, SubscriptionEntry, SubscriptionResult,
    SubscriptionStatus,
};
use crate::types::{Header, Headers};

/// Read a length-prefixed string (u16 BE length + UTF-8 bytes).
fn read_string(buf: &mut impl Buf) -> Result<String, Error> {
    if buf.remaining() < 2 {
        return Err(Error::Codec("truncated string length".into()));
    }
    let len = buf.get_u16() as usize;
    if buf.remaining() < len {
        return Err(Error::Codec(format!(
            "truncated string: expected {len} bytes, got {}",
            buf.remaining()
        )));
    }
    let mut data = vec![0u8; len];
    buf.copy_to_slice(&mut data);
    String::from_utf8(data).map_err(|e| Error::Codec(format!("invalid UTF-8 in string: {e}")))
}

/// Read a length-prefixed payload (u32 BE length + raw bytes) as zero-copy `Bytes`.
fn read_payload(buf: &mut BytesMut) -> Result<Bytes, Error> {
    if buf.remaining() < 4 {
        return Err(Error::Codec("truncated payload length".into()));
    }
    let len = buf.get_u32() as usize;
    if buf.remaining() < len {
        return Err(Error::Codec(format!(
            "truncated payload: expected {len} bytes, got {}",
            buf.remaining()
        )));
    }
    Ok(buf.split_to(len).freeze())
}

/// Read a group string field: empty string → None, non-empty → Some(GroupName).
fn read_group(buf: &mut impl Buf) -> Result<Option<GroupName>, Error> {
    let s = read_string(buf)?;
    if s.is_empty() {
        Ok(None)
    } else {
        Ok(Some(GroupName(s)))
    }
}

/// Read a header section: [u16 header_count] followed by each header as
/// [u16 key_len][key bytes][u32 value_len][value bytes].
fn read_headers(buf: &mut impl Buf) -> Result<Headers, Error> {
    if buf.remaining() < 2 {
        return Err(Error::Codec("truncated header count".into()));
    }
    let count = buf.get_u16() as usize;
    let mut headers = Vec::with_capacity(count.min(256));
    for _ in 0..count {
        if buf.remaining() < 2 {
            return Err(Error::Codec("truncated header key length".into()));
        }
        let key_len = buf.get_u16() as usize;
        if buf.remaining() < key_len {
            return Err(Error::Codec(format!(
                "truncated header key: expected {key_len} bytes, got {}",
                buf.remaining()
            )));
        }
        let mut key = vec![0u8; key_len];
        buf.copy_to_slice(&mut key);

        if buf.remaining() < 4 {
            return Err(Error::Codec("truncated header value length".into()));
        }
        let value_len = buf.get_u32() as usize;
        if buf.remaining() < value_len {
            return Err(Error::Codec(format!(
                "truncated header value: expected {value_len} bytes, got {}",
                buf.remaining()
            )));
        }
        let mut value = vec![0u8; value_len];
        buf.copy_to_slice(&mut value);

        headers.push(Header { key, value });
    }
    Ok(headers)
}

impl Decoder for MessageCodec {
    type Item = Frame;
    type Error = Error;

    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
        // Need at least 4 bytes for the frame length prefix.
        if src.len() < 4 {
            return Ok(None);
        }

        // Peek frame length via indexing (do NOT consume yet).
        let frame_length = u32::from_be_bytes([src[0], src[1], src[2], src[3]]) as usize;

        // Frame length 0 is invalid (minimum 1 byte for message type).
        if frame_length == 0 {
            tracing::debug!("rejecting frame with length 0");
            return Err(Error::Codec("frame length is 0".into()));
        }

        // Check max frame size.
        if frame_length > self.max_frame_size as usize {
            tracing::warn!(
                frame_length,
                max = self.max_frame_size,
                "rejecting oversized frame"
            );
            metrics::counter!("fafb_frame_rejected_total", "reason" => "oversized").increment(1);
            return Err(Error::Codec(format!(
                "frame too large: {frame_length} bytes (max {})",
                self.max_frame_size
            )));
        }

        // Check if we have the complete frame.
        let total = 4 + frame_length;
        if src.len() < total {
            src.reserve(total - src.len());
            return Ok(None);
        }

        // Consume the length prefix.
        let _ = src.split_to(4);

        // Isolate the frame body into a separate BytesMut.
        let mut body = src.split_to(frame_length);

        // Read message type byte.
        let type_byte = body.get_u8();
        let msg_type = MessageType::try_from(type_byte)?;

        // Parse payload fields per message type.
        let frame = match msg_type {
            MessageType::Connect => {
                if body.remaining() < 1 {
                    return Err(Error::Codec("truncated Connect: missing version".into()));
                }
                let version = body.get_u8();
                let namespace = read_string(&mut body)?;
                let api_key = read_string(&mut body)?;
                // Backward-compatible: parse capabilities if 4+ bytes remain,
                // otherwise default to 0 (legacy v1 client).
                let capabilities = if body.remaining() >= 4 {
                    body.get_u32()
                } else {
                    0
                };
                Frame::Connect {
                    version,
                    namespace: NamespaceId(namespace),
                    api_key: ApiKey(api_key),
                    capabilities,
                }
            }
            MessageType::ConnectAck => {
                if body.remaining() < 1 {
                    return Err(Error::Codec("truncated ConnectAck: missing status".into()));
                }
                let status = body.get_u8();
                let message = read_string(&mut body)?;
                // Backward-compatible: parse server_capabilities if 4+ bytes remain.
                let server_capabilities = if body.remaining() >= 4 {
                    body.get_u32()
                } else {
                    0
                };
                Frame::ConnectAck {
                    success: status == 0x00,
                    message,
                    server_capabilities,
                }
            }
            MessageType::Subscribe => {
                let topic = read_string(&mut body)?;
                let group = read_group(&mut body)?;
                Frame::Subscribe {
                    topic: TopicName(topic),
                    group,
                }
            }
            MessageType::SubscribeAck => {
                let topic = read_string(&mut body)?;
                if body.remaining() < 8 {
                    return Err(Error::Codec(
                        "truncated SubscribeAck: missing start_offset".into(),
                    ));
                }
                let start_offset = body.get_u64();
                let group = read_group(&mut body)?;
                // Backward compatible: matched_count only present when CAP_WILDCARD negotiated.
                let matched_count = if body.remaining() >= 2 {
                    body.get_u16()
                } else {
                    0
                };
                Frame::SubscribeAck {
                    topic: TopicName(topic),
                    group,
                    start_offset: TopicOffset(start_offset),
                    matched_count,
                }
            }
            MessageType::Unsubscribe => {
                let topic = read_string(&mut body)?;
                let group = read_group(&mut body)?;
                Frame::Unsubscribe {
                    topic: TopicName(topic),
                    group,
                }
            }
            MessageType::UnsubscribeAck => {
                let topic = read_string(&mut body)?;
                let group = read_group(&mut body)?;
                Frame::UnsubscribeAck {
                    topic: TopicName(topic),
                    group,
                }
            }
            MessageType::Publish => {
                let topic = read_string(&mut body)?;
                let ttl_ms = if self.ttl_enabled {
                    if body.remaining() < 4 {
                        return Err(Error::Codec("truncated Publish: missing ttl_ms".into()));
                    }
                    body.get_u32()
                } else {
                    0
                };
                let headers = if self.headers_enabled {
                    let h = read_headers(&mut body)?;
                    super::frame::validate_headers(&h, &self.header_limits)?;
                    h
                } else {
                    vec![]
                };
                let payload = read_payload(&mut body)?;
                Frame::Publish {
                    topic: TopicName(topic),
                    headers,
                    ttl_ms,
                    payload,
                }
            }
            MessageType::PublishAck => {
                if body.remaining() < 8 {
                    return Err(Error::Codec(
                        "truncated PublishAck: missing message_id".into(),
                    ));
                }
                let message_id = body.get_u64();
                Frame::PublishAck {
                    message_id: MessageId(message_id),
                }
            }
            MessageType::Deliver => {
                if body.remaining() < 24 {
                    return Err(Error::Codec(
                        "truncated Deliver: missing message_id/offset/timestamp".into(),
                    ));
                }
                let message_id = body.get_u64();
                let offset = body.get_u64();
                let timestamp_millis = body.get_u64();
                let ttl_ms = if self.ttl_enabled {
                    if body.remaining() < 4 {
                        return Err(Error::Codec("truncated Deliver: missing ttl_ms".into()));
                    }
                    body.get_u32()
                } else {
                    0
                };
                let topic = read_string(&mut body)?;
                let headers = if self.headers_enabled {
                    read_headers(&mut body)?
                } else {
                    vec![]
                };
                let payload = read_payload(&mut body)?;
                Frame::Deliver {
                    message_id: MessageId(message_id),
                    offset: TopicOffset(offset),
                    timestamp_millis,
                    ttl_ms,
                    topic: TopicName(topic),
                    headers,
                    payload,
                }
            }
            MessageType::Ack => {
                if body.remaining() < 8 {
                    return Err(Error::Codec("truncated Ack: missing message_id".into()));
                }
                let message_id = body.get_u64();
                Frame::Ack {
                    message_id: MessageId(message_id),
                }
            }
            MessageType::Error => {
                if body.remaining() < 2 {
                    return Err(Error::Codec("truncated Error: missing error code".into()));
                }
                let code = body.get_u16();
                let code = ErrorCode::try_from(code)?;
                let message = read_string(&mut body)?;
                Frame::Error { code, message }
            }
            MessageType::SubscribeAt => {
                let topic = read_string(&mut body)?;
                if body.remaining() < 8 {
                    return Err(Error::Codec("truncated SubscribeAt: missing offset".into()));
                }
                let offset = body.get_u64();
                let group = read_group(&mut body)?;
                Frame::SubscribeAt {
                    topic: TopicName(topic),
                    offset: TopicOffset(offset),
                    group,
                }
            }
            MessageType::SubscribeFrom => {
                let topic = read_string(&mut body)?;
                if body.remaining() < 8 {
                    return Err(Error::Codec(
                        "truncated SubscribeFrom: missing timestamp".into(),
                    ));
                }
                let timestamp_millis = body.get_u64();
                let group = read_group(&mut body)?;
                Frame::SubscribeFrom {
                    topic: TopicName(topic),
                    timestamp_millis,
                    group,
                }
            }
            MessageType::GoingAway => {
                if body.remaining() < 5 {
                    return Err(Error::Codec(
                        "truncated GoingAway: missing reason/deadline".into(),
                    ));
                }
                let reason_byte = body.get_u8();
                let deadline_ms = body.get_u32();
                let message = read_string(&mut body)?;
                Frame::GoingAway {
                    reason: super::frame::GoingAwayReason(reason_byte),
                    deadline_ms,
                    message,
                }
            }
            MessageType::Reconnect => {
                if body.remaining() < 1 {
                    return Err(Error::Codec("truncated Reconnect: missing version".into()));
                }
                let version = body.get_u8();
                let namespace = read_string(&mut body)?;
                let api_key = read_string(&mut body)?;
                if body.remaining() < 4 {
                    return Err(Error::Codec(
                        "truncated Reconnect: missing capabilities".into(),
                    ));
                }
                let capabilities = body.get_u32();
                if body.remaining() < 2 {
                    return Err(Error::Codec(
                        "truncated Reconnect: missing subscription_count".into(),
                    ));
                }
                let sub_count = body.get_u16() as usize;
                let mut subscriptions = Vec::with_capacity(sub_count.min(1024));
                for _ in 0..sub_count {
                    let topic = read_string(&mut body)?;
                    let group = read_group(&mut body)?;
                    if body.remaining() < 8 {
                        return Err(Error::Codec(
                            "truncated Reconnect: missing last_acked_offset".into(),
                        ));
                    }
                    let last_acked_offset = body.get_u64();
                    subscriptions.push(SubscriptionEntry {
                        topic: TopicName(topic),
                        group,
                        last_acked_offset: TopicOffset(last_acked_offset),
                    });
                }
                Frame::Reconnect {
                    version,
                    namespace: NamespaceId(namespace),
                    api_key: ApiKey(api_key),
                    capabilities,
                    subscriptions,
                }
            }
            MessageType::ReconnectAck => {
                if body.remaining() < 1 {
                    return Err(Error::Codec(
                        "truncated ReconnectAck: missing status".into(),
                    ));
                }
                let status = ReconnectStatus::try_from(body.get_u8())?;
                let message = read_string(&mut body)?;
                if body.remaining() < 4 {
                    return Err(Error::Codec(
                        "truncated ReconnectAck: missing server_capabilities".into(),
                    ));
                }
                let server_capabilities = body.get_u32();
                if body.remaining() < 2 {
                    return Err(Error::Codec(
                        "truncated ReconnectAck: missing result_count".into(),
                    ));
                }
                let result_count = body.get_u16() as usize;
                let mut results = Vec::with_capacity(result_count.min(1024));
                for _ in 0..result_count {
                    let topic = read_string(&mut body)?;
                    let group = read_group(&mut body)?;
                    if body.remaining() < 1 {
                        return Err(Error::Codec(
                            "truncated ReconnectAck: missing sub_status".into(),
                        ));
                    }
                    let sub_status = SubscriptionStatus::try_from(body.get_u8())?;
                    if body.remaining() < 8 {
                        return Err(Error::Codec(
                            "truncated ReconnectAck: missing actual_start_offset".into(),
                        ));
                    }
                    let actual_start_offset = body.get_u64();
                    results.push(SubscriptionResult {
                        topic: TopicName(topic),
                        group,
                        status: sub_status,
                        actual_start_offset: TopicOffset(actual_start_offset),
                    });
                }
                Frame::ReconnectAck {
                    status,
                    message,
                    server_capabilities,
                    results,
                }
            }
            MessageType::Fetch => {
                let topic = read_string(&mut body)?;
                if topic.is_empty() {
                    return Err(Error::Codec("Fetch: topic must not be empty".into()));
                }
                if body.remaining() < 12 {
                    return Err(Error::Codec(
                        "truncated Fetch: missing offset/max_count".into(),
                    ));
                }
                let offset = body.get_u64();
                let max_count = body.get_u32();
                if max_count == 0 {
                    return Err(Error::Codec("Fetch: max_count must be > 0".into()));
                }
                Frame::Fetch {
                    topic: TopicName(topic),
                    offset: TopicOffset(offset),
                    max_count,
                }
            }
            MessageType::FetchResponse => {
                let topic = read_string(&mut body)?;
                if body.remaining() < 20 {
                    return Err(Error::Codec(
                        "truncated FetchResponse: missing next_offset/head_offset/count".into(),
                    ));
                }
                let next_offset = body.get_u64();
                let head_offset = body.get_u64();
                let count = body.get_u32() as usize;
                let mut messages = Vec::with_capacity(count.min(1024));
                for _ in 0..count {
                    if body.remaining() < 16 {
                        return Err(Error::Codec(
                            "truncated FetchResponse: incomplete message entry".into(),
                        ));
                    }
                    let offset = body.get_u64();
                    let timestamp_millis = body.get_u64();
                    let ttl_ms = if self.ttl_enabled {
                        if body.remaining() < 4 {
                            return Err(Error::Codec(
                                "truncated FetchResponse: missing ttl_ms".into(),
                            ));
                        }
                        body.get_u32()
                    } else {
                        0
                    };
                    let headers = if self.headers_enabled {
                        read_headers(&mut body)?
                    } else {
                        vec![]
                    };
                    let payload = read_payload(&mut body)?;
                    messages.push(super::frame::FetchedMessage {
                        offset: TopicOffset(offset),
                        timestamp_millis,
                        ttl_ms,
                        headers,
                        payload,
                    });
                }
                Frame::FetchResponse {
                    topic: TopicName(topic),
                    next_offset: TopicOffset(next_offset),
                    head_offset: TopicOffset(head_offset),
                    messages,
                }
            }
            MessageType::FetchFrom => {
                let topic = read_string(&mut body)?;
                if topic.is_empty() {
                    return Err(Error::Codec("FetchFrom: topic must not be empty".into()));
                }
                if body.remaining() < 12 {
                    return Err(Error::Codec(
                        "truncated FetchFrom: missing timestamp/max_count".into(),
                    ));
                }
                let timestamp_millis = body.get_u64();
                let max_count = body.get_u32();
                if max_count == 0 {
                    return Err(Error::Codec("FetchFrom: max_count must be > 0".into()));
                }
                Frame::FetchFrom {
                    topic: TopicName(topic),
                    timestamp_millis,
                    max_count,
                }
            }
            MessageType::Nack => {
                if body.remaining() < 8 {
                    return Err(Error::Codec("truncated Nack: missing message_id".into()));
                }
                let message_id = body.get_u64();
                let reason = read_string(&mut body)?;
                // Defense in depth: truncate reason to 1024 bytes max even if
                // a non-standard client sends a longer string. The encoder does
                // not truncate — it relies on the decoder to enforce the limit.
                let reason = if reason.len() > 1024 {
                    reason[..1024].to_string()
                } else {
                    reason
                };
                Frame::Nack {
                    message_id: MessageId(message_id),
                    reason,
                }
            }
            MessageType::NackAck => {
                if body.remaining() < 9 {
                    return Err(Error::Codec(
                        "truncated NackAck: missing message_id/status".into(),
                    ));
                }
                let message_id = body.get_u64();
                let status = body.get_u8();
                Frame::NackAck {
                    message_id: MessageId(message_id),
                    status,
                }
            }
            MessageType::BatchAck => {
                if body.remaining() < 4 {
                    return Err(Error::Codec("truncated BatchAck: missing count".into()));
                }
                let count = body.get_u32() as usize;
                if count > 10_000 {
                    return Err(Error::Codec(format!(
                        "BatchAck count too large: {count} (max 10000)"
                    )));
                }
                if body.remaining() < count * 8 {
                    return Err(Error::Codec(format!(
                        "truncated BatchAck: expected {count} message_ids"
                    )));
                }
                let mut message_ids = Vec::with_capacity(count);
                for _ in 0..count {
                    message_ids.push(MessageId(body.get_u64()));
                }
                Frame::BatchAck { message_ids }
            }
            MessageType::BatchAckResponse => {
                if body.remaining() < 8 {
                    return Err(Error::Codec(
                        "truncated BatchAckResponse: missing counts".into(),
                    ));
                }
                let acked_count = body.get_u32();
                let unknown_count = body.get_u32();
                Frame::BatchAckResponse {
                    acked_count,
                    unknown_count,
                }
            }
            MessageType::NotLeader => {
                let leader_addr = read_string(&mut body)?;
                Frame::NotLeader { leader_addr }
            }
            MessageType::ClusterUnavailable => {
                if body.remaining() < 4 {
                    return Err(Error::Codec(
                        "truncated ClusterUnavailable: missing retry_after_ms".into(),
                    ));
                }
                let retry_after_ms = body.get_u32();
                let message = read_string(&mut body)?;
                Frame::ClusterUnavailable {
                    retry_after_ms,
                    message,
                }
            }
            MessageType::Ping => Frame::Ping,
            MessageType::Pong => Frame::Pong,
        };

        // Strict: reject trailing bytes in frame body.
        if body.has_remaining() {
            let extra = body.remaining();
            tracing::debug!(
                frame_type = ?msg_type,
                trailing_bytes = extra,
                "rejecting frame with trailing bytes"
            );
            return Err(Error::Codec(format!(
                "trailing bytes in frame: {extra} extra byte(s)",
            )));
        }

        Ok(Some(frame))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::codec::frame::FetchedMessage;
    use bytes::BufMut;
    use tokio_util::codec::Encoder;

    fn make_codec() -> MessageCodec {
        MessageCodec::new(1_048_576)
    }

    /// Encode a frame and return the raw bytes.
    fn encode_frame(frame: Frame) -> BytesMut {
        let mut codec = make_codec();
        let mut buf = BytesMut::new();
        codec.encode(frame, &mut buf).unwrap();
        buf
    }

    // --- Decoding tests ---

    #[test]
    fn decode_ping() {
        let mut buf = encode_frame(Frame::Ping);
        let frame = make_codec().decode(&mut buf).unwrap().unwrap();
        assert_eq!(frame, Frame::Ping);
    }

    #[test]
    fn decode_pong() {
        let mut buf = encode_frame(Frame::Pong);
        let frame = make_codec().decode(&mut buf).unwrap().unwrap();
        assert_eq!(frame, Frame::Pong);
    }

    #[test]
    fn decode_connect() {
        let original = Frame::Connect {
            version: 1,
            namespace: NamespaceId("my-ns".into()),
            api_key: ApiKey("secret-key-123".into()),
            capabilities: 0,
        };
        let mut buf = encode_frame(original.clone());
        let decoded = make_codec().decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, original);
    }

    #[test]
    fn decode_connect_ack_success() {
        let original = Frame::ConnectAck {
            success: true,
            message: String::new(),
            server_capabilities: 0,
        };
        let mut buf = encode_frame(original.clone());
        let decoded = make_codec().decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, original);
    }

    #[test]
    fn decode_connect_ack_failure() {
        let original = Frame::ConnectAck {
            success: false,
            message: "invalid credentials".into(),
            server_capabilities: 0,
        };
        let mut buf = encode_frame(original.clone());
        let decoded = make_codec().decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, original);
    }

    #[test]
    fn decode_subscribe_with_group() {
        let original = Frame::Subscribe {
            topic: TopicName("events".into()),
            group: Some(GroupName("workers".into())),
        };
        let mut buf = encode_frame(original.clone());
        let decoded = make_codec().decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, original);
    }

    #[test]
    fn decode_subscribe_without_group() {
        let original = Frame::Subscribe {
            topic: TopicName("events".into()),
            group: None,
        };
        let mut buf = encode_frame(original.clone());
        let decoded = make_codec().decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, original);
    }

    #[test]
    fn decode_publish() {
        let original = Frame::Publish {
            topic: TopicName("logs".into()),
            payload: Bytes::from_static(b"log entry data"),
            headers: vec![],
            ttl_ms: 0,
        };
        let mut buf = encode_frame(original.clone());
        let decoded = make_codec().decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, original);
    }

    #[test]
    fn decode_publish_empty_payload() {
        let original = Frame::Publish {
            topic: TopicName("t".into()),
            payload: Bytes::new(),
            headers: vec![],
            ttl_ms: 0,
        };
        let mut buf = encode_frame(original.clone());
        let decoded = make_codec().decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, original);
    }

    #[test]
    fn decode_publish_ack() {
        let original = Frame::PublishAck {
            message_id: MessageId(12345),
        };
        let mut buf = encode_frame(original.clone());
        let decoded = make_codec().decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, original);
    }

    #[test]
    fn decode_deliver() {
        let original = Frame::Deliver {
            message_id: MessageId(999),
            offset: TopicOffset(50),
            timestamp_millis: 1_700_000_000_000,
            ttl_ms: 0,
            topic: TopicName("events".into()),
            payload: Bytes::from_static(b"message body"),
            headers: vec![],
        };
        let mut buf = encode_frame(original.clone());
        let decoded = make_codec().decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, original);
    }

    #[test]
    fn decode_ack() {
        let original = Frame::Ack {
            message_id: MessageId(42),
        };
        let mut buf = encode_frame(original.clone());
        let decoded = make_codec().decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, original);
    }

    #[test]
    fn decode_error() {
        let original = Frame::Error {
            code: ErrorCode::PermissionDenied,
            message: "not allowed".into(),
        };
        let mut buf = encode_frame(original.clone());
        let decoded = make_codec().decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, original);
    }

    // --- Partial buffer tests ---

    #[test]
    fn partial_too_few_bytes() {
        let mut codec = make_codec();
        // 0 bytes
        let mut buf = BytesMut::new();
        assert!(codec.decode(&mut buf).unwrap().is_none());
        // 1 byte
        buf.put_u8(0);
        assert!(codec.decode(&mut buf).unwrap().is_none());
        // 3 bytes
        buf.put_u8(0);
        buf.put_u8(0);
        assert!(codec.decode(&mut buf).unwrap().is_none());
    }

    #[test]
    fn partial_length_present_body_incomplete() {
        let mut codec = make_codec();
        let mut buf = BytesMut::new();
        // Frame length says 10, but we only provide the length prefix + 5 bytes
        buf.put_u32(10);
        buf.put_slice(&[0x0C, 0, 0, 0, 0]);
        assert!(codec.decode(&mut buf).unwrap().is_none());
        // Buffer should be untouched for retry
        assert_eq!(buf.len(), 9);
    }

    #[test]
    fn two_complete_frames() {
        let mut codec = make_codec();
        let mut buf = BytesMut::new();
        // Encode two frames into the same buffer
        codec.encode(Frame::Ping, &mut buf).unwrap();
        codec.encode(Frame::Pong, &mut buf).unwrap();

        let first = codec.decode(&mut buf).unwrap().unwrap();
        assert_eq!(first, Frame::Ping);

        let second = codec.decode(&mut buf).unwrap().unwrap();
        assert_eq!(second, Frame::Pong);

        assert!(buf.is_empty());
    }

    #[test]
    fn complete_plus_partial() {
        let mut codec = make_codec();
        let mut buf = BytesMut::new();
        codec.encode(Frame::Ping, &mut buf).unwrap();
        // Add partial frame: length prefix but no body
        buf.put_u32(5);
        buf.put_u8(0x0D);

        let first = codec.decode(&mut buf).unwrap().unwrap();
        assert_eq!(first, Frame::Ping);

        // Second decode should return None (incomplete)
        assert!(codec.decode(&mut buf).unwrap().is_none());
    }

    // --- Malformed input tests ---

    #[test]
    fn malformed_frame_length_zero() {
        let mut codec = make_codec();
        let mut buf = BytesMut::new();
        buf.put_u32(0);
        assert!(codec.decode(&mut buf).is_err());
    }

    #[test]
    fn malformed_frame_too_large() {
        let mut codec = MessageCodec::new(100);
        let mut buf = BytesMut::new();
        buf.put_u32(200); // exceeds max
        buf.put_slice(&vec![0u8; 200]);
        assert!(codec.decode(&mut buf).is_err());
    }

    #[test]
    fn malformed_unknown_message_type() {
        let mut codec = make_codec();
        let mut buf = BytesMut::new();
        buf.put_u32(1); // frame length = 1
        buf.put_u8(0xFF); // unknown type
        assert!(codec.decode(&mut buf).is_err());
    }

    #[test]
    fn malformed_non_utf8_string() {
        let mut codec = make_codec();
        let mut buf = BytesMut::new();
        // Build a Connect frame with invalid UTF-8 in namespace
        let body_len = 1 + 1 + 2 + 2 + 2 + 3; // type + ver + ns_len + ns_bytes(invalid) + key_len + key
        buf.put_u32(body_len as u32);
        buf.put_u8(0x01); // Connect
        buf.put_u8(1); // version
        buf.put_u16(2); // namespace length
        buf.put_slice(&[0xFF, 0xFE]); // invalid UTF-8
        buf.put_u16(3); // key length
        buf.put_slice(b"key");
        assert!(codec.decode(&mut buf).is_err());
    }

    #[test]
    fn malformed_truncated_string() {
        let mut codec = make_codec();
        let mut buf = BytesMut::new();
        // Connect frame where namespace claims 100 bytes but frame ends
        let body_len = 1 + 1 + 2; // type + ver + ns_len (but no ns data)
        buf.put_u32(body_len as u32);
        buf.put_u8(0x01); // Connect
        buf.put_u8(1); // version
        buf.put_u16(100); // claims 100 bytes
        assert!(codec.decode(&mut buf).is_err());
    }

    #[test]
    fn malformed_truncated_payload() {
        let mut codec = make_codec();
        let mut buf = BytesMut::new();
        // Publish frame with topic but payload length claims more than available
        let body_len = 1 + 2 + 1 + 4; // type + topic_len + topic("t") + payload_len (no payload data)
        buf.put_u32(body_len as u32);
        buf.put_u8(0x07); // Publish
        buf.put_u16(1);
        buf.put_u8(b't');
        buf.put_u32(999); // claims 999 payload bytes
        assert!(codec.decode(&mut buf).is_err());
    }

    #[test]
    fn malformed_trailing_bytes() {
        let mut codec = make_codec();
        let mut buf = BytesMut::new();
        // Ping frame with extra trailing bytes
        buf.put_u32(3); // frame length = 3 (type byte + 2 extra)
        buf.put_u8(0x0C); // Ping
        buf.put_u8(0xAA); // trailing
        buf.put_u8(0xBB); // trailing
        assert!(codec.decode(&mut buf).is_err());
    }

    // --- Roundtrip tests ---

    fn roundtrip(frame: Frame) {
        let mut codec = make_codec();
        let mut buf = BytesMut::new();
        codec.encode(frame.clone(), &mut buf).unwrap();
        let decoded = codec.decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, frame);
        assert!(buf.is_empty());
    }

    #[test]
    fn roundtrip_connect() {
        roundtrip(Frame::Connect {
            version: 1,
            namespace: NamespaceId("production".into()),
            api_key: ApiKey("sk-abcdef1234567890".into()),
            capabilities: 0,
        });
    }

    #[test]
    fn roundtrip_connect_ack_success() {
        roundtrip(Frame::ConnectAck {
            success: true,
            message: String::new(),
            server_capabilities: 0,
        });
    }

    #[test]
    fn roundtrip_connect_ack_failure() {
        roundtrip(Frame::ConnectAck {
            success: false,
            message: "auth failed".into(),
            server_capabilities: 0,
        });
    }

    #[test]
    fn roundtrip_subscribe_with_group() {
        roundtrip(Frame::Subscribe {
            topic: TopicName("events.user".into()),
            group: Some(GroupName("worker-pool".into())),
        });
    }

    #[test]
    fn roundtrip_subscribe_without_group() {
        roundtrip(Frame::Subscribe {
            topic: TopicName("events.user".into()),
            group: None,
        });
    }

    #[test]
    fn roundtrip_subscribe_ack() {
        roundtrip(Frame::SubscribeAck {
            topic: TopicName("t".into()),
            group: Some(GroupName("g".into())),
            start_offset: TopicOffset(42),
            matched_count: 0,
        });
    }

    #[test]
    fn roundtrip_subscribe_ack_zero_offset() {
        roundtrip(Frame::SubscribeAck {
            topic: TopicName("t".into()),
            group: None,
            start_offset: TopicOffset(0),
            matched_count: 0,
        });
    }

    #[test]
    fn roundtrip_subscribe_at_with_group() {
        roundtrip(Frame::SubscribeAt {
            topic: TopicName("events".into()),
            offset: TopicOffset(42),
            group: Some(GroupName("workers".into())),
        });
    }

    #[test]
    fn roundtrip_subscribe_at_no_group() {
        roundtrip(Frame::SubscribeAt {
            topic: TopicName("events".into()),
            offset: TopicOffset(0),
            group: None,
        });
    }

    #[test]
    fn roundtrip_subscribe_from_with_group() {
        roundtrip(Frame::SubscribeFrom {
            topic: TopicName("events".into()),
            timestamp_millis: 1_700_000_000_000,
            group: Some(GroupName("workers".into())),
        });
    }

    #[test]
    fn roundtrip_subscribe_from_no_group() {
        roundtrip(Frame::SubscribeFrom {
            topic: TopicName("logs".into()),
            timestamp_millis: 0,
            group: None,
        });
    }

    #[test]
    fn roundtrip_unsubscribe() {
        roundtrip(Frame::Unsubscribe {
            topic: TopicName("t".into()),
            group: None,
        });
    }

    #[test]
    fn roundtrip_unsubscribe_ack() {
        roundtrip(Frame::UnsubscribeAck {
            topic: TopicName("t".into()),
            group: Some(GroupName("g".into())),
        });
    }

    #[test]
    fn roundtrip_publish() {
        roundtrip(Frame::Publish {
            topic: TopicName("logs".into()),
            payload: Bytes::from_static(b"some log data here"),
            headers: vec![],
            ttl_ms: 0,
        });
    }

    #[test]
    fn roundtrip_publish_empty_payload() {
        roundtrip(Frame::Publish {
            topic: TopicName("t".into()),
            payload: Bytes::new(),
            headers: vec![],
            ttl_ms: 0,
        });
    }

    #[test]
    fn roundtrip_publish_ack() {
        roundtrip(Frame::PublishAck {
            message_id: MessageId(1234567890),
        });
    }

    #[test]
    fn roundtrip_deliver() {
        roundtrip(Frame::Deliver {
            message_id: MessageId(42),
            offset: TopicOffset(100),
            timestamp_millis: 1_700_000_000_000,
            ttl_ms: 0,
            topic: TopicName("events".into()),
            payload: Bytes::from(vec![1, 2, 3, 4, 5]),
            headers: vec![],
        });
    }

    #[test]
    fn roundtrip_ack() {
        roundtrip(Frame::Ack {
            message_id: MessageId(99),
        });
    }

    #[test]
    fn roundtrip_error() {
        roundtrip(Frame::Error {
            code: ErrorCode::InternalError,
            message: "something went wrong".into(),
        });
    }

    #[test]
    fn roundtrip_ping() {
        roundtrip(Frame::Ping);
    }

    #[test]
    fn roundtrip_pong() {
        roundtrip(Frame::Pong);
    }

    #[test]
    fn roundtrip_going_away_upgrade() {
        roundtrip(Frame::GoingAway {
            reason: crate::codec::frame::GoingAwayReason::UPGRADE,
            deadline_ms: 30000,
            message: "deploying v2.1.0".into(),
        });
    }

    #[test]
    fn roundtrip_going_away_maintenance() {
        roundtrip(Frame::GoingAway {
            reason: crate::codec::frame::GoingAwayReason::MAINTENANCE,
            deadline_ms: 0,
            message: String::new(),
        });
    }

    #[test]
    fn roundtrip_going_away_shutdown() {
        roundtrip(Frame::GoingAway {
            reason: crate::codec::frame::GoingAwayReason::SHUTDOWN,
            deadline_ms: u32::MAX,
            message: "permanent shutdown".into(),
        });
    }

    #[test]
    fn roundtrip_going_away_unknown_reason() {
        // Forward compatibility: unknown reason codes are accepted, never rejected.
        roundtrip(Frame::GoingAway {
            reason: crate::codec::frame::GoingAwayReason(0xFF),
            deadline_ms: 5000,
            message: "unknown reason".into(),
        });
    }

    #[test]
    fn roundtrip_reconnect_empty_subscriptions() {
        roundtrip(Frame::Reconnect {
            version: 1,
            namespace: NamespaceId("production".into()),
            api_key: ApiKey("sk-key-123456".into()),
            capabilities: 0x03,
            subscriptions: vec![],
        });
    }

    #[test]
    fn roundtrip_reconnect_with_subscriptions() {
        use crate::codec::frame::SubscriptionEntry;
        roundtrip(Frame::Reconnect {
            version: 1,
            namespace: NamespaceId("ns".into()),
            api_key: ApiKey("key".into()),
            capabilities: 0x03,
            subscriptions: vec![
                SubscriptionEntry {
                    topic: TopicName("events".into()),
                    group: None,
                    last_acked_offset: TopicOffset(499),
                },
                SubscriptionEntry {
                    topic: TopicName("logs".into()),
                    group: Some(GroupName("workers".into())),
                    last_acked_offset: TopicOffset(1234),
                },
            ],
        });
    }

    #[test]
    fn roundtrip_reconnect_ack_success() {
        use crate::codec::frame::{ReconnectStatus, SubscriptionResult, SubscriptionStatus};
        roundtrip(Frame::ReconnectAck {
            status: ReconnectStatus::Success,
            message: String::new(),
            server_capabilities: 0x03,
            results: vec![
                SubscriptionResult {
                    topic: TopicName("events".into()),
                    group: None,
                    status: SubscriptionStatus::Restored,
                    actual_start_offset: TopicOffset(500),
                },
                SubscriptionResult {
                    topic: TopicName("logs".into()),
                    group: Some(GroupName("workers".into())),
                    status: SubscriptionStatus::OffsetClamped,
                    actual_start_offset: TopicOffset(100),
                },
            ],
        });
    }

    #[test]
    fn roundtrip_reconnect_ack_partial() {
        use crate::codec::frame::{ReconnectStatus, SubscriptionResult, SubscriptionStatus};
        roundtrip(Frame::ReconnectAck {
            status: ReconnectStatus::Partial,
            message: "some topics failed".into(),
            server_capabilities: 0x01,
            results: vec![
                SubscriptionResult {
                    topic: TopicName("events".into()),
                    group: None,
                    status: SubscriptionStatus::Restored,
                    actual_start_offset: TopicOffset(500),
                },
                SubscriptionResult {
                    topic: TopicName("deleted-topic".into()),
                    group: None,
                    status: SubscriptionStatus::TopicNotFound,
                    actual_start_offset: TopicOffset(0),
                },
            ],
        });
    }

    #[test]
    fn roundtrip_reconnect_ack_auth_failed() {
        use crate::codec::frame::ReconnectStatus;
        roundtrip(Frame::ReconnectAck {
            status: ReconnectStatus::AuthFailed,
            message: "invalid api key".into(),
            server_capabilities: 0,
            results: vec![],
        });
    }

    // --- Fetch roundtrip tests ---

    #[test]
    fn encode_decode_fetch_roundtrip() {
        roundtrip(Frame::Fetch {
            topic: TopicName("events".into()),
            offset: TopicOffset(42),
            max_count: 100,
        });
    }

    #[test]
    fn encode_decode_fetch_from_roundtrip() {
        roundtrip(Frame::FetchFrom {
            topic: TopicName("events".into()),
            timestamp_millis: 1_700_000_000_000,
            max_count: 50,
        });
    }

    #[test]
    fn encode_decode_fetch_response_empty() {
        let frame = Frame::FetchResponse {
            topic: TopicName("events".into()),
            next_offset: TopicOffset(10),
            head_offset: TopicOffset(10),
            messages: vec![],
        };
        roundtrip(frame);
    }

    #[test]
    fn encode_decode_fetch_response_single() {
        use crate::codec::frame::FetchedMessage;
        roundtrip(Frame::FetchResponse {
            topic: TopicName("events".into()),
            next_offset: TopicOffset(1),
            head_offset: TopicOffset(10),
            messages: vec![FetchedMessage {
                offset: TopicOffset(0),
                timestamp_millis: 1_700_000_000_000,
                ttl_ms: 0,
                payload: Bytes::from_static(b"hello"),
                headers: vec![],
            }],
        });
    }

    #[test]
    fn encode_decode_fetch_response_multi() {
        use crate::codec::frame::FetchedMessage;
        let messages = vec![
            FetchedMessage {
                offset: TopicOffset(5),
                timestamp_millis: 1_700_000_000_000,
                ttl_ms: 0,
                payload: Bytes::from_static(b"first"),
                headers: vec![],
            },
            FetchedMessage {
                offset: TopicOffset(6),
                timestamp_millis: 1_700_000_000_100,
                ttl_ms: 0,
                payload: Bytes::from_static(b"second"),
                headers: vec![],
            },
            FetchedMessage {
                offset: TopicOffset(7),
                timestamp_millis: 1_700_000_000_200,
                ttl_ms: 0,
                payload: Bytes::from_static(b"third"),
                headers: vec![],
            },
        ];
        let frame = Frame::FetchResponse {
            topic: TopicName("events".into()),
            next_offset: TopicOffset(8),
            head_offset: TopicOffset(42),
            messages,
        };
        let mut codec = make_codec();
        let mut buf = BytesMut::new();
        codec.encode(frame.clone(), &mut buf).unwrap();
        let decoded = codec.decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, frame);
        if let Frame::FetchResponse { messages, .. } = &decoded {
            assert_eq!(messages.len(), 3);
            assert_eq!(messages[0].offset, TopicOffset(5));
            assert_eq!(messages[1].offset, TopicOffset(6));
            assert_eq!(messages[2].offset, TopicOffset(7));
            assert_eq!(messages[0].payload, Bytes::from_static(b"first"));
            assert_eq!(messages[1].payload, Bytes::from_static(b"second"));
            assert_eq!(messages[2].payload, Bytes::from_static(b"third"));
        } else {
            panic!("expected FetchResponse");
        }
    }

    #[test]
    fn encode_decode_fetch_response_head_offset() {
        use crate::codec::frame::FetchedMessage;
        let frame = Frame::FetchResponse {
            topic: TopicName("t".into()),
            next_offset: TopicOffset(5),
            head_offset: TopicOffset(1000),
            messages: vec![FetchedMessage {
                offset: TopicOffset(4),
                timestamp_millis: 100,
                ttl_ms: 0,
                payload: Bytes::from_static(b"x"),
                headers: vec![],
            }],
        };
        let mut codec = make_codec();
        let mut buf = BytesMut::new();
        codec.encode(frame, &mut buf).unwrap();
        let decoded = codec.decode(&mut buf).unwrap().unwrap();
        if let Frame::FetchResponse {
            next_offset,
            head_offset,
            ..
        } = decoded
        {
            assert_eq!(next_offset, TopicOffset(5));
            assert_eq!(head_offset, TopicOffset(1000));
        } else {
            panic!("expected FetchResponse");
        }
    }

    #[test]
    fn decode_fetch_zero_max_count() {
        let mut codec = make_codec();
        let mut buf = BytesMut::new();
        // Build a Fetch frame with max_count=0 manually
        let topic = b"events";
        let body_len = 1 + 2 + topic.len() + 8 + 4; // type + topic_len + topic + offset + max_count
        buf.put_u32(body_len as u32);
        buf.put_u8(0x13); // Fetch
        buf.put_u16(topic.len() as u16);
        buf.put_slice(topic);
        buf.put_u64(0); // offset
        buf.put_u32(0); // max_count = 0
        assert!(codec.decode(&mut buf).is_err());
    }

    #[test]
    fn decode_fetch_from_zero_max_count() {
        let mut codec = make_codec();
        let mut buf = BytesMut::new();
        let topic = b"events";
        let body_len = 1 + 2 + topic.len() + 8 + 4;
        buf.put_u32(body_len as u32);
        buf.put_u8(0x15); // FetchFrom
        buf.put_u16(topic.len() as u16);
        buf.put_slice(topic);
        buf.put_u64(1_000_000); // timestamp
        buf.put_u32(0); // max_count = 0
        assert!(codec.decode(&mut buf).is_err());
    }

    #[test]
    fn decode_fetch_empty_topic() {
        let mut codec = make_codec();
        let mut buf = BytesMut::new();
        let body_len = 1 + 2 + 0 + 8 + 4; // type + empty topic + offset + max_count
        buf.put_u32(body_len as u32);
        buf.put_u8(0x13); // Fetch
        buf.put_u16(0); // empty topic
        buf.put_u64(0);
        buf.put_u32(100);
        assert!(codec.decode(&mut buf).is_err());
    }

    #[test]
    fn decode_fetch_from_empty_topic() {
        let mut codec = make_codec();
        let mut buf = BytesMut::new();
        let body_len = 1 + 2 + 0 + 8 + 4;
        buf.put_u32(body_len as u32);
        buf.put_u8(0x15); // FetchFrom
        buf.put_u16(0); // empty topic
        buf.put_u64(1_000_000);
        buf.put_u32(100);
        assert!(codec.decode(&mut buf).is_err());
    }

    #[test]
    fn decode_fetch_trailing_bytes() {
        let mut codec = make_codec();
        let mut buf = BytesMut::new();
        let topic = b"t";
        // Normal Fetch body + 1 extra trailing byte
        let body_len = 1 + 2 + topic.len() + 8 + 4 + 1;
        buf.put_u32(body_len as u32);
        buf.put_u8(0x13); // Fetch
        buf.put_u16(topic.len() as u16);
        buf.put_slice(topic);
        buf.put_u64(0);
        buf.put_u32(100);
        buf.put_u8(0xFF); // trailing
        assert!(codec.decode(&mut buf).is_err());
    }

    #[test]
    fn decode_fetch_response_truncated() {
        let mut codec = make_codec();
        let mut buf = BytesMut::new();
        let topic = b"t";
        // FetchResponse claiming 1 message but with incomplete message entry
        let body_len = 1 + 2 + topic.len() + 8 + 8 + 4 + 4; // type + topic + next + head + count + 4 bytes (incomplete entry)
        buf.put_u32(body_len as u32);
        buf.put_u8(0x14); // FetchResponse
        buf.put_u16(topic.len() as u16);
        buf.put_slice(topic);
        buf.put_u64(1); // next_offset
        buf.put_u64(10); // head_offset
        buf.put_u32(1); // count = 1 message
        buf.put_u32(0); // only 4 bytes of what should be a 20+ byte entry
        assert!(codec.decode(&mut buf).is_err());
    }

    #[test]
    fn error_code_message_too_large_non_fatal() {
        assert!(!ErrorCode::MessageTooLarge.is_fatal());
        // Also verify roundtrip through u16
        let code: u16 = ErrorCode::MessageTooLarge.into();
        assert_eq!(code, 0x000E);
        let back = ErrorCode::try_from(code).unwrap();
        assert_eq!(back, ErrorCode::MessageTooLarge);
    }

    #[test]
    fn roundtrip_max_length_string() {
        let long = "x".repeat(65535);
        roundtrip(Frame::ConnectAck {
            success: true,
            message: long,
            server_capabilities: 0,
        });
    }

    #[test]
    fn roundtrip_nack() {
        roundtrip(Frame::Nack {
            message_id: MessageId(42),
            reason: "invalid JSON".into(),
        });
    }

    #[test]
    fn roundtrip_nack_empty_reason() {
        roundtrip(Frame::Nack {
            message_id: MessageId(1),
            reason: String::new(),
        });
    }

    #[test]
    fn roundtrip_nack_truncates_long_reason() {
        let long_reason = "x".repeat(2000);
        let frame = Frame::Nack {
            message_id: MessageId(7),
            reason: long_reason,
        };
        let mut codec = MessageCodec::new(1_048_576);
        let mut buf = BytesMut::new();
        codec.encode(frame, &mut buf).unwrap();
        let decoded = codec.decode(&mut buf).unwrap().unwrap();
        match decoded {
            Frame::Nack { message_id, reason } => {
                assert_eq!(message_id, MessageId(7));
                assert_eq!(reason.len(), 1024);
            }
            other => panic!("expected Nack, got {other:?}"),
        }
    }

    #[test]
    fn roundtrip_nack_ack_dead_lettered() {
        roundtrip(Frame::NackAck {
            message_id: MessageId(42),
            status: 0x00,
        });
    }

    #[test]
    fn roundtrip_nack_ack_dropped() {
        roundtrip(Frame::NackAck {
            message_id: MessageId(42),
            status: 0x01,
        });
    }

    #[test]
    fn roundtrip_nack_ack_write_failed() {
        roundtrip(Frame::NackAck {
            message_id: MessageId(42),
            status: 0x02,
        });
    }

    // --- Static assertion ---

    #[test]
    fn message_codec_satisfies_framed_bounds() {
        fn assert_encoder<T: tokio_util::codec::Encoder<Frame>>() {}
        fn assert_decoder<T: tokio_util::codec::Decoder>() {}
        assert_encoder::<MessageCodec>();
        assert_decoder::<MessageCodec>();
    }

    // --- Header codec tests ---

    fn make_headers_codec() -> MessageCodec {
        let mut codec = MessageCodec::new(1_048_576);
        codec.set_headers_enabled(true);
        codec
    }

    fn encode_with_headers(frame: Frame) -> BytesMut {
        let mut codec = make_headers_codec();
        let mut buf = BytesMut::new();
        codec.encode(frame, &mut buf).unwrap();
        buf
    }

    fn roundtrip_headers(frame: Frame) {
        let mut buf = encode_with_headers(frame.clone());
        let decoded = make_headers_codec().decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, frame);
    }

    #[test]
    fn publish_roundtrip_no_headers_with_cap() {
        roundtrip_headers(Frame::Publish {
            topic: TopicName("events".into()),
            headers: vec![],
            ttl_ms: 0,
            payload: Bytes::from_static(b"hello"),
        });
    }

    #[test]
    fn publish_roundtrip_one_header() {
        roundtrip_headers(Frame::Publish {
            topic: TopicName("events".into()),
            headers: vec![Header {
                key: b"Content-Type".to_vec(),
                value: b"application/json".to_vec(),
            }],
            ttl_ms: 0,
            payload: Bytes::from_static(b"{\"ok\":true}"),
        });
    }

    #[test]
    fn publish_roundtrip_many_headers() {
        let headers = vec![
            Header {
                key: b"Content-Type".to_vec(),
                value: b"application/json".to_vec(),
            },
            Header {
                key: b"Correlation-Id".to_vec(),
                value: b"req-12345".to_vec(),
            },
            Header {
                key: b"Trace-Parent".to_vec(),
                value: b"00-abcdef0123456789-abcdef01-01".to_vec(),
            },
            Header {
                key: b"binary-val".to_vec(),
                value: vec![0x00, 0xFF, 0x80, 0x7F],
            },
        ];
        roundtrip_headers(Frame::Publish {
            topic: TopicName("events".into()),
            headers,
            ttl_ms: 0,
            payload: Bytes::from_static(b"data"),
        });
    }

    #[test]
    fn deliver_roundtrip_with_headers() {
        roundtrip_headers(Frame::Deliver {
            message_id: MessageId(42),
            offset: TopicOffset(100),
            timestamp_millis: 1_700_000_000_000,
            ttl_ms: 0,
            topic: TopicName("orders".into()),
            headers: vec![
                Header {
                    key: b"Content-Type".to_vec(),
                    value: b"application/json".to_vec(),
                },
                Header {
                    key: b"Source".to_vec(),
                    value: b"order-svc".to_vec(),
                },
            ],
            payload: Bytes::from_static(b"{\"id\":1}"),
        });
    }

    #[test]
    fn fetch_response_roundtrip_with_headers() {
        use crate::codec::frame::FetchedMessage;
        roundtrip_headers(Frame::FetchResponse {
            topic: TopicName("events".into()),
            next_offset: TopicOffset(3),
            head_offset: TopicOffset(10),
            messages: vec![
                FetchedMessage {
                    offset: TopicOffset(1),
                    timestamp_millis: 1_000_000,
                    ttl_ms: 0,
                    headers: vec![Header {
                        key: b"key1".to_vec(),
                        value: b"val1".to_vec(),
                    }],
                    payload: Bytes::from_static(b"msg1"),
                },
                FetchedMessage {
                    offset: TopicOffset(2),
                    timestamp_millis: 2_000_000,
                    ttl_ms: 0,
                    headers: vec![],
                    payload: Bytes::from_static(b"msg2"),
                },
            ],
        });
    }

    #[test]
    fn legacy_publish_no_headers_in_wire_format() {
        // With headers_enabled=false, the wire format should be identical to before.
        let frame = Frame::Publish {
            topic: TopicName("t".into()),
            headers: vec![Header {
                key: b"ignored".to_vec(),
                value: b"value".to_vec(),
            }],
            ttl_ms: 0,
            payload: Bytes::from_static(b"hello"),
        };
        let mut codec = MessageCodec::new(1_048_576); // headers_enabled=false
        let mut buf = BytesMut::new();
        codec.encode(frame, &mut buf).unwrap();
        // type: 1, topic: 2+1=3, payload: 4+5=9 => 13
        // Headers should NOT be in the wire format.
        let frame_len = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]);
        assert_eq!(frame_len, 13);
    }

    #[test]
    fn headers_in_wire_format_increase_frame_size() {
        let frame = Frame::Publish {
            topic: TopicName("t".into()),
            headers: vec![Header {
                key: b"k".to_vec(),
                value: b"v".to_vec(),
            }],
            ttl_ms: 0,
            payload: Bytes::from_static(b"hello"),
        };
        let buf = encode_with_headers(frame);
        // type: 1, topic: 2+1=3, header_count: 2, header: 2+1+4+1=8, payload: 4+5=9 => 23
        let frame_len = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]);
        assert_eq!(frame_len, 23);
    }

    #[test]
    fn validate_headers_empty() {
        use crate::codec::frame::{HeaderLimits, validate_headers};
        assert!(validate_headers(&[], &HeaderLimits::default()).is_ok());
    }

    #[test]
    fn validate_headers_too_many() {
        use crate::codec::frame::{HeaderLimits, validate_headers};
        let limits = HeaderLimits {
            max_count: 2,
            ..HeaderLimits::default()
        };
        let headers = vec![
            Header {
                key: b"a".to_vec(),
                value: b"1".to_vec(),
            },
            Header {
                key: b"b".to_vec(),
                value: b"2".to_vec(),
            },
            Header {
                key: b"c".to_vec(),
                value: b"3".to_vec(),
            },
        ];
        let err = validate_headers(&headers, &limits).unwrap_err();
        assert!(err.to_string().contains("too many headers"));
    }

    #[test]
    fn validate_headers_key_too_large() {
        use crate::codec::frame::{HeaderLimits, validate_headers};
        let limits = HeaderLimits {
            max_key_bytes: 4,
            ..HeaderLimits::default()
        };
        let headers = vec![Header {
            key: b"toolong".to_vec(),
            value: b"v".to_vec(),
        }];
        let err = validate_headers(&headers, &limits).unwrap_err();
        assert!(err.to_string().contains("header key too large"));
    }

    #[test]
    fn validate_headers_value_too_large() {
        use crate::codec::frame::{HeaderLimits, validate_headers};
        let limits = HeaderLimits {
            max_value_bytes: 2,
            ..HeaderLimits::default()
        };
        let headers = vec![Header {
            key: b"k".to_vec(),
            value: b"toolong".to_vec(),
        }];
        let err = validate_headers(&headers, &limits).unwrap_err();
        assert!(err.to_string().contains("header value too large"));
    }

    #[test]
    fn validate_headers_total_too_large() {
        use crate::codec::frame::{HeaderLimits, validate_headers};
        let limits = HeaderLimits {
            max_total_bytes: 10,
            ..HeaderLimits::default()
        };
        let headers = vec![
            Header {
                key: b"aaaa".to_vec(),
                value: b"1111".to_vec(),
            },
            Header {
                key: b"bb".to_vec(),
                value: b"22".to_vec(),
            },
        ];
        // total = 4+4+2+2 = 12 > 10
        let err = validate_headers(&headers, &limits).unwrap_err();
        assert!(err.to_string().contains("total header bytes too large"));
    }

    #[test]
    fn validate_headers_reserved_prefix() {
        use crate::codec::frame::{HeaderLimits, validate_headers};
        let headers = vec![Header {
            key: b"_fafb.dlq.source_topic".to_vec(),
            value: b"orders".to_vec(),
        }];
        let err = validate_headers(&headers, &HeaderLimits::default()).unwrap_err();
        assert!(err.to_string().contains("reserved prefix"));
    }

    #[test]
    fn validate_headers_pass_within_limits() {
        use crate::codec::frame::{HeaderLimits, validate_headers};
        let headers = vec![
            Header {
                key: b"Content-Type".to_vec(),
                value: b"application/json".to_vec(),
            },
            Header {
                key: b"Correlation-Id".to_vec(),
                value: b"abc-123".to_vec(),
            },
        ];
        assert!(validate_headers(&headers, &HeaderLimits::default()).is_ok());
    }

    #[test]
    fn header_bytes_count_toward_max_frame_size() {
        // Create a codec with a small max frame size.
        let mut codec = MessageCodec::new(30);
        codec.set_headers_enabled(true);
        let frame = Frame::Publish {
            topic: TopicName("t".into()),
            headers: vec![Header {
                key: b"big-key".to_vec(),
                value: vec![0u8; 100], // way too large
            }],
            ttl_ms: 0,
            payload: Bytes::from_static(b"x"),
        };
        let mut buf = BytesMut::new();
        // Encode succeeds (encoder doesn't check frame size — that's the decoder's job).
        codec.encode(frame, &mut buf).unwrap();
        // Trying to decode with the same small max_frame_size should fail.
        let err = codec.decode(&mut buf).unwrap_err();
        assert!(err.to_string().contains("frame too large"));
    }

    #[test]
    fn decode_rejects_publish_with_too_many_headers() {
        use crate::codec::frame::HeaderLimits;
        let mut enc = make_headers_codec();
        let frame = Frame::Publish {
            topic: TopicName("t".into()),
            headers: vec![
                Header {
                    key: b"a".to_vec(),
                    value: b"1".to_vec(),
                },
                Header {
                    key: b"b".to_vec(),
                    value: b"2".to_vec(),
                },
                Header {
                    key: b"c".to_vec(),
                    value: b"3".to_vec(),
                },
            ],
            ttl_ms: 0,
            payload: Bytes::from_static(b"x"),
        };
        let mut buf = BytesMut::new();
        enc.encode(frame, &mut buf).unwrap();

        // Decoder with strict limit of 2 headers
        let mut dec = MessageCodec::new(1_048_576);
        dec.set_headers_enabled(true);
        dec.set_header_limits(HeaderLimits {
            max_count: 2,
            ..HeaderLimits::default()
        });
        let err = dec.decode(&mut buf).unwrap_err();
        assert!(err.to_string().contains("too many headers"));
    }

    #[test]
    fn decode_rejects_publish_with_reserved_prefix() {
        use crate::codec::frame::HeaderLimits;
        let mut enc = make_headers_codec();
        let frame = Frame::Publish {
            topic: TopicName("t".into()),
            headers: vec![Header {
                key: b"_fafb.internal".to_vec(),
                value: b"val".to_vec(),
            }],
            ttl_ms: 0,
            payload: Bytes::from_static(b"x"),
        };
        let mut buf = BytesMut::new();
        enc.encode(frame, &mut buf).unwrap();

        let mut dec = MessageCodec::new(1_048_576);
        dec.set_headers_enabled(true);
        dec.set_header_limits(HeaderLimits::default());
        let err = dec.decode(&mut buf).unwrap_err();
        assert!(err.to_string().contains("reserved prefix"));
    }

    #[test]
    fn decode_accepts_publish_with_valid_headers() {
        use crate::codec::frame::HeaderLimits;
        let mut enc = make_headers_codec();
        let frame = Frame::Publish {
            topic: TopicName("t".into()),
            headers: vec![Header {
                key: b"Content-Type".to_vec(),
                value: b"application/json".to_vec(),
            }],
            ttl_ms: 0,
            payload: Bytes::from_static(b"{}"),
        };
        let mut buf = BytesMut::new();
        enc.encode(frame.clone(), &mut buf).unwrap();

        let mut dec = MessageCodec::new(1_048_576);
        dec.set_headers_enabled(true);
        dec.set_header_limits(HeaderLimits::default());
        let decoded = dec.decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, frame);
    }

    // --- TTL codec tests ---

    fn make_ttl_codec() -> MessageCodec {
        let mut codec = MessageCodec::new(1_048_576);
        codec.set_ttl_enabled(true);
        codec
    }

    fn make_ttl_and_headers_codec() -> MessageCodec {
        let mut codec = MessageCodec::new(1_048_576);
        codec.set_ttl_enabled(true);
        codec.set_headers_enabled(true);
        codec
    }

    #[test]
    fn publish_roundtrip_with_ttl() {
        let frame = Frame::Publish {
            topic: TopicName("events".into()),
            headers: vec![],
            ttl_ms: 30_000,
            payload: Bytes::from_static(b"hello"),
        };
        let mut codec = make_ttl_codec();
        let mut buf = BytesMut::new();
        codec.encode(frame.clone(), &mut buf).unwrap();
        let decoded = codec.decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, frame);
    }

    #[test]
    fn publish_roundtrip_zero_ttl() {
        let frame = Frame::Publish {
            topic: TopicName("t".into()),
            headers: vec![],
            ttl_ms: 0,
            payload: Bytes::from_static(b"data"),
        };
        let mut codec = make_ttl_codec();
        let mut buf = BytesMut::new();
        codec.encode(frame.clone(), &mut buf).unwrap();
        let decoded = codec.decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, frame);
    }

    #[test]
    fn publish_roundtrip_ttl_and_headers() {
        let frame = Frame::Publish {
            topic: TopicName("events".into()),
            headers: vec![Header {
                key: b"Content-Type".to_vec(),
                value: b"application/json".to_vec(),
            }],
            ttl_ms: 60_000,
            payload: Bytes::from_static(b"{\"ok\":true}"),
        };
        let mut codec = make_ttl_and_headers_codec();
        let mut buf = BytesMut::new();
        codec.encode(frame.clone(), &mut buf).unwrap();
        let decoded = codec.decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, frame);
    }

    #[test]
    fn publish_ttl_not_in_wire_when_disabled() {
        // With ttl_enabled=false, ttl_ms is not on the wire.
        let frame = Frame::Publish {
            topic: TopicName("t".into()),
            headers: vec![],
            ttl_ms: 99_000, // this value should be ignored by the decoder
            payload: Bytes::from_static(b"hello"),
        };
        let mut codec = MessageCodec::new(1_048_576); // ttl_enabled=false
        let mut buf = BytesMut::new();
        codec.encode(frame, &mut buf).unwrap();
        // type: 1, topic: 2+1=3, payload: 4+5=9 => 13
        // No TTL field in the wire format.
        let frame_len = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]);
        assert_eq!(frame_len, 13);

        // Decoder should produce ttl_ms=0 since it wasn't on the wire.
        let decoded = codec.decode(&mut buf).unwrap().unwrap();
        match decoded {
            Frame::Publish { ttl_ms, .. } => assert_eq!(ttl_ms, 0),
            _ => panic!("expected Publish frame"),
        }
    }

    #[test]
    fn publish_ttl_in_wire_when_enabled() {
        let frame = Frame::Publish {
            topic: TopicName("t".into()),
            headers: vec![],
            ttl_ms: 30_000,
            payload: Bytes::from_static(b"hello"),
        };
        let mut codec = make_ttl_codec();
        let mut buf = BytesMut::new();
        codec.encode(frame, &mut buf).unwrap();
        // type: 1, topic: 2+1=3, ttl: 4, payload: 4+5=9 => 17
        let frame_len = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]);
        assert_eq!(frame_len, 17);
    }

    #[test]
    fn deliver_roundtrip_with_ttl() {
        let frame = Frame::Deliver {
            message_id: MessageId(42),
            offset: TopicOffset(100),
            timestamp_millis: 1_700_000_000_000,
            ttl_ms: 30_000,
            topic: TopicName("orders".into()),
            headers: vec![],
            payload: Bytes::from_static(b"payload"),
        };
        let mut codec = make_ttl_codec();
        let mut buf = BytesMut::new();
        codec.encode(frame.clone(), &mut buf).unwrap();
        let decoded = codec.decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, frame);
    }

    #[test]
    fn deliver_roundtrip_zero_ttl() {
        let frame = Frame::Deliver {
            message_id: MessageId(1),
            offset: TopicOffset(0),
            timestamp_millis: 1_700_000_000_000,
            ttl_ms: 0,
            topic: TopicName("t".into()),
            headers: vec![],
            payload: Bytes::from_static(b"data"),
        };
        let mut codec = make_ttl_codec();
        let mut buf = BytesMut::new();
        codec.encode(frame.clone(), &mut buf).unwrap();
        let decoded = codec.decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, frame);
    }

    #[test]
    fn deliver_roundtrip_ttl_and_headers() {
        let frame = Frame::Deliver {
            message_id: MessageId(7),
            offset: TopicOffset(50),
            timestamp_millis: 1_700_000_000_000,
            ttl_ms: 60_000,
            topic: TopicName("events".into()),
            headers: vec![Header {
                key: b"key".to_vec(),
                value: b"val".to_vec(),
            }],
            payload: Bytes::from_static(b"body"),
        };
        let mut codec = make_ttl_and_headers_codec();
        let mut buf = BytesMut::new();
        codec.encode(frame.clone(), &mut buf).unwrap();
        let decoded = codec.decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, frame);
    }

    #[test]
    fn deliver_ttl_not_in_wire_when_disabled() {
        let frame = Frame::Deliver {
            message_id: MessageId(1),
            offset: TopicOffset(0),
            timestamp_millis: 1_700_000_000_000,
            ttl_ms: 99_000,
            topic: TopicName("t".into()),
            headers: vec![],
            payload: Bytes::from_static(b"hi"),
        };
        let mut codec = MessageCodec::new(1_048_576); // ttl_enabled=false
        let mut buf = BytesMut::new();
        codec.encode(frame, &mut buf).unwrap();
        // Decoder should produce ttl_ms=0 since TTL wasn't on the wire.
        let decoded = codec.decode(&mut buf).unwrap().unwrap();
        match decoded {
            Frame::Deliver { ttl_ms, .. } => assert_eq!(ttl_ms, 0),
            _ => panic!("expected Deliver frame"),
        }
    }

    #[test]
    fn deliver_ttl_in_wire_increases_size() {
        let frame = Frame::Deliver {
            message_id: MessageId(1),
            offset: TopicOffset(0),
            timestamp_millis: 0,
            ttl_ms: 30_000,
            topic: TopicName("t".into()),
            headers: vec![],
            payload: Bytes::from_static(b"x"),
        };
        // Encode without TTL
        let mut codec_no_ttl = MessageCodec::new(1_048_576);
        let mut buf_no_ttl = BytesMut::new();
        codec_no_ttl.encode(frame.clone(), &mut buf_no_ttl).unwrap();
        let len_no_ttl =
            u32::from_be_bytes([buf_no_ttl[0], buf_no_ttl[1], buf_no_ttl[2], buf_no_ttl[3]]);

        // Encode with TTL
        let mut codec_ttl = make_ttl_codec();
        let mut buf_ttl = BytesMut::new();
        codec_ttl.encode(frame, &mut buf_ttl).unwrap();
        let len_ttl = u32::from_be_bytes([buf_ttl[0], buf_ttl[1], buf_ttl[2], buf_ttl[3]]);

        // TTL adds 4 bytes
        assert_eq!(len_ttl, len_no_ttl + 4);
    }

    #[test]
    fn fetch_response_roundtrip_with_ttl() {
        let frame = Frame::FetchResponse {
            topic: TopicName("orders".into()),
            next_offset: TopicOffset(3),
            head_offset: TopicOffset(10),
            messages: vec![
                FetchedMessage {
                    offset: TopicOffset(0),
                    timestamp_millis: 1_700_000_000_000,
                    ttl_ms: 30_000,
                    headers: vec![],
                    payload: Bytes::from_static(b"msg1"),
                },
                FetchedMessage {
                    offset: TopicOffset(1),
                    timestamp_millis: 1_700_000_000_001,
                    ttl_ms: 0, // no TTL
                    headers: vec![],
                    payload: Bytes::from_static(b"msg2"),
                },
                FetchedMessage {
                    offset: TopicOffset(2),
                    timestamp_millis: 1_700_000_000_002,
                    ttl_ms: 60_000,
                    headers: vec![],
                    payload: Bytes::from_static(b"msg3"),
                },
            ],
        };
        let mut codec = make_ttl_codec();
        let mut buf = BytesMut::new();
        codec.encode(frame.clone(), &mut buf).unwrap();
        let decoded = codec.decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, frame);
    }

    #[test]
    fn fetch_response_ttl_not_in_wire_when_disabled() {
        let frame = Frame::FetchResponse {
            topic: TopicName("t".into()),
            next_offset: TopicOffset(1),
            head_offset: TopicOffset(1),
            messages: vec![FetchedMessage {
                offset: TopicOffset(0),
                timestamp_millis: 1_700_000_000_000,
                ttl_ms: 99_000,
                headers: vec![],
                payload: Bytes::from_static(b"data"),
            }],
        };
        let mut codec = MessageCodec::new(1_048_576); // ttl_enabled=false
        let mut buf = BytesMut::new();
        codec.encode(frame, &mut buf).unwrap();
        let decoded = codec.decode(&mut buf).unwrap().unwrap();
        match decoded {
            Frame::FetchResponse { messages, .. } => {
                assert_eq!(messages[0].ttl_ms, 0);
            }
            _ => panic!("expected FetchResponse"),
        }
    }

    #[test]
    fn fetch_response_roundtrip_ttl_and_headers() {
        let frame = Frame::FetchResponse {
            topic: TopicName("t".into()),
            next_offset: TopicOffset(1),
            head_offset: TopicOffset(1),
            messages: vec![FetchedMessage {
                offset: TopicOffset(0),
                timestamp_millis: 1_700_000_000_000,
                ttl_ms: 5_000,
                headers: vec![Header {
                    key: b"k".to_vec(),
                    value: b"v".to_vec(),
                }],
                payload: Bytes::from_static(b"body"),
            }],
        };
        let mut codec = make_ttl_and_headers_codec();
        let mut buf = BytesMut::new();
        codec.encode(frame.clone(), &mut buf).unwrap();
        let decoded = codec.decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, frame);
    }

    #[test]
    fn decode_batch_ack_roundtrip() {
        let frame = Frame::BatchAck {
            message_ids: vec![MessageId(1), MessageId(2), MessageId(3)],
        };
        let mut buf = encode_frame(frame.clone());
        let decoded = make_codec().decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, frame);
    }

    #[test]
    fn decode_batch_ack_empty() {
        let frame = Frame::BatchAck {
            message_ids: vec![],
        };
        let mut buf = encode_frame(frame.clone());
        let decoded = make_codec().decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, frame);
    }

    #[test]
    fn decode_batch_ack_response_roundtrip() {
        let frame = Frame::BatchAckResponse {
            acked_count: 42,
            unknown_count: 7,
        };
        let mut buf = encode_frame(frame.clone());
        let decoded = make_codec().decode(&mut buf).unwrap().unwrap();
        assert_eq!(decoded, frame);
    }

    #[test]
    fn decode_batch_ack_too_many_ids_rejected() {
        // Manually craft a frame with count > 10,000
        let mut buf = BytesMut::new();
        let body_len: u32 = 1 + 4; // type + count (no actual IDs needed, decoder checks count first)
        buf.put_u32(body_len);
        buf.put_u8(0x18); // BatchAck
        buf.put_u32(10_001); // count exceeds limit
        let result = make_codec().decode(&mut buf);
        assert!(result.is_err());
    }

    #[test]
    fn decode_batch_ack_boundary_10000_accepted() {
        // Exactly 10,000 IDs — the maximum allowed.
        let ids: Vec<MessageId> = (1..=10_000).map(MessageId).collect();
        let frame = Frame::BatchAck {
            message_ids: ids.clone(),
        };
        let mut buf = encode_frame(frame);
        let decoded = make_codec().decode(&mut buf).unwrap().unwrap();
        match decoded {
            Frame::BatchAck { message_ids } => {
                assert_eq!(message_ids.len(), 10_000);
                assert_eq!(message_ids[0], MessageId(1));
                assert_eq!(message_ids[9_999], MessageId(10_000));
            }
            other => panic!("expected BatchAck, got: {other:?}"),
        }
    }

    #[test]
    fn decode_batch_ack_boundary_10001_rejected() {
        // Manually craft a frame with exactly 10,001 IDs.
        let mut buf = BytesMut::new();
        let body_len: u32 = 1 + 4 + 10_001 * 8; // type + count + IDs
        buf.put_u32(body_len);
        buf.put_u8(0x18); // BatchAck
        buf.put_u32(10_001);
        for i in 0..10_001u64 {
            buf.put_u64(i);
        }
        let result = make_codec().decode(&mut buf);
        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(
            err.contains("10001"),
            "error should mention the count, got: {err}"
        );
    }

    #[test]
    fn roundtrip_not_leader() {
        roundtrip(Frame::NotLeader {
            leader_addr: "192.168.1.2:9876".into(),
        });
    }

    #[test]
    fn roundtrip_not_leader_empty_addr() {
        roundtrip(Frame::NotLeader {
            leader_addr: String::new(),
        });
    }

    #[test]
    fn roundtrip_cluster_unavailable() {
        roundtrip(Frame::ClusterUnavailable {
            retry_after_ms: 5000,
            message: "failover in progress".into(),
        });
    }

    #[test]
    fn roundtrip_cluster_unavailable_zero_retry() {
        roundtrip(Frame::ClusterUnavailable {
            retry_after_ms: 0,
            message: String::new(),
        });
    }

    // --- Property-based tests ---

    mod proptest_tests {
        use super::*;
        use proptest::prelude::*;

        fn arb_topic() -> impl Strategy<Value = TopicName> {
            "[a-z][a-z0-9._-]{0,30}".prop_map(|s| TopicName(s))
        }

        fn arb_namespace() -> impl Strategy<Value = NamespaceId> {
            "[a-z][a-z0-9_-]{0,30}".prop_map(|s| NamespaceId(s))
        }

        fn arb_api_key() -> impl Strategy<Value = ApiKey> {
            "[a-zA-Z0-9]{8,64}".prop_map(|s| ApiKey(s))
        }

        fn arb_group() -> impl Strategy<Value = Option<GroupName>> {
            prop_oneof![
                Just(None),
                "[a-z][a-z0-9_-]{0,20}".prop_map(|s| Some(GroupName(s))),
            ]
        }

        fn arb_payload() -> impl Strategy<Value = Bytes> {
            prop::collection::vec(any::<u8>(), 0..1024).prop_map(Bytes::from)
        }

        fn arb_message_id() -> impl Strategy<Value = MessageId> {
            any::<u64>().prop_map(MessageId)
        }

        fn arb_error_code() -> impl Strategy<Value = ErrorCode> {
            prop_oneof![
                Just(ErrorCode::AuthFailed),
                Just(ErrorCode::PermissionDenied),
                Just(ErrorCode::NamespaceNotFound),
                Just(ErrorCode::TopicNotFound),
                Just(ErrorCode::InvalidFrame),
                Just(ErrorCode::FrameTooLarge),
                Just(ErrorCode::UnsupportedVersion),
                Just(ErrorCode::NotAuthenticated),
                Just(ErrorCode::InternalError),
                Just(ErrorCode::MessageTooLarge),
            ]
        }

        fn arb_frame() -> impl Strategy<Value = Frame> {
            prop_oneof![
                (any::<u8>(), arb_namespace(), arb_api_key()).prop_map(
                    |(version, namespace, api_key)| Frame::Connect {
                        version,
                        namespace,
                        api_key,
                        capabilities: 0,
                    }
                ),
                (any::<bool>(), "[a-z]{0,50}").prop_map(|(success, message)| {
                    Frame::ConnectAck {
                        success,
                        message,
                        server_capabilities: 0,
                    }
                }),
                (arb_topic(), arb_group())
                    .prop_map(|(topic, group)| Frame::Subscribe { topic, group }),
                (arb_topic(), arb_group(), any::<u64>(), any::<u16>()).prop_map(
                    |(topic, group, start_offset, matched_count)| Frame::SubscribeAck {
                        topic,
                        group,
                        start_offset: TopicOffset(start_offset),
                        matched_count,
                    },
                ),
                (arb_topic(), arb_group())
                    .prop_map(|(topic, group)| Frame::Unsubscribe { topic, group }),
                (arb_topic(), arb_group())
                    .prop_map(|(topic, group)| Frame::UnsubscribeAck { topic, group }),
                (arb_topic(), arb_payload()).prop_map(|(topic, payload)| Frame::Publish {
                    topic,
                    payload,
                    headers: vec![],
                    ttl_ms: 0,
                }),
                arb_message_id().prop_map(|message_id| Frame::PublishAck { message_id }),
                (
                    arb_message_id(),
                    any::<u64>(),
                    any::<u64>(),
                    arb_topic(),
                    arb_payload()
                )
                    .prop_map(
                        |(message_id, offset, timestamp_millis, topic, payload)| Frame::Deliver {
                            message_id,
                            offset: TopicOffset(offset),
                            timestamp_millis,
                            ttl_ms: 0,
                            topic,
                            payload,
                            headers: vec![],
                        }
                    ),
                arb_message_id().prop_map(|message_id| Frame::Ack { message_id }),
                (arb_error_code(), "[a-z ]{0,50}")
                    .prop_map(|(code, message)| Frame::Error { code, message }),
                (arb_topic(), any::<u64>(), arb_group()).prop_map(|(topic, offset, group)| {
                    Frame::SubscribeAt {
                        topic,
                        offset: TopicOffset(offset),
                        group,
                    }
                },),
                (arb_topic(), any::<u64>(), arb_group()).prop_map(
                    |(topic, timestamp_millis, group)| Frame::SubscribeFrom {
                        topic,
                        timestamp_millis,
                        group,
                    },
                ),
                (any::<u8>(), any::<u32>(), "[a-z ]{0,50}").prop_map(
                    |(reason, deadline_ms, message)| Frame::GoingAway {
                        reason: crate::codec::frame::GoingAwayReason(reason),
                        deadline_ms,
                        message,
                    },
                ),
                (
                    arb_namespace(),
                    arb_api_key(),
                    any::<u32>(),
                    prop::collection::vec(
                        (arb_topic(), arb_group(), any::<u64>()).prop_map(
                            |(topic, group, offset)| {
                                crate::codec::frame::SubscriptionEntry {
                                    topic,
                                    group,
                                    last_acked_offset: TopicOffset(offset),
                                }
                            }
                        ),
                        0..5,
                    )
                )
                    .prop_map(
                        |(namespace, api_key, capabilities, subscriptions)| Frame::Reconnect {
                            version: 1,
                            namespace,
                            api_key,
                            capabilities,
                            subscriptions,
                        }
                    ),
                (
                    "[a-z]{0,30}",
                    any::<u32>(),
                    prop::collection::vec(
                        (
                            arb_topic(),
                            arb_group(),
                            prop_oneof![
                                Just(crate::codec::frame::SubscriptionStatus::Restored),
                                Just(crate::codec::frame::SubscriptionStatus::OffsetClamped),
                                Just(crate::codec::frame::SubscriptionStatus::TopicNotFound),
                                Just(crate::codec::frame::SubscriptionStatus::PermissionDenied),
                                Just(crate::codec::frame::SubscriptionStatus::GroupFull),
                            ],
                            any::<u64>()
                        )
                            .prop_map(|(topic, group, status, offset)| {
                                crate::codec::frame::SubscriptionResult {
                                    topic,
                                    group,
                                    status,
                                    actual_start_offset: TopicOffset(offset),
                                }
                            }),
                        0..5,
                    )
                )
                    .prop_map(|(message, server_capabilities, results)| {
                        Frame::ReconnectAck {
                            status: crate::codec::frame::ReconnectStatus::Success,
                            message,
                            server_capabilities,
                            results,
                        }
                    }),
                (arb_topic(), any::<u64>(), 1..10000u32).prop_map(|(topic, offset, max_count)| {
                    Frame::Fetch {
                        topic,
                        offset: TopicOffset(offset),
                        max_count,
                    }
                },),
                (arb_topic(), any::<u64>(), 1..10000u32).prop_map(
                    |(topic, timestamp_millis, max_count)| Frame::FetchFrom {
                        topic,
                        timestamp_millis,
                        max_count,
                    },
                ),
                (
                    arb_topic(),
                    any::<u64>(),
                    any::<u64>(),
                    prop::collection::vec(
                        (any::<u64>(), any::<u64>(), arb_payload()).prop_map(
                            |(offset, ts, payload)| crate::codec::frame::FetchedMessage {
                                offset: TopicOffset(offset),
                                timestamp_millis: ts,
                                ttl_ms: 0,
                                payload,
                                headers: vec![],
                            },
                        ),
                        0..5,
                    ),
                )
                    .prop_map(|(topic, next_offset, head_offset, messages)| {
                        Frame::FetchResponse {
                            topic,
                            next_offset: TopicOffset(next_offset),
                            head_offset: TopicOffset(head_offset),
                            messages,
                        }
                    },),
                (arb_message_id(), "[a-z ]{0,100}")
                    .prop_map(|(message_id, reason)| { Frame::Nack { message_id, reason } }),
                (arb_message_id(), 0..3u8)
                    .prop_map(|(message_id, status)| Frame::NackAck { message_id, status }),
                prop::collection::vec(arb_message_id(), 0..100)
                    .prop_map(|message_ids| Frame::BatchAck { message_ids }),
                (any::<u32>(), any::<u32>()).prop_map(|(acked_count, unknown_count)| {
                    Frame::BatchAckResponse {
                        acked_count,
                        unknown_count,
                    }
                }),
                Just(Frame::Ping),
                Just(Frame::Pong),
                "[a-z0-9.:]{0,50}".prop_map(|leader_addr| Frame::NotLeader { leader_addr }),
                (any::<u32>(), "[a-z ]{0,50}").prop_map(|(retry_after_ms, message)| {
                    Frame::ClusterUnavailable {
                        retry_after_ms,
                        message,
                    }
                }),
            ]
        }

        proptest! {
            #[test]
            fn roundtrip_any_frame(frame in arb_frame()) {
                let mut codec = MessageCodec::new(1_048_576);
                let mut buf = BytesMut::new();
                codec.encode(frame.clone(), &mut buf).unwrap();
                let decoded = codec.decode(&mut buf).unwrap().unwrap();
                prop_assert_eq!(decoded, frame);
                prop_assert!(buf.is_empty());
            }

            #[test]
            fn decoder_never_panics(data in prop::collection::vec(any::<u8>(), 0..512)) {
                let mut codec = MessageCodec::new(1_048_576);
                let mut buf = BytesMut::from(&data[..]);
                // Must not panic — Ok or Err only
                let _ = codec.decode(&mut buf);
            }
        }
    }
}
