Tutorial: Build a Real-Time Analytics Dashboard

import http from "node:http";
import crypto from "node:crypto";

// ===========================================
// 1. EVENT DATA MODEL
// ===========================================
interface AnalyticsEvent {
  name: string;
  visitorId: string;
  userId?: string;
  properties: Record<string, string>;
  timestamp: number;
  url?: string;
  userAgent?: string;
}

type Granularity = "minute" | "hour" | "day";

interface TimeBucket {
  key: string; // "2024-01-15T10:30" for minute
  timestamp: number;
  granularity: Granularity;
  totalEvents: number;
  uniqueVisitors: Set<string>;
  eventCounts: Map<string, number>;
}

// ===========================================
// 2. TIME-SERIES AGGREGATOR
// ===========================================
class TimeSeriesStore {
  private minuteBuckets = new Map<string, TimeBucket>();
  private hourBuckets = new Map<string, TimeBucket>();
  private dayBuckets = new Map<string, TimeBucket>();

  private getBucketKey(timestamp: number, granularity: Granularity): string {
    const d = new Date(timestamp);
    switch (granularity) {
      case "minute":
        return `${d.getUTCFullYear()}-${String(d.getUTCMonth() + 1).padStart(2, "0")}-${String(d.getUTCDate()).padStart(2, "0")}T${String(d.getUTCHours()).padStart(2, "0")}:${String(d.getUTCMinutes()).padStart(2, "0")}`;
      case "hour":
        return `${d.getUTCFullYear()}-${String(d.getUTCMonth() + 1).padStart(2, "0")}-${String(d.getUTCDate()).padStart(2, "0")}T${String(d.getUTCHours()).padStart(2, "0")}`;
      case "day":
        return `${d.getUTCFullYear()}-${String(d.getUTCMonth() + 1).padStart(2, "0")}-${String(d.getUTCDate()).padStart(2, "0")}`;
    }
  }

  private getStore(g: Granularity): Map<string, TimeBucket> {
    switch (g) {
      case "minute": return this.minuteBuckets;
      case "hour": return this.hourBuckets;
      case "day": return this.dayBuckets;
    }
  }

  record(event: AnalyticsEvent): void {
    for (const granularity of ["minute", "hour", "day"] as Granularity[]) {
      const key = this.getBucketKey(event.timestamp, granularity);
      const store = this.getStore(granularity);

      let bucket = store.get(key);
      if (!bucket) {
        bucket = {
          key, timestamp: event.timestamp, granularity,
          totalEvents: 0, uniqueVisitors: new Set(), eventCounts: new Map(),
        };
        store.set(key, bucket);
      }

      bucket.totalEvents++;
      bucket.uniqueVisitors.add(event.visitorId);
      bucket.eventCounts.set(event.name, (bucket.eventCounts.get(event.name) || 0) + 1);
    }
  }

  query(from: number, to: number, granularity: Granularity): object[] {
    const store = this.getStore(granularity);
    const results: object[] = [];

    for (const [key, bucket] of store) {
      const bucketTime = new Date(key).getTime();
      if (bucketTime >= from && bucketTime <= to) {
        results.push({
          key: bucket.key,
          totalEvents: bucket.totalEvents,
          uniqueVisitors: bucket.uniqueVisitors.size,
          eventCounts: Object.fromEntries(bucket.eventCounts),
        });
      }
    }

    return results.sort((a: any, b: any) => a.key.localeCompare(b.key));
  }

  getCurrentMinute(): object | null {
    const key = this.getBucketKey(Date.now(), "minute");
    const bucket = this.minuteBuckets.get(key);
    if (!bucket) return null;
    return {
      key: bucket.key,
      totalEvents: bucket.totalEvents,
      uniqueVisitors: bucket.uniqueVisitors.size,
      eventCounts: Object.fromEntries(bucket.eventCounts),
    };
  }
}

// ===========================================
// 3. ACTIVE USER TRACKER (Sliding Window)
// ===========================================
class ActiveUserTracker {
  private visitors = new Map<string, number>(); // visitorId -> lastSeen timestamp
  private readonly windowMs: number;

  constructor(windowMinutes = 5) {
    this.windowMs = windowMinutes * 60 * 1000;
    // Cleanup expired entries every 30 seconds
    setInterval(() => this.cleanup(), 30000);
  }

  track(visitorId: string): void {
    this.visitors.set(visitorId, Date.now());
  }

  getActiveCount(): number {
    const cutoff = Date.now() - this.windowMs;
    let count = 0;
    for (const lastSeen of this.visitors.values()) {
      if (lastSeen > cutoff) count++;
    }
    return count;
  }

  private cleanup(): void {
    const cutoff = Date.now() - this.windowMs;
    for (const [id, lastSeen] of this.visitors) {
      if (lastSeen <= cutoff) this.visitors.delete(id);
    }
  }
}

// ===========================================
// 4. EVENT BUFFER (Batch Processing)
// ===========================================
class EventBuffer {
  private buffer: AnalyticsEvent[] = [];
  private timeSeries: TimeSeriesStore;
  private activeUsers: ActiveUserTracker;
  private wsHub: WebSocketHub;
  private flushInterval: ReturnType<typeof setInterval>;

  constructor(ts: TimeSeriesStore, au: ActiveUserTracker, ws: WebSocketHub) {
    this.timeSeries = ts;
    this.activeUsers = au;
    this.wsHub = ws;
    this.flushInterval = setInterval(() => this.flush(), 1000);
  }

