///|
pub(all) struct RetentionPolicy {
  name : String
  topic : TopicPattern?
  min_timestamp_ms : Int?
  max_timestamp_ms : Int?
  max_events : Int
  max_events_per_topic : Int
  compact_fingerprints : Bool
} derive(Eq, @debug.Debug)

///|
pub(all) struct RetentionDecision {
  index : Int
  sequence : Int
  event_id : String
  topic : String
  timestamp_ms : Int
  fingerprint : String
  keep : Bool
  reason : String
} derive(Eq, @debug.Debug)

///|
pub(all) struct RetentionTopicSummary {
  topic : String
  kept : Int
  dropped : Int
} derive(Eq, @debug.Debug)

///|
pub(all) struct RetentionReport {
  policy : String
  source : String
  scanned : Int
  kept : Int
  dropped : Int
  topic_summaries : Array[RetentionTopicSummary]
  decisions : Array[RetentionDecision]
} derive(Eq, @debug.Debug)

///|
struct RetentionCandidate {
  index : Int
  sequence : Int
  event : Envelope
} derive(Eq, @debug.Debug)

///|
struct RetentionCounter {
  topic : String
  kept : Int
} derive(Eq, @debug.Debug)

///|
pub fn retention_policy(
  name? : StringView = "retention",
  topic? : StringView = "",
  min_timestamp_ms? : Int,
  max_timestamp_ms? : Int,
  max_events? : Int = 0,
  max_events_per_topic? : Int = 0,
  compact_fingerprints? : Bool = false,
) -> Result[RetentionPolicy, EventRailError] {
  let parsed = if topic == "" {
    None
  } else {
    match topic_pattern(topic) {
      Ok(value) => Some(value)
      Err(err) => return Err(err)
    }
  }
  Ok({
    name: name.to_owned(),
    topic: parsed,
    min_timestamp_ms,
    max_timestamp_ms,
    max_events: clamp_non_negative(max_events),
    max_events_per_topic: clamp_non_negative(max_events_per_topic),
    compact_fingerprints,
  })
}

///|
pub fn RetentionPolicy::with_topic(
  self : RetentionPolicy,
  topic : StringView,
) -> Result[RetentionPolicy, EventRailError] {
  match topic_pattern(topic) {
    Ok(value) => Ok({ ..self, topic: Some(value) })
    Err(err) => Err(err)
  }
}

///|
pub fn RetentionPolicy::without_topic(
  self : RetentionPolicy,
) -> RetentionPolicy {
  { ..self, topic: None }
}

///|
pub fn RetentionPolicy::with_time_range(
  self : RetentionPolicy,
  min_timestamp_ms? : Int,
  max_timestamp_ms? : Int,
) -> RetentionPolicy {
  { ..self, min_timestamp_ms, max_timestamp_ms }
}

///|
pub fn RetentionPolicy::with_limits(
  self : RetentionPolicy,
  max_events? : Int = 0,
  max_events_per_topic? : Int = 0,
) -> RetentionPolicy {
  {
    ..self,
    max_events: clamp_non_negative(max_events),
    max_events_per_topic: clamp_non_negative(max_events_per_topic),
  }
}

///|
pub fn RetentionPolicy::with_compaction(
  self : RetentionPolicy,
  compact_fingerprints : Bool,
) -> RetentionPolicy {
  { ..self, compact_fingerprints, }
}

///|
pub fn RetentionPolicy::to_wire(self : RetentionPolicy) -> String {
  let topic = match self.topic {
    Some(pattern) => pattern.raw
    None => "*"
  }
  let min_value = match self.min_timestamp_ms {
    Some(value) => value.to_string()
    None => "*"
  }
  let max_value = match self.max_timestamp_ms {
    Some(value) => value.to_string()
    None => "*"
  }
  "retention=\{escape_wire_text(self.name)};topic=\{escape_wire_text(topic)};min=\{min_value};max=\{max_value};max_events=\{self.max_events};max_per_topic=\{self.max_events_per_topic};compact=\{self.compact_fingerprints}"
}

///|
pub fn RetentionPolicy::bounded(self : RetentionPolicy) -> Bool {
  self.topic is Some(_) ||
  self.min_timestamp_ms is Some(_) ||
  self.max_timestamp_ms is Some(_) ||
  self.max_events > 0 ||
  self.max_events_per_topic > 0 ||
  self.compact_fingerprints
}

///|
pub fn EventBatch::retention_plan(
  self : EventBatch,
  policy : RetentionPolicy,
) -> Result[RetentionReport, EventRailError] {
  let candidates : Array[RetentionCandidate] = []
  for index, event in self.events {
    candidates.push({ index, sequence: index, event })
  }
  build_retention_report("batch:\{self.name}", candidates, policy)
}

