krafka 0.8.0

use bytes::{Buf, BufMut};

use super::{VersionedDecode, VersionedEncode};
use crate::error::{ErrorCode, KrafkaError, Result};
use crate::protocol::api::ApiKey;
use crate::protocol::check_compact_array_len;
use crate::protocol::primitives::{Decode, Encode, KafkaString, TaggedFields, TryEncode};

// ---------------------------------------------------------------------------
// ConsumerGroupHeartbeat (API key 68) — KIP-848
// ---------------------------------------------------------------------------

/// Topic-partition pair using topic IDs for the KIP-848 consumer group protocol.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConsumerGroupTopicPartitions {
    /// The topic ID (16-byte UUID).
    pub topic_id: [u8; 16],
    /// The partition indices.
    pub partitions: Vec<i32>,
}

/// ConsumerGroupHeartbeat request (API key 68, KIP-848).
///
/// Members use this to join, leave, and maintain their session with the group
/// coordinator. All versions use flexible encoding (compact strings, compact
/// arrays, tagged fields).
///
/// **Wire format:** Flexible versions 0+.
/// - `MemberEpoch = 0`  → join the group
/// - `MemberEpoch = -1` → leave the group
/// - `MemberEpoch = -2` → static member temporary leave (KIP-345)
///
/// Nullable fields that have not changed since the last heartbeat should be
/// sent as null to reduce bandwidth.
#[derive(Debug, Clone)]
pub struct ConsumerGroupHeartbeatRequest {
    /// The group identifier.
    pub group_id: String,
    /// The member ID (generated by the consumer; must persist for the
    /// lifetime of the consumer process).
    pub member_id: String,
    /// The current member epoch; 0 to join, -1 to leave, -2 for static
    /// member temporary leave.
    pub member_epoch: i32,
    /// The instance ID for static membership (null if not provided or unchanged).
    pub instance_id: Option<String>,
    /// The rack ID of the consumer (null if not provided or unchanged).
    pub rack_id: Option<String>,
    /// The maximum time in milliseconds that the coordinator will wait for
    /// the member to revoke its partitions. -1 if unchanged.
    pub rebalance_timeout_ms: i32,
    /// The subscribed topic names (null if unchanged since last heartbeat).
    pub subscribed_topic_names: Option<Vec<String>>,
    /// The subscribed topic regex (null if unchanged since last heartbeat).
    /// Only present in version 1+ (KIP-848).
    pub subscribed_topic_regex: Option<String>,
    /// The server-side assignor to use (null if not used or unchanged).
    pub server_assignor: Option<String>,
    /// The partitions owned by the member (null if unchanged since last heartbeat).
    pub topic_partitions: Option<Vec<ConsumerGroupTopicPartitions>>,
}

impl ConsumerGroupHeartbeatRequest {
    /// Get the API key.
    pub fn api_key() -> ApiKey {
        ApiKey::ConsumerGroupHeartbeat
    }

    /// Encode for version 0 (flexible).
    pub fn encode_v0(&self, buf: &mut impl BufMut) -> Result<()> {
        // GroupId — compact non-nullable string
        KafkaString::new(&self.group_id).try_encode_compact(buf)?;
        // MemberId — compact non-nullable string
        KafkaString::new(&self.member_id).try_encode_compact(buf)?;
        // MemberEpoch
        self.member_epoch.encode(buf);
        // InstanceId — compact nullable string
        match &self.instance_id {
            Some(id) => KafkaString::new(id).try_encode_compact(buf)?,
            None => KafkaString::null().try_encode_compact(buf)?,
        }
        // RackId — compact nullable string
        match &self.rack_id {
            Some(id) => KafkaString::new(id).try_encode_compact(buf)?,
            None => KafkaString::null().try_encode_compact(buf)?,
        }
        // RebalanceTimeoutMs
        self.rebalance_timeout_ms.encode(buf);
        // SubscribedTopicNames — compact nullable array of compact strings
        Self::encode_subscribed_topic_names(&self.subscribed_topic_names, buf)?;
        // ServerAssignor — compact nullable string
        match &self.server_assignor {
            Some(a) => KafkaString::new(a).try_encode_compact(buf)?,
            None => KafkaString::null().try_encode_compact(buf)?,
        }
        // TopicPartitions — compact nullable array of structs
        self.encode_topic_partitions(buf)?;
        // Tagged fields (none defined)
        TaggedFields::default().try_encode(buf)?;
        Ok(())
    }

