///|
fn print_measurement(
  algorithm : String,
  input_count : Int,
  estimate : Double,
  tracked_items : Int,
  serialized_bytes : Int,
  repeat_count : Int,
  elapsed_ms : Int64,
  throughput : Double,
) -> Unit {
  println(
    algorithm +
    "," +
    input_count.to_string() +
    "," +
    estimate.to_string() +
    "," +
    tracked_items.to_string() +
    "," +
    serialized_bytes.to_string() +
    "," +
    repeat_count.to_string() +
    "," +
    elapsed_ms.to_string() +
    "," +
    throughput.to_string(),
  )
}

///|
fn run_benchmark() -> Int {
  let n = 2000
  let repeat_count = 3
  let input_count = n * repeat_count
  println(
    "algorithm,input_count,estimate,tracked_items,serialized_bytes," +
    "repeat_count,elapsed_ms,throughput_events_per_second",
  )

  let hll = @hll.HyperLogLog::new(12)
  let hllp = @hllplus.HyperLogLogPlus::new(12, 18)
  let cms = @countmin.CountMinSketch::new(5, 512)
  let topk = @topk.SpaceSaving::new(20, 0x42454E4348UL)
  let quantile = @quantile.QuantileSummary::new(
    @quantile.QuantileConfig::new(0.01, 64),
  )
  let histogram = @quantile.Histogram::new([1.0, 10.0, 100.0, 1000.0])
  let stream = @stream.StreamAggregator::new(
    @stream.StreamConfig::new(60000L, 5000L, 10, 10),
  )
  let bloom = @bloom.BloomFilter::new(
    @bloom.BloomConfig::new(n, 0.01, 0x424C4F4FUL),
  )

  let start_ms = @env.now()
  let mut i = 0
  while i < input_count {
    let key = "benchmark-key-" + (i % 500).to_string()
    hll.add_string(key)
    hllp.add_string(key)
    cms.add(key, 1L)
    let _ = topk.add(key, i.to_int64())
    quantile.add((i % 1000).to_double())
    histogram.observe((i % 1000).to_double())
    let _ = stream.ingest(
      @stream.StreamRecord::new(i.to_int64(), "bench", "api", key, 1L),
    )
    let _ = bloom.add_string(key)
    i = i + 1
  }
  let elapsed_ms = @env.now() - start_ms
  let elapsed_int = elapsed_ms.reinterpret_as_int64()
  let throughput = if elapsed_ms == 0UL {
    0.0
  } else {
    input_count.to_double() * 1000.0 / elapsed_ms.to_double()
  }
  print_measurement(
    "hll",
    input_count,
    hll.cardinality(),
    0,
    @serde.serialize_hll(hll).length(),
    repeat_count,
    elapsed_int,
    throughput,
  )
  print_measurement(
    "hllplus",
    input_count,
    hllp.cardinality(),
    0,
    0,
    repeat_count,
    elapsed_int,
    throughput,
  )
  print_measurement(
    "countmin",
    input_count,
    cms.estimate_frequency("benchmark-key-1").to_double(),
    0,
    0,
    repeat_count,
    elapsed_int,
    throughput,
  )
  print_measurement(
    "space_saving",
    input_count,
    topk.total_weight().to_double(),
    topk.tracked_items(),
    0,
    repeat_count,
    elapsed_int,
    throughput,
  )
  print_measurement(
    "quantile",
    input_count,
    quantile.quantile(0.5).unwrap_or(0.0),
    quantile.entry_count(),
    @serde.serialize_quantile(quantile).length(),
    repeat_count,
    elapsed_int,
    throughput,
  )
  print_measurement(
    "histogram",
    input_count,
    histogram.mean().unwrap_or(0.0),
    histogram.bucket_counts().length(),
    @serde.serialize_histogram(histogram).length(),
    repeat_count,
    elapsed_int,
    throughput,
  )
  print_measurement(
    "bloom",
    input_count,
    bloom.estimated_false_positive_rate(),
    bloom.capacity(),
    0,
    repeat_count,
    elapsed_int,
    throughput,
  )
  print_measurement(
    "stream",
    input_count,
    stream
    .tenant_snapshot("bench")
    .unwrap_or({
      tenant: "bench",
      dimension: "api",
      window_start_ms: 0L,
      window_end_ms: 60000L,
      distinct: 0.0,
      total_weight: 0L,
      top_keys: [],
      histogram: None,
      accepted_events: 0L,
    }).distinct,
    stream.global_snapshots().length(),
    0,
    repeat_count,
    elapsed_int,
    throughput,
  )
  0
}