use bytes::Bytes;

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

/// Bitmask of protocol capabilities negotiated during the Connect handshake.
pub type Capabilities = u32;

/// Client/server understands GoingAway frame (0x10).
pub const CAP_GOING_AWAY: Capabilities = 1 << 0;
/// Client/server understands Reconnect/ReconnectAck (0x11, 0x12).
pub const CAP_RECONNECT: Capabilities = 1 << 1;
/// Client/server understands Nack/NackAck (0x16, 0x17).
pub const CAP_NACK: Capabilities = 1 << 2;
/// Client/server understands message headers on Publish/Deliver/FetchResponse.
pub const CAP_HEADERS: Capabilities = 1 << 3;
/// Client/server understands wildcard/pattern subscriptions.
pub const CAP_WILDCARD: Capabilities = 1 << 4;
/// Client/server understands BatchAck/BatchAckResponse (0x18, 0x19).
pub const CAP_BATCH_ACK: Capabilities = 1 << 5;
/// Client/server understands per-message TTL on Publish/Deliver/FetchResponse.
pub const CAP_MESSAGE_TTL: Capabilities = 1 << 6;
/// Client/server understands cluster-aware frames (NotLeader, ClusterUnavailable).
pub const CAP_CLUSTER: Capabilities = 1 << 7;

/// All capabilities this server version supports.
/// Note: CAP_CLUSTER is NOT included here — it is added dynamically when the
/// cluster feature is enabled and configured. See Phase 24.6.
pub const SERVER_CAPABILITIES: Capabilities = CAP_GOING_AWAY
    | CAP_RECONNECT
    | CAP_NACK
    | CAP_HEADERS
    | CAP_WILDCARD
    | CAP_BATCH_ACK
    | CAP_MESSAGE_TTL;

/// Reason code for GoingAway frame. Newtype u8 for forward compatibility —
/// unknown reason values are accepted and logged, never rejected.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct GoingAwayReason(pub u8);

impl GoingAwayReason {
    /// Server is being upgraded to a new version.
    pub const UPGRADE: Self = Self(0x01);
    /// Server is being taken down for maintenance.
    pub const MAINTENANCE: Self = Self(0x02);
    /// Server is overloaded and shedding connections (reserved for future load-shedding).
    pub const OVERLOADED: Self = Self(0x03);
    /// Server is shutting down permanently.
    pub const SHUTDOWN: Self = Self(0x04);

    /// Parse a reason string from config into a GoingAwayReason.
    /// Unknown strings log a warning and default to UPGRADE.
    pub fn from_config(s: &str) -> Self {
        match s {
            "upgrade" => Self::UPGRADE,
            "maintenance" => Self::MAINTENANCE,
            "shutdown" => Self::SHUTDOWN,
            _ => {
                tracing::warn!(
                    reason = s,
                    "unknown going_away_reason, defaulting to \"upgrade\""
                );
                Self::UPGRADE
            }
        }
    }
}

/// Wire message type identifiers.
#[repr(u8)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MessageType {
    Connect = 0x01,
    ConnectAck = 0x02,
    Subscribe = 0x03,
    SubscribeAck = 0x04,
    Unsubscribe = 0x05,
    UnsubscribeAck = 0x06,
    Publish = 0x07,
    PublishAck = 0x08,
    Deliver = 0x09,
    Ack = 0x0A,
    Error = 0x0B,
    Ping = 0x0C,
    Pong = 0x0D,
    SubscribeAt = 0x0E,
    SubscribeFrom = 0x0F,
    GoingAway = 0x10,
    Reconnect = 0x11,
    ReconnectAck = 0x12,
    Fetch = 0x13,
    FetchResponse = 0x14,
    FetchFrom = 0x15,
    Nack = 0x16,
    NackAck = 0x17,
    BatchAck = 0x18,
    BatchAckResponse = 0x19,
    NotLeader = 0x1A,
    ClusterUnavailable = 0x1B,
}

impl TryFrom<u8> for MessageType {
    type Error = Error;