    /// Encode for version 1 (flexible).
    ///
    /// Same as v0 but adds `SubscribedTopicRegex` (compact nullable string)
    /// between `SubscribedTopicNames` and `ServerAssignor`.
    pub fn encode_v1(&self, buf: &mut impl BufMut) -> Result<()> {
        // GroupId — compact non-nullable string
        KafkaString::new(&self.group_id).try_encode_compact(buf)?;
        // MemberId — compact non-nullable string
        KafkaString::new(&self.member_id).try_encode_compact(buf)?;
        // MemberEpoch
        self.member_epoch.encode(buf);
        // InstanceId — compact nullable string
        match &self.instance_id {
            Some(id) => KafkaString::new(id).try_encode_compact(buf)?,
            None => KafkaString::null().try_encode_compact(buf)?,
        }
        // RackId — compact nullable string
        match &self.rack_id {
            Some(id) => KafkaString::new(id).try_encode_compact(buf)?,
            None => KafkaString::null().try_encode_compact(buf)?,
        }
        // RebalanceTimeoutMs
        self.rebalance_timeout_ms.encode(buf);
        // SubscribedTopicNames — compact nullable array of compact strings
        Self::encode_subscribed_topic_names(&self.subscribed_topic_names, buf)?;
        // SubscribedTopicRegex — compact nullable string (v1+ only)
        match &self.subscribed_topic_regex {
            Some(r) => KafkaString::new(r).try_encode_compact(buf)?,
            None => KafkaString::null().try_encode_compact(buf)?,
        }
        // ServerAssignor — compact nullable string
        match &self.server_assignor {
            Some(a) => KafkaString::new(a).try_encode_compact(buf)?,
            None => KafkaString::null().try_encode_compact(buf)?,
        }
        // TopicPartitions — compact nullable array of structs
        self.encode_topic_partitions(buf)?;
        // Tagged fields (none defined)
        TaggedFields::default().try_encode(buf)?;
        Ok(())
    }

    /// Encode `SubscribedTopicNames` as a compact nullable array of compact
    /// strings without allocating an intermediate `Vec<KafkaString>`.
    fn encode_subscribed_topic_names(
        names: &Option<Vec<String>>,
        buf: &mut impl BufMut,
    ) -> Result<()> {
        match names {
            None => {
                // null compact array: varint 0
                crate::util::varint::encode_unsigned_varint(0, buf);
            }
            Some(names) => {
                let len_plus_one = u32::try_from(names.len().saturating_add(1)).map_err(|_| {
                    KrafkaError::protocol(format!(
                        "subscribed topic names array length {} exceeds u32 limit",
                        names.len()
                    ))
                })?;
                crate::util::varint::encode_unsigned_varint(len_plus_one, buf);
                for name in names {
                    KafkaString::new(name).try_encode_compact(buf)?;
                }
            }
        }
        Ok(())
    }

    /// Encode the `TopicPartitions` field as a compact nullable array.
    fn encode_topic_partitions(&self, buf: &mut impl BufMut) -> Result<()> {
        match &self.topic_partitions {
            None => {
                // null compact array: varint 0
                crate::util::varint::encode_unsigned_varint(0, buf);
            }
            Some(tps) => {
                let len_plus_one = u32::try_from(tps.len().saturating_add(1)).map_err(|_| {
                    KrafkaError::protocol(format!(
                        "topic partitions array length {} exceeds u32 limit",
                        tps.len()
                    ))
                })?;
                crate::util::varint::encode_unsigned_varint(len_plus_one, buf);
                for tp in tps {
                    // TopicId — 16-byte UUID
                    buf.put_slice(&tp.topic_id);
                    // Partitions — compact array of i32
                    let part_len_plus_one = u32::try_from(tp.partitions.len().saturating_add(1))
                        .map_err(|_| {
                            KrafkaError::protocol(format!(
                                "partitions array length {} exceeds u32 limit",
                                tp.partitions.len()
                            ))
                        })?;
                    crate::util::varint::encode_unsigned_varint(part_len_plus_one, buf);
                    for &p in &tp.partitions {
                        p.encode(buf);
                    }
                    // Tagged fields for the struct
                    TaggedFields::default().try_encode(buf)?;
                }
            }
        }
        Ok(())
    }
}

impl VersionedEncode for ConsumerGroupHeartbeatRequest {
    fn encode_versioned(&self, version: i16, buf: &mut impl BufMut) -> Result<()> {
        match version {
            0 => self.encode_v0(buf)?,
            1 => self.encode_v1(buf)?,
            _ => return unsupported_encode!("ConsumerGroupHeartbeatRequest", version),
        }
        Ok(())
    }
}

/// Assignment in the ConsumerGroupHeartbeat response.
#[derive(Debug, Clone, Default)]
pub struct ConsumerGroupAssignment {
    /// The partitions assigned to the member.
    pub topic_partitions: Vec<ConsumerGroupTopicPartitions>,
}

/// ConsumerGroupHeartbeat response (API key 68, KIP-848).
///
/// The coordinator returns the member's current epoch and assignment.
/// The assignment field is null until the coordinator has computed an
/// assignment for the member.
#[derive(Debug, Clone)]
pub struct ConsumerGroupHeartbeatResponse {
    /// The duration in milliseconds for which the request was throttled.
    pub throttle_time_ms: i32,
    /// The top-level error code, or 0 if there was no error.
    pub error_code: ErrorCode,
    /// The top-level error message, or None if there was no error.
    pub error_message: Option<String>,
    /// The member ID (assigned by the coordinator in v0, generated by
    /// the consumer starting from v1).
    pub member_id: Option<String>,
    /// The member epoch.
    pub member_epoch: i32,
    /// The heartbeat interval in milliseconds.
    pub heartbeat_interval_ms: i32,
    /// The assignment for the member, or None if not yet assigned.
    pub assignment: Option<ConsumerGroupAssignment>,
}