///|
pub fn EventTape::retention_plan(
  self : EventTape,
  policy : RetentionPolicy,
) -> Result[RetentionReport, EventRailError] {
  let candidates : Array[RetentionCandidate] = []
  for index, entry in self.entries {
    candidates.push({ index, sequence: entry.sequence, event: entry.event })
  }
  build_retention_report("event-tape", candidates, policy)
}

///|
pub fn EventBatch::apply_retention(
  self : EventBatch,
  policy : RetentionPolicy,
) -> Result[(EventBatch, RetentionReport), EventRailError] {
  match self.retention_plan(policy) {
    Err(err) => Err(err)
    Ok(report) => {
      let kept = report.kept_indices()
      let events : Array[Envelope] = []
      for index, event in self.events {
        if kept.contains(index) {
          events.push(event)
        }
      }
      Ok(({ ..self, events, }, report))
    }
  }
}

///|
pub fn EventTape::apply_retention(
  self : EventTape,
  policy : RetentionPolicy,
) -> Result[(EventTape, RetentionReport), EventRailError] {
  match self.retention_plan(policy) {
    Err(err) => Err(err)
    Ok(report) => {
      let kept = report.kept_sequences()
      let entries : Array[TapeEntry] = []
      for entry in self.entries {
        if kept.contains(entry.sequence) {
          entries.push({ ..entry, sequence: entries.length() })
        }
      }
      Ok(({ entries, next_sequence: entries.length() }, report))
    }
  }
}

///|
pub fn RetentionReport::ok(self : RetentionReport) -> Bool {
  self.kept > 0 || self.scanned == 0
}

///|
pub fn RetentionReport::summary(self : RetentionReport) -> String {
  "retention=\{escape_wire_text(self.policy)} source=\{escape_wire_text(self.source)} scanned=\{self.scanned} kept=\{self.kept} dropped=\{self.dropped} topics=\{self.topic_summaries.length()}"
}

///|
pub fn RetentionReport::kept_indices(self : RetentionReport) -> Array[Int] {
  self.decisions
  .filter(decision => decision.keep)
  .map(decision => decision.index)
}

///|
pub fn RetentionReport::dropped_indices(self : RetentionReport) -> Array[Int] {
  self.decisions
  .filter(decision => !decision.keep)
  .map(decision => decision.index)
}

///|
pub fn RetentionReport::kept_sequences(self : RetentionReport) -> Array[Int] {
  self.decisions
  .filter(decision => decision.keep)
  .map(decision => decision.sequence)
}

///|
pub fn RetentionReport::dropped_sequences(self : RetentionReport) -> Array[Int] {
  self.decisions
  .filter(decision => !decision.keep)
  .map(decision => decision.sequence)
}

///|
pub fn RetentionReport::kept_event_ids(self : RetentionReport) -> Array[String] {
  self.decisions
  .filter(decision => decision.keep)
  .map(decision => decision.event_id)
}

///|
pub fn RetentionReport::dropped_event_ids(
  self : RetentionReport,
) -> Array[String] {
  self.decisions
  .filter(decision => !decision.keep)
  .map(decision => decision.event_id)
}

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

///|
pub fn RetentionReport::keep_lines(self : RetentionReport) -> Array[String] {
  self.decisions
  .filter(decision => decision.keep)
  .map(decision => decision.to_wire())
}

///|
pub fn RetentionReport::drop_lines(self : RetentionReport) -> Array[String] {
  self.decisions
  .filter(decision => !decision.keep)
  .map(decision => decision.to_wire())
}

///|
pub fn RetentionReport::topic_lines(self : RetentionReport) -> Array[String] {
  self.topic_summaries.map(row => row.to_wire())
}

///|
pub fn RetentionReport::manifest_lines(self : RetentionReport) -> Array[String] {
  let lines : Array[String] = []
  lines.push(self.summary())
  for topic in self.topic_summaries {
    lines.push(topic.to_wire())
  }
  for decision in self.decisions {
    lines.push(decision.to_wire())
  }
  lines
}

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

///|
pub fn RetentionDecision::to_wire(self : RetentionDecision) -> String {
  "idx=\{self.index};seq=\{self.sequence};event=\{escape_wire_text(self.event_id)};topic=\{escape_wire_text(self.topic)};timestamp=\{self.timestamp_ms};keep=\{self.keep};reason=\{escape_wire_text(self.reason)}"
}

///|
pub fn RetentionTopicSummary::to_wire(self : RetentionTopicSummary) -> String {
  "topic=\{escape_wire_text(self.topic)};kept=\{self.kept};dropped=\{self.dropped}"
}

