How would you design ad click event aggregation?

// Ad click aggregation: event-time tumbling windows with a watermark, allowed lateness,
// dedup by click ID, top ads per minute, and a batch reconciliation of the raw log.

// ---- 0. capacity estimate (assumptions) ----
const clicksPerDay = 1e9, peakFactor = 5, bytesPerClick = 200, activeAds = 2e6, activeShare = 0.1;
const avg = Math.round(clicksPerDay / 86400);
console.log("== 0. capacity ==");
console.log(`  average ${avg} clicks/s, peak ${avg * peakFactor} clicks/s`);
console.log(`  raw log ${(clicksPerDay * bytesPerClick) / 1e9} GB/day before replication`);
console.log(`  aggregate rows per day (ad x minute, ${activeShare * 100}% of ${activeAds / 1e6}M ads active each minute): ${activeAds * activeShare * 1440 / 1e6}M`);

let seed = 2026; // 32-bit LCG so every language draws the same events
const rand = () => { seed = (seed * 1664525 + 1013904223) % 4294967296; return seed / 4294967296; };
function pickAd() {
  const r = rand();
  if (r < 0.4) return "ad-1"; if (r < 0.65) return "ad-2"; if (r < 0.8) return "ad-3";
  if (r < 0.9) return "ad-4"; if (r < 0.96) return "ad-5"; return "ad-6";
}
// 6000 clicks over 3 minutes of event time; most arrive within 5 s, some 20-59 s late, a few 100-199 s late;
// 2% are re-sent by a client retry (same click ID).
const events = [];
for (let i = 0; i < 6000; i++) {
  const et = Math.floor((i * 180) / 6000), ad = pickAd();
  const r = rand();
  let delay;
  if (r < 0.94) delay = Math.floor(rand() * 6);
  else if (r < 0.98) delay = 20 + Math.floor(rand() * 40);
  else delay = 100 + Math.floor(rand() * 100);
  events.push({ id: "c" + i, ad, et, at: et + delay, seq: 2 * i });
  if (rand() < 0.02) events.push({ id: "c" + i, ad, et, at: et + delay + 1 + Math.floor(rand() * 10), seq: 2 * i + 1 });
}
events.sort((a, b) => a.at - b.at || a.seq - b.seq); // arrival order

const W = 60, OUT_OF_ORDER = 10, LATENESS = 30;
const windows = new Map(); // start -> { counts: Map, fired, firedAt }
const seen = new Set();
let watermark = -Infinity, dups = 0, onTime = 0, corrections = 0, tooLate = 0;
const top = (counts, n) => [...counts.entries()].sort((a, b) => b[1] - a[1] || (a[0] < b[0] ? -1 : 1)).slice(0, n).map(([k, v]) => k + "=" + v).join(" ");
const total = (counts) => [...counts.values()].reduce((s, v) => s + v, 0);
const firstResults = [];
function fire(upTo, at) {
  for (const start of [...windows.keys()].sort((a, b) => a - b)) {
    const w = windows.get(start);
    if (!w.fired && start + W <= upTo) { w.fired = true; firstResults.push(`  [${start}s,${start + W}s) emitted at t=${at}: total=${total(w.counts)}  top3: ${top(w.counts, 3)}`); }
  }
}
for (const e of events) {
  if (seen.has(e.id)) { dups++; continue; }
  seen.add(e.id);
  const start = Math.floor(e.et / W) * W;
  if (!windows.has(start)) windows.set(start, { counts: new Map(), fired: false });
  const w = windows.get(start);
  if (w.fired && watermark >= start + W + LATENESS) { tooLate++; continue; } // side output -> batch path
  w.counts.set(e.ad, (w.counts.get(e.ad) || 0) + 1);
  if (w.fired) corrections++; else onTime++;
  watermark = Math.max(watermark, e.et - OUT_OF_ORDER);
  fire(watermark, e.at);
}
fire(Infinity, "end");

console.log("== 1. streaming job (event time, 60 s windows, watermark = max event time - 10 s, lateness 30 s) ==");
console.log(`  received ${events.length}  duplicates dropped ${dups}  on time ${onTime}  late but corrected ${corrections}  too late ${tooLate}`);
for (const line of firstResults) console.log(line);

// ---- 2. ground truth from the raw log (what the batch job computes) ----
const truth = new Map(), byArrival = new Map(), once = new Set();
for (const e of events) {
  if (once.has(e.id)) continue; once.add(e.id);
  const s = Math.floor(e.et / W) * W, a = Math.floor(e.at / W) * W;
  truth.set(s, (truth.get(s) || 0) + 1);
  byArrival.set(a, (byArrival.get(a) || 0) + 1);
}
console.log("== 2. per window: streaming final vs batch truth vs processing-time windows ==");
const starts = [...new Set([...truth.keys(), ...byArrival.keys()])].sort((a, b) => a - b);
for (const s of starts) {
  const stream = windows.has(s) ? total(windows.get(s).counts) : 0;
  console.log(`  [${s}s,${s + W}s)  streaming=${stream}  batch=${truth.get(s) || 0}  missing=${(truth.get(s) || 0) - stream}  by-arrival-minute=${byArrival.get(s) || 0}`);
}
const sumTruth = [...truth.values()].reduce((x, y) => x + y, 0);
const sumStream = [...windows.values()].reduce((x, w) => x + total(w.counts), 0);
console.log(`  totals: streaming ${sumStream} + too late ${tooLate} = ${sumStream + tooLate}, batch ${sumTruth}`);