///|
pub(all) enum BatchFailureMode {
  BatchContinue
  BatchStopOnFailure
  BatchStopOnUndelivered
  BatchStopOnStopped
} derive(Eq, Debug)

///|
pub(all) struct BatchOptions {
  max_events : Int
  failure_mode : BatchFailureMode
} derive(Eq, Debug)

///|
pub(all) enum BatchHaltReason {
  BatchNotHalted
  BatchLimitReached
  BatchDispatchFailed(Int)
  BatchUndelivered(Int)
  BatchDispatchStopped(Int)
} derive(Eq, Debug)

///|
pub(all) struct BatchReport {
  input_count : Int
  reports : Array[DispatchReport]
  remaining : EventQueue
  halt_reason : BatchHaltReason
} derive(Eq, Debug)

///|
pub(all) struct EventBatch {
  index : Int
  events : Array[Event]
} derive(Eq, Debug)

///|
pub fn BatchOptions::new(
  max_events? : Int = 0,
  failure_mode? : BatchFailureMode = BatchContinue,
) -> BatchOptions {
  { max_events, failure_mode }
}

///|
fn batch_should_halt(
  mode : BatchFailureMode,
  report : DispatchReport,
  index : Int,
) -> BatchHaltReason {
  match mode {
    BatchContinue => BatchNotHalted
    BatchStopOnFailure =>
      if report.failed > 0 {
        BatchDispatchFailed(index)
      } else {
        BatchNotHalted
      }
    BatchStopOnUndelivered =>
      if report.delivered == 0 {
        BatchUndelivered(index)
      } else {
        BatchNotHalted
      }
    BatchStopOnStopped =>
      if report.stopped {
        BatchDispatchStopped(index)
      } else {
        BatchNotHalted
      }
  }
}

///|
fn append_remaining(
  target : Array[Event],
  events : Array[Event],
  start : Int,
) -> Unit {
  let mut index = 0
  for event in events {
    if index >= start {
      target.push(event)
    }
    index = index + 1
  }
}

///|
pub fn EventBus::publish_batch(
  self : EventBus,
  events : Array[Event],
  options? : BatchOptions = BatchOptions::new(),
) -> (EventBus, BatchReport) {
  let reports : Array[DispatchReport] = []
  let remaining : Array[Event] = []
  let mut bus = self
  let mut index = 0
  let mut halted = false
  let mut reason = BatchNotHalted
  while index < events.length() {
    if options.max_events > 0 && reports.length() >= options.max_events {
      append_remaining(remaining, events, index)
      reason = BatchLimitReached
      halted = true
      break
    }
    let (next, report) = bus.publish(events[index])
    bus = next
    reports.push(report)
    let decision = batch_should_halt(options.failure_mode, report, index)
    if decision != BatchNotHalted {
      append_remaining(remaining, events, index + 1)
      reason = decision
      halted = true
      break
    }
    index = index + 1
  }
  if !halted {
    reason = BatchNotHalted
  }
  (
    bus,
    {
      input_count: events.length(),
      reports,
      remaining: EventQueue::from(remaining),
      halt_reason: reason,
    },
  )
}

///|
pub fn BatchReport::processed(self : BatchReport) -> Int {
  self.reports.length()
}

///|
pub fn BatchReport::remaining_count(self : BatchReport) -> Int {
  self.remaining.length()
}

///|
pub fn BatchReport::completed(self : BatchReport) -> Bool {
  self.remaining.is_empty() && self.halt_reason == BatchNotHalted
}

///|
pub fn BatchReport::failed(self : BatchReport) -> Int {
  let mut count = 0
  for report in self.reports {
    if report.failed > 0 {
      count = count + 1
    }
  }
  count
}

///|
pub fn BatchReport::undelivered(self : BatchReport) -> Int {
  let mut count = 0
  for report in self.reports {
    if report.delivered == 0 {
      count = count + 1
    }
  }
  count
}

///|
pub fn BatchReport::metrics(self : BatchReport) -> BusMetrics {
  metrics_from_reports(self.reports)
}

///|
pub fn BatchReport::continue_with(
  self : BatchReport,
  bus : EventBus,
  options? : BatchOptions = BatchOptions::new(),
) -> (EventBus, BatchReport) {
  bus.publish_batch(self.remaining.events, options~)
}

///|
pub fn chunk_events(events : Array[Event], size : Int) -> Array[EventBatch] {
  let batches : Array[EventBatch] = []
  if size <= 0 {
    batches.push({ index: 0, events })
    return batches
  }
  let mut current : Array[Event] = []
  let mut batch_index = 0
  for event in events {
    current.push(event)
    if current.length() == size {
      batches.push({ index: batch_index, events: current })
      current = []
      batch_index = batch_index + 1
    }
  }
  if current.length() > 0 {
    batches.push({ index: batch_index, events: current })
  }
  batches
}

///|
pub fn EventBatch::length(self : EventBatch) -> Int {
  self.events.length()
}

///|
pub fn EventBatch::publish(
  self : EventBatch,
  bus : EventBus,
  options? : BatchOptions = BatchOptions::new(),
) -> (EventBus, BatchReport) {
  bus.publish_batch(self.events, options~)
}

///|
pub fn flatten_batches(batches : Array[EventBatch]) -> Array[Event] {
  let events : Array[Event] = []
  for batch in batches {
    for event in batch.events {
      events.push(event)
    }
  }
  events
}