///|
fn build_retention_report(
  source : StringView,
  candidates : Array[RetentionCandidate],
  policy : RetentionPolicy,
) -> Result[RetentionReport, EventRailError] {
  let decisions : Array[RetentionDecision] = []
  let counters : Array[RetentionCounter] = []
  let seen_fingerprints : Array[String] = []
  let mut kept = 0
  let mut dropped = 0
  for offset in 0.. return Err(err)
      Ok((true, reason, fingerprint)) => {
        kept += 1
        increment_retention_counter(counters, candidate.event.topic)
        if policy.compact_fingerprints &&
          !seen_fingerprints.contains(fingerprint) {
          seen_fingerprints.push(fingerprint)
        }
        decisions.push(retention_decision(candidate, true, reason, fingerprint))
      }
      Ok((false, reason, fingerprint)) => {
        dropped += 1
        decisions.push(
          retention_decision(candidate, false, reason, fingerprint),
        )
      }
    }
  }
  decisions.sort_by(compare_retention_decision)
  Ok({
    policy: policy.name,
    source: source.to_owned(),
    scanned: candidates.length(),
    kept,
    dropped,
    topic_summaries: summarize_retention_topics(decisions),
    decisions,
  })
}

///|
fn decide_retention(
  candidate : RetentionCandidate,
  policy : RetentionPolicy,
  kept_total : Int,
  counters : Array[RetentionCounter],
  seen_fingerprints : Array[String],
) -> Result[(Bool, String, String), EventRailError] {
  match topic_segments(candidate.event.topic) {
    Err(err) => return Err(err)
    Ok(_) => ()
  }
  match policy.topic {
    Some(pattern) =>
      match pattern.matches_topic(candidate.event.topic) {
        Err(err) => return Err(err)
        Ok(false) =>
          return Ok((false, "topic mismatch", candidate.event.fingerprint()))
        Ok(true) => ()
      }
    None => ()
  }
  match policy.min_timestamp_ms {
    Some(minimum) if candidate.event.timestamp_ms < minimum =>
      return Ok(
        (false, "timestamp before minimum", candidate.event.fingerprint()),
      )
    _ => ()
  }
  match policy.max_timestamp_ms {
    Some(maximum) if candidate.event.timestamp_ms > maximum =>
      return Ok(
        (false, "timestamp after maximum", candidate.event.fingerprint()),
      )
    _ => ()
  }
  let fingerprint = candidate.event.fingerprint()
  if policy.compact_fingerprints && seen_fingerprints.contains(fingerprint) {
    return Ok((false, "duplicate fingerprint", fingerprint))
  }
  if policy.max_events > 0 && kept_total >= policy.max_events {
    return Ok((false, "total limit reached", fingerprint))
  }
  if policy.max_events_per_topic > 0 &&
    retained_topic_count(counters, candidate.event.topic) >=
    policy.max_events_per_topic {
    return Ok((false, "topic limit reached", fingerprint))
  }
  Ok((true, "retained", fingerprint))
}

///|
fn retention_decision(
  candidate : RetentionCandidate,
  keep : Bool,
  reason : String,
  fingerprint : String,
) -> RetentionDecision {
  {
    index: candidate.index,
    sequence: candidate.sequence,
    event_id: candidate.event.id,
    topic: candidate.event.topic,
    timestamp_ms: candidate.event.timestamp_ms,
    fingerprint,
    keep,
    reason,
  }
}

///|
fn summarize_retention_topics(
  decisions : Array[RetentionDecision],
) -> Array[RetentionTopicSummary] {
  let rows : Array[RetentionTopicSummary] = []
  for decision in decisions {
    record_retention_topic(rows, decision.topic, decision.keep)
  }
  rows.sort_by(compare_retention_topic_summary)
  rows
}

///|
fn retained_topic_count(
  counters : Array[RetentionCounter],
  topic : StringView,
) -> Int {
  let wanted = topic.to_owned()
  for counter in counters {
    if counter.topic == wanted {
      return counter.kept
    }
  }
  0
}

///|
fn increment_retention_counter(
  counters : Array[RetentionCounter],
  topic : String,
) -> Unit {
  for index, counter in counters {
    if counter.topic == topic {
      counters[index] = { ..counter, kept: counter.kept + 1 }
      return
    }
  }
  counters.push({ topic, kept: 1 })
}

///|
fn record_retention_topic(
  rows : Array[RetentionTopicSummary],
  topic : String,
  keep : Bool,
) -> Unit {
  for index, row in rows {
    if row.topic == topic {
      rows[index] = if keep {
        { ..row, kept: row.kept + 1 }
      } else {
        { ..row, dropped: row.dropped + 1 }
      }
      return
    }
  }
  rows.push(
    if keep {
      { topic, kept: 1, dropped: 0 }
    } else {
      { topic, kept: 0, dropped: 1 }
    },
  )
}

///|
fn compare_retention_decision(
  left : RetentionDecision,
  right : RetentionDecision,
) -> Int {
  if left.index != right.index {
    left.index - right.index
  } else {
    left.sequence - right.sequence
  }
}

///|
fn compare_retention_topic_summary(
  left : RetentionTopicSummary,
  right : RetentionTopicSummary,
) -> Int {
  left.topic.compare(right.topic)
}

///|
fn clamp_non_negative(value : Int) -> Int {
  if value < 0 {
    0
  } else {
    value
  }
}