///|
/// A logical batch inside a stream. Window text uses `|` or `;` as separators.
pub(all) struct EventWindow {
  index : Int
  label : String
  events : Array[String]
} derive(Eq, Debug)

///|
pub(all) struct WindowSnapshot {
  index : Int
  label : String
  events : Int
  exact_unique : Int
  hll_unique : Double
  bloom_false_positive_rate : Double
  dominant_key : String
  dominant_count : Int
  topk : Array[TopKItem]
} derive(Eq, Debug)

///|
pub(all) struct WindowChange {
  index : Int
  previous_label : String
  current_label : String
  event_delta : Int
  unique_delta : Int
  top_overlap : Double
  minhash_similarity : Double
  dominant_key : String
  dominant_count : Int
} derive(Eq, Debug)

///|
pub(all) struct WindowedStreamReport {
  windows : Array[WindowSnapshot]
  changes : Array[WindowChange]
  total_events : Int
  max_event_spike : Int
  min_similarity : Double
} derive(Eq, Debug)

///|
pub fn parse_event_windows(input : String) -> Array[EventWindow] {
  let windows : Array[EventWindow] = Array::new()
  let token = StringBuilder()
  let mut current : Array[String] = Array::new()
  let mut index = 0
  for ch in input.iter() {
    if window_separator(ch) {
      window_flush_token(current, token)
      if current.length() > 0 {
        windows.push({ index, label: window_label(index), events: current })
        index += 1
      }
      current = Array::new()
    } else if window_token_separator(ch) {
      window_flush_token(current, token)
    } else {
      token.write_char(ch)
    }
  }
  window_flush_token(current, token)
  if current.length() > 0 {
    windows.push({ index, label: window_label(index), events: current })
  }
  windows
}

///|
pub fn windowed_report_from_text(input : String) -> WindowedStreamReport {
  windowed_report(parse_event_windows(input))
}

///|
pub fn windowed_report(windows : Array[EventWindow]) -> WindowedStreamReport {
  let snapshots : Array[WindowSnapshot] = Array::new()
  let changes : Array[WindowChange] = Array::new()
  let mut total = 0
  for window in windows {
    let snapshot = window_snapshot(window)
    total += snapshot.events
    snapshots.push(snapshot)
  }
  for i in 1.. WindowSnapshot {
  let counter = exact_counter_from_items(window.events)
  let topk = exact_counter_topk(counter, 5)
  let bloom = bloom_stats(bloom_from_items(window.events, 256, 4))
  let dominant = if topk.length() == 0 {
    { key: "", count: 0, error: 0 }
  } else {
    topk[0]
  }
  {
    index: window.index,
    label: window.label,
    events: window.events.length(),
    exact_unique: exact_counter_unique(counter),
    hll_unique: hll_estimate(hll_from_items(window.events, 64)),
    bloom_false_positive_rate: bloom.estimated_false_positive_rate,
    dominant_key: dominant.key,
    dominant_count: dominant.count,
    topk,
  }
}

///|
pub fn window_change(
  previous : WindowSnapshot,
  current : WindowSnapshot,
) -> WindowChange {
  {
    index: current.index,
    previous_label: previous.label,
    current_label: current.label,
    event_delta: current.events - previous.events,
    unique_delta: current.exact_unique - previous.exact_unique,
    top_overlap: window_top_overlap(previous.topk, current.topk),
    minhash_similarity: window_top_minhash(previous.topk, current.topk),
    dominant_key: current.dominant_key,
    dominant_count: current.dominant_count,
  }
}

///|
pub fn windowed_report_markdown(report : WindowedStreamReport) -> String {
  let out = StringBuilder()
  out.write_string("# Windowed Stream Report\n\n")
  out.write_string("| metric | value |\n| --- | ---: |\n")
  out.write_string(
    "| windows | " + report.windows.length().to_string() + " |\n",
  )
  out.write_string(
    "| total events | " + report.total_events.to_string() + " |\n",
  )
  out.write_string(
    "| max event spike | " + report.max_event_spike.to_string() + " |\n",
  )
  out.write_string(
    "| min top similarity | " +
    sketch_double_text(report.min_similarity) +
    " |\n\n",
  )
  out.write_string("## Windows\n\n")
  out.write_string(
    "| window | events | unique | hll estimate | dominant key | dominant count | bloom fpr |\n| --- | ---: | ---: | ---: | --- | ---: | ---: |\n",
  )
  for snapshot in report.windows {
    out.write_string(
      "| " +
      snapshot.label +
      " | " +
      snapshot.events.to_string() +
      " | " +
      snapshot.exact_unique.to_string() +
      " | " +
      sketch_double_text(snapshot.hll_unique) +
      " | " +
      snapshot.dominant_key +
      " | " +
      snapshot.dominant_count.to_string() +
      " | " +
      sketch_double_text(snapshot.bloom_false_positive_rate) +
      " |\n",
    )
  }
  out.write_string("\n## Changes\n\n")
  if report.changes.length() == 0 {
    out.write_string("No adjacent window changes.\n")
  } else {
    out.write_string(
      "| from | to | event delta | unique delta | top overlap | top minhash | dominant key |\n| --- | --- | ---: | ---: | ---: | ---: | --- |\n",
    )
    for change in report.changes {
      out.write_string(
        "| " +
        change.previous_label +
        " | " +
        change.current_label +
        " | " +
        change.event_delta.to_string() +
        " | " +
        change.unique_delta.to_string() +
        " | " +
        sketch_double_text(change.top_overlap) +
        " | " +
        sketch_double_text(change.minhash_similarity) +
        " | " +
        change.dominant_key +
        " |\n",
      )
    }
  }
  out.to_string()
}

///|
pub fn windowed_report_json(report : WindowedStreamReport) -> String {
  let out = StringBuilder()
  out.write_string("{")
  out.write_string("\"windows\":[")
  for i in 0.. 0 {
      out.write_string(",")
    }
    let snapshot = report.windows[i]
    out.write_string(
      "{\"label\":\"" +
      sketch_escape_json(snapshot.label) +
      "\",\"events\":" +
      snapshot.events.to_string() +
      ",\"unique\":" +
      snapshot.exact_unique.to_string() +
      ",\"hll_unique\":" +
      sketch_double_text(snapshot.hll_unique) +
      ",\"dominant_key\":\"" +
      sketch_escape_json(snapshot.dominant_key) +
      "\",\"dominant_count\":" +
      snapshot.dominant_count.to_string() +
      "}",
    )
  }
  out.write_string("],\"changes\":[")
  for i in 0.. 0 {
      out.write_string(",")
    }
    let change = report.changes[i]
    out.write_string(
      "{\"from\":\"" +
      sketch_escape_json(change.previous_label) +
      "\",\"to\":\"" +
      sketch_escape_json(change.current_label) +
      "\",\"event_delta\":" +
      change.event_delta.to_string() +
      ",\"unique_delta\":" +
      change.unique_delta.to_string() +
      ",\"top_overlap\":" +
      sketch_double_text(change.top_overlap) +
      ",\"minhash_similarity\":" +
      sketch_double_text(change.minhash_similarity) +
      "}",
    )
  }
  out.write_string(
    "],\"total_events\":" +
    report.total_events.to_string() +
    ",\"max_event_spike\":" +
    report.max_event_spike.to_string() +
    ",\"min_similarity\":" +
    sketch_double_text(report.min_similarity) +
    "}",
  )
  out.to_string()
}

