sinusoidal 0.10.0

use crate::proto::streaming_api::Datagram;
use std::io::Error;

const AGGREGATION_WINDOW_NS: i64 = 1_000_000_000;
const AGGREGATION_MIDPOINT_NS: i64 = AGGREGATION_WINDOW_NS / 2;
const OUTPUT_SCALE: f64 = 1_000_000.0;

struct Template {
  sample_count: u32,
  clock_synch: u32,
  gm_identity: Vec<u8>,
}

impl Template {
  fn from_datagram(d: &Datagram) -> Self {
    Self {
      sample_count: d.sample_count,
      clock_synch: d.clock_synch,
      gm_identity: d.gm_identity.clone(),
    }
  }
}

struct HavePacketsAggregate {
  window_start: i64,
  sample_count: u64,
  sums: Vec<f64>,
  last_template: Template,
}

impl HavePacketsAggregate {
  fn new(window_start: i64, values: &[f64], template: &Datagram) -> Self {
    Self {
      window_start,
      sample_count: 1,
      sums: values.to_vec(),
      last_template: Template::from_datagram(template),
    }
  }

  fn add_sample(&mut self, values: &[f64], template: &Datagram) {
    self.sample_count += 1;
    for (sum, value) in self.sums.iter_mut().zip(values.iter()) {
      *sum += *value;
    }
    self.last_template = Template::from_datagram(template);
  }

  fn to_datagram(&self) -> Datagram {
    let values = self
      .sums
      .iter()
      .map(|sum| ((sum / self.sample_count as f64) * OUTPUT_SCALE).round() as i64)
      .collect();

    Datagram {
      timestamp: self.window_start + AGGREGATION_MIDPOINT_NS,
      sample_count: self.last_template.sample_count,
      values,
      clock_synch: self.last_template.clock_synch,
      flags: Vec::new(),
      gm_identity: self.last_template.gm_identity.clone(),
    }
  }
}

#[derive(Default)]
pub struct SecondMeanAggregate {
  state: Option<HavePacketsAggregate>,
}

impl SecondMeanAggregate {
  pub fn new() -> Self {
    Self { state: None }
  }

  fn aligned_window_start(timestamp: i64) -> i64 {
    timestamp.saturating_sub(timestamp.rem_euclid(AGGREGATION_WINDOW_NS))
  }

  pub fn ingest_sample(
    &mut self,
    timestamp: i64,
    values: &[f64],
    template: &Datagram,
  ) -> Result<Option<Datagram>, Error> {
    if values.is_empty() {
      return Ok(None);
    }

    let sample_window_start = Self::aligned_window_start(timestamp);

    match self.state.as_mut() {
      None => {
        self.state = Some(HavePacketsAggregate::new(
          sample_window_start,
          values,
          template,
        ));
        Ok(None)
      }
      Some(aggregate) => {
        if values.len() != aggregate.sums.len() {
          return Err(Error::other(format!(
            "SecondMeanAggregate: sample width {} does not match aggregate width {}",
            values.len(),
            aggregate.sums.len()
          )));
        }

        if sample_window_start < aggregate.window_start {
          return Ok(None);
        }

        if sample_window_start == aggregate.window_start {
          aggregate.add_sample(values, template);
          return Ok(None);
        }

        let packet_to_send = aggregate.to_datagram();
        self.state = Some(HavePacketsAggregate::new(
          sample_window_start,
          values,
          template,
        ));
        Ok(Some(packet_to_send))
      }
    }
  }
}

#[cfg(test)]
mod tests {
  use super::*;
  use quickcheck::TestResult;
  use quickcheck_macros::quickcheck;

  fn template(ts: i64, width: usize) -> Datagram {
    Datagram {
      timestamp: ts,
      sample_count: 7,
      values: vec![0; width],
      clock_synch: 3,
      flags: Vec::new(),
      gm_identity: vec![1, 2, 3],
    }
  }