    fn try_from(value: u8) -> Result<Self, <Self as TryFrom<u8>>::Error> {
        match value {
            0x01 => Ok(Self::Connect),
            0x02 => Ok(Self::ConnectAck),
            0x03 => Ok(Self::Subscribe),
            0x04 => Ok(Self::SubscribeAck),
            0x05 => Ok(Self::Unsubscribe),
            0x06 => Ok(Self::UnsubscribeAck),
            0x07 => Ok(Self::Publish),
            0x08 => Ok(Self::PublishAck),
            0x09 => Ok(Self::Deliver),
            0x0A => Ok(Self::Ack),
            0x0B => Ok(Self::Error),
            0x0C => Ok(Self::Ping),
            0x0D => Ok(Self::Pong),
            0x0E => Ok(Self::SubscribeAt),
            0x0F => Ok(Self::SubscribeFrom),
            0x10 => Ok(Self::GoingAway),
            0x11 => Ok(Self::Reconnect),
            0x12 => Ok(Self::ReconnectAck),
            0x13 => Ok(Self::Fetch),
            0x14 => Ok(Self::FetchResponse),
            0x15 => Ok(Self::FetchFrom),
            0x16 => Ok(Self::Nack),
            0x17 => Ok(Self::NackAck),
            0x18 => Ok(Self::BatchAck),
            0x19 => Ok(Self::BatchAckResponse),
            0x1A => Ok(Self::NotLeader),
            0x1B => Ok(Self::ClusterUnavailable),
            _ => Err(Error::Codec(format!("unknown message type: 0x{value:02X}"))),
        }
    }
}

impl From<MessageType> for u8 {
    fn from(mt: MessageType) -> Self {
        mt as u8
    }
}

/// Wire error codes sent in Error frames.
#[repr(u16)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ErrorCode {
    AuthFailed = 0x0001,
    PermissionDenied = 0x0002,
    NamespaceNotFound = 0x0003,
    TopicNotFound = 0x0004,
    InvalidFrame = 0x0005,
    FrameTooLarge = 0x0006,
    UnsupportedVersion = 0x0007,
    NotAuthenticated = 0x0008,
    InternalError = 0x0009,
    ResourceExhausted = 0x000A,
    SlowConsumer = 0x000B,
    OffsetNotAvailable = 0x000C,
    ServerDraining = 0x000D,
    MessageTooLarge = 0x000E,
    InvalidTopic = 0x000F,
    InvalidOperation = 0x0010,
}

impl TryFrom<u16> for ErrorCode {
    type Error = crate::error::Error;

    fn try_from(value: u16) -> Result<Self, crate::error::Error> {
        match value {
            0x0001 => Ok(Self::AuthFailed),
            0x0002 => Ok(Self::PermissionDenied),
            0x0003 => Ok(Self::NamespaceNotFound),
            0x0004 => Ok(Self::TopicNotFound),
            0x0005 => Ok(Self::InvalidFrame),
            0x0006 => Ok(Self::FrameTooLarge),
            0x0007 => Ok(Self::UnsupportedVersion),
            0x0008 => Ok(Self::NotAuthenticated),
            0x0009 => Ok(Self::InternalError),
            0x000A => Ok(Self::ResourceExhausted),
            0x000B => Ok(Self::SlowConsumer),
            0x000C => Ok(Self::OffsetNotAvailable),
            0x000D => Ok(Self::ServerDraining),
            0x000E => Ok(Self::MessageTooLarge),
            0x000F => Ok(Self::InvalidTopic),
            0x0010 => Ok(Self::InvalidOperation),
            _ => Err(Error::Codec(format!("unknown error code: 0x{value:04X}"))),
        }
    }
}

impl From<ErrorCode> for u16 {
    fn from(code: ErrorCode) -> Self {
        code as u16
    }
}

impl ErrorCode {
    /// Returns true if this error code is fatal (server closes connection).
    pub fn is_fatal(&self) -> bool {
        matches!(
            self,
            Self::AuthFailed
                | Self::InvalidFrame
                | Self::FrameTooLarge
                | Self::UnsupportedVersion
                | Self::NotAuthenticated
                | Self::SlowConsumer
        )
    }