///|
pub fn windowed_topk_timeline_markdown(report : WindowedStreamReport) -> String {
  let out = StringBuilder()
  out.write_string("# Top-K Timeline\n\n")
  out.write_string(
    "| window | rank | key | count |\n| --- | ---: | --- | ---: |\n",
  )
  for snapshot in report.windows {
    for i in 0.. Unit {
  let value = token.to_string()
  if !value.is_empty() {
    events.push(value)
    token.reset()
  }
}

///|
fn window_label(index : Int) -> String {
  "window-" + (index + 1).to_string()
}

///|
fn window_separator(ch : Char) -> Bool {
  ch == '|' || ch == ';'
}

///|
fn window_token_separator(ch : Char) -> Bool {
  ch == ',' || ch == '\n' || ch == '\r' || ch == '\t' || ch == ' '
}

///|
fn window_max_spike(changes : Array[WindowChange]) -> Int {
  let mut spike = 0
  for change in changes {
    if change.event_delta > spike {
      spike = change.event_delta
    }
  }
  spike
}

///|
fn window_min_similarity(changes : Array[WindowChange]) -> Double {
  if changes.length() == 0 {
    return 1.0
  }
  let mut value = 1.0
  for change in changes {
    if change.minhash_similarity < value {
      value = change.minhash_similarity
    }
  }
  value
}

///|
fn window_top_overlap(
  left : Array[TopKItem],
  right : Array[TopKItem],
) -> Double {
  if left.length() == 0 && right.length() == 0 {
    return 1.0
  }
  let union : Array[String] = Array::new()
  let intersection : Array[String] = Array::new()
  for item in left {
    if !sketch_contains(union, item.key) {
      union.push(item.key)
    }
  }
  for item in right {
    if !sketch_contains(union, item.key) {
      union.push(item.key)
    }
    if window_top_has_key(left, item.key) &&
      !sketch_contains(intersection, item.key) {
      intersection.push(item.key)
    }
  }
  if union.length() == 0 {
    0.0
  } else {
    intersection.length().to_double() / union.length().to_double()
  }
}

///|
fn window_top_minhash(
  left : Array[TopKItem],
  right : Array[TopKItem],
) -> Double {
  let left_keys : Array[String] = Array::new()
  let right_keys : Array[String] = Array::new()
  for item in left {
    left_keys.push(item.key)
  }
  for item in right {
    right_keys.push(item.key)
  }
  minhash_similarity(
    minhash_from_items(left_keys, 32),
    minhash_from_items(right_keys, 32),
  )
}

///|
fn window_top_has_key(items : Array[TopKItem], key : String) -> Bool {
  for item in items {
    if item.key == key {
      return true
    }
  }
  false
}