impl ConsumerGroupHeartbeatResponse {
    /// Decode from version 0 (flexible).
    pub fn decode_v0(buf: &mut impl Buf) -> Result<Self> {
        let throttle_time_ms = i32::decode(buf)?;
        let error_code = ErrorCode::from_i16(i16::decode(buf)?);
        let error_message = KafkaString::decode_compact(buf)?.0;
        let member_id = KafkaString::decode_compact(buf)?.0;
        let member_epoch = i32::decode(buf)?;
        let heartbeat_interval_ms = i32::decode(buf)?;

        // Assignment — nullable struct
        let assignment = Self::decode_assignment(buf)?;

        // Skip tagged fields
        let _ = TaggedFields::decode(buf)?;

        Ok(Self {
            throttle_time_ms,
            error_code,
            error_message,
            member_id,
            member_epoch,
            heartbeat_interval_ms,
            assignment,
        })
    }

    /// Decode the assignment field.
    ///
    /// Non-tagged nullable structs in flexible versions use a single signed
    /// byte as presence marker: a negative value (broker writes `-1`) means
    /// null, `1` means the struct fields follow.  This matches the Kafka
    /// generator's reader:
    /// `if (_readable.readByte() < 0) { … null … } else { … read struct … }`.
    fn decode_assignment(buf: &mut impl Buf) -> Result<Option<ConsumerGroupAssignment>> {
        if buf.remaining() < 1 {
            return Err(KrafkaError::protocol(
                "not enough bytes for assignment presence tag",
            ));
        }
        let presence = buf.get_i8();
        if presence < 0 {
            return Ok(None);
        }
        if presence != 1 {
            return Err(KrafkaError::protocol(format!(
                "invalid assignment presence tag: expected negative for null or 1 for present, got {presence}"
            )));
        }

        // Struct is present — decode TopicPartitions compact array + tagged fields.
        let tp_count_raw = crate::util::varint::decode_unsigned_varint(buf)?;
        let topic_partitions = Self::decode_topic_partitions_from_count(tp_count_raw, buf)?;

        // Tagged fields for the Assignment struct
        let _ = TaggedFields::decode(buf)?;

        Ok(Some(ConsumerGroupAssignment { topic_partitions }))
    }

    /// Decode topic partitions given the already-decoded compact array count.
    fn decode_topic_partitions_from_count(
        count: u32,
        buf: &mut impl Buf,
    ) -> Result<Vec<ConsumerGroupTopicPartitions>> {
        let len = check_compact_array_len(count)?;
        let mut result = Vec::with_capacity(len);
        for _ in 0..len {
            // TopicId — 16-byte UUID
            if buf.remaining() < 16 {
                return Err(KrafkaError::protocol("not enough bytes for topic ID UUID"));
            }
            let mut topic_id = [0u8; 16];
            buf.copy_to_slice(&mut topic_id);

            // Partitions — compact array of i32
            let part_count = crate::util::varint::decode_unsigned_varint(buf)?;
            let part_len = check_compact_array_len(part_count)?;
            let mut partitions = Vec::with_capacity(part_len);
            for _ in 0..part_len {
                partitions.push(i32::decode(buf)?);
            }

            // Tagged fields for the struct
            let _ = TaggedFields::decode(buf)?;

            result.push(ConsumerGroupTopicPartitions {
                topic_id,
                partitions,
            });
        }
        Ok(result)
    }
}

