// OffsetForLeaderEpoch v4 — leader-epoch end offsets, the truncation
// detection the fetcher (Phase 4) runs after UNKNOWN_LEADER_EPOCH /
// FENCED_LEADER_EPOCH fetch errors. Pinned from
// OffsetForLeaderEpochRequest.json / OffsetForLeaderEpochResponse.json in
// the Kafka 4.3 tree: v4 is the first flexible version; consumers send
// ReplicaId -1 and address topics by name at every in-range version.

///|
/// One partition's epoch end offset.
pub struct EpochEndOffset {
  partition : Int
  error_code : Int
  leader_epoch : Int
  end_offset : Int64
} derive(@debug.Debug)

///|
/// Encode an OffsetForLeaderEpoch v4 request body: one topic, one entry
/// per partition, asking for the last offset of `leader_epoch` (the
/// consumer sends its current leader epoch for fencing).
pub fn encode_offset_for_leader_epoch_request(
  topic : String,
  partitions : Array[(Int, Int, Int)],
  timeout_ms? : Int = 0,
) -> Bytes {
  let body = @buf.Encoder::new()
  body.write_i32(-1) // replica_id: consumer
  body.write_compact_len(1) // one topic
  body.write_compact_string(topic)
  body.write_compact_len(partitions.length())
  for entry in partitions {
    let (partition, current_leader_epoch, leader_epoch) = entry
    body.write_i32(partition)
    body.write_i32(current_leader_epoch)
    body.write_i32(leader_epoch)
    body.write_tag_buffer()
  }
  body.write_tag_buffer()
  body.write_tag_buffer()
  ignore(timeout_ms)
  body.to_bytes()
}

///|
/// Decode an OffsetForLeaderEpoch v4 response body: per-partition epoch
/// end offsets plus the throttle hint. UNDEFINED_EPOCH end offset (-1)
/// means the broker has no data for the requested epoch.
pub fn decode_offset_for_leader_epoch_response(
  d : @buf.Decoder,
) -> (Array[EpochEndOffset], Int) raise {
  let throttle = d.read_i32()
  let out : Array[EpochEndOffset] = []
  let n = d.read_compact_len()
  for _ in 0.. Array[EpochEndOffset] {
  let version = self.api_version(API_OFFSET_FOR_LEADER_EPOCH)
  let d = self.request(
    API_OFFSET_FOR_LEADER_EPOCH,
    version,
    encode_offset_for_leader_epoch_request(topic, partitions),
    timeout_ms~,
  )
  let (results, throttle) = decode_offset_for_leader_epoch_response(d)
  self.note_throttle(throttle)
  results
}