#!/usr/bin/env python3
"""Windowing by processing time invents a spike that never happened, DDIA 2e Ch. 12.

A steady stream of one event per second flows for 120 seconds -- the real event
rate is flat at 1/second the whole time. But the consumer that windows the stream
suffers a 60-second outage (a redeploy): while it is down the events pile up in
the broker's backlog, and when it restarts all 60 buffered events are processed
at nearly the SAME instant.

The surprise is which clock you count by. Bucket the events by PROCESSING time
(when the consumer got to each event) and the 60 buffered events all land in one
per-second bucket at the restart instant -- a single ~60 events/second bar, a 60x
phantom spike (DDIA Figure 12-8), flanked by empty buckets during the outage.
Re-bucket the SAME events by their embedded EVENT-time timestamp (when they
happened in the world) and the spike vanishes: a flat 1/second across all 60
buckets. Nothing in the data spiked; the consumer's own schedule did.

Timing is modeled explicitly -- no wall-clock, no sleeps -- so the transcript is
fully deterministic.
"""

# --- The world: a steady stream, one event per second, for 120 seconds. ---

DURATION = 120          # seconds of real traffic
OUTAGE_START = 30       # event-time second the consumer goes down (inclusive)
OUTAGE_END = 90         # event-time second the consumer comes back (exclusive)
RESTART_INSTANT = 90    # processing time at which the backlog is drained
BUCKET_LO = 85          # window we zoom the histograms into
BUCKET_HI = 95


def make_events():
    """One event per second. Each carries an event_time and a processing_time.

    event_time  = when it happened in the world (0, 1, 2, ... 119) -- steady.
    processing_time = when the consumer got to it. Normally equal to event_time,
    BUT every event that arrives during the outage [30, 90) is buffered in the
    broker and only processed when the consumer restarts, so its processing_time
    collapses onto the restart instant. The backlog compresses 60 distinct
    event-times into one processing-time.
    """
    events = []
    for event_time in range(DURATION):
        if OUTAGE_START <= event_time < OUTAGE_END:
            # Consumer is down: event waits in the backlog, drained at restart.
            processing_time = RESTART_INSTANT
        else:
            # Consumer is keeping up: processed the second it arrives.
            processing_time = event_time
        events.append({"event_time": event_time, "processing_time": processing_time})
    return events


def bucket_by(events, key):
    """Count events into per-second buckets keyed by event_time or processing_time."""
    counts = {}
    for e in events:
        second = e[key]
        counts[second] = counts.get(second, 0) + 1
    return counts


def histogram(counts, lo, hi):
    """Render per-second buckets [lo, hi] as a labeled bar chart."""
    lines = []
    for second in range(lo, hi + 1):
        n = counts.get(second, 0)
        bar = "#" * n
        lines.append(f"    t={second:>3}s | {bar:<60} {n}")
    return "\n".join(lines)


def main():
    events = make_events()

    print("A steady stream: one event per second for 120 seconds (true rate = 1/s, flat).")
    print(f"    The consumer is DOWN for a redeploy over event-times [{OUTAGE_START}, {OUTAGE_END}) --")
    print(f"    those {OUTAGE_END - OUTAGE_START} events buffer in the backlog and all get processed at the")
    print(f"    restart instant, processing_time = {RESTART_INSTANT}s.\n")

    proc = bucket_by(events, "processing_time")
    evt = bucket_by(events, "event_time")

    # Bucket by PROCESSING time -- the consumer's schedule, not the world's.
    print(f"per-second rate bucketed by PROCESSING time (zoom on t={BUCKET_LO}..{BUCKET_HI}s):")
    print(histogram(proc, BUCKET_LO, BUCKET_HI))
    peak_proc_second = max(proc, key=proc.get)
    peak_proc_rate = proc[peak_proc_second]
    print(f"    peak = {peak_proc_rate} events/second at processing-time t={peak_proc_second}s"
          f" -- a {peak_proc_rate}x PHANTOM SPIKE.\n")

    # Bucket the SAME events by EVENT time -- when they actually happened.
    print(f"per-second rate bucketed by EVENT time (same window t={BUCKET_LO}..{BUCKET_HI}s):")
    print(histogram(evt, BUCKET_LO, BUCKET_HI))
    peak_evt_rate = max(evt.values())
    min_evt_rate = min(evt[s] for s in range(DURATION))
    print(f"    peak = {peak_evt_rate} event/second, floor = {min_evt_rate} event/second"
          f" -- FLAT 1/s, no spike anywhere.\n")

    print("verdict:")
    print(f"    processing-time peak rate = {peak_proc_rate}/s  (invented by the backlog)")
    print(f"    event-time      peak rate = {peak_evt_rate}/s  (the real traffic)")
    print(f"    same {len(events)} events, same data -- the {peak_proc_rate}x spike is an artifact of WHEN")
    print("    the consumer looked, not of anything that happened in the world.")


if __name__ == "__main__":
    main()