///|
pub(all) enum DedupStrategy {
  DedupCorrelation
  DedupNameAndCorrelation
  DedupTopicAndPayload
  DedupNameTopicAndPayload
} derive(Eq, Debug)

///|
pub(all) enum DedupDecision {
  DedupAccept(String)
  DedupDuplicate(String, Int)
  DedupUnkeyed
} derive(Eq, Debug)

///|
pub(all) struct SeenEvent {
  key : String
  sequence : Int
  event_name : String
} derive(Eq, Debug)

///|
pub(all) struct DedupCache {
  strategy : DedupStrategy
  capacity : Int
  seen : Array[SeenEvent]
  next_sequence : Int
} derive(Eq, Debug)

///|
pub(all) struct DedupBatch {
  accepted : Array[Event]
  duplicates : Array[Event]
  unkeyed : Array[Event]
} derive(Eq, Debug)

///|
pub(all) struct UniquePublishResult {
  reports : Array[DispatchReport]
  accepted : Int
  duplicates : Int
  unkeyed : Int
} derive(Eq, Debug)

///|
pub fn DedupCache::new(
  strategy? : DedupStrategy = DedupNameAndCorrelation,
  capacity? : Int = 1024,
) -> DedupCache {
  { strategy, capacity, seen: [], next_sequence: 1 }
}

///|
pub fn dedup_key(strategy : DedupStrategy, event : Event) -> String {
  match strategy {
    DedupCorrelation => event.correlation_id
    DedupNameAndCorrelation =>
      if event.correlation_id == "" {
        ""
      } else {
        event.name + "|" + event.correlation_id
      }
    DedupTopicAndPayload =>
      if event.topic == "" && event.payload == "" {
        ""
      } else {
        event.topic + "|" + event.payload
      }
    DedupNameTopicAndPayload =>
      if event.name == "" && event.topic == "" && event.payload == "" {
        ""
      } else {
        event.name + "|" + event.topic + "|" + event.payload
      }
  }
}

///|
pub fn DedupCache::find(self : DedupCache, key : String) -> SeenEvent? {
  for item in self.seen {
    if item.key == key {
      return Some(item)
    }
  }
  None
}

///|
pub fn DedupCache::check(self : DedupCache, event : Event) -> DedupDecision {
  let key = dedup_key(self.strategy, event)
  if key == "" {
    return DedupUnkeyed
  }
  match self.find(key) {
    Some(item) => DedupDuplicate(key, item.sequence)
    None => DedupAccept(key)
  }
}

///|
fn trim_seen(items : Array[SeenEvent], capacity : Int) -> Array[SeenEvent] {
  if capacity <= 0 || items.length() <= capacity {
    return items
  }
  let kept : Array[SeenEvent] = []
  let skip = items.length() - capacity
  let mut index = 0
  for item in items {
    if index >= skip {
      kept.push(item)
    }
    index = index + 1
  }
  kept
}

///|
pub fn DedupCache::remember(self : DedupCache, event : Event) -> DedupCache {
  match self.check(event) {
    DedupAccept(key) => {
      let seen : Array[SeenEvent] = []
      for item in self.seen {
        seen.push(item)
      }
      seen.push({ key, sequence: self.next_sequence, event_name: event.name })
      {
        ..self,
        seen: trim_seen(seen, self.capacity),
        next_sequence: self.next_sequence + 1,
      }
    }
    DedupDuplicate(_, _) | DedupUnkeyed => self
  }
}

///|
pub fn DedupCache::forget(self : DedupCache, key : String) -> DedupCache {
  let seen : Array[SeenEvent] = []
  for item in self.seen {
    if item.key != key {
      seen.push(item)
    }
  }
  { ..self, seen, }
}

///|
pub fn DedupCache::clear(self : DedupCache) -> DedupCache {
  { ..self, seen: [] }
}

///|
pub fn DedupCache::length(self : DedupCache) -> Int {
  self.seen.length()
}

///|
pub fn DedupCache::filter(
  self : DedupCache,
  events : Array[Event],
  accept_unkeyed? : Bool = true,
) -> (DedupCache, DedupBatch) {
  let accepted : Array[Event] = []
  let duplicates : Array[Event] = []
  let unkeyed : Array[Event] = []
  let mut cache = self
  for event in events {
    match cache.check(event) {
      DedupAccept(_) => {
        accepted.push(event)
        cache = cache.remember(event)
      }
      DedupDuplicate(_, _) => duplicates.push(event)
      DedupUnkeyed => {
        unkeyed.push(event)
        if accept_unkeyed {
          accepted.push(event)
        }
      }
    }
  }
  (cache, { accepted, duplicates, unkeyed })
}

///|
pub fn DedupCache::publish_unique(
  self : DedupCache,
  bus : EventBus,
  events : Array[Event],
  accept_unkeyed? : Bool = true,
) -> (DedupCache, EventBus, UniquePublishResult) {
  let (cache, batch) = self.filter(events, accept_unkeyed~)
  let (bus, reports) = bus.publish_many(batch.accepted)
  (
    cache,
    bus,
    {
      reports,
      accepted: batch.accepted.length(),
      duplicates: batch.duplicates.length(),
      unkeyed: batch.unkeyed.length(),
    },
  )
}

///|
pub fn DedupBatch::total(self : DedupBatch) -> Int {
  self.accepted.length() + self.duplicates.length()
}