///|
/// A timestamped scalar record accepted by the streaming runtime.
pub struct StreamRecord {
  timestamp : Int
  value : Double
  source : String
  sequence : Int
  valid : Bool
} derive(Debug)

///|
pub fn StreamRecord::new(
  timestamp : Int,
  value : Double,
  source : String,
  sequence : Int,
) -> StreamRecord {
  {
    timestamp,
    value,
    source,
    sequence,
    valid: !value.is_nan() && !value.is_inf(),
  }
}

///|
pub fn StreamRecord::timestamp(self : StreamRecord) -> Int {
  self.timestamp
}

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

///|
pub fn StreamRecord::source(self : StreamRecord) -> String {
  self.source
}

///|
pub fn StreamRecord::sequence(self : StreamRecord) -> Int {
  self.sequence
}

///|
pub fn StreamRecord::valid(self : StreamRecord) -> Bool {
  self.valid
}

///|
pub fn StreamRecord::age(self : StreamRecord, now : Int) -> Int {
  now - self.timestamp
}

///|
pub fn StreamRecord::with_value(
  self : StreamRecord,
  value : Double,
) -> StreamRecord {
  StreamRecord::new(self.timestamp, value, self.source, self.sequence)
}

///|
/// Window semantics for event-time processing.
pub struct StreamWindowSpec {
  width : Int
  lateness : Int
  capacity : Int
  min_samples : Int
  allow_out_of_order : Bool
} derive(Debug)

///|
pub fn StreamWindowSpec::new(
  width : Int,
  lateness : Int,
  capacity : Int,
  min_samples : Int,
) -> StreamWindowSpec {
  {
    width: width.max(1),
    lateness: lateness.max(0),
    capacity: capacity.max(1),
    min_samples: min_samples.max(1),
    allow_out_of_order: true,
  }
}

///|
pub fn StreamWindowSpec::width(self : StreamWindowSpec) -> Int {
  self.width
}

///|
pub fn StreamWindowSpec::lateness(self : StreamWindowSpec) -> Int {
  self.lateness
}

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

///|
pub fn StreamWindowSpec::min_samples(self : StreamWindowSpec) -> Int {
  self.min_samples
}

///|
pub fn StreamWindowSpec::allow_out_of_order(self : StreamWindowSpec) -> Bool {
  self.allow_out_of_order
}

///|
pub fn StreamWindowSpec::with_out_of_order(
  self : StreamWindowSpec,
  enabled : Bool,
) -> StreamWindowSpec {
  {
    width: self.width,
    lateness: self.lateness,
    capacity: self.capacity,
    min_samples: self.min_samples,
    allow_out_of_order: enabled,
  }
}

///|
/// Event-time window with bounded memory and watermark accounting.
pub struct StreamWindow {
  spec : StreamWindowSpec
  records : Array[StreamRecord]
  mut watermark : Int?
  mut max_timestamp : Int?
  mut late_count : Int
  mut invalid_count : Int
  mut evicted_count : Int
} derive(Debug)

///|
pub fn StreamWindow::new(spec : StreamWindowSpec) -> StreamWindow {
  {
    spec,
    records: [],
    watermark: None,
    max_timestamp: None,
    late_count: 0,
    invalid_count: 0,
    evicted_count: 0,
  }
}

///|
pub fn StreamWindow::spec(self : StreamWindow) -> StreamWindowSpec {
  self.spec
}

///|
pub fn StreamWindow::records(self : StreamWindow) -> Array[StreamRecord] {
  self.records.copy()
}

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

///|
pub fn StreamWindow::watermark(self : StreamWindow) -> Int? {
  self.watermark
}

///|
pub fn StreamWindow::late_count(self : StreamWindow) -> Int {
  self.late_count
}

///|
pub fn StreamWindow::invalid_count(self : StreamWindow) -> Int {
  self.invalid_count
}

///|
pub fn StreamWindow::evicted_count(self : StreamWindow) -> Int {
  self.evicted_count
}

///|
fn StreamWindow::stream_window_is_late(
  self : StreamWindow,
  record : StreamRecord,
) -> Bool {
  match self.watermark {
    None => false
    Some(mark) => record.timestamp() < mark - self.spec.lateness()
  }
}

///|
fn StreamWindow::stream_window_insert_sorted(
  self : StreamWindow,
  record : StreamRecord,
) -> Unit {
  let mut index = self.records.length()
  for i, existing in self.records {
    if record.timestamp() < existing.timestamp() {
      index = i
      break
    }
  }
  self.records.push(record)
  let last = self.records.length() - 1
  let mut i = last
  while i > index {
    self.records[i] = self.records[i - 1]
    i = i - 1
  }
  self.records[index] = record
}