impl VersionedDecode for ConsumerGroupHeartbeatResponse {
    fn decode_versioned(version: i16, buf: &mut impl Buf) -> Result<Self> {
        match version {
            // v0 and v1 have identical response wire format.
            0 | 1 => Self::decode_v0(buf),
            _ => unsupported_decode!("ConsumerGroupHeartbeatResponse", version),
        }
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod tests {
    use super::*;
    use crate::protocol::primitives::KafkaArray;
    use crate::protocol::*;

    use bytes::BytesMut;

    // -----------------------------------------------------------------------
    // ConsumerGroupHeartbeat (API key 68, KIP-848)
    // -----------------------------------------------------------------------

    /// Helper: encode a compact string into `buf`.
    /// Non-null string: varint(len + 1) then bytes.
    /// Null string: varint(0).
    fn put_compact_string(buf: &mut BytesMut, s: Option<&str>) {
        match s {
            Some(val) => {
                // len + 1 fits in one byte for small strings
                buf.put_u8((val.len() + 1) as u8);
                buf.put_slice(val.as_bytes());
            }
            None => buf.put_u8(0),
        }
    }

    /// Helper: encode a compact array count (count + 1) as unsigned varint.
    fn put_compact_array_count(buf: &mut BytesMut, count: Option<usize>) {
        match count {
            Some(n) => buf.put_u8((n + 1) as u8),
            None => buf.put_u8(0),
        }
    }

    /// Helper: write empty tagged fields (varint 0).
    fn put_tagged_fields(buf: &mut BytesMut) {
        buf.put_u8(0);
    }

    #[test]
    fn test_consumer_group_heartbeat_request_encode_v0_all_fields() {
        let topic_id: [u8; 16] = [
            0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e,
            0x0f, 0x10,
        ];
        let request = ConsumerGroupHeartbeatRequest {
            group_id: "grp".to_string(),
            member_id: "m1".to_string(),
            member_epoch: 5,
            instance_id: Some("inst".to_string()),
            rack_id: Some("rack-a".to_string()),
            rebalance_timeout_ms: 30_000,
            subscribed_topic_names: Some(vec!["topicA".to_string()]),
            subscribed_topic_regex: None,
            server_assignor: Some("uniform".to_string()),
            topic_partitions: Some(vec![ConsumerGroupTopicPartitions {
                topic_id,
                partitions: vec![0, 1, 2],
            }]),
        };

        let mut buf = BytesMut::new();
        request.encode_v0(&mut buf).unwrap();

        // Decode the buffer and verify field-by-field
        let mut r = buf.freeze();

        // group_id — compact string: varint(3+1)=4, "grp"
        assert_eq!(r.get_u8(), 4); // len+1
        let mut gid = vec![0u8; 3];
        r.copy_to_slice(&mut gid);
        assert_eq!(&gid, b"grp");

        // member_id — compact string: varint(2+1)=3, "m1"
        assert_eq!(r.get_u8(), 3);
        let mut mid = vec![0u8; 2];
        r.copy_to_slice(&mut mid);
        assert_eq!(&mid, b"m1");

        // member_epoch
        assert_eq!(r.get_i32(), 5);

        // instance_id — compact string: varint(4+1)=5, "inst"
        assert_eq!(r.get_u8(), 5);
        let mut iid = vec![0u8; 4];
        r.copy_to_slice(&mut iid);
        assert_eq!(&iid, b"inst");

        // rack_id — compact string: varint(6+1)=7, "rack-a"
        assert_eq!(r.get_u8(), 7);
        let mut rid = vec![0u8; 6];
        r.copy_to_slice(&mut rid);
        assert_eq!(&rid, b"rack-a");

        // rebalance_timeout_ms
        assert_eq!(r.get_i32(), 30_000);

        // subscribed_topic_names — compact array: varint(1+1)=2, then 1 compact string
        assert_eq!(r.get_u8(), 2); // count+1
        assert_eq!(r.get_u8(), 7); // "topicA" len+1
        let mut tn = vec![0u8; 6];
        r.copy_to_slice(&mut tn);
        assert_eq!(&tn, b"topicA");

        // server_assignor — compact string: varint(7+1)=8, "uniform"
        assert_eq!(r.get_u8(), 8);
        let mut sa = vec![0u8; 7];
        r.copy_to_slice(&mut sa);
        assert_eq!(&sa, b"uniform");

        // topic_partitions — compact array: varint(1+1)=2
        assert_eq!(r.get_u8(), 2); // count+1
        // element: 16-byte UUID
        let mut tid = [0u8; 16];
        r.copy_to_slice(&mut tid);
        assert_eq!(tid, topic_id);
        // partitions compact array: varint(3+1)=4
        assert_eq!(r.get_u8(), 4);
        assert_eq!(r.get_i32(), 0);
        assert_eq!(r.get_i32(), 1);
        assert_eq!(r.get_i32(), 2);
        // element tagged fields
        assert_eq!(r.get_u8(), 0);

        // top-level tagged fields
        assert_eq!(r.get_u8(), 0);

        // buffer fully consumed
        assert_eq!(r.remaining(), 0);
    }

    #[test]
    fn test_consumer_group_heartbeat_request_encode_v0_null_optionals() {
        let request = ConsumerGroupHeartbeatRequest {
            group_id: "g".to_string(),
            member_id: "m".to_string(),
            member_epoch: 0,
            instance_id: None,
            rack_id: None,
            rebalance_timeout_ms: -1,
            subscribed_topic_names: None,
            subscribed_topic_regex: None,
            server_assignor: None,
            topic_partitions: None,
        };

        let mut buf = BytesMut::new();
        request.encode_v0(&mut buf).unwrap();

        let mut r = buf.freeze();

        // group_id: varint(2), "g"
        assert_eq!(r.get_u8(), 2);
        assert_eq!(r.get_u8(), b'g');

        // member_id: varint(2), "m"
        assert_eq!(r.get_u8(), 2);
        assert_eq!(r.get_u8(), b'm');

        // member_epoch: 0
        assert_eq!(r.get_i32(), 0);

        // instance_id: null compact string → varint(0)
        assert_eq!(r.get_u8(), 0);

        // rack_id: null compact string → varint(0)
        assert_eq!(r.get_u8(), 0);

        // rebalance_timeout_ms: -1
        assert_eq!(r.get_i32(), -1);

        // subscribed_topic_names: null compact array → varint(0)
        assert_eq!(r.get_u8(), 0);

        // server_assignor: null compact string → varint(0)
        assert_eq!(r.get_u8(), 0);

        // topic_partitions: null compact array → varint(0)
        assert_eq!(r.get_u8(), 0);

        // tagged fields
        assert_eq!(r.get_u8(), 0);

        assert_eq!(r.remaining(), 0);
    }

    #[test]
    fn test_consumer_group_heartbeat_request_leave_epoch() {
        // Epoch -1 means "leave the group"
        let request = ConsumerGroupHeartbeatRequest {
            group_id: "g".to_string(),
            member_id: "m".to_string(),
            member_epoch: -1,
            instance_id: None,
            rack_id: None,
            rebalance_timeout_ms: -1,
            subscribed_topic_names: None,
            subscribed_topic_regex: None,
            server_assignor: None,
            topic_partitions: None,
        };

        let mut buf = BytesMut::new();
        request.encode_v0(&mut buf).unwrap();

        let mut r = buf.freeze();
        // Skip group_id + member_id
        let _ = r.get_u8();
        let _ = r.get_u8(); // "g"
        let _ = r.get_u8();
        let _ = r.get_u8(); // "m"
        // member_epoch
        assert_eq!(r.get_i32(), -1);
    }

    #[test]
    fn test_consumer_group_heartbeat_request_versioned_encode_v0() {
        let request = ConsumerGroupHeartbeatRequest {
            group_id: "g".to_string(),
            member_id: "m".to_string(),
            member_epoch: 0,
            instance_id: None,
            rack_id: None,
            rebalance_timeout_ms: -1,
            subscribed_topic_names: None,
            subscribed_topic_regex: None,
            server_assignor: None,
            topic_partitions: None,
        };

        let mut buf_direct = BytesMut::new();
        request.encode_v0(&mut buf_direct).unwrap();

        let mut buf_versioned = BytesMut::new();
        request.encode_versioned(0, &mut buf_versioned).unwrap();

        assert_eq!(buf_direct, buf_versioned);
    }

    #[test]
    fn test_consumer_group_heartbeat_request_versioned_encode_unsupported() {
        let request = ConsumerGroupHeartbeatRequest {
            group_id: "g".to_string(),
            member_id: "m".to_string(),
            member_epoch: 0,
            instance_id: None,
            rack_id: None,
            rebalance_timeout_ms: -1,
            subscribed_topic_names: None,
            subscribed_topic_regex: None,
            server_assignor: None,
            topic_partitions: None,
        };

        let mut buf = BytesMut::new();
        let result = request.encode_versioned(2, &mut buf);
        assert!(result.is_err());
        let msg = result.unwrap_err().to_string();
        assert!(msg.contains("unsupported"), "got: {msg}");
    }

    #[test]
    fn test_consumer_group_heartbeat_response_decode_v0_with_assignment() {
        let mut buf = BytesMut::new();

        // throttle_time_ms
        buf.put_i32(100);
        // error_code
        buf.put_i16(0);
        // error_message — null compact string
        put_compact_string(&mut buf, None);
        // member_id — "member-1"
        put_compact_string(&mut buf, Some("member-1"));
        // member_epoch
        buf.put_i32(3);
        // heartbeat_interval_ms
        buf.put_i32(5000);

        // assignment — present (presence byte = 0x01)
        buf.put_i8(1);
        // topic_partitions compact array: 1 element → varint(1+1)=2
        put_compact_array_count(&mut buf, Some(1));
        // element: topic_id (16 bytes)
        let topic_id: [u8; 16] = [
            0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff, 0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77,
            0x88, 0x99,
        ];
        buf.put_slice(&topic_id);
        // partitions compact array: 2 elements → varint(2+1)=3
        put_compact_array_count(&mut buf, Some(2));
        buf.put_i32(0);
        buf.put_i32(1);
        // element tagged fields
        put_tagged_fields(&mut buf);
        // assignment tagged fields
        put_tagged_fields(&mut buf);

        // top-level tagged fields
        put_tagged_fields(&mut buf);

        let resp = ConsumerGroupHeartbeatResponse::decode_v0(&mut buf.freeze()).unwrap();
        assert_eq!(resp.throttle_time_ms, 100);
        assert!(resp.error_code.is_ok());
        assert!(resp.error_message.is_none());
        assert_eq!(resp.member_id.as_deref(), Some("member-1"));
        assert_eq!(resp.member_epoch, 3);
        assert_eq!(resp.heartbeat_interval_ms, 5000);

        let assignment = resp.assignment.expect("assignment should be present");
        assert_eq!(assignment.topic_partitions.len(), 1);
        assert_eq!(assignment.topic_partitions[0].topic_id, topic_id);
        assert_eq!(assignment.topic_partitions[0].partitions, vec![0, 1]);
    }

    #[test]
    fn test_consumer_group_heartbeat_response_decode_v0_null_assignment() {
        let mut buf = BytesMut::new();

        // throttle_time_ms
        buf.put_i32(0);
        // error_code
        buf.put_i16(0);
        // error_message — null
        put_compact_string(&mut buf, None);
        // member_id — "m"
        put_compact_string(&mut buf, Some("m"));
        // member_epoch
        buf.put_i32(1);
        // heartbeat_interval_ms
        buf.put_i32(3000);
        // assignment — null (presence byte = 0xff = -1 as i8)
        buf.put_i8(-1);
        // top-level tagged fields
        put_tagged_fields(&mut buf);

        let resp = ConsumerGroupHeartbeatResponse::decode_v0(&mut buf.freeze()).unwrap();
        assert_eq!(resp.throttle_time_ms, 0);
        assert!(resp.error_code.is_ok());
        assert!(resp.member_id.as_deref() == Some("m"));
        assert_eq!(resp.member_epoch, 1);
        assert_eq!(resp.heartbeat_interval_ms, 3000);
        assert!(resp.assignment.is_none());
    }

    #[test]
    fn test_consumer_group_heartbeat_response_decode_v0_invalid_assignment_presence() {
        let mut buf = BytesMut::new();

        // throttle_time_ms
        buf.put_i32(0);
        // error_code
        buf.put_i16(0);
        // error_message — null
        put_compact_string(&mut buf, None);
        // member_id — "m"
        put_compact_string(&mut buf, Some("m"));
        // member_epoch
        buf.put_i32(1);
        // heartbeat_interval_ms
        buf.put_i32(3000);
        // assignment — invalid presence byte (0 is non-negative but != 1)
        buf.put_i8(0);
        // top-level tagged fields
        put_tagged_fields(&mut buf);

        let err = ConsumerGroupHeartbeatResponse::decode_v0(&mut buf.freeze()).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("invalid assignment presence tag"),
            "expected presence tag error, got: {msg}"
        );
    }

    #[test]
    fn test_consumer_group_heartbeat_response_decode_v0_with_error() {
        let mut buf = BytesMut::new();

        // throttle_time_ms
        buf.put_i32(0);
        // error_code: FENCED_MEMBER_EPOCH (110)
        buf.put_i16(110);
        // error_message — "Fenced"
        put_compact_string(&mut buf, Some("Fenced"));
        // member_id — null
        put_compact_string(&mut buf, None);
        // member_epoch
        buf.put_i32(-1);
        // heartbeat_interval_ms
        buf.put_i32(0);
        // assignment — null
        buf.put_i8(-1);
        // tagged fields
        put_tagged_fields(&mut buf);

        let resp = ConsumerGroupHeartbeatResponse::decode_v0(&mut buf.freeze()).unwrap();
        assert!(!resp.error_code.is_ok());
        assert_eq!(resp.error_message.as_deref(), Some("Fenced"));
        assert!(resp.member_id.is_none());
        assert_eq!(resp.member_epoch, -1);
    }

    #[test]
    fn test_consumer_group_heartbeat_response_decode_v0_empty_assignment() {
        let mut buf = BytesMut::new();

        // throttle_time_ms
        buf.put_i32(0);
        // error_code
        buf.put_i16(0);
        // error_message — null
        put_compact_string(&mut buf, None);
        // member_id — "m"
        put_compact_string(&mut buf, Some("m"));
        // member_epoch
        buf.put_i32(2);
        // heartbeat_interval_ms
        buf.put_i32(5000);
        // assignment — present with empty topic_partitions
        buf.put_i8(1);
        // topic_partitions compact array: 0 elements → varint(0+1)=1
        put_compact_array_count(&mut buf, Some(0));
        // assignment tagged fields
        put_tagged_fields(&mut buf);
        // top-level tagged fields
        put_tagged_fields(&mut buf);

        let resp = ConsumerGroupHeartbeatResponse::decode_v0(&mut buf.freeze()).unwrap();
        let assignment = resp.assignment.expect("assignment should be present");
        assert!(assignment.topic_partitions.is_empty());
    }

    #[test]
    fn test_consumer_group_heartbeat_response_versioned_decode_v0() {
        let mut buf = BytesMut::new();
        buf.put_i32(0); // throttle
        buf.put_i16(0); // error_code
        put_compact_string(&mut buf, None); // error_message
        put_compact_string(&mut buf, Some("m")); // member_id
        buf.put_i32(1); // member_epoch
        buf.put_i32(5000); // heartbeat_interval_ms
        buf.put_i8(-1); // assignment null
        put_tagged_fields(&mut buf);

        let resp = ConsumerGroupHeartbeatResponse::decode_versioned(0, &mut buf.freeze()).unwrap();
        assert!(resp.error_code.is_ok());
        assert!(resp.assignment.is_none());
    }

    #[test]
    fn test_consumer_group_heartbeat_response_versioned_decode_unsupported() {
        let mut buf = BytesMut::new();
        buf.put_u8(0); // dummy byte
        let result = ConsumerGroupHeartbeatResponse::decode_versioned(2, &mut buf.freeze());
        assert!(result.is_err());
        let msg = result.unwrap_err().to_string();
        assert!(msg.contains("unsupported"), "got: {msg}");
    }

    #[test]
    fn test_consumer_group_heartbeat_request_encode_decode_roundtrip() {
        // Encode the request, then manually decode it field-by-field to verify
        // the wire format is self-consistent.
        let topic_id = [0xab_u8; 16];
        let request = ConsumerGroupHeartbeatRequest {
            group_id: "test-grp".to_string(),
            member_id: "consumer-1".to_string(),
            member_epoch: 7,
            instance_id: Some("static-1".to_string()),
            rack_id: None,
            rebalance_timeout_ms: 60_000,
            subscribed_topic_names: Some(vec!["t1".to_string(), "t2".to_string()]),
            subscribed_topic_regex: None,
            server_assignor: None,
            topic_partitions: Some(vec![
                ConsumerGroupTopicPartitions {
                    topic_id,
                    partitions: vec![0],
                },
                ConsumerGroupTopicPartitions {
                    topic_id: [0xcd; 16],
                    partitions: vec![1, 2, 3],
                },
            ]),
        };

        let mut buf = BytesMut::new();
        request.encode_v0(&mut buf).unwrap();
        let mut r = buf.freeze();

        // group_id
        let gid = KafkaString::decode_compact(&mut r).unwrap().0.unwrap();
        assert_eq!(gid, "test-grp");
        // member_id
        let mid = KafkaString::decode_compact(&mut r).unwrap().0.unwrap();
        assert_eq!(mid, "consumer-1");
        // member_epoch
        assert_eq!(i32::decode(&mut r).unwrap(), 7);
        // instance_id
        let iid = KafkaString::decode_compact(&mut r).unwrap().0;
        assert_eq!(iid.as_deref(), Some("static-1"));
        // rack_id
        let rid = KafkaString::decode_compact(&mut r).unwrap().0;
        assert!(rid.is_none());
        // rebalance_timeout_ms
        assert_eq!(i32::decode(&mut r).unwrap(), 60_000);
        // subscribed_topic_names
        let arr = KafkaArray::<KafkaString>::decode_compact(&mut r).unwrap();
        let names: Vec<String> = arr.0.unwrap().into_iter().map(|s| s.0.unwrap()).collect();
        assert_eq!(names, vec!["t1", "t2"]);
        // server_assignor
        let sa = KafkaString::decode_compact(&mut r).unwrap().0;
        assert!(sa.is_none());
        // topic_partitions — compact array with 2 elements
        let tp_count = crate::util::varint::decode_unsigned_varint(&mut r).unwrap();
        assert_eq!(tp_count, 3); // 2 + 1
        // first element
        let mut tid1 = [0u8; 16];
        r.copy_to_slice(&mut tid1);
        assert_eq!(tid1, [0xab; 16]);
        let pc1 = crate::util::varint::decode_unsigned_varint(&mut r).unwrap();
        assert_eq!(pc1, 2); // 1 + 1
        assert_eq!(i32::decode(&mut r).unwrap(), 0);
        let _ = TaggedFields::decode(&mut r).unwrap();
        // second element
        let mut tid2 = [0u8; 16];
        r.copy_to_slice(&mut tid2);
        assert_eq!(tid2, [0xcd; 16]);
        let pc2 = crate::util::varint::decode_unsigned_varint(&mut r).unwrap();
        assert_eq!(pc2, 4); // 3 + 1
        assert_eq!(i32::decode(&mut r).unwrap(), 1);
        assert_eq!(i32::decode(&mut r).unwrap(), 2);
        assert_eq!(i32::decode(&mut r).unwrap(), 3);
        let _ = TaggedFields::decode(&mut r).unwrap();
        // top-level tagged fields
        let _ = TaggedFields::decode(&mut r).unwrap();

        assert_eq!(r.remaining(), 0);
    }

    #[test]
    fn test_consumer_group_heartbeat_response_multi_topic_assignment() {
        let mut buf = BytesMut::new();

        buf.put_i32(50); // throttle
        buf.put_i16(0); // error_code
        put_compact_string(&mut buf, None); // error_message
        put_compact_string(&mut buf, Some("mem")); // member_id
        buf.put_i32(10); // member_epoch
        buf.put_i32(4000); // heartbeat_interval_ms

        // assignment present with 2 topics
        buf.put_i8(1);
        put_compact_array_count(&mut buf, Some(2));

        // topic 1: 1 partition
        buf.put_slice(&[0x11; 16]); // topic_id
        put_compact_array_count(&mut buf, Some(1));
        buf.put_i32(5);
        put_tagged_fields(&mut buf);

        // topic 2: 3 partitions
        buf.put_slice(&[0x22; 16]); // topic_id
        put_compact_array_count(&mut buf, Some(3));
        buf.put_i32(0);
        buf.put_i32(1);
        buf.put_i32(2);
        put_tagged_fields(&mut buf);

        // assignment tagged fields
        put_tagged_fields(&mut buf);
        // top-level tagged fields
        put_tagged_fields(&mut buf);

        let resp = ConsumerGroupHeartbeatResponse::decode_v0(&mut buf.freeze()).unwrap();
        assert_eq!(resp.throttle_time_ms, 50);
        assert_eq!(resp.member_epoch, 10);

        let assignment = resp.assignment.unwrap();
        assert_eq!(assignment.topic_partitions.len(), 2);
        assert_eq!(assignment.topic_partitions[0].topic_id, [0x11; 16]);
        assert_eq!(assignment.topic_partitions[0].partitions, vec![5]);
        assert_eq!(assignment.topic_partitions[1].topic_id, [0x22; 16]);
        assert_eq!(assignment.topic_partitions[1].partitions, vec![0, 1, 2]);
    }

    #[test]
    fn test_consumer_group_heartbeat_request_encode_v1_with_regex() {
        let request = ConsumerGroupHeartbeatRequest {
            group_id: "g".to_string(),
            member_id: "m".to_string(),
            member_epoch: 1,
            instance_id: None,
            rack_id: None,
            rebalance_timeout_ms: -1,
            subscribed_topic_names: Some(vec!["t1".to_string()]),
            subscribed_topic_regex: Some("topic-.*".to_string()),
            server_assignor: None,
            topic_partitions: None,
        };

        let mut buf = BytesMut::new();
        request.encode_v1(&mut buf).unwrap();
        let mut r = buf.freeze();

        // group_id
        let gid = KafkaString::decode_compact(&mut r).unwrap().0.unwrap();
        assert_eq!(gid, "g");
        // member_id
        let mid = KafkaString::decode_compact(&mut r).unwrap().0.unwrap();
        assert_eq!(mid, "m");
        // member_epoch
        assert_eq!(i32::decode(&mut r).unwrap(), 1);
        // instance_id (null)
        assert!(KafkaString::decode_compact(&mut r).unwrap().0.is_none());
        // rack_id (null)
        assert!(KafkaString::decode_compact(&mut r).unwrap().0.is_none());
        // rebalance_timeout_ms
        assert_eq!(i32::decode(&mut r).unwrap(), -1);
        // subscribed_topic_names: 1 element
        let stn_count = crate::util::varint::decode_unsigned_varint(&mut r).unwrap();
        assert_eq!(stn_count, 2); // 1 + 1
        let t = KafkaString::decode_compact(&mut r).unwrap().0.unwrap();
        assert_eq!(t, "t1");
        // subscribed_topic_regex: "topic-.*"
        let regex = KafkaString::decode_compact(&mut r).unwrap().0.unwrap();
        assert_eq!(regex, "topic-.*");
        // server_assignor (null)
        assert!(KafkaString::decode_compact(&mut r).unwrap().0.is_none());
        // topic_partitions (null)
        let tp = crate::util::varint::decode_unsigned_varint(&mut r).unwrap();
        assert_eq!(tp, 0); // null compact array
        // tagged fields
        let _ = TaggedFields::decode(&mut r).unwrap();
        assert_eq!(r.remaining(), 0);
    }

    #[test]
    fn test_consumer_group_heartbeat_request_encode_v1_null_regex() {
        let request = ConsumerGroupHeartbeatRequest {
            group_id: "g".to_string(),
            member_id: "m".to_string(),
            member_epoch: 0,
            instance_id: None,
            rack_id: None,
            rebalance_timeout_ms: -1,
            subscribed_topic_names: None,
            subscribed_topic_regex: None,
            server_assignor: None,
            topic_partitions: None,
        };

        let mut buf_v0 = BytesMut::new();
        request.encode_v0(&mut buf_v0).unwrap();
        let mut buf_v1 = BytesMut::new();
        request.encode_v1(&mut buf_v1).unwrap();

        // v1 should be longer by exactly one byte (the null regex compact string = varint 0)
        assert_eq!(buf_v1.len(), buf_v0.len() + 1);
    }

    #[test]
    fn test_consumer_group_heartbeat_request_versioned_encode_v1() {
        let request = ConsumerGroupHeartbeatRequest {
            group_id: "g".to_string(),
            member_id: "m".to_string(),
            member_epoch: 0,
            instance_id: None,
            rack_id: None,
            rebalance_timeout_ms: -1,
            subscribed_topic_names: None,
            subscribed_topic_regex: None,
            server_assignor: None,
            topic_partitions: None,
        };

        let mut buf_direct = BytesMut::new();
        request.encode_v1(&mut buf_direct).unwrap();

        let mut buf_versioned = BytesMut::new();
        request.encode_versioned(1, &mut buf_versioned).unwrap();

        assert_eq!(buf_direct, buf_versioned);
    }

    #[test]
    fn test_consumer_group_heartbeat_response_versioned_decode_v1() {
        // v1 response has the same wire format as v0
        let mut buf = BytesMut::new();
        buf.put_i32(0); // throttle
        buf.put_i16(0); // no error
        put_compact_string(&mut buf, None); // error_message
        put_compact_string(&mut buf, Some("consumer-gen-id")); // member_id
        buf.put_i32(3); // member_epoch
        buf.put_i32(5000); // heartbeat_interval_ms
        buf.put_i8(-1); // null assignment
        put_tagged_fields(&mut buf);

        let resp = ConsumerGroupHeartbeatResponse::decode_versioned(1, &mut buf.freeze()).unwrap();
        assert_eq!(resp.error_code, ErrorCode::None);
        assert_eq!(resp.member_id.as_deref(), Some("consumer-gen-id"));
        assert_eq!(resp.member_epoch, 3);
        assert_eq!(resp.heartbeat_interval_ms, 5000);
        assert!(resp.assignment.is_none());
    }
}