Live Streaming Platform Development

Live Streaming Platform Development

Live streaming is a solution to delivering video to thousands and millions of viewers in real-time with minimal latency. Twitch maintains ~2–5 seconds of latency, YouTube Live 5–30 depending on mode. Building the same from scratch is a task for several engineers over several months. Let's break down what's inside.

Delivery Scheme and Protocols

Streamer → Ingest → Transcoding → CDN → Viewer

Incoming stream from streamer is typically RTMP (OBS, StreamLabs, XSplit all support RTMP out of the box). On output to viewers: HLS or DASH for browser, WebRTC for ultra-low-latency (< 1 sec).

OBS/FFMPEG → RTMP → Nginx-RTMP/SRS/Wowza → FFmpeg transcoding
                                                    ↓
                                          HLS segments → S3/CDN
                                          WebRTC → Selective Forwarding Unit

Ingest Server Based on SRS

SRS (Simple Realtime Server) — open-source, Go, handles load well:

# docker-compose.yml fragment
services:
  srs:
    image: ossrs/srs:5
    ports:
      - "1935:1935"   # RTMP
      - "1985:1985"   # HTTP API
      - "8080:8080"   # HLS
    volumes:
      - ./srs.conf:/usr/local/srs/conf/srs.conf

# srs.conf
listen              1935;
max_connections     1000;
daemon              off;

http_server {
    enabled     on;
    listen      8080;
    dir         ./objs/nginx/html;
}

vhost __defaultVhost__ {
    # Hook: notify backend of stream start/stop
    http_hooks {
        enabled on;
        on_publish  http://api:8000/hooks/stream/start;
        on_unpublish http://api:8000/hooks/stream/stop;
        on_play     http://api:8000/hooks/stream/view;
    }

    hls {
        enabled     on;
        hls_path    ./objs/nginx/html;
        hls_fragment 2;    # 2 seconds — balance latency and stability
        hls_window  10;    # 10 segments in window
    }

    transcode {
        enabled on;
        ffmpeg /usr/local/bin/ffmpeg;

        engine hd {
            enabled on;
            vcodec  libx264;
            vbitrate 2000;
            vfps    30;
            vwidth  1280; vheight 720;
            acodec  aac;
            abitrate 128;
            output rtmp://localhost:1935/[app]/[stream]_720p;
        }

        engine sd {
            enabled on;
            vcodec  libx264;
            vbitrate 800;
            vfps    30;
            vwidth  854; vheight 480;
            acodec  aac;
            abitrate 96;
            output rtmp://localhost:1935/[app]/[stream]_480p;
        }
    }
}

Streamer Authentication

Streamer publishes stream using stream key. Cannot accept RTMP from unknown sources:

# FastAPI: hook for SRS on_publish
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

class PublishHook(BaseModel):
    action: str
    app: str
    stream: str  # stream key from streamer
    param: str   # query string

@app.post("/hooks/stream/start")
async def on_stream_start(hook: PublishHook):
    # Validate stream key
    streamer = await db.fetchrow(
        "SELECT id, user_id, is_active FROM stream_keys WHERE key = $1",
        hook.stream
    )

    if not streamer or not streamer['is_active']:
        raise HTTPException(status_code=403, detail="Invalid stream key")

    # Start broadcast in DB
    await db.execute("""
        INSERT INTO live_streams (user_id, stream_key_id, started_at, status)
        VALUES ($1, $2, NOW(), 'live')
        ON CONFLICT (stream_key_id) DO UPDATE SET started_at = NOW(), status = 'live'
    """, streamer['user_id'], streamer['id'])

    # Notify followers via WebSocket
    await notify_followers(streamer['user_id'], 'stream_started')

    return {"code": 0}  # SRS expects code=0 to allow

HLS Segments to S3

SRS writes segments locally, they need to be synced to S3 for CDN:

# inotifywait: monitor segment directory, push to S3
inotifywait -m -e close_write /var/srs/hls/ --format '%f' |
while read filename; do
  if [[ "$filename" == *.ts || "$filename" == *.m3u8 ]]; then
    # m3u8 with short cache — changes often
    # ts with long — immutable
    if [[ "$filename" == *.m3u8 ]]; then
      cache="max-age=2"
    else
      cache="max-age=86400,immutable"
    fi
    aws s3 cp "/var/srs/hls/$filename" "s3://live-streams/hls/$filename" \
      --cache-control "$cache" \
      --content-type "$(get_mime $filename)"
  fi
