///|
/// Time-aware aggregations for event streams.
///
/// These structures make time semantics explicit: callers choose a window
/// width, late-event policy, and watermark rather than relying on wall-clock
/// state hidden inside a model. All aggregators expose bounded counters that
/// are suitable for service health dashboards.
pub(all) enum WindowKind {
  Tumbling
  Sliding
  Session
} derive(Debug, Eq)

///|
pub fn window_kind_catalog() -> Array[WindowKind] {
  [Tumbling, Sliding, Session]
}

///|
pub struct AggregateSummary {
  start : Int64
  end : Int64
  mut count : Int
  mut weight : Double
  mut sum : Double
  mut square_sum : Double
  mut minimum : Double
  mut maximum : Double
}

///|
pub fn AggregateSummary::new(start : Int64, end : Int64) -> AggregateSummary {
  {
    start,
    end,
    count: 0,
    weight: 0.0,
    sum: 0.0,
    square_sum: 0.0,
    minimum: 0.0,
    maximum: 0.0,
  }
}

///|
pub fn AggregateSummary::observe(
  self : AggregateSummary,
  value : Double,
  weight? : Double = 1.0,
) -> Bool {
  if value.is_nan() || value.is_inf() || weight.is_nan() || weight <= 0.0 {
    false
  } else {
    self.count += 1
    self.weight += weight
    self.sum += value * weight
    self.square_sum += value * value * weight
    if self.count == 1 {
      self.minimum = value
      self.maximum = value
    } else {
      if value < self.minimum {
        self.minimum = value
      }
      if value > self.maximum {
        self.maximum = value
      }
    }
    true
  }
}

///|
pub fn AggregateSummary::start(self : AggregateSummary) -> Int64 {
  self.start
}

///|
pub fn AggregateSummary::end(self : AggregateSummary) -> Int64 {
  self.end
}

///|
pub fn AggregateSummary::count(self : AggregateSummary) -> Int {
  self.count
}

///|
pub fn AggregateSummary::weight(self : AggregateSummary) -> Double {
  self.weight
}

///|
pub fn AggregateSummary::sum(self : AggregateSummary) -> Double {
  self.sum
}

///|
pub fn AggregateSummary::mean(self : AggregateSummary) -> Double {
  if self.weight <= 0.0 {
    0.0
  } else {
    self.sum / self.weight
  }
}

///|
pub fn AggregateSummary::variance(self : AggregateSummary) -> Double {
  if self.weight <= 0.0 {
    0.0
  } else {
    let value = self.square_sum / self.weight - self.mean() * self.mean()
    if value < 0.0 {
      0.0
    } else {
      value
    }
  }
}

///|
pub fn AggregateSummary::minimum(self : AggregateSummary) -> Double {
  if self.count == 0 {
    0.0
  } else {
    self.minimum
  }
}

///|
pub fn AggregateSummary::maximum(self : AggregateSummary) -> Double {
  if self.count == 0 {
    0.0
  } else {
    self.maximum
  }
}

///|
pub fn AggregateSummary::merge(
  self : AggregateSummary,
  other : AggregateSummary,
) -> Unit {
  if other.count > 0 {
    if self.count == 0 {
      self.minimum = other.minimum
      self.maximum = other.maximum
    } else {
      if other.minimum < self.minimum {
        self.minimum = other.minimum
      }
      if other.maximum > self.maximum {
        self.maximum = other.maximum
      }
    }
  }
  self.count += other.count
  self.weight += other.weight
  self.sum += other.sum
  self.square_sum += other.square_sum
}

///|
pub fn AggregateSummary::reset(self : AggregateSummary) -> Unit {
  self.count = 0
  self.weight = 0.0
  self.sum = 0.0
  self.square_sum = 0.0
  self.minimum = 0.0
  self.maximum = 0.0
}

///|
pub struct TumblingWindowAggregator {
  width : Int64
  allowed_lateness : Int64
  windows : Map[Int64, AggregateSummary]
  mut watermark : Int64
  mut accepted : Int
  mut late : Int
  mut invalid : Int
}

///|
pub fn TumblingWindowAggregator::new(
  width : Int64,
  allowed_lateness? : Int64 = 0,
) -> TumblingWindowAggregator {
  {
    width: if width <= 0 {
      1
    } else {
      width
    },
    allowed_lateness: if allowed_lateness < 0 {
      0
    } else {
      allowed_lateness
    },
    windows: {},
    watermark: -1,
    accepted: 0,
    late: 0,
    invalid: 0,
  }
}