///|
fn StreamWindow::stream_window_evict(self : StreamWindow) -> Unit {
  match self.watermark {
    None => ()
    Some(mark) => {
      let cutoff = mark - self.spec.width()
      while self.records.length() > 0 && self.records[0].timestamp() < cutoff {
        self.records.remove(0) |> ignore
        self.evicted_count = self.evicted_count + 1
      }
    }
  }
  while self.records.length() > self.spec.capacity() {
    self.records.remove(0) |> ignore
    self.evicted_count = self.evicted_count + 1
  }
}

///|
/// Push a record, returning whether it participates in the active window.
pub fn StreamWindow::push(self : StreamWindow, record : StreamRecord) -> Bool {
  if !record.valid() {
    self.invalid_count = self.invalid_count + 1
    return false
  }
  if self.stream_window_is_late(record) && !self.spec.allow_out_of_order() {
    self.late_count = self.late_count + 1
    return false
  }
  if self.stream_window_is_late(record) {
    self.late_count = self.late_count + 1
  }
  match self.max_timestamp {
    None => self.max_timestamp = Some(record.timestamp())
    Some(value) =>
      if record.timestamp() > value {
        self.max_timestamp = Some(record.timestamp())
      }
  }
  self.stream_window_insert_sorted(record)
  match self.max_timestamp {
    None => ()
    Some(maximum) => self.watermark = Some(maximum - self.spec.lateness())
  }
  self.stream_window_evict()
  true
}

///|
pub fn StreamWindow::advance_watermark(
  self : StreamWindow,
  timestamp : Int,
) -> Unit {
  match self.watermark {
    None => self.watermark = Some(timestamp)
    Some(value) => if timestamp > value { self.watermark = Some(timestamp) }
  }
  self.stream_window_evict()
}

///|
pub fn StreamWindow::clear(self : StreamWindow) -> Unit {
  self.records.clear()
  self.watermark = None
  self.max_timestamp = None
}

///|
pub fn StreamWindow::span(self : StreamWindow) -> Int {
  if self.records.length() < 2 {
    0
  } else {
    self.records[self.records.length() - 1].timestamp() -
    self.records[0].timestamp()
  }
}

///|
pub fn StreamWindow::contains_timestamp(
  self : StreamWindow,
  timestamp : Int,
) -> Bool {
  for record in self.records {
    if record.timestamp() == timestamp {
      return true
    }
  }
  false
}

///|
/// Aggregate statistics for a window, computed without retaining mutable state.
pub struct StreamStatistics {
  count : Int
  mean : Double
  variance : Double
  minimum : Double
  maximum : Double
  first_timestamp : Int?
  last_timestamp : Int?
  slope : Double
  valid : Bool
} derive(Debug)

///|
pub fn StreamStatistics::empty() -> StreamStatistics {
  {
    count: 0,
    mean: 0.0,
    variance: 0.0,
    minimum: 0.0,
    maximum: 0.0,
    first_timestamp: None,
    last_timestamp: None,
    slope: 0.0,
    valid: false,
  }
}

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

///|
pub fn StreamStatistics::mean(self : StreamStatistics) -> Double {
  self.mean
}

///|
pub fn StreamStatistics::variance(self : StreamStatistics) -> Double {
  self.variance
}

///|
pub fn StreamStatistics::standard_deviation(self : StreamStatistics) -> Double {
  self.variance.max(0.0).sqrt()
}

///|
pub fn StreamStatistics::minimum(self : StreamStatistics) -> Double {
  self.minimum
}

///|
pub fn StreamStatistics::maximum(self : StreamStatistics) -> Double {
  self.maximum
}

///|
pub fn StreamStatistics::first_timestamp(self : StreamStatistics) -> Int? {
  self.first_timestamp
}

///|
pub fn StreamStatistics::last_timestamp(self : StreamStatistics) -> Int? {
  self.last_timestamp
}

///|
pub fn StreamStatistics::slope(self : StreamStatistics) -> Double {
  self.slope
}

///|
pub fn StreamStatistics::valid(self : StreamStatistics) -> Bool {
  self.valid
}

