///|
pub(all) struct EventDigest {
  index : Int
  event_id : String
  topic : String
  timestamp_ms : Int
  attempt : Int
  fingerprint : String
  header_count : Int
  trace_count : Int
  payload_fields : Int
} derive(Eq, @debug.Debug)

///|
pub(all) struct SnapshotDeliveryRow {
  subscription_id : String
  stats : DeliveryStats
} derive(Eq, @debug.Debug)

///|
pub(all) struct EventSnapshot {
  name : String
  source : String
  timestamp_ms : Int
  event_count : Int
  delivery_count : Int
  dead_letter_count : Int
  digests : Array[EventDigest]
  topic_counts : Array[TopicCount]
  delivery_stats : DeliveryStats
  subscription_stats : Array[SnapshotDeliveryRow]
} derive(Eq, @debug.Debug)

///|
pub(all) enum SnapshotChangeKind {
  SnapshotAdded
  SnapshotRemoved
  SnapshotChanged
} derive(Eq, @debug.Debug)

///|
pub(all) struct SnapshotChange {
  kind : SnapshotChangeKind
  event_id : String
  before_fingerprint : String
  after_fingerprint : String
} derive(Eq, @debug.Debug)

///|
pub(all) struct SnapshotDiff {
  left : String
  right : String
  added : Int
  removed : Int
  changed : Int
  changes : Array[SnapshotChange]
} derive(Eq, @debug.Debug)

///|
pub fn event_snapshot(
  name : StringView,
  source : StringView,
  events : ArrayView[Envelope],
  bus : Bus,
  timestamp_ms? : Int = 0,
) -> EventSnapshot {
  let tape = event_tape().append_many(events)
  {
    name: name.to_owned(),
    source: source.to_owned(),
    timestamp_ms,
    event_count: events.length(),
    delivery_count: bus.delivery_count(),
    dead_letter_count: bus.dead_letter_count(),
    digests: build_event_digests(events),
    topic_counts: tape.topic_counts(),
    delivery_stats: bus.delivery_stats(),
    subscription_stats: snapshot_subscription_stats(bus),
  }
}

///|
pub fn EventBatch::snapshot(
  self : EventBatch,
  bus : Bus,
  name? : StringView = "batch-snapshot",
  timestamp_ms? : Int = 0,
) -> EventSnapshot {
  event_snapshot(name, self.name, self.events, bus, timestamp_ms~)
}

///|
pub fn EventTape::snapshot(
  self : EventTape,
  bus : Bus,
  name? : StringView = "tape-snapshot",
  timestamp_ms? : Int = 0,
) -> EventSnapshot {
  event_snapshot(
    name,
    "event-tape",
    self.entries.map(entry => entry.event),
    bus,
    timestamp_ms~,
  )
}

///|
pub fn EventSnapshot::summary(self : EventSnapshot) -> String {
  "snapshot=\{escape_wire_text(self.name)} source=\{escape_wire_text(self.source)} events=\{self.event_count} deliveries=\{self.delivery_count} dead_letters=\{self.dead_letter_count} topics=\{self.topic_counts.length()}"
}

///|
pub fn EventSnapshot::event_ids(self : EventSnapshot) -> Array[String] {
  self.digests.map(digest => digest.event_id)
}

///|
pub fn EventSnapshot::find_event(
  self : EventSnapshot,
  event_id : StringView,
) -> EventDigest? {
  let wanted = event_id.to_owned()
  for digest in self.digests {
    if digest.event_id == wanted {
      return Some(digest)
    }
  }
  None
}

///|
pub fn EventSnapshot::topic_count(
  self : EventSnapshot,
  topic : StringView,
) -> Int {
  let wanted = topic.to_owned()
  for row in self.topic_counts {
    if row.topic == wanted {
      return row.count
    }
  }
  0
}

///|
pub fn EventSnapshot::subscription_stats_for(
  self : EventSnapshot,
  subscription_id : StringView,
) -> DeliveryStats? {
  let wanted = subscription_id.to_owned()
  for row in self.subscription_stats {
    if row.subscription_id == wanted {
      return Some(row.stats)
    }
  }
  None
}