  add(event: AnalyticsEvent): void {
    this.buffer.push(event);
  }

  private flush(): void {
    if (this.buffer.length === 0) return;
    const batch = this.buffer.splice(0);

    for (const event of batch) {
      this.timeSeries.record(event);
      this.activeUsers.track(event.visitorId);
    }

    // Broadcast current stats to dashboard clients
    const current = this.timeSeries.getCurrentMinute();
    const activeCount = this.activeUsers.getActiveCount();
    this.wsHub.broadcast({
      type: "stats_update",
      data: { currentMinute: current, activeUsers: activeCount, batchSize: batch.length },
    });

    console.log(`[FLUSH] Processed ${batch.length} events | Active users: ${activeCount}`);
  }

  stop(): void { clearInterval(this.flushInterval); }
}

// ===========================================
// 5. WEBSOCKET HUB (Simulated)
// ===========================================
class WebSocketHub {
  private clients = new Set<string>();
  private messages: object[] = [];

  connect(clientId: string): void {
    this.clients.add(clientId);
    console.log(`[WS] Client ${clientId} connected (${this.clients.size} total)`);
  }

  disconnect(clientId: string): void {
    this.clients.delete(clientId);
  }

  broadcast(message: object): void {
    this.messages.push(message);
    // In production: push to all WebSocket connections
    // For demo: store for polling via /ws/dashboard endpoint
    if (this.messages.length > 100) this.messages.shift();
  }

  getRecent(since = 0): object[] {
    return this.messages.slice(since);
  }

  getClientCount(): number {
    return this.clients.size;
  }
}

// ===========================================
// 6. HTTP SERVER
// ===========================================
const timeSeries = new TimeSeriesStore();
const activeUsers = new ActiveUserTracker(5);
const wsHub = new WebSocketHub();
const eventBuffer = new EventBuffer(timeSeries, activeUsers, wsHub);

function parseBody(req: http.IncomingMessage): Promise<unknown> {
  return new Promise((resolve, reject) => {
    const chunks: Buffer[] = [];
    req.on("data", (c) => chunks.push(c));
    req.on("end", () => {
      try { resolve(JSON.parse(Buffer.concat(chunks).toString())); }
      catch { reject(new Error("Invalid JSON")); }
    });
  });
}

function json(res: http.ServerResponse, status: number, data: unknown): void {
  res.writeHead(status, { "Content-Type": "application/json" });
  res.end(JSON.stringify(data));
}

const server = http.createServer(async (req, res) => {
  const url = new URL(req.url || "/", `http://${req.headers.host}`);
  const method = req.method || "GET";

  try {
    // POST /api/events — Single event ingestion
    if (url.pathname === "/api/events" && method === "POST") {
      const body = await parseBody(req) as any;
      if (!body.name || !body.visitorId) {
        json(res, 400, { error: "name and visitorId required" }); return;
      }
      const event: AnalyticsEvent = {
        name: body.name, visitorId: body.visitorId,
        userId: body.userId, properties: body.properties || {},
        timestamp: body.timestamp || Date.now(),
        url: body.url, userAgent: body.userAgent,
      };
      eventBuffer.add(event);
      json(res, 202, { status: "accepted" }); return;
    }

    // POST /api/events/batch — Batch event ingestion
    if (url.pathname === "/api/events/batch" && method === "POST") {
      const body = await parseBody(req) as any;
      if (!Array.isArray(body.events)) {
        json(res, 400, { error: "events array required" }); return;
      }
      for (const e of body.events) {
        if (e.name && e.visitorId) {
          eventBuffer.add({
            name: e.name, visitorId: e.visitorId,
            userId: e.userId, properties: e.properties || {},
            timestamp: e.timestamp || Date.now(),
            url: e.url, userAgent: e.userAgent,
          });
        }
      }
      json(res, 202, { accepted: body.events.length }); return;
    }

    // GET /api/metrics — Query time-series data
    if (url.pathname === "/api/metrics" && method === "GET") {
      const from = parseInt(url.searchParams.get("from") || String(Date.now() - 3600000));
      const to = parseInt(url.searchParams.get("to") || String(Date.now()));
      const granularity = (url.searchParams.get("granularity") || "minute") as Granularity;
      const data = timeSeries.query(from, to, granularity);
      json(res, 200, { data, granularity, from, to }); return;
    }

    // GET /api/active-users
    if (url.pathname === "/api/active-users" && method === "GET") {
      json(res, 200, { activeUsers: activeUsers.getActiveCount() }); return;
    }

    // GET /api/dashboard/stream — Simulated WebSocket (polling fallback)
    if (url.pathname === "/api/dashboard/stream" && method === "GET") {
      const since = parseInt(url.searchParams.get("since") || "0");
      json(res, 200, { messages: wsHub.getRecent(since), clients: wsHub.getClientCount() }); return;
    }

    if (url.pathname === "/health") { json(res, 200, { status: "ok" }); return; }
    json(res, 404, { error: "Not found" });
  } catch (err: any) {
    json(res, 500, { error: err.message || "Internal server error" });
  }
});

const PORT = parseInt(process.env.PORT || "3000");
server.listen(PORT, () => console.log(`Analytics Dashboard on http://localhost:${PORT}`));
process.on("SIGTERM", () => { eventBuffer.stop(); server.close(); });