///|
pub(all) struct EventBatch {
  name : String
  events : Array[Envelope]
} derive(Eq, @debug.Debug)

///|
pub(all) struct BatchValidationIssue {
  index : Int
  event_id : String
  topic : String
  schema : String
  path : String
  code : String
  message : String
} derive(Eq, @debug.Debug)

///|
pub(all) struct BatchValidationReport {
  name : String
  checked : Int
  valid : Int
  invalid : Int
  issues : Array[BatchValidationIssue]
} derive(Eq, @debug.Debug)

///|
pub(all) struct BatchDecision {
  index : Int
  event_id : String
  topic : String
  decision : RuleDecision
} derive(Eq, @debug.Debug)

///|
pub(all) struct BatchRuleReport {
  name : String
  checked : Int
  accepted : Int
  rejected : Int
  decisions : Array[BatchDecision]
} derive(Eq, @debug.Debug)

///|
pub(all) struct BatchRouteRow {
  index : Int
  event_id : String
  topic : String
  routed : Int
  selected : Array[String]
} derive(Eq, @debug.Debug)

///|
pub(all) struct BatchRouteReport {
  name : String
  events : Int
  routed_events : Int
  unmatched_events : Int
  deliveries_planned : Int
  rows : Array[BatchRouteRow]
} derive(Eq, @debug.Debug)

///|
pub(all) struct BatchPublishReport {
  name : String
  events : Int
  routed_events : Int
  unmatched_events : Int
  deliveries : Int
  dead_letters_added : Int
  lines : Array[String]
} derive(Eq, @debug.Debug)

///|
pub(all) struct BatchTopicCount {
  topic : String
  count : Int
  first_index : Int
  last_index : Int
} derive(Eq, @debug.Debug)

///|
pub fn event_batch(
  name? : StringView = "batch",
  events? : ArrayView[Envelope] = [],
) -> EventBatch {
  { name: name.to_owned(), events: events.to_owned() }
}

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

///|
pub fn EventBatch::is_empty(self : EventBatch) -> Bool {
  self.events.length() == 0
}

///|
pub fn EventBatch::append(self : EventBatch, event : Envelope) -> EventBatch {
  let events = self.events.copy()
  events.push(event)
  { ..self, events, }
}

///|
pub fn EventBatch::append_many(
  self : EventBatch,
  events : ArrayView[Envelope],
) -> EventBatch {
  let appended = self.events.copy()
  for event in events {
    appended.push(event)
  }
  { ..self, events: appended }
}

///|
pub fn EventBatch::with_name(
  self : EventBatch,
  name : StringView,
) -> EventBatch {
  { ..self, name: name.to_owned() }
}

///|
pub fn EventBatch::filter(
  self : EventBatch,
  pattern : StringView,
) -> Result[EventBatch, EventRailError] {
  match topic_pattern(pattern) {
    Err(err) => Err(err)
    Ok(parsed) => {
      let events : Array[Envelope] = []
      for event in self.events {
        match parsed.matches_topic(event.topic) {
          Err(err) => return Err(err)
          Ok(true) => events.push(event)
          Ok(false) => ()
        }
      }
      Ok({ ..self, events, })
    }
  }
}

///|
pub fn EventBatch::to_tape(self : EventBatch) -> EventTape {
  event_tape().append_many(self.events)
}

///|
pub fn EventBatch::compact_by_fingerprint(self : EventBatch) -> EventBatch {
  let seen : Array[String] = []
  let events : Array[Envelope] = []
  for event in self.events {
    let fingerprint = event.fingerprint()
    if !seen.contains(fingerprint) {
      seen.push(fingerprint)
      events.push(event)
    }
  }
  { ..self, events, }
}

///|
pub fn EventBatch::sort_by_timestamp(self : EventBatch) -> EventBatch {
  let events = self.events.copy()
  events.sort_by(compare_event_timestamp)
  { ..self, events, }
}

