Case Study: Live Streaming Platform

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

// ===========================================
// 1. TYPES
// ===========================================
type StreamStatus = "idle" | "live" | "ended";

interface Stream {
  id: string;
  title: string;
  streamKey: string;
  status: StreamStatus;
  startedAt: string | null;
  endedAt: string | null;
  broadcasterId: string;
  quality: string[];
  viewerCount: number;
  chatEnabled: boolean;
  recordingEnabled: boolean;
}

interface Segment {
  streamId: string;
  quality: string;
  sequenceNum: number;
  duration: number; // seconds
  createdAt: number;
  data: string; // placeholder for actual .ts data
}

interface ChatMessage {
  id: string;
  streamId: string;
  userId: string;
  username: string;
  content: string;
  timestamp: number;
  streamTime: number; // seconds since stream start
}

interface ViewerSession {
  userId: string;
  streamId: string;
  quality: string;
  connectedAt: number;
  lastHeartbeat: number;
}

// ===========================================
// 2. STREAM MANAGER
// ===========================================
class StreamManager {
  private streams = new Map<string, Stream>();
  private streamsByKey = new Map<string, string>(); // streamKey -> streamId

  create(broadcasterId: string, title: string): Stream {
    const stream: Stream = {
      id: crypto.randomUUID().slice(0, 8),
      title,
      streamKey: `live_${crypto.randomBytes(16).toString("hex")}`,
      status: "idle",
      startedAt: null, endedAt: null,
      broadcasterId,
      quality: ["1080p", "720p", "480p", "360p"],
      viewerCount: 0,
      chatEnabled: true, recordingEnabled: true,
    };
    this.streams.set(stream.id, stream);
    this.streamsByKey.set(stream.streamKey, stream.id);
    return stream;
  }

  authenticate(streamKey: string): Stream | null {
    const streamId = this.streamsByKey.get(streamKey);
    if (!streamId) return null;
    return this.streams.get(streamId) || null;
  }

  goLive(streamId: string): Stream {
    const stream = this.streams.get(streamId);
    if (!stream) throw new Error("Stream not found");
    if (stream.status === "live") throw new Error("Already live");
    stream.status = "live";
    stream.startedAt = new Date().toISOString();
    console.log(`[STREAM] ${stream.id} is LIVE — ${stream.title}`);
    return stream;
  }

  endStream(streamId: string): Stream {
    const stream = this.streams.get(streamId);
    if (!stream) throw new Error("Stream not found");
    if (stream.status !== "live") throw new Error("Not live");
    stream.status = "ended";
    stream.endedAt = new Date().toISOString();
    console.log(`[STREAM] ${stream.id} ended`);
    return stream;
  }

  get(id: string): Stream | null {
    return this.streams.get(id) || null;
  }

  getLive(): Stream[] {
    return [...this.streams.values()].filter(s => s.status === "live");
  }
}

// ===========================================
// 3. SEGMENT BUFFER (Sliding Window HLS)
// ===========================================
class SegmentBuffer {
  private segments = new Map<string, Segment[]>(); // "streamId:quality" -> segments
  private readonly windowSize = 15; // keep last 15 segments (~30-60 seconds)
  private sequenceCounters = new Map<string, number>();

  addSegment(streamId: string, quality: string, duration: number): Segment {
    const key = `${streamId}:${quality}`;
    const seqNum = (this.sequenceCounters.get(key) || 0) + 1;
    this.sequenceCounters.set(key, seqNum);

    const segment: Segment = {
      streamId, quality, sequenceNum: seqNum,
      duration, createdAt: Date.now(),
      data: `segment_${seqNum}_${quality}.ts`,
    };

    const list = this.segments.get(key) || [];
    list.push(segment);

    // Sliding window: remove old segments
    while (list.length > this.windowSize) list.shift();
    this.segments.set(key, list);

    return segment;
  }

  getSegments(streamId: string, quality: string): Segment[] {
    return this.segments.get(`${streamId}:${quality}`) || [];
  }

  generatePlaylist(streamId: string, quality: string): string {
    const segments = this.getSegments(streamId, quality);
    if (segments.length === 0) return "";

    const firstSeq = segments[0].sequenceNum;
    let playlist = "#EXTM3U\n#EXT-X-VERSION:3\n";
    playlist += `#EXT-X-TARGETDURATION:4\n`;
    playlist += `#EXT-X-MEDIA-SEQUENCE:${firstSeq}\n\n`;

    for (const seg of segments) {
      playlist += `#EXTINF:${seg.duration.toFixed(3)},\n`;
      playlist += `${seg.data}\n`;
    }
    return playlist;
  }