    /// Maps an `Error` variant to a wire error code.
    pub fn from_error(err: &Error) -> Self {
        match err {
            Error::AuthFailed => Self::AuthFailed,
            Error::PermissionDenied => Self::PermissionDenied,
            Error::NamespaceNotFound(_) => Self::NamespaceNotFound,
            Error::TopicNotFound(_) => Self::TopicNotFound,
            Error::Codec(_) => Self::InvalidFrame,
            Error::NotSubscribed { .. } => Self::InternalError,
            Error::DuplicateSubscription { .. } => Self::InternalError,
            Error::GroupNotFound { .. } => Self::InternalError,
            Error::QueueFull { .. } => Self::InternalError,
            Error::Storage(_) | Error::Serialization(_) => Self::InternalError,
            Error::Config(_) | Error::Tls(_) => Self::InternalError,
            Error::ResourceExhausted(_) => Self::ResourceExhausted,
            Error::MessageTooLarge { .. } => Self::MessageTooLarge,
            Error::InvalidTopic(_) => Self::InvalidTopic,
            Error::InvalidOperation(_) => Self::InvalidOperation,
        }
    }
}

/// Configuration limits for header validation, passed from config layer
/// to keep the codec pure (no global state).
#[derive(Debug, Clone, Copy)]
pub struct HeaderLimits {
    pub max_count: usize,
    pub max_key_bytes: usize,
    pub max_value_bytes: usize,
    pub max_total_bytes: usize,
}

impl Default for HeaderLimits {
    fn default() -> Self {
        Self {
            max_count: 64,
            max_key_bytes: 256,
            max_value_bytes: 8_192,
            max_total_bytes: 32_768,
        }
    }
}

use crate::dlq::RESERVED_HEADER_PREFIX;

/// Validate message headers against configured limits.
///
/// Returns `Ok(())` if all headers pass validation. Returns an `Error::Codec`
/// with a descriptive message on the first violation found. Increments the
/// `fafb_header_rejected_total` metric with the appropriate reason label.
pub fn validate_headers(headers: &[Header], limits: &HeaderLimits) -> Result<(), Error> {
    if headers.len() > limits.max_count {
        metrics::counter!("fafb_header_rejected_total", "reason" => "too_many").increment(1);
        return Err(Error::Codec(format!(
            "too many headers: {} (max {})",
            headers.len(),
            limits.max_count
        )));
    }

    let mut total_bytes: usize = 0;
    for header in headers {
        if header.key.len() > limits.max_key_bytes {
            metrics::counter!("fafb_header_rejected_total", "reason" => "key_too_large")
                .increment(1);
            return Err(Error::Codec(format!(
                "header key too large: {} bytes (max {})",
                header.key.len(),
                limits.max_key_bytes
            )));
        }
        if header.value.len() > limits.max_value_bytes {
            metrics::counter!("fafb_header_rejected_total", "reason" => "value_too_large")
                .increment(1);
            return Err(Error::Codec(format!(
                "header value too large: {} bytes (max {})",
                header.value.len(),
                limits.max_value_bytes
            )));
        }
        if header.key.starts_with(RESERVED_HEADER_PREFIX) {
            metrics::counter!("fafb_header_rejected_total", "reason" => "reserved_prefix")
                .increment(1);
            return Err(Error::Codec(format!(
                "header key uses reserved prefix '_fafb.': {:?}",
                String::from_utf8_lossy(&header.key)
            )));
        }
        total_bytes += header.key.len() + header.value.len();
    }

    if total_bytes > limits.max_total_bytes {
        metrics::counter!("fafb_header_rejected_total", "reason" => "total_too_large").increment(1);
        return Err(Error::Codec(format!(
            "total header bytes too large: {total_bytes} (max {})",
            limits.max_total_bytes
        )));
    }

    Ok(())
}

/// Overall status of a Reconnect attempt.
#[repr(u8)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReconnectStatus {
    /// All subscriptions restored at requested offsets.
    Success = 0x00,
    /// Some subscriptions restored, others failed. Check per-subscription results.
    Partial = 0x01,
    /// API key invalid. Connection closed.
    AuthFailed = 0x02,
}

