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());