///|
pub fn stream_statistics(records : Array[StreamRecord]) -> StreamStatistics {
  if records.length() == 0 {
    return StreamStatistics::empty()
  }
  let mut sum = 0.0
  let mut minimum = 1.0e300
  let mut maximum = -1.0e300
  let mut valid_count = 0
  for record in records {
    if record.valid() {
      sum = sum + record.value()
      minimum = minimum.min(record.value())
      maximum = maximum.max(record.value())
      valid_count = valid_count + 1
    }
  }
  if valid_count == 0 {
    return StreamStatistics::empty()
  }
  let mean = sum / valid_count.to_double()
  let mut variance_sum = 0.0
  for record in records {
    if record.valid() {
      let error = record.value() - mean
      variance_sum = variance_sum + error * error
    }
  }
  let variance = if valid_count < 2 {
    0.0
  } else {
    variance_sum / (valid_count - 1).to_double()
  }
  let mut time_sum = 0.0
  let mut value_time_sum = 0.0
  let mut time_square_sum = 0.0
  for record in records {
    if record.valid() {
      let time = record.timestamp().to_double()
      time_sum = time_sum + time
      value_time_sum = value_time_sum + time * record.value()
      time_square_sum = time_square_sum + time * time
    }
  }
  let denominator = valid_count.to_double() * time_square_sum -
    time_sum * time_sum
  let slope = if denominator.abs() < 1.0e-12 {
    0.0
  } else {
    (valid_count.to_double() * value_time_sum - time_sum * sum) / denominator
  }
  {
    count: valid_count,
    mean,
    variance,
    minimum,
    maximum,
    first_timestamp: Some(records[0].timestamp()),
    last_timestamp: Some(records[records.length() - 1].timestamp()),
    slope,
    valid: valid_count > 0,
  }
}

///|
pub fn stream_statistics_merge(
  left : StreamStatistics,
  right : StreamStatistics,
) -> StreamStatistics {
  if !left.valid() {
    return right
  }
  if !right.valid() {
    return left
  }
  let count = left.count() + right.count()
  let delta = right.mean() - left.mean()
  let variance_sum = left.variance() * (left.count() - 1).max(0).to_double() +
    right.variance() * (right.count() - 1).max(0).to_double() +
    delta *
    delta *
    left.count().to_double() *
    right.count().to_double() /
    count.to_double()
  {
    count,
    mean: (
      left.mean() * left.count().to_double() +
      right.mean() * right.count().to_double()
    ) /
    count.to_double(),
    variance: if count < 2 {
      0.0
    } else {
      variance_sum / (count - 1).to_double()
    },
    minimum: left.minimum().min(right.minimum()),
    maximum: left.maximum().max(right.maximum()),
    first_timestamp: left.first_timestamp(),
    last_timestamp: right.last_timestamp(),
    slope: (left.slope() + right.slope()) * 0.5,
    valid: true,
  }
}

///|
pub fn stream_window_statistics(window : StreamWindow) -> StreamStatistics {
  stream_statistics(window.records())
}

///|
/// A cursor tracks consumption and supports replay-safe sequence checks.
pub struct StreamCursor {
  mut next_sequence : Int
  mut last_timestamp : Int?
  mut accepted : Int
  mut rejected : Int
  mut duplicate : Int
  mut out_of_order : Int
} derive(Debug)

///|
pub fn StreamCursor::new(start_sequence : Int) -> StreamCursor {
  {
    next_sequence: start_sequence.max(0),
    last_timestamp: None,
    accepted: 0,
    rejected: 0,
    duplicate: 0,
    out_of_order: 0,
  }
}

///|
pub fn StreamCursor::next_sequence(self : StreamCursor) -> Int {
  self.next_sequence
}

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

///|
pub fn StreamCursor::rejected(self : StreamCursor) -> Int {
  self.rejected
}

///|
pub fn StreamCursor::duplicate(self : StreamCursor) -> Int {
  self.duplicate
}

///|
pub fn StreamCursor::out_of_order(self : StreamCursor) -> Int {
  self.out_of_order
}

///|
pub fn StreamCursor::accept(self : StreamCursor, record : StreamRecord) -> Bool {
  if record.sequence() < self.next_sequence {
    self.duplicate = self.duplicate + 1
    self.rejected = self.rejected + 1
    return false
  }
  match self.last_timestamp {
    Some(previous) =>
      if record.timestamp() < previous {
        self.out_of_order = self.out_of_order + 1
      }
    None => ()
  }
  self.last_timestamp = Some(record.timestamp())
  self.next_sequence = record.sequence() + 1
  self.accepted = self.accepted + 1
  true
}

///|
pub fn StreamCursor::reset(self : StreamCursor, sequence : Int) -> Unit {
  self.next_sequence = sequence.max(0)
  self.last_timestamp = None
  self.accepted = 0
  self.rejected = 0
  self.duplicate = 0
  self.out_of_order = 0
}

///|
/// Operational decisions emitted by a stream processor.
pub enum StreamDecision {
  AcceptedRecord
  RejectedInvalid
  RejectedDuplicate
  Backpressure
} derive(Debug, Eq)

///|
/// A production-oriented scalar stream processor.
pub struct StreamRuntime {
  window : StreamWindow
  cursor : StreamCursor
  max_queue : Int
  queue : Array[StreamRecord]
  mut processed : Int
  mut dropped : Int
  mut emitted : Int
  mut backpressure : Int
} derive(Debug)