impl TryFrom<u8> for ReconnectStatus {
    type Error = Error;

    fn try_from(value: u8) -> Result<Self, Error> {
        match value {
            0x00 => Ok(Self::Success),
            0x01 => Ok(Self::Partial),
            0x02 => Ok(Self::AuthFailed),
            _ => Err(Error::Codec(format!(
                "unknown reconnect status: 0x{value:02X}"
            ))),
        }
    }
}

/// Per-subscription status in a ReconnectAck.
#[repr(u8)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SubscriptionStatus {
    /// Subscription restored at the requested offset.
    Restored = 0x00,
    /// Requested offset was below the oldest available; clamped forward.
    OffsetClamped = 0x01,
    /// Topic does not exist.
    TopicNotFound = 0x02,
    /// Client does not have subscribe permission.
    PermissionDenied = 0x03,
    /// Consumer group is at capacity.
    GroupFull = 0x04,
}

impl TryFrom<u8> for SubscriptionStatus {
    type Error = Error;

    fn try_from(value: u8) -> Result<Self, Error> {
        match value {
            0x00 => Ok(Self::Restored),
            0x01 => Ok(Self::OffsetClamped),
            0x02 => Ok(Self::TopicNotFound),
            0x03 => Ok(Self::PermissionDenied),
            0x04 => Ok(Self::GroupFull),
            _ => Err(Error::Codec(format!(
                "unknown subscription status: 0x{value:02X}"
            ))),
        }
    }
}

/// A subscription entry in a Reconnect frame (client → server).
#[derive(Debug, Clone, PartialEq)]
pub struct SubscriptionEntry {
    pub topic: TopicName,
    pub group: Option<GroupName>,
    pub last_acked_offset: TopicOffset,
}

/// A per-subscription result in a ReconnectAck frame (server → client).
#[derive(Debug, Clone, PartialEq)]
pub struct SubscriptionResult {
    pub topic: TopicName,
    pub group: Option<GroupName>,
    pub status: SubscriptionStatus,
    pub actual_start_offset: TopicOffset,
}

/// Per-message wire overhead in a FetchResponse frame (base + TTL):
/// offset (8 bytes) + timestamp_millis (8 bytes) + ttl_ms (4 bytes) + payload_len (4 bytes) = 24 bytes.
/// Includes TTL overhead unconditionally for conservative budget estimation —
/// non-TTL clients use 20 bytes on wire, but 4 extra bytes per message is negligible
/// and avoids threading ttl_enabled through the fetch API.
/// When CAP_HEADERS is negotiated, add 2 bytes for header_count plus per-header overhead.
/// Shared between the manager (budget tracking) and the encoder.
pub const FETCH_MESSAGE_OVERHEAD: usize = 8 + 8 + 4 + 4;

/// Additional per-message overhead when CAP_HEADERS is negotiated:
/// header_count (2 bytes).
pub const FETCH_MESSAGE_HEADER_OVERHEAD: usize = 2;

/// Per-header wire overhead: key_len (2 bytes) + value_len (4 bytes) = 6 bytes.
pub const PER_HEADER_OVERHEAD: usize = 2 + 4;

/// A single message within a FetchResponse frame.
/// Intentionally lighter than PublishedMessage — no MessageId (not needed
/// for pull consumers who don't ack).
#[derive(Debug, Clone, PartialEq)]
pub struct FetchedMessage {
    pub offset: TopicOffset,
    pub timestamp_millis: u64,
    pub ttl_ms: u32,
    pub headers: Headers,
    pub payload: Bytes,
}