///|
pub fn EventSnapshot::manifest_lines(self : EventSnapshot) -> Array[String] {
  let lines : Array[String] = []
  lines.push(self.summary())
  lines.push("delivery=\{self.delivery_stats.to_wire()}")
  for topic in self.topic_counts {
    lines.push(
      "topic=\{escape_wire_text(topic.topic)};count=\{topic.count};first=\{topic.first_sequence};last=\{topic.last_sequence}",
    )
  }
  for row in self.subscription_stats {
    lines.push(row.to_wire())
  }
  for digest in self.digests {
    lines.push(digest.to_wire())
  }
  lines
}

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

///|
pub fn EventSnapshot::diff(
  self : EventSnapshot,
  other : EventSnapshot,
) -> SnapshotDiff {
  let changes : Array[SnapshotChange] = []
  let mut added = 0
  let mut removed = 0
  let mut changed = 0
  for digest in self.digests {
    match other.find_event(digest.event_id) {
      None => {
        removed += 1
        changes.push({
          kind: SnapshotRemoved,
          event_id: digest.event_id,
          before_fingerprint: digest.fingerprint,
          after_fingerprint: "",
        })
      }
      Some(next) =>
        if next.fingerprint != digest.fingerprint {
          changed += 1
          changes.push({
            kind: SnapshotChanged,
            event_id: digest.event_id,
            before_fingerprint: digest.fingerprint,
            after_fingerprint: next.fingerprint,
          })
        }
    }
  }
  for digest in other.digests {
    match self.find_event(digest.event_id) {
      None => {
        added += 1
        changes.push({
          kind: SnapshotAdded,
          event_id: digest.event_id,
          before_fingerprint: "",
          after_fingerprint: digest.fingerprint,
        })
      }
      Some(_) => ()
    }
  }
  { left: self.name, right: other.name, added, removed, changed, changes }
}

///|
pub fn EventDigest::to_wire(self : EventDigest) -> String {
  "idx=\{self.index};event=\{escape_wire_text(self.event_id)};topic=\{escape_wire_text(self.topic)};timestamp=\{self.timestamp_ms};attempt=\{self.attempt};headers=\{self.header_count};trace=\{self.trace_count};fields=\{self.payload_fields};fingerprint=\{escape_wire_text(self.fingerprint)}"
}

///|
pub fn SnapshotDeliveryRow::to_wire(self : SnapshotDeliveryRow) -> String {
  "subscription=\{escape_wire_text(self.subscription_id)};\{self.stats.to_wire()}"
}

///|
pub fn SnapshotChangeKind::to_wire(self : SnapshotChangeKind) -> String {
  match self {
    SnapshotAdded => "added"
    SnapshotRemoved => "removed"
    SnapshotChanged => "changed"
  }
}

///|
pub fn SnapshotChange::to_wire(self : SnapshotChange) -> String {
  "kind=\{self.kind.to_wire()};event=\{escape_wire_text(self.event_id)};before=\{escape_wire_text(self.before_fingerprint)};after=\{escape_wire_text(self.after_fingerprint)}"
}

///|
pub fn SnapshotDiff::summary(self : SnapshotDiff) -> String {
  "diff=\{escape_wire_text(self.left)}..\{escape_wire_text(self.right)} added=\{self.added} removed=\{self.removed} changed=\{self.changed}"
}

///|
pub fn SnapshotDiff::change_lines(self : SnapshotDiff) -> Array[String] {
  self.changes.map(change => change.to_wire())
}

///|
fn build_event_digests(events : ArrayView[Envelope]) -> Array[EventDigest] {
  let digests : Array[EventDigest] = []
  for index, event in events {
    digests.push({
      index,
      event_id: event.id,
      topic: event.topic,
      timestamp_ms: event.timestamp_ms,
      attempt: event.attempt,
      fingerprint: event.fingerprint(),
      header_count: event.headers.length(),
      trace_count: event.trace.length(),
      payload_fields: event.payload.field_count(),
    })
  }
  digests
}

///|
fn snapshot_subscription_stats(bus : Bus) -> Array[SnapshotDeliveryRow] {
  let rows : Array[SnapshotDeliveryRow] = []
  for row in bus.subscription_delivery_stats() {
    rows.push({ subscription_id: row.subscription_id, stats: row.stats })
  }
  rows.sort_by(compare_snapshot_delivery_row)
  rows
}

///|
fn compare_snapshot_delivery_row(
  left : SnapshotDeliveryRow,
  right : SnapshotDeliveryRow,
) -> Int {
  left.subscription_id.compare(right.subscription_id)
}