Design an Ad Click Aggregator

interface Click { clickId: string; adId: string; eventTs: number }

const DEDUP_TTL_MS = 15 * 60 * 1000;
const LATENESS_MS = 30 * 1000;

const seen = new Map<string, number>();            // clickId -> firstSeen (SQLite in real MVP)
const windows = new Map<string, number>();          // "adId|minuteTs" -> count
let watermark = 0;

export function onEvent(c: Click, flush: (adId: string, minuteTs: number, count: number) => void) {
  const now = Date.now();
  if (seen.has(c.clickId)) return;                  // duplicate retry, drop
  seen.set(c.clickId, now);

  const minuteTs = Math.floor(c.eventTs / 60000) * 60000;
  watermark = Math.max(watermark, c.eventTs - LATENESS_MS);

  if (minuteTs + 60000 <= watermark) {
    // window already closed: route to late side-output for reconciliation
    lateOutput(c);
    return;
  }
  const key = c.adId + "|" + minuteTs;
  windows.set(key, (windows.get(key) ?? 0) + 1);

  // Close every window whose end is behind the watermark.
  for (const [k, count] of windows) {
    const ts = Number(k.split("|")[1]);
    if (ts + 60000 <= watermark) {
      const adId = k.split("|")[0];
      flush(adId, ts, count);                       // idempotent upsert downstream
      windows.delete(k);
    }
  }
  // TTL sweep for dedup state (run on a timer in real code)
  if (seen.size > 1_000_000) {
    for (const [id, t] of seen) if (now - t > DEDUP_TTL_MS) seen.delete(id);
  }
}

function lateOutput(c: Click) {
  // append to late_clicks NDJSON; nightly reconciliation picks it up
}

CREATE TABLE minute_aggregates (
  ad_id      TEXT NOT NULL,
  minute_ts  TIMESTAMPTZ NOT NULL,
  clicks     BIGINT NOT NULL,
  version    BIGINT NOT NULL,       -- bumped on each re-emission of the window
  updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  PRIMARY KEY (ad_id, minute_ts)
);

-- Replaying the same closed window overwrites instead of double-counting.
-- Version guard means an old replay can never clobber a newer correction.
INSERT INTO minute_aggregates (ad_id, minute_ts, clicks, version)
VALUES (:ad_id, :minute_ts, :clicks, :version)
ON CONFLICT (ad_id, minute_ts) DO UPDATE
SET clicks = EXCLUDED.clicks,
    version = EXCLUDED.version,
    updated_at = now()
WHERE minute_aggregates.version <= EXCLUDED.version;

import json
from collections import defaultdict

def recompute_from_raw(paths):
    seen = set()
    counts = defaultdict(int)  # (ad_id, minute_ts) -> clicks
    for path in paths:
        with open(path) as f:
            for line in f:
                e = json.loads(line)
                if e["click_id"] in seen:      # exact full-day dedup
                    continue
                seen.add(e["click_id"])
                minute_ts = (e["event_ts"] // 60000) * 60000
                counts[(e["ad_id"], minute_ts)] += 1
    return counts

def reconcile(raw_counts, db_rows, threshold=0.001):
    drifts = []
    db = {(r["ad_id"], r["minute_ts"]): r["clicks"] for r in db_rows}
    for key in set(raw_counts) | set(db):
        truth, rt = raw_counts.get(key, 0), db.get(key, 0)
        if truth == rt:
            continue
        drift = abs(truth - rt) / max(truth, 1)
        drifts.append({"key": key, "batch": truth, "realtime": rt, "drift": drift,
                       "page_oncall": drift > threshold})
    return drifts  # apply corrections: upsert batch value with bumped version