  #[test]
  fn test_emits_arithmetic_mean_for_two_values() {
    let mut aggregate = SecondMeanAggregate::new();

    assert!(
      aggregate
        .ingest_sample(100_000_000, &[-0.75, -0.02], &template(1, 2))
        .unwrap()
        .is_none()
    );
    assert!(
      aggregate
        .ingest_sample(200_000_000, &[0.0, 0.0], &template(2, 2))
        .unwrap()
        .is_none()
    );
    let out = aggregate
      .ingest_sample(1_100_000_000, &[0.75, 0.02], &template(3, 2))
      .unwrap()
      .expect("expected packet");

    assert_eq!(out.values, vec![-375000, -10000]);
    assert_eq!(out.sample_count, 7);
    assert_eq!(out.clock_synch, 3);
    assert_eq!(out.gm_identity, vec![1, 2, 3]);
  }

  #[test]
  fn test_width_mismatch_returns_error() {
    let mut aggregate = SecondMeanAggregate::new();
    aggregate
      .ingest_sample(100_000_000, &[1.0, 2.0], &template(1, 2))
      .unwrap();
    // Feeding a sample with different width should error
    let result = aggregate.ingest_sample(200_000_000, &[1.0], &template(2, 1));
    assert!(result.is_err());
  }

  #[quickcheck]
  fn prop_aligned_window_contains_timestamp(ts: u64) -> bool {
    let ts = (ts % (i64::MAX as u64)) as i64;
    let start = SecondMeanAggregate::aligned_window_start(ts);
    let delta = ts - start;
    (0..AGGREGATION_WINDOW_NS).contains(&delta)
  }

  #[quickcheck]
  fn prop_emitted_timestamp_is_second_midpoint(
    base_second: u32,
    initial_offset_ns: u32,
    step_ms: Vec<u16>,
  ) -> TestResult {
    if step_ms.len() < 2 {
      return TestResult::discard();
    }

    let mut aggregate = SecondMeanAggregate::new();
    let start_second = 10_i64 + i64::from(base_second % 120);
    let mut timestamp =
      start_second * AGGREGATION_WINDOW_NS + i64::from(initial_offset_ns % 900_000_000);

    for step in step_ms {
      let delta_ms = i64::from(step % 7_001) + 1;
      timestamp += delta_ms * 1_000_000;

      match aggregate.ingest_sample(timestamp, &[1.0], &template(timestamp, 1)) {
        Ok(Some(packet))
          if packet.timestamp.rem_euclid(AGGREGATION_WINDOW_NS) != AGGREGATION_MIDPOINT_NS =>
        {
          return TestResult::failed();
        }
        Err(_) => return TestResult::failed(),
        _ => {}
      }
    }

    TestResult::passed()
  }

  #[quickcheck]
  fn prop_constant_stream_yields_constant_mean(value: f64, n: u8) -> TestResult {
    if !value.is_finite() || n == 0 {
      return TestResult::discard();
    }
    // Bound the magnitude to keep floating-point summation error below 0.5 ULP
    // (rounding error ≈ eps * n * |v| * OUTPUT_SCALE; with n≤50, |v|≤1e3,
    // OUTPUT_SCALE=1e6 → error ≈ 1e-15 * 50 * 1e3 * 1e6 = 0.05 < 0.5)
    if value.abs() > 1e3 {
      return TestResult::discard();
    }
    // Feed n samples in the same 1-second window, then one in the next to flush
    let mut aggregate = SecondMeanAggregate::new();
    let n = (n % 50) as i64 + 1;
    for i in 0..n {
      let ts = 100_000_000 + i * 10_000_000;
      aggregate
        .ingest_sample(ts, &[value], &template(ts, 1))
        .unwrap();
    }
    // Trigger flush with a sample in the next second
    let flush_result = aggregate
      .ingest_sample(1_100_000_000, &[value], &template(1_100_000_000, 1))
      .unwrap();
    let packet = match flush_result {
      Some(p) => p,
      None => return TestResult::discard(),
    };
    // Mean of a constant stream should equal the constant (scaled by OUTPUT_SCALE)
    let expected = (value * OUTPUT_SCALE).round() as i64;
    if packet.values[0] == expected {
      TestResult::passed()
    } else {
      TestResult::failed()
    }
  }
}