  generateMasterPlaylist(streamId: string, qualities: string[]): string {
    const bandwidths: Record<string, number> = {
      "1080p": 5000000, "720p": 2500000, "480p": 1000000, "360p": 500000,
    };
    const resolutions: Record<string, string> = {
      "1080p": "1920x1080", "720p": "1280x720", "480p": "854x480", "360p": "640x360",
    };

    let manifest = "#EXTM3U\n#EXT-X-VERSION:3\n\n";
    for (const q of qualities) {
      manifest += `#EXT-X-STREAM-INF:BANDWIDTH=${bandwidths[q] || 1000000},RESOLUTION=${resolutions[q] || "640x360"}\n`;
      manifest += `${q}/playlist.m3u8\n\n`;
    }
    return manifest;
  }
}

// ===========================================
// 4. VIEWER MANAGER
// ===========================================
class ViewerManager {
  private viewers = new Map<string, ViewerSession>(); // viewerId -> session
  private readonly timeout = 30000; // 30s heartbeat timeout

  join(userId: string, streamId: string, quality: string): void {
    this.viewers.set(`${userId}:${streamId}`, {
      userId, streamId, quality, connectedAt: Date.now(), lastHeartbeat: Date.now(),
    });
  }

  leave(userId: string, streamId: string): void {
    this.viewers.delete(`${userId}:${streamId}`);
  }

  heartbeat(userId: string, streamId: string): void {
    const key = `${userId}:${streamId}`;
    const session = this.viewers.get(key);
    if (session) session.lastHeartbeat = Date.now();
  }

  getCount(streamId: string): number {
    const now = Date.now();
    let count = 0;
    for (const [, session] of this.viewers) {
      if (session.streamId === streamId && now - session.lastHeartbeat < this.timeout) count++;
    }
    return count;
  }
}

// ===========================================
// 5. LIVE CHAT
// ===========================================
class LiveChat {
  private messages = new Map<string, ChatMessage[]>(); // streamId -> messages
  private rateLimiter = new Map<string, number>(); // "userId:streamId" -> lastMessage timestamp
  private readonly slowModeMs = 3000; // 1 message per 3 seconds
  private readonly maxHistory = 200;

  send(streamId: string, userId: string, username: string, content: string, streamStartTime: number): ChatMessage {
    const key = `${userId}:${streamId}`;
    const lastMsg = this.rateLimiter.get(key) || 0;
    if (Date.now() - lastMsg < this.slowModeMs) {
      throw new Error("Slow mode: wait before sending another message");
    }

    const msg: ChatMessage = {
      id: crypto.randomUUID().slice(0, 8),
      streamId, userId, username, content,
      timestamp: Date.now(),
      streamTime: streamStartTime > 0 ? (Date.now() - streamStartTime) / 1000 : 0,
    };

    const list = this.messages.get(streamId) || [];
    list.push(msg);
    if (list.length > this.maxHistory) list.shift();
    this.messages.set(streamId, list);
    this.rateLimiter.set(key, Date.now());

    return msg;
  }

  getRecent(streamId: string, limit = 50): ChatMessage[] {
    const list = this.messages.get(streamId) || [];
    return list.slice(-limit);
  }
}

// ===========================================
// 6. STREAM HEALTH MONITOR
// ===========================================
class HealthMonitor {
  private metrics = new Map<string, { bitrate: number; fps: number; droppedFrames: number; lastUpdate: number }>();

  update(streamId: string, bitrate: number, fps: number, droppedFrames: number): void {
    this.metrics.set(streamId, { bitrate, fps, droppedFrames, lastUpdate: Date.now() });

    if (bitrate < 500000) console.log(`[HEALTH] ⚠ Stream ${streamId}: low bitrate ${(bitrate/1000).toFixed(0)}kbps`);
    if (fps < 20) console.log(`[HEALTH] ⚠ Stream ${streamId}: low FPS ${fps}`);
    if (droppedFrames > 100) console.log(`[HEALTH] ⚠ Stream ${streamId}: ${droppedFrames} dropped frames`);
  }

  get(streamId: string) { return this.metrics.get(streamId); }
}

// ===========================================
// 7. TRANSCODER SIMULATOR
// ===========================================
class LiveTranscoder {
  private segmentBuffer: SegmentBuffer;
  private intervals = new Map<string, ReturnType<typeof setInterval>>();

  constructor(buffer: SegmentBuffer) { this.segmentBuffer = buffer; }

  startTranscoding(streamId: string, qualities: string[]): void {
    // Simulate generating segments every 2 seconds
    const interval = setInterval(() => {
      for (const quality of qualities) {
        this.segmentBuffer.addSegment(streamId, quality, 2.0);
      }
    }, 2000);
    this.intervals.set(streamId, interval);
    console.log(`[TRANSCODE] Started for stream ${streamId} — ${qualities.length} qualities`);
  }