done

Or via s3fs / rclone mount — simpler but less controlled.

Real-Time Chat

Stream chat is a mandatory element. WebSocket via Redis Pub/Sub:

// Node.js: WebSocket server for chat
import { WebSocketServer } from 'ws';
import { createClient } from 'redis';

const wss = new WebSocketServer({ port: 3001 });
const redis = createClient({ url: process.env.REDIS_URL });
const redisSub = redis.duplicate();

await redis.connect();
await redisSub.connect();

interface ChatMessage {
  type: 'message' | 'emote' | 'sub' | 'ban';
  streamId: string;
  userId: string;
  username: string;
  text: string;
  badges: string[];
  timestamp: number;
}

// Subscribe to stream channel
wss.on('connection', (ws, req) => {
  const streamId = new URL(req.url!, 'ws://x').searchParams.get('stream');
  if (!streamId) return ws.close();

  const channel = `chat:${streamId}`;

  // Listen to Redis Pub/Sub for this stream
  redisSub.subscribe(channel, (message) => {
    if (ws.readyState === ws.OPEN) {
      ws.send(message);
    }
  });

  ws.on('message', async (data) => {
    const msg: ChatMessage = JSON.parse(data.toString());

    // Anti-spam: rate limit per user
    const key = `chat_limit:${msg.userId}:${streamId}`;
    const count = await redis.incr(key);
    if (count === 1) await redis.expire(key, 5);
    if (count > 20) { // 20 messages per 5 sec — too much
      ws.send(JSON.stringify({ type: 'slowmode', waitMs: 5000 }));
      return;
    }

    // Save to Redis Stream (sliding window 1000 messages)
    await redis.xAdd(`stream_chat:${streamId}`, '*', msg as any, {
      TRIM: { strategy: 'MAXLEN', threshold: 1000 }
    });

    // Publish to all connected
    await redis.publish(channel, JSON.stringify(msg));
  });

  ws.on('close', () => {
    redisSub.unsubscribe(channel);
  });
});

Stream Recording

VOD (Video on Demand) after stream ends is standard expectation:

# Celery task: after stream ends convert to VOD
@app.task
def process_vod(stream_id: int):
    stream = LiveStream.objects.get(id=stream_id)

    # Collect TS segments into one file
    segments = sorted(
        glob(f"/var/srs/hls/{stream.stream_key}/*.ts"),
        key=lambda f: int(Path(f).stem.split('_')[-1])
    )

    concat_list = "/tmp/vod_concat.txt"
    with open(concat_list, 'w') as f:
        for s in segments: f.write(f"file '{s}'\n")

    raw_mp4 = f"/tmp/vod_{stream_id}_raw.mp4"
    subprocess.run([
        'ffmpeg', '-f', 'concat', '-safe', '0',
        '-i', concat_list,
        '-c', 'copy',
        raw_mp4
    ], check=True)

    # Re-encode for VOD (web optimization: faststart)
    vod_mp4 = f"/var/vod/{stream_id}.mp4"
    subprocess.run([
        'ffmpeg', '-i', raw_mp4,
        '-c:v', 'libx264', '-preset', 'fast', '-crf', '23',
        '-c:a', 'aac', '-b:a', '128k',
        '-movflags', '+faststart',  # moov atom to start for pseudo-streaming
        vod_mp4
    ], check=True)

    stream.vod_path = vod_mp4
    stream.status = 'ended'
    stream.save()

Scaling: Multi-Server Ingest

Single ingest server is SPOF. For production need a cluster with load balancing:

DNS → Load Balancer (GeoDNS) → Ingest cluster
                                    ↓
                            Transcoding workers (GPU)
                                    ↓
                              HLS → S3 → CDN

Streamers are routed to nearest ingest via GeoDNS. Each ingest writes to shared object storage or replicates segments synchronously.

Timeline

MVP with RTMP ingest, HLS delivery, WebSocket chat and VOD recording: 10–12 weeks. Adding multi-quality transcoding, gift subscriptions, chat moderation, mobile player: another 8–10 weeks. Scaling to 10k+ concurrent viewers, ingest balancing, CDN with origin shield: separate architectural design phase.