/// A decoded protocol frame.
#[derive(Debug, Clone, PartialEq)]
pub enum Frame {
    Connect {
        version: u8,
        namespace: NamespaceId,
        api_key: ApiKey,
        capabilities: Capabilities,
    },
    ConnectAck {
        success: bool,
        message: String,
        server_capabilities: Capabilities,
    },
    Subscribe {
        topic: TopicName,
        group: Option<GroupName>,
    },
    SubscribeAck {
        topic: TopicName,
        group: Option<GroupName>,
        start_offset: TopicOffset,
        /// Number of currently matching topics (wildcard subscriptions only).
        /// 0 for exact-match subscriptions. Only encoded/decoded when
        /// `CAP_WILDCARD` is negotiated; old decoders ignore trailing bytes.
        matched_count: u16,
    },
    Unsubscribe {
        topic: TopicName,
        group: Option<GroupName>,
    },
    UnsubscribeAck {
        topic: TopicName,
        group: Option<GroupName>,
    },
    Publish {
        topic: TopicName,
        headers: Headers,
        ttl_ms: u32,
        payload: Bytes,
    },
    PublishAck {
        message_id: MessageId,
    },
    Deliver {
        message_id: MessageId,
        offset: TopicOffset,
        timestamp_millis: u64,
        ttl_ms: u32,
        topic: TopicName,
        headers: Headers,
        payload: Bytes,
    },
    Ack {
        message_id: MessageId,
    },
    Error {
        code: ErrorCode,
        message: String,
    },
    Ping,
    Pong,
    /// Subscribe starting from a specific offset.
    SubscribeAt {
        topic: TopicName,
        offset: TopicOffset,
        group: Option<GroupName>,
    },
    /// Subscribe starting from the first message at or after a timestamp.
    SubscribeFrom {
        topic: TopicName,
        timestamp_millis: u64,
        group: Option<GroupName>,
    },
    /// Server → Client: planned shutdown notification with deadline.
    GoingAway {
        reason: GoingAwayReason,
        deadline_ms: u32,
        message: String,
    },
    /// Client → Server: restore auth + subscriptions in a single round-trip.
    Reconnect {
        version: u8,
        namespace: NamespaceId,
        api_key: ApiKey,
        capabilities: Capabilities,
        subscriptions: Vec<SubscriptionEntry>,
    },
    /// Server → Client: result of a Reconnect attempt.
    ReconnectAck {
        status: ReconnectStatus,
        message: String,
        server_capabilities: Capabilities,
        results: Vec<SubscriptionResult>,
    },
    /// Client → Server: pull a batch of messages starting from an offset.
    Fetch {
        topic: TopicName,
        offset: TopicOffset,
        max_count: u32,
    },
    /// Client → Server: pull a batch of messages starting from a timestamp.
    FetchFrom {
        topic: TopicName,
        timestamp_millis: u64,
        max_count: u32,
    },
    /// Server → Client: response to a Fetch or FetchFrom request.
    FetchResponse {
        topic: TopicName,
        next_offset: TopicOffset,
        head_offset: TopicOffset,
        messages: Vec<FetchedMessage>,
    },
    /// Client → Server: explicitly reject a delivered message.
    /// The server routes the message to the DLQ (if enabled) and responds
    /// with `NackAck`.
    Nack {
        message_id: MessageId,
        reason: String,
    },
    /// Server → Client: result of a Nack.
    /// Status: 0x00 = dead-lettered, 0x01 = DLQ disabled (dropped), 0x02 = DLQ write failed.
    NackAck {
        message_id: MessageId,
        status: u8,
    },
    /// Client → Server: acknowledge multiple messages in one frame.
    /// Requires `CAP_BATCH_ACK`. Max 10,000 message_ids per frame.
    BatchAck {
        message_ids: Vec<MessageId>,
    },
    /// Server → Client: result of a BatchAck.
    BatchAckResponse {
        acked_count: u32,
        unknown_count: u32,
    },
    /// Server → Client: you connected to a follower, here's the leader.
    /// Only sent to clients with `CAP_CLUSTER`.
    NotLeader {
        leader_addr: String,
    },
    /// Server → Client: cluster is in failover, no leader currently.
    /// Only sent to clients with `CAP_CLUSTER`.
    ClusterUnavailable {
        retry_after_ms: u32,
        message: String,
    },
}

