///|
/// Tracks when an idempotency key was first seen and when it expires.
pub(all) struct IdempotencyRecord {
  event_id : String
  first_seen_ms : Int
  expires_at_ms : Int
} derive(Eq, @debug.Debug)

///|
/// In-memory idempotency store. Keys are derived from event identity by the
/// caller, typically through a SHA-256 digest of the event id and type.
pub struct IdempotencyStore {
  records : @map.HashMap[String, IdempotencyRecord]
} derive(@debug.Debug)

///|
pub fn IdempotencyStore::new() -> IdempotencyStore {
  { records: @map.HashMap([]), }
}

///|
/// Returns true when `key` has not been seen inside the TTL window, and records
/// it in that case. Returns false for duplicate keys inside the window.
pub fn IdempotencyStore::check_and_mark(
  self : IdempotencyStore,
  key : String,
  event_id : String,
  now_ms : Int,
  ttl_ms : Int,
) -> Bool {
  let ttl = if ttl_ms < 0 { 0 } else { ttl_ms }
  let expires_at = now_ms + ttl
  match self.records.get(key) {
    Some(record) =>
      if record.expires_at_ms < now_ms {
        self.records.set(key, {
          event_id,
          first_seen_ms: now_ms,
          expires_at_ms: expires_at,
        })
        true
      } else {
        false
      }
    None => {
      self.records.set(key, {
        event_id,
        first_seen_ms: now_ms,
        expires_at_ms: expires_at,
      })
      true
    }
  }
}

///|
pub fn IdempotencyStore::get(
  self : IdempotencyStore,
  key : String,
) -> IdempotencyRecord? {
  self.records.get(key)
}

///|
pub fn IdempotencyStore::length(self : IdempotencyStore) -> Int {
  self.records.length()
}

///|
/// Removes records whose TTL has passed. Call this from the scheduler before
/// processing a batch so the table does not grow without bound.
pub fn IdempotencyStore::clear_expired(
  self : IdempotencyStore,
  now_ms : Int,
) -> Unit {
  let expired : Array[String] = []
  for item in self.records.iter() {
    if item.1.expires_at_ms < now_ms {
      expired.push(item.0)
    }
  }
  for key in expired {
    self.records.remove(key)
  }
}