///|
pub fn StreamRuntime::new(
  spec : StreamWindowSpec,
  start_sequence : Int,
  max_queue : Int,
) -> StreamRuntime {
  {
    window: StreamWindow::new(spec),
    cursor: StreamCursor::new(start_sequence),
    max_queue: max_queue.max(1),
    queue: [],
    processed: 0,
    dropped: 0,
    emitted: 0,
    backpressure: 0,
  }
}

///|
pub fn StreamRuntime::window(self : StreamRuntime) -> StreamWindow {
  self.window
}

///|
pub fn StreamRuntime::cursor(self : StreamRuntime) -> StreamCursor {
  self.cursor
}

///|
pub fn StreamRuntime::queue_length(self : StreamRuntime) -> Int {
  self.queue.length()
}

///|
pub fn StreamRuntime::processed(self : StreamRuntime) -> Int {
  self.processed
}

///|
pub fn StreamRuntime::dropped(self : StreamRuntime) -> Int {
  self.dropped
}

///|
pub fn StreamRuntime::emitted(self : StreamRuntime) -> Int {
  self.emitted
}

///|
pub fn StreamRuntime::backpressure(self : StreamRuntime) -> Int {
  self.backpressure
}

///|
pub fn StreamRuntime::ingest(
  self : StreamRuntime,
  record : StreamRecord,
) -> StreamDecision {
  if !record.valid() {
    self.dropped = self.dropped + 1
    return RejectedInvalid
  }
  if self.queue.length() >= self.max_queue {
    self.backpressure = self.backpressure + 1
    self.dropped = self.dropped + 1
    return Backpressure
  }
  if !self.cursor.accept(record) {
    self.dropped = self.dropped + 1
    return RejectedDuplicate
  }
  self.queue.push(record)
  AcceptedRecord
}

///|
pub fn StreamRuntime::drain(self : StreamRuntime, limit : Int) -> Int {
  let target = limit.max(0)
  let mut count = 0
  while count < target && self.queue.length() > 0 {
    let record = self.queue.remove(0)
    if self.window.push(record) {
      self.processed = self.processed + 1
    } else {
      self.dropped = self.dropped + 1
    }
    count = count + 1
  }
  count
}

///|
pub fn StreamRuntime::flush(self : StreamRuntime) -> Int {
  self.drain(self.queue.length())
}

///|
pub fn StreamRuntime::emit_if_ready(self : StreamRuntime) -> StreamStatistics? {
  let stats = stream_window_statistics(self.window)
  if stats.count() >= self.window.spec().min_samples() {
    self.emitted = self.emitted + 1
    Some(stats)
  } else {
    None
  }
}

///|
pub fn StreamRuntime::reset(self : StreamRuntime, sequence : Int) -> Unit {
  self.queue.clear()
  self.window.clear()
  self.cursor.reset(sequence)
  self.processed = 0
  self.dropped = 0
  self.emitted = 0
  self.backpressure = 0
}

///|
pub fn stream_records_for_source(
  records : Array[StreamRecord],
  source : String,
) -> Array[StreamRecord] {
  records.filter(record => record.source() == source)
}

///|
pub fn stream_records_between(
  records : Array[StreamRecord],
  start : Int,
  end : Int,
) -> Array[StreamRecord] {
  records.filter(record => {
    record.timestamp() >= start && record.timestamp() <= end
  })
}

///|
pub fn stream_records_values(records : Array[StreamRecord]) -> Array[Double] {
  records.filter(record => record.valid()).map(record => record.value())
}

///|
pub fn stream_outlier_fraction(
  records : Array[StreamRecord],
  center : Double,
  scale : Double,
) -> Double {
  if records.length() == 0 {
    0.0
  } else {
    let threshold = scale.max(1.0e-12) * 3.0
    let mut outliers = 0
    for record in records {
      if record.valid() && (record.value() - center).abs() > threshold {
        outliers = outliers + 1
      }
    }
    outliers.to_double() / records.length().to_double()
  }
}

///|
pub fn stream_jitter(records : Array[StreamRecord]) -> Double {
  if records.length() < 2 {
    0.0
  } else {
    let mut sum = 0.0
    let mut count = 0
    for i in 1.. 0 {
        sum = sum + delta.to_double()
        count = count + 1
      }
    }
    if count == 0 {
      0.0
    } else {
      sum / count.to_double()
    }
  }
}

///|
pub fn stream_health_score(
  runtime : StreamRuntime,
  expected_count : Int,
) -> Double {
  let accepted_ratio = if expected_count <= 0 {
    1.0
  } else {
    runtime.processed().to_double() / expected_count.to_double()
  }
  let pressure = if runtime.processed() + runtime.dropped() == 0 {
    0.0
  } else {
    runtime.backpressure().to_double() /
    (runtime.processed() + runtime.dropped()).to_double()
  }
  (accepted_ratio.min(1.0) * (1.0 - pressure)).clamp(min=0.0, max=1.0)
}