impl Frame {
    /// Returns the message type byte for this frame.
    pub fn message_type(&self) -> MessageType {
        match self {
            Self::Connect { .. } => MessageType::Connect,
            Self::ConnectAck { .. } => MessageType::ConnectAck,
            Self::Subscribe { .. } => MessageType::Subscribe,
            Self::SubscribeAck { .. } => MessageType::SubscribeAck,
            Self::Unsubscribe { .. } => MessageType::Unsubscribe,
            Self::UnsubscribeAck { .. } => MessageType::UnsubscribeAck,
            Self::Publish { .. } => MessageType::Publish,
            Self::PublishAck { .. } => MessageType::PublishAck,
            Self::Deliver { .. } => MessageType::Deliver,
            Self::Ack { .. } => MessageType::Ack,
            Self::Error { .. } => MessageType::Error,
            Self::Ping => MessageType::Ping,
            Self::Pong => MessageType::Pong,
            Self::SubscribeAt { .. } => MessageType::SubscribeAt,
            Self::SubscribeFrom { .. } => MessageType::SubscribeFrom,
            Self::GoingAway { .. } => MessageType::GoingAway,
            Self::Reconnect { .. } => MessageType::Reconnect,
            Self::ReconnectAck { .. } => MessageType::ReconnectAck,
            Self::Fetch { .. } => MessageType::Fetch,
            Self::FetchFrom { .. } => MessageType::FetchFrom,
            Self::FetchResponse { .. } => MessageType::FetchResponse,
            Self::Nack { .. } => MessageType::Nack,
            Self::NackAck { .. } => MessageType::NackAck,
            Self::BatchAck { .. } => MessageType::BatchAck,
            Self::BatchAckResponse { .. } => MessageType::BatchAckResponse,
            Self::NotLeader { .. } => MessageType::NotLeader,
            Self::ClusterUnavailable { .. } => MessageType::ClusterUnavailable,
        }
    }
}

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

    #[test]
    fn message_type_u8_roundtrip() {
        let variants = [
            (0x01u8, MessageType::Connect),
            (0x02, MessageType::ConnectAck),
            (0x03, MessageType::Subscribe),
            (0x04, MessageType::SubscribeAck),
            (0x05, MessageType::Unsubscribe),
            (0x06, MessageType::UnsubscribeAck),
            (0x07, MessageType::Publish),
            (0x08, MessageType::PublishAck),
            (0x09, MessageType::Deliver),
            (0x0A, MessageType::Ack),
            (0x0B, MessageType::Error),
            (0x0C, MessageType::Ping),
            (0x0D, MessageType::Pong),
            (0x0E, MessageType::SubscribeAt),
            (0x0F, MessageType::SubscribeFrom),
            (0x10, MessageType::GoingAway),
            (0x11, MessageType::Reconnect),
            (0x12, MessageType::ReconnectAck),
            (0x13, MessageType::Fetch),
            (0x14, MessageType::FetchResponse),
            (0x15, MessageType::FetchFrom),
            (0x16, MessageType::Nack),
            (0x17, MessageType::NackAck),
            (0x18, MessageType::BatchAck),
            (0x19, MessageType::BatchAckResponse),
            (0x1A, MessageType::NotLeader),
            (0x1B, MessageType::ClusterUnavailable),
        ];
        for (byte, expected) in variants {
            let mt = MessageType::try_from(byte).unwrap();
            assert_eq!(mt, expected);
            let back: u8 = mt.into();
            assert_eq!(back, byte);
        }
    }

    #[test]
    fn message_type_unknown_byte() {
        assert!(MessageType::try_from(0x00).is_err());
        assert!(MessageType::try_from(0x1C).is_err());
        assert!(MessageType::try_from(0xFF).is_err());
    }

    #[test]
    fn error_code_u16_roundtrip() {
        let variants = [
            (0x0001u16, ErrorCode::AuthFailed),
            (0x0002, ErrorCode::PermissionDenied),
            (0x0003, ErrorCode::NamespaceNotFound),
            (0x0004, ErrorCode::TopicNotFound),
            (0x0005, ErrorCode::InvalidFrame),
            (0x0006, ErrorCode::FrameTooLarge),
            (0x0007, ErrorCode::UnsupportedVersion),
            (0x0008, ErrorCode::NotAuthenticated),
            (0x0009, ErrorCode::InternalError),
            (0x000A, ErrorCode::ResourceExhausted),
            (0x000B, ErrorCode::SlowConsumer),
            (0x000C, ErrorCode::OffsetNotAvailable),
            (0x000D, ErrorCode::ServerDraining),
            (0x000E, ErrorCode::MessageTooLarge),
            (0x000F, ErrorCode::InvalidTopic),
            (0x0010, ErrorCode::InvalidOperation),
        ];
        for (code, expected) in variants {
            let ec = ErrorCode::try_from(code).unwrap();
            assert_eq!(ec, expected);
            let back: u16 = ec.into();
            assert_eq!(back, code);
        }
    }

    #[test]
    fn error_code_unknown_value() {
        assert!(ErrorCode::try_from(0x0000).is_err());
        assert!(ErrorCode::try_from(0x0011).is_err());
        assert!(ErrorCode::try_from(0xFFFF).is_err());
    }

    #[test]
    fn error_code_is_fatal() {
        assert!(ErrorCode::AuthFailed.is_fatal());
        assert!(ErrorCode::InvalidFrame.is_fatal());
        assert!(ErrorCode::FrameTooLarge.is_fatal());
        assert!(ErrorCode::UnsupportedVersion.is_fatal());
        assert!(ErrorCode::NotAuthenticated.is_fatal());
        assert!(ErrorCode::SlowConsumer.is_fatal());

        assert!(!ErrorCode::PermissionDenied.is_fatal());
        assert!(!ErrorCode::NamespaceNotFound.is_fatal());
        assert!(!ErrorCode::TopicNotFound.is_fatal());
        assert!(!ErrorCode::InternalError.is_fatal());
        assert!(!ErrorCode::ResourceExhausted.is_fatal());
        assert!(!ErrorCode::OffsetNotAvailable.is_fatal());
        assert!(!ErrorCode::ServerDraining.is_fatal());
        assert!(!ErrorCode::MessageTooLarge.is_fatal());
        assert!(!ErrorCode::InvalidTopic.is_fatal());
        assert!(!ErrorCode::InvalidOperation.is_fatal());
    }

    #[test]
    fn error_code_from_error() {
        assert_eq!(
            ErrorCode::from_error(&Error::AuthFailed),
            ErrorCode::AuthFailed
        );
        assert_eq!(
            ErrorCode::from_error(&Error::PermissionDenied),
            ErrorCode::PermissionDenied
        );
        assert_eq!(
            ErrorCode::from_error(&Error::NamespaceNotFound(NamespaceId("ns".into()))),
            ErrorCode::NamespaceNotFound
        );
        assert_eq!(
            ErrorCode::from_error(&Error::TopicNotFound(TopicName("t".into()))),
            ErrorCode::TopicNotFound
        );
        assert_eq!(
            ErrorCode::from_error(&Error::Codec("bad".into())),
            ErrorCode::InvalidFrame
        );
        assert_eq!(
            ErrorCode::from_error(&Error::Config("bad".into())),
            ErrorCode::InternalError
        );
        assert_eq!(
            ErrorCode::from_error(&Error::QueueFull {
                namespace: NamespaceId("ns".into()),
                topic: TopicName("t".into()),
            }),
            ErrorCode::InternalError
        );
        assert_eq!(
            ErrorCode::from_error(&Error::ResourceExhausted("test".into())),
            ErrorCode::ResourceExhausted
        );
        assert_eq!(
            ErrorCode::from_error(&Error::InvalidTopic("bad topic".into())),
            ErrorCode::InvalidTopic
        );
        assert_eq!(
            ErrorCode::from_error(&Error::InvalidOperation("not supported".into())),
            ErrorCode::InvalidOperation
        );
    }

    #[test]
    fn frame_message_type() {
        assert_eq!(
            Frame::Connect {
                version: 1,
                namespace: NamespaceId("ns".into()),
                api_key: ApiKey("key".into()),
                capabilities: 0,
            }
            .message_type(),
            MessageType::Connect
        );
        assert_eq!(
            Frame::ConnectAck {
                success: true,
                message: String::new(),
                server_capabilities: 0,
            }
            .message_type(),
            MessageType::ConnectAck
        );
        assert_eq!(
            Frame::Subscribe {
                topic: TopicName("t".into()),
                group: None,
            }
            .message_type(),
            MessageType::Subscribe
        );
        assert_eq!(
            Frame::SubscribeAck {
                topic: TopicName("t".into()),
                group: None,
                start_offset: TopicOffset(0),
                matched_count: 0,
            }
            .message_type(),
            MessageType::SubscribeAck
        );
        assert_eq!(
            Frame::Unsubscribe {
                topic: TopicName("t".into()),
                group: None,
            }
            .message_type(),
            MessageType::Unsubscribe
        );
        assert_eq!(
            Frame::UnsubscribeAck {
                topic: TopicName("t".into()),
                group: None,
            }
            .message_type(),
            MessageType::UnsubscribeAck
        );
        assert_eq!(
            Frame::Publish {
                topic: TopicName("t".into()),
                payload: Bytes::new(),
                headers: vec![],
                ttl_ms: 0,
            }
            .message_type(),
            MessageType::Publish
        );
        assert_eq!(
            Frame::PublishAck {
                message_id: MessageId(1),
            }
            .message_type(),
            MessageType::PublishAck
        );
        assert_eq!(
            Frame::Deliver {
                message_id: MessageId(1),
                offset: TopicOffset(0),
                timestamp_millis: 0,
                ttl_ms: 0,
                topic: TopicName("t".into()),
                payload: Bytes::new(),
                headers: vec![],
            }
            .message_type(),
            MessageType::Deliver
        );
        assert_eq!(
            Frame::Ack {
                message_id: MessageId(1),
            }
            .message_type(),
            MessageType::Ack
        );
        assert_eq!(
            Frame::Error {
                code: ErrorCode::InternalError,
                message: "oops".into(),
            }
            .message_type(),
            MessageType::Error
        );
        assert_eq!(Frame::Ping.message_type(), MessageType::Ping);
        assert_eq!(Frame::Pong.message_type(), MessageType::Pong);
        assert_eq!(
            Frame::GoingAway {
                reason: GoingAwayReason::UPGRADE,
                deadline_ms: 30000,
                message: "deploying".into(),
            }
            .message_type(),
            MessageType::GoingAway
        );
        assert_eq!(
            Frame::Reconnect {
                version: 1,
                namespace: NamespaceId("ns".into()),
                api_key: ApiKey("key".into()),
                capabilities: 0,
                subscriptions: vec![],
            }
            .message_type(),
            MessageType::Reconnect
        );
        assert_eq!(
            Frame::ReconnectAck {
                status: ReconnectStatus::Success,
                message: String::new(),
                server_capabilities: 0,
                results: vec![],
            }
            .message_type(),
            MessageType::ReconnectAck
        );
        assert_eq!(
            Frame::Fetch {
                topic: TopicName("t".into()),
                offset: TopicOffset(0),
                max_count: 100,
            }
            .message_type(),
            MessageType::Fetch
        );
        assert_eq!(
            Frame::FetchFrom {
                topic: TopicName("t".into()),
                timestamp_millis: 1_000_000,
                max_count: 100,
            }
            .message_type(),
            MessageType::FetchFrom
        );
        assert_eq!(
            Frame::FetchResponse {
                topic: TopicName("t".into()),
                next_offset: TopicOffset(0),
                head_offset: TopicOffset(0),
                messages: vec![],
            }
            .message_type(),
            MessageType::FetchResponse
        );
    }

    #[test]
    fn frame_partial_eq_with_bytes() {
        let a = Frame::Publish {
            topic: TopicName("t".into()),
            payload: Bytes::from_static(b"hello"),
            headers: vec![],
            ttl_ms: 0,
        };
        let b = Frame::Publish {
            topic: TopicName("t".into()),
            payload: Bytes::from(b"hello".to_vec()),
            headers: vec![],
            ttl_ms: 0,
        };
        assert_eq!(a, b);
    }
}