  stopTranscoding(streamId: string): void {
    const interval = this.intervals.get(streamId);
    if (interval) clearInterval(interval);
    this.intervals.delete(streamId);
  }
}

// ===========================================
// 8. HTTP SERVER
// ===========================================
const streams = new StreamManager();
const segmentBuffer = new SegmentBuffer();
const viewers = new ViewerManager();
const chat = new LiveChat();
const health = new HealthMonitor();
const transcoder = new LiveTranscoder(segmentBuffer);

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/streams — Create stream
    if (url.pathname === "/api/streams" && method === "POST") {
      const body = await parseBody(req) as any;
      const stream = streams.create(body.broadcasterId || "anon", body.title || "Untitled");
      json(res, 201, stream); return;
    }

    // POST /api/streams/:id/start — Go live
    const startMatch = url.pathname.match(/^\/api\/streams\/([^/]+)\/start$/);
    if (startMatch && method === "POST") {
      const body = await parseBody(req) as any;
      const stream = streams.authenticate(body.streamKey);
      if (!stream || stream.id !== startMatch[1]) {
        json(res, 401, { error: "Invalid stream key" }); return;
      }
      const live = streams.goLive(stream.id);
      transcoder.startTranscoding(stream.id, live.quality);
      json(res, 200, live); return;
    }

    // POST /api/streams/:id/end — End stream
    const endMatch = url.pathname.match(/^\/api\/streams\/([^/]+)\/end$/);
    if (endMatch && method === "POST") {
      transcoder.stopTranscoding(endMatch[1]);
      const stream = streams.endStream(endMatch[1]);
      json(res, 200, stream); return;
    }

    // GET /api/streams/:id/master.m3u8 — HLS master playlist
    const masterMatch = url.pathname.match(/^\/api\/streams\/([^/]+)\/master\.m3u8$/);
    if (masterMatch && method === "GET") {
      const stream = streams.get(masterMatch[1]);
      if (!stream || stream.status !== "live") { json(res, 404, { error: "Stream not live" }); return; }
      const manifest = segmentBuffer.generateMasterPlaylist(stream.id, stream.quality);
      res.writeHead(200, { "Content-Type": "application/vnd.apple.mpegurl" });
      res.end(manifest); return;
    }

    // GET /api/streams/:id/:quality/playlist.m3u8 — Variant playlist
    const variantMatch = url.pathname.match(/^\/api\/streams\/([^/]+)\/(\w+)\/playlist\.m3u8$/);
    if (variantMatch && method === "GET") {
      const playlist = segmentBuffer.generatePlaylist(variantMatch[1], variantMatch[2]);
      res.writeHead(200, { "Content-Type": "application/vnd.apple.mpegurl" });
      res.end(playlist); return;
    }

    // GET /api/streams/:id/viewers
    const viewerMatch = url.pathname.match(/^\/api\/streams\/([^/]+)\/viewers$/);
    if (viewerMatch && method === "GET") {
      json(res, 200, { count: viewers.getCount(viewerMatch[1]) }); return;
    }

    // POST /api/streams/:id/chat — Send chat message
    const chatMatch = url.pathname.match(/^\/api\/streams\/([^/]+)\/chat$/);
    if (chatMatch && method === "POST") {
      const body = await parseBody(req) as any;
      const stream = streams.get(chatMatch[1]);
      const startTime = stream?.startedAt ? new Date(stream.startedAt).getTime() : 0;
      const msg = chat.send(chatMatch[1], body.userId || "anon", body.username || "Anonymous", body.content, startTime);
      json(res, 201, msg); return;
    }

    // GET /api/streams/:id/chat
    if (chatMatch && method === "GET") {
      const limit = parseInt(url.searchParams.get("limit") || "50");
      json(res, 200, { messages: chat.getRecent(chatMatch[1], limit) }); return;
    }

    // GET /api/streams/live — List live streams
    if (url.pathname === "/api/streams/live" && method === "GET") {
      const live = streams.getLive().map(s => ({
        ...s, viewerCount: viewers.getCount(s.id), streamKey: undefined,
      }));
      json(res, 200, { streams: live }); return;
    }

    // GET /api/streams/:id
    const streamMatch = url.pathname.match(/^\/api\/streams\/([^/]+)$/);
    if (streamMatch && method === "GET") {
      const stream = streams.get(streamMatch[1]);
      if (!stream) { json(res, 404, { error: "Stream not found" }); return; }
      json(res, 200, { ...stream, viewerCount: viewers.getCount(stream.id), streamKey: undefined }); return;
    }

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

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