///|
pub fn EventBatch::topic_counts(self : EventBatch) -> Array[BatchTopicCount] {
  let counts : Array[BatchTopicCount] = []
  for index, event in self.events {
    increment_batch_topic_count(counts, event.topic, index)
  }
  counts
}

///|
pub fn EventBatch::fingerprints(self : EventBatch) -> Array[String] {
  self.events.map(event => event.fingerprint())
}

///|
pub fn EventBatch::manifest_lines(self : EventBatch) -> Array[String] {
  let lines : Array[String] = []
  for index, event in self.events {
    lines.push("idx=\{index};\{event.summary()}")
  }
  lines
}

///|
pub fn EventBatch::manifest(self : EventBatch) -> String {
  self.manifest_lines().join("\n")
}

///|
pub fn EventBatch::validate(
  self : EventBatch,
  schema : EventSchema,
) -> BatchValidationReport {
  let issues : Array[BatchValidationIssue] = []
  let mut valid = 0
  let mut invalid = 0
  for index, event in self.events {
    let report = schema.validate(event)
    if report.valid {
      valid += 1
    } else {
      invalid += 1
      for issue in report.issues {
        issues.push({
          index,
          event_id: event.id,
          topic: event.topic,
          schema: schema.name,
          path: issue.path,
          code: issue.code,
          message: issue.message,
        })
      }
    }
  }
  { name: self.name, checked: self.events.length(), valid, invalid, issues }
}

///|
pub fn EventBatch::evaluate_rules(
  self : EventBatch,
  rules : RuleSet,
) -> BatchRuleReport {
  let decisions : Array[BatchDecision] = []
  let mut accepted = 0
  let mut rejected = 0
  for index, event in self.events {
    let decision = rules.decide(event)
    match decision {
      Accept(_) => accepted += 1
      Reject(_) => rejected += 1
    }
    decisions.push({ index, event_id: event.id, topic: event.topic, decision })
  }
  {
    name: self.name,
    checked: self.events.length(),
    accepted,
    rejected,
    decisions,
  }
}

///|
pub fn EventBatch::route_preview(
  self : EventBatch,
  bus : Bus,
  mode? : DeliveryMode = Fanout,
) -> Result[BatchRouteReport, EventRailError] {
  let rows : Array[BatchRouteRow] = []
  let mut routed_events = 0
  let mut unmatched_events = 0
  let mut deliveries_planned = 0
  for index, event in self.events {
    match bus.route_with_mode(event, mode) {
      Err(err) => return Err(err)
      Ok(routes) => {
        if routes.length() == 0 {
          unmatched_events += 1
        } else {
          routed_events += 1
        }
        deliveries_planned += routes.length()
        rows.push({
          index,
          event_id: event.id,
          topic: event.topic,
          routed: routes.length(),
          selected: routes.map(sub => sub.id),
        })
      }
    }
  }
  Ok({
    name: self.name,
    events: self.events.length(),
    routed_events,
    unmatched_events,
    deliveries_planned,
    rows,
  })
}

///|
pub fn EventBatch::publish(
  self : EventBatch,
  bus : Bus,
  handler : (Subscription, Envelope) -> HandlerResult,
  mode? : DeliveryMode = Fanout,
) -> Result[(Bus, BatchPublishReport), EventRailError] {
  let mut current = bus
  let lines : Array[String] = []
  let mut routed_events = 0
  let mut unmatched_events = 0
  let mut deliveries = 0
  let mut dead_letters_added = 0
  for index, event in self.events {
    match current.publish_with_mode(event, mode, handler) {
      Err(err) => return Err(err)
      Ok((next, report)) => {
        current = next
        if report.routed == 0 {
          unmatched_events += 1
        } else {
          routed_events += 1
        }
        deliveries += report.deliveries.length()
        dead_letters_added += report.dead_letters_added
        lines.push("idx=\{index} \{report.status_line()}")
      }
    }
  }
  Ok(
    (
      current,
      {
        name: self.name,
        events: self.events.length(),
        routed_events,
        unmatched_events,
        deliveries,
        dead_letters_added,
        lines,
      },
    ),
  )
}

