File: ad-click-event-aggregation/click_aggregator.py

Date: 2026-06-05

Time: 13:22

ad-click-event-aggregation/click_aggregator.py

Purpose

This file implements an ad click event aggregation system — a classic system design interview problem. It owns the responsibility of ingesting raw click events, deduplicating them, bucketing them into fixed-size time windows (tumbling windows), and producing aggregated counts per ad. This is the kind of component that sits between a raw event stream (Kafka, Kinesis) and a queryable analytics store in a production ad-tech pipeline.

Key Components

Data Classes

• ClickEvent — The input unit. Carries eventid (dedup key), adid (partition key), user_id (for unique-user counting), timestamp (event time, not processing time), and optional metadata.
• AggregationResult — The output unit. A materialized view of one window: ad, time range, total clicks, and unique user count. This is what downstream consumers (dashboards, billing) receive.
• _Window — Internal mutable accumulator. The underscore prefix signals it's not part of the public API. Tracks raw count, a set of user IDs, and a WindowState lifecycle.

WindowState Enum

Three-state lifecycle: OPEN → CLOSED → FINALIZED. This is a one-way ratchet — windows never reopen. OPEN accepts all events, CLOSED accepts late events within the lateness allowance, FINALIZED rejects everything.

ClickAggregator

The core class. Its constructor takes two parameters that control the fundamental tradeoff between latency and completeness:

• windowsizeseconds (default 60) — the tumbling window width.
• allowedlatenessseconds (default 300) — how long after a window closes it will still accept late arrivals.

Key methods:

| Method | Contract |

|--------|----------|

| processevent(event) | Idempotent per eventid. Returns True if accepted, False if deduped or rejected. Mutates window state. |

| process_batch(events) | Convenience wrapper. Returns a breakdown dict of accepted/deduped/rejected counts. |

| advance_watermark(ts) | Monotonic — ignores timestamps ≤ current watermark. Triggers window lifecycle transitions and dedup pruning. Returns newly finalized AggregationResults. |

| query(ad_id, start, end) | Read-only range scan over windows for a single ad. |

| querytopads(start, end, k) | Cross-ad aggregation. Returns top-K ads by total click count. |

Patterns

Tumbling windows — Non-overlapping, fixed-size time buckets. windowstart uses math.floor(ts / size) * size to snap any timestamp to its window boundary. This is deterministic and partition-safe — two nodes processing the same event will compute the same window.

Event-time processing with watermarks — The system separates event time (when the click happened) from processing time (when process_event is called). The watermark is an explicit "I believe all events before this time have arrived" signal. This is the same model as Apache Flink/Beam.

Dedup via seen-event registry — seenevents is a dict keyed by eventid. This is an at-least-once-to-exactly-once conversion layer. The registry is pruned on watermark advance to bound memory (events older than 2 * allowed_lateness are evicted).

Two-level dict for window lookup — windows[adid][windowstart] gives O(1) access to any window. The ad_id is the partition key; in a distributed version, each partition would hold its own subset.

Dependencies

Imports: Only stdlib — math, dataclasses, enum. No external dependencies. This is intentional for an interview implementation: the design concepts are the focus, not framework integration.

Imported by: testclickaggregator.py — the test suite is the sole consumer.

Flow

1. Ingestion: process_event is called with a ClickEvent.

2. Dedup gate: If eventid already in seenevents, reject immediately.

3. Window resolution: Compute windowstart from event timestamp, get-or-create the Window.

4. Lifecycle gate: If FINALIZED, reject. If CLOSED, check if watermark - window.end > allowed_lateness — if so, finalize and reject; otherwise accept as a late event.

5. Accumulate: Add eventid to seenevents, increment count, add user_id to the window's user set.

6. Watermark advance (separate call): Scan all windows. Those past watermark - allowedlateness get finalized and emitted as AggregationResult. Those past watermark (but not yet past lateness) get closed. Dedup entries older than 2 * allowedlateness are pruned.

The separation of event processing from watermark advance is critical — it means the caller controls when finalization happens, which maps to how stream processors work in practice.

Invariants

• Watermark monotonicity: advance_watermark is a no-op if the new timestamp isn't strictly greater than the current watermark.
• Window lifecycle is one-directional: OPEN → CLOSED → FINALIZED. There's no path back.
• Dedup is global across ads: seenevents is keyed by eventid alone, not (adid, eventid). An event ID that appears for two different ads will still be deduped.
• Finalization is irreversible: Once a window is FINALIZED, no further events can modify its counts. This is the "exactly once" guarantee for downstream consumers.
• Late acceptance window: A late event is accepted only if watermark - window.end ≤ allowedlateness. This means the actual acceptance window extends allowedlateness seconds beyond the window's end time.

Error Handling

There is essentially none — and that's appropriate for this domain. The system uses return-value signaling (True/False from process_event) rather than exceptions. Invalid states (duplicate events, late arrivals) are expected in stream processing, not exceptional. The stats dict provides observability into rejection reasons without interrupting the processing pipeline.

One subtlety: processbatch infers the rejection reason by diffing stats counters before and after each call. This is fragile if processevent's internal accounting ever changes — the batch method's classification logic is coupled to the ordering of checks in process_event.