// Track acknowledgments inside a broker batch. A batch entry must not be
// acknowledged as a whole until every contained message was acknowledged.

///|
priv struct BatchAckState {
  acknowledged : Array[Bool]
  generation : Int
}

///|
struct BatchAckTracker {
  states : Map[MessageId, BatchAckState]
  enabled : Bool
  mu : @async.Mutex
  mut next_generation : Int
}

///|
fn BatchAckTracker::new(enabled : Bool) -> BatchAckTracker {
  { states: {}, enabled, mu: @async.Mutex::Mutex(), next_generation: 0, }
}

///|
fn batch_entry_id(id : MessageId) -> MessageId {
  { ..id, batch_index: -1, }
}

///|
fn BatchAckTracker::register(
  self : BatchAckTracker,
  id : MessageId,
  size : Int,
  broker_ack_set : Array[Int64],
) -> Unit {
  let key = batch_entry_id(id)
  let acknowledged : Array[Bool] = []
  for _ in 0..= broker_ack_set.length() ||
        (
          (broker_ack_set[index / 64].reinterpret_as_uint64() >> (index % 64)) &
          1UL
        ) ==
        0UL {
        acknowledged[index] = true
      }
    }
  }
  self.next_generation += 1
  self.states[key] = { acknowledged, generation: self.next_generation, }
}

///|
fn BatchAckTracker::clear(self : BatchAckTracker) -> Unit {
  self.states.clear()
}

///|
fn batch_ack_set(
  acknowledged : Array[Bool],
  through : Int,
  cumulative : Bool,
  only_current? : Bool = false,
) -> Array[Int64] {
  let words : Array[Int64] = []
  for _ in 0..<((acknowledged.length() + 63) / 64) {
    words.push(0L)
  }
  for index in 0.. Unit {
  self.mu.acquire()
  defer self.mu.release()
  let key = batch_entry_id(id)
  guard self.states.get(key) is Some(state) &&
    id.batch_index >= 0 &&
    id.batch_index < state.acknowledged.length() else {
    raise InvalidFrame("batch ACK requires a received message and batch size")
  }
  if ack_type is @proto.CommandAck_AckType::Individual &&
    state.acknowledged[id.batch_index] {
    return
  }
  let cumulative = ack_type is @proto.CommandAck_AckType::Cumulative
  let mut all_acked = true
  for index in 0..