///|
pub fn BatchValidationIssue::to_wire(self : BatchValidationIssue) -> String {
  "idx=\{self.index};event=\{escape_wire_text(self.event_id)};topic=\{escape_wire_text(self.topic)};schema=\{escape_wire_text(self.schema)};path=\{escape_wire_text(self.path)};code=\{escape_wire_text(self.code)};message=\{escape_wire_text(self.message)}"
}

///|
pub fn BatchValidationReport::summary(self : BatchValidationReport) -> String {
  "batch=\{escape_wire_text(self.name)} checked=\{self.checked} valid=\{self.valid} invalid=\{self.invalid} issues=\{self.issues.length()}"
}

///|
pub fn BatchValidationReport::issue_lines(
  self : BatchValidationReport,
) -> Array[String] {
  self.issues.map(issue => issue.to_wire())
}

///|
pub fn BatchDecision::to_wire(self : BatchDecision) -> String {
  "idx=\{self.index};event=\{escape_wire_text(self.event_id)};topic=\{escape_wire_text(self.topic)};decision=\{self.decision.to_wire()}"
}

///|
pub fn BatchRuleReport::summary(self : BatchRuleReport) -> String {
  "batch=\{escape_wire_text(self.name)} checked=\{self.checked} accepted=\{self.accepted} rejected=\{self.rejected}"
}

///|
pub fn BatchRuleReport::decision_lines(self : BatchRuleReport) -> Array[String] {
  self.decisions.map(decision => decision.to_wire())
}

///|
pub fn BatchRouteRow::to_wire(self : BatchRouteRow) -> String {
  "idx=\{self.index};event=\{escape_wire_text(self.event_id)};topic=\{escape_wire_text(self.topic)};routed=\{self.routed};selected=\{self.selected.join(",")}"
}

///|
pub fn BatchRouteReport::summary(self : BatchRouteReport) -> String {
  "batch=\{escape_wire_text(self.name)} events=\{self.events} routed=\{self.routed_events} unmatched=\{self.unmatched_events} deliveries=\{self.deliveries_planned}"
}

///|
pub fn BatchRouteReport::route_lines(self : BatchRouteReport) -> Array[String] {
  self.rows.map(row => row.to_wire())
}

///|
pub fn BatchPublishReport::summary(self : BatchPublishReport) -> String {
  "batch=\{escape_wire_text(self.name)} events=\{self.events} routed=\{self.routed_events} unmatched=\{self.unmatched_events} deliveries=\{self.deliveries} dead_letters=\{self.dead_letters_added}"
}

///|
pub fn BatchPublishReport::manifest(self : BatchPublishReport) -> String {
  self.lines.join("\n")
}

///|
pub fn BatchTopicCount::to_wire(self : BatchTopicCount) -> String {
  "topic=\{escape_wire_text(self.topic)};count=\{self.count};first=\{self.first_index};last=\{self.last_index}"
}

///|
pub fn RuleDecision::to_wire(self : RuleDecision) -> String {
  match self {
    Accept(message) => "accept:\{escape_wire_text(message)}"
    Reject(reason) => "reject:\{escape_wire_text(reason)}"
  }
}

///|
fn increment_batch_topic_count(
  counts : Array[BatchTopicCount],
  topic : String,
  index : Int,
) -> Unit {
  for row_index, item in counts {
    if item.topic == topic {
      counts[row_index] = {
        topic: item.topic,
        count: item.count + 1,
        first_index: item.first_index,
        last_index: index,
      }
      return
    }
  }
  counts.push({ topic, count: 1, first_index: index, last_index: index })
}

///|
fn compare_event_timestamp(left : Envelope, right : Envelope) -> Int {
  if left.timestamp_ms != right.timestamp_ms {
    left.timestamp_ms - right.timestamp_ms
  } else if left.topic != right.topic {
    left.topic.compare(right.topic)
  } else {
    left.id.compare(right.id)
  }
}