///|
fn tumbling_start(timestamp : Int64, width : Int64) -> Int64 {
  timestamp - timestamp % width
}

///|
pub fn TumblingWindowAggregator::observe(
  self : TumblingWindowAggregator,
  timestamp : Int64,
  value : Double,
  weight? : Double = 1.0,
) -> Bool {
  if timestamp < 0 || value.is_nan() || value.is_inf() {
    self.invalid += 1
    false
  } else if self.watermark >= 0 &&
    timestamp + self.allowed_lateness < self.watermark {
    self.late += 1
    false
  } else {
    let start = tumbling_start(timestamp, self.width)
    let window = self.windows.get_or_init(start, () => {
      AggregateSummary::new(start, start + self.width)
    })
    let accepted = window.observe(value, weight~)
    if accepted {
      self.accepted += 1
    } else {
      self.invalid += 1
    }
    accepted
  }
}

///|
pub fn TumblingWindowAggregator::advance_watermark(
  self : TumblingWindowAggregator,
  watermark : Int64,
) -> Bool {
  if watermark < self.watermark {
    false
  } else {
    self.watermark = watermark
    true
  }
}

///|
pub fn TumblingWindowAggregator::window(
  self : TumblingWindowAggregator,
  start : Int64,
) -> AggregateSummary? {
  self.windows.get(start)
}

///|
pub fn TumblingWindowAggregator::ready_windows(
  self : TumblingWindowAggregator,
) -> Array[AggregateSummary] {
  let output : Array[AggregateSummary] = []
  self.windows.each((start, summary) => {
    if start + self.width + self.allowed_lateness <= self.watermark {
      output.push(summary)
    }
  })
  output.sort_by((left, right) => {
    if left.start() < right.start() {
      -1
    } else if left.start() > right.start() {
      1
    } else {
      0
    }
  })
  output
}

///|
pub fn TumblingWindowAggregator::close_before(
  self : TumblingWindowAggregator,
  timestamp : Int64,
) -> Array[AggregateSummary] {
  let output : Array[AggregateSummary] = []
  let keys : Array[Int64] = []
  for start in self.windows.keys() {
    if start + self.width <= timestamp {
      keys.push(start)
    }
  }
  keys.sort()
  for start in keys {
    let summary = self.windows.get(start)
    self.windows.remove(start)
    match summary {
      None => ()
      Some(summary) => output.push(summary)
    }
  }
  output
}

///|
pub fn TumblingWindowAggregator::window_count(
  self : TumblingWindowAggregator,
) -> Int {
  self.windows.length()
}

///|
pub fn TumblingWindowAggregator::watermark(
  self : TumblingWindowAggregator,
) -> Int64 {
  self.watermark
}

///|
pub fn TumblingWindowAggregator::accepted(
  self : TumblingWindowAggregator,
) -> Int {
  self.accepted
}

///|
pub fn TumblingWindowAggregator::late(self : TumblingWindowAggregator) -> Int {
  self.late
}

///|
pub fn TumblingWindowAggregator::invalid(
  self : TumblingWindowAggregator,
) -> Int {
  self.invalid
}

///|
pub fn TumblingWindowAggregator::reset(self : TumblingWindowAggregator) -> Unit {
  self.windows.clear()
  self.watermark = -1
  self.accepted = 0
  self.late = 0
  self.invalid = 0
}

///|
pub struct SlidingWindowAggregator {
  width : Int64
  capacity : Int
  timestamps : Array[Int64]
  values : Array[Double]
  weights : Array[Double]
  mut accepted : Int
  mut evicted : Int
  mut invalid : Int
}

///|
pub fn SlidingWindowAggregator::new(
  width : Int64,
  capacity? : Int = 4096,
) -> SlidingWindowAggregator {
  {
    width: if width <= 0 {
      1
    } else {
      width
    },
    capacity: if capacity < 1 {
      1
    } else {
      capacity
    },
    timestamps: [],
    values: [],
    weights: [],
    accepted: 0,
    evicted: 0,
    invalid: 0,
  }
}

///|
fn SlidingWindowAggregator::evict(
  self : SlidingWindowAggregator,
  timestamp : Int64,
) -> Unit {
  while !self.timestamps.is_empty() &&
        timestamp - self.timestamps[0] >= self.width {
    let _ = self.timestamps.remove(0)
    let _ = self.values.remove(0)
    let _ = self.weights.remove(0)
    self.evicted += 1
  }
  while self.timestamps.length() > self.capacity {
    let _ = self.timestamps.remove(0)
    let _ = self.values.remove(0)
    let _ = self.weights.remove(0)
    self.evicted += 1
  }
}

///|
pub fn SlidingWindowAggregator::observe(
  self : SlidingWindowAggregator,
  timestamp : Int64,
  value : Double,
  weight? : Double = 1.0,
) -> Bool {
  if timestamp < 0 ||
    value.is_nan() ||
    value.is_inf() ||
    weight.is_nan() ||
    weight <= 0.0 {
    self.invalid += 1
    false
  } else {
    self.evict(timestamp)
    self.timestamps.push(timestamp)
    self.values.push(value)
    self.weights.push(weight)
    self.accepted += 1
    true
  }
}

///|
pub fn SlidingWindowAggregator::summary(
  self : SlidingWindowAggregator,
  end : Int64,
) -> AggregateSummary {
  let result = AggregateSummary::new(end - self.width, end)
  for i in 0.. Int {
  self.values.length()
}

///|
pub fn SlidingWindowAggregator::capacity(self : SlidingWindowAggregator) -> Int {
  self.capacity
}

///|
pub fn SlidingWindowAggregator::accepted(self : SlidingWindowAggregator) -> Int {
  self.accepted
}

///|
pub fn SlidingWindowAggregator::evicted(self : SlidingWindowAggregator) -> Int {
  self.evicted
}

///|
pub fn SlidingWindowAggregator::invalid(self : SlidingWindowAggregator) -> Int {
  self.invalid
}

///|
pub fn SlidingWindowAggregator::timestamps(
  self : SlidingWindowAggregator,
) -> Array[Int64] {
  self.timestamps.map(value => value)
}

///|
pub fn SlidingWindowAggregator::reset(self : SlidingWindowAggregator) -> Unit {
  self.timestamps.clear()
  self.values.clear()
  self.weights.clear()
  self.accepted = 0
  self.evicted = 0
  self.invalid = 0
}

///|
pub struct SessionSummary {
  start : Int64
  end : Int64
  events : Int
  value : Double
}

///|
pub fn SessionSummary::new(
  start : Int64,
  end : Int64,
  events : Int,
  value : Double,
) -> SessionSummary {
  { start, end, events, value }
}

///|
pub fn SessionSummary::start(self : SessionSummary) -> Int64 {
  self.start
}

///|
pub fn SessionSummary::end(self : SessionSummary) -> Int64 {
  self.end
}

///|
pub fn SessionSummary::events(self : SessionSummary) -> Int {
  self.events
}

///|
pub fn SessionSummary::value(self : SessionSummary) -> Double {
  self.value
}

///|
pub struct SessionWindowAggregator {
  gap : Int64
  mut current_start : Int64?
  mut current_end : Int64?
  mut current_events : Int
  mut current_value : Double
  sessions : Array[SessionSummary]
  mut accepted : Int
  mut invalid : Int
}

///|
pub fn SessionWindowAggregator::new(gap : Int64) -> SessionWindowAggregator {
  {
    gap: if gap <= 0 {
      1
    } else {
      gap
    },
    current_start: None,
    current_end: None,
    current_events: 0,
    current_value: 0.0,
    sessions: [],
    accepted: 0,
    invalid: 0,
  }
}

///|
fn SessionWindowAggregator::finish_current(
  self : SessionWindowAggregator,
) -> Unit {
  match (self.current_start, self.current_end) {
    (Some(start), Some(end)) =>
      self.sessions.push(
        SessionSummary::new(start, end, self.current_events, self.current_value),
      )
    _ => ()
  }
  self.current_start = None
  self.current_end = None
  self.current_events = 0
  self.current_value = 0.0
}

///|
pub fn SessionWindowAggregator::observe(
  self : SessionWindowAggregator,
  timestamp : Int64,
  value : Double,
) -> Bool {
  if timestamp < 0 || value.is_nan() || value.is_inf() {
    self.invalid += 1
    false
  } else {
    match self.current_end {
      Some(end) =>
        if timestamp - end > self.gap {
          self.finish_current()
          self.current_start = Some(timestamp)
        } else {
          ()
        }
      None => self.current_start = Some(timestamp)
    }
    self.current_end = Some(timestamp)
    self.current_events += 1
    self.current_value += value
    self.accepted += 1
    true
  }
}

///|
pub fn SessionWindowAggregator::close(
  self : SessionWindowAggregator,
  timestamp : Int64,
) -> Array[SessionSummary] {
  match self.current_end {
    Some(end) =>
      if timestamp - end > self.gap {
        self.finish_current()
      } else {
        ()
      }
    None => ()
  }
  let result = self.sessions.map(value => value)
  self.sessions.clear()
  result
}

///|
pub fn SessionWindowAggregator::current_events(
  self : SessionWindowAggregator,
) -> Int {
  self.current_events
}

///|
pub fn SessionWindowAggregator::session_count(
  self : SessionWindowAggregator,
) -> Int {
  self.sessions.length()
}

///|
pub fn SessionWindowAggregator::accepted(self : SessionWindowAggregator) -> Int {
  self.accepted
}

///|
pub fn SessionWindowAggregator::invalid(self : SessionWindowAggregator) -> Int {
  self.invalid
}

///|
pub fn SessionWindowAggregator::reset(self : SessionWindowAggregator) -> Unit {
  self.finish_current()
  self.sessions.clear()
  self.accepted = 0
  self.invalid = 0
}

///|
pub struct CountMinSketch {
  width : Int
  depth : Int
  table : Array[Array[Int]]
  seeds : Array[Int]
  mut updates : Int
}

///|
pub fn CountMinSketch::new(width : Int, depth : Int) -> CountMinSketch {
  let safe_width = if width < 1 { 1 } else { width }
  let safe_depth = if depth < 1 { 1 } else { depth }
  {
    width: safe_width,
    depth: safe_depth,
    table: Array::makei(safe_depth, _ => Array::make(safe_width, 0)),
    seeds: Array::makei(safe_depth, i => 17 + i * 7919),
    updates: 0,
  }
}

///|
fn sketch_hash(value : String, seed : Int, width : Int) -> Int {
  let raw = value.hash() + seed * 31
  (if raw < 0 { -raw } else { raw }) % width
}

///|
pub fn CountMinSketch::update(
  self : CountMinSketch,
  key : String,
  amount? : Int = 1,
) -> Unit {
  let value = if amount < 0 { 0 } else { amount }
  for row in 0.. Int {
  if self.depth == 0 {
    0
  } else {
    let mut result = self.table[0][sketch_hash(key, self.seeds[0], self.width)]
    for row in 1.. Int {
  self.width
}

///|
pub fn CountMinSketch::depth(self : CountMinSketch) -> Int {
  self.depth
}

///|
pub fn CountMinSketch::updates(self : CountMinSketch) -> Int {
  self.updates
}

///|
pub fn CountMinSketch::reset(self : CountMinSketch) -> Unit {
  for row in self.table {
    row.fill(0)
  }
  self.updates = 0
}

///|
pub struct HeavyHitter {
  key : String
  count : Int
  error : Int
}

///|
pub fn HeavyHitter::new(key : String, count : Int, error : Int) -> HeavyHitter {
  { key, count, error }
}

///|
pub fn HeavyHitter::key(self : HeavyHitter) -> String {
  self.key
}

///|
pub fn HeavyHitter::count(self : HeavyHitter) -> Int {
  self.count
}

///|
pub fn HeavyHitter::error(self : HeavyHitter) -> Int {
  self.error
}

///|
pub struct HeavyHitters {
  capacity : Int
  entries : Map[String, Int]
  mut updates : Int
  mut evictions : Int
}

///|
pub fn HeavyHitters::new(capacity : Int) -> HeavyHitters {
  {
    capacity: if capacity < 1 {
      1
    } else {
      capacity
    },
    entries: {},
    updates: 0,
    evictions: 0,
  }
}

///|
pub fn HeavyHitters::update(
  self : HeavyHitters,
  key : String,
  amount? : Int = 1,
) -> Unit {
  let value = if amount < 0 { 0 } else { amount }
  self.entries[key] = self.entries.get(key).unwrap_or(0) + value
  self.updates += value
  if self.entries.length() > self.capacity {
    let mut smallest_key = ""
    let mut smallest = 1 << 30
    self.entries.each((candidate, count) => {
      if count < smallest {
        smallest = count
        smallest_key = candidate
      }
    })
    if smallest_key != "" {
      self.entries.remove(smallest_key)
      self.evictions += 1
    }
  }
}

///|
pub fn HeavyHitters::top(
  self : HeavyHitters,
  limit : Int,
) -> Array[HeavyHitter] {
  let result : Array[HeavyHitter] = []
  self.entries.each((key, count) => result.push(HeavyHitter::new(key, count, 0)))
  result.sort_by((left, right) => {
    if left.count() > right.count() {
      -1
    } else if left.count() < right.count() {
      1
    } else {
      left.key().compare(right.key())
    }
  })
  let size = if limit < 0 {
    0
  } else if limit > result.length() {
    result.length()
  } else {
    limit
  }
  result[:size].to_owned()
}

///|
pub fn HeavyHitters::estimate(self : HeavyHitters, key : String) -> Int {
  self.entries.get(key).unwrap_or(0)
}

///|
pub fn HeavyHitters::size(self : HeavyHitters) -> Int {
  self.entries.length()
}

///|
pub fn HeavyHitters::updates(self : HeavyHitters) -> Int {
  self.updates
}

///|
pub fn HeavyHitters::evictions(self : HeavyHitters) -> Int {
  self.evictions
}

///|
pub fn HeavyHitters::reset(self : HeavyHitters) -> Unit {
  self.entries.clear()
  self.updates = 0
  self.evictions = 0
}

///|
pub struct DistinctCounter {
  buckets : Int
  bits : Array[Bool]
  mut observations : Int
}

///|
pub fn DistinctCounter::new(buckets : Int) -> DistinctCounter {
  let size = if buckets < 1 { 1 } else { buckets }
  { buckets: size, bits: Array::make(size, false), observations: 0 }
}

///|
pub fn DistinctCounter::observe(self : DistinctCounter, value : String) -> Unit {
  let hash = value.hash()
  let index = (if hash < 0 { -hash } else { hash }) % self.buckets
  self.bits[index] = true
  self.observations += 1
}

///|
pub fn DistinctCounter::estimate(self : DistinctCounter) -> Double {
  let empty = self.bits.count_if(value => !value)
  if empty == 0 {
    self.buckets.to_double() * 4.0
  } else {
    -self.buckets.to_double() *
    @math.ln(empty.to_double() / self.buckets.to_double())
  }
}

///|
pub fn DistinctCounter::fill_rate(self : DistinctCounter) -> Double {
  self.bits.count_if(value => value).to_double() / self.buckets.to_double()
}

///|
pub fn DistinctCounter::observations(self : DistinctCounter) -> Int {
  self.observations
}

///|
pub fn DistinctCounter::reset(self : DistinctCounter) -> Unit {
  self.bits.fill(false)
  self.observations = 0
}

///|
pub struct EventRateWindow {
  width : Int64
  capacity : Int
  timestamps : Array[Int64]
  mut invalid : Int
}

///|
pub fn EventRateWindow::new(
  width : Int64,
  capacity? : Int = 4096,
) -> EventRateWindow {
  {
    width: if width <= 0 {
      1
    } else {
      width
    },
    capacity: if capacity < 1 {
      1
    } else {
      capacity
    },
    timestamps: Array::make(0, 0L),
    invalid: 0,
  }
}

///|
pub fn EventRateWindow::observe(
  self : EventRateWindow,
  timestamp : Int64,
) -> Bool {
  if timestamp < 0 {
    self.invalid += 1
    false
  } else {
    self.timestamps.push(timestamp)
    self.expire(timestamp)
    while self.timestamps.length() > self.capacity {
      let _ = self.timestamps.remove(0)
    }
    true
  }
}

///|
fn EventRateWindow::expire(self : EventRateWindow, now : Int64) -> Unit {
  while !self.timestamps.is_empty() && now - self.timestamps[0] >= self.width {
    let _ = self.timestamps.remove(0)
  }
}

///|
pub fn EventRateWindow::rate(self : EventRateWindow, now : Int64) -> Double {
  self.expire(now)
  self.timestamps.length().to_double() / self.width.to_double()
}

///|
pub fn EventRateWindow::size(self : EventRateWindow) -> Int {
  self.timestamps.length()
}

///|
pub fn EventRateWindow::capacity(self : EventRateWindow) -> Int {
  self.capacity
}

///|
pub fn EventRateWindow::invalid(self : EventRateWindow) -> Int {
  self.invalid
}

///|
pub fn EventRateWindow::reset(self : EventRateWindow) -> Unit {
  self.timestamps.clear()
  self.invalid = 0
}