Розробка платформи для стримінгу прямих трансляцій

Розробка платформи live-стримінгу

Live-стриминг — це рішення проблеми доставки відео тисячам і мільйонам глядачів у реальному часі з мінімальною затримкою. Twitch підтримує ~2–5 секунд затримки, YouTube Live — 5–30 залежно від режиму. Побудувати те ж саме з нуля — завдання для кількох інженерів на кілька місяців. Розберімо, що всередині.

Схема доставки та протоколи

Стример → Ingest → Transcoding → CDN → Viewer

Вхідний потік від стримера — зазвичай RTMP (OBS, StreamLabs, XSplit всі підтримують RTMP з коробки). На виході до глядача — HLS або DASH для браузера, WebRTC для ultra-low-latency (< 1 сек).

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

Ingest-сервер на базі SRS

SRS (Simple Realtime Server) — open-source, Go, добре тримає навантаження:

# docker-compose.yml фрагмент
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__ {
    # Хук: повідомляємо backend про початок/кінець трансляції
    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 секунди — баланс затримки та стабільності
        hls_window  10;    # 10 сегментів у вікні
    }

    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;
        }
    }
}

Аутентифікація стримера

Стример публікує потік за допомогою stream key. Не можна приймати RTMP від невідомих джерел:

# FastAPI: хук для SRS on_publish
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

class PublishHook(BaseModel):
    action: str
    app: str
    stream: str  # stream key від стримера
    param: str   # query string

@app.post("/hooks/stream/start")
async def on_stream_start(hook: PublishHook):
    # Валідуємо 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")

    # Запускаємо трансляцію в БД
    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'])

    # Повідомляємо підписників через WebSocket
    await notify_followers(streamer['user_id'], 'stream_started')

    return {"code": 0}  # SRS очікує code=0 для дозволу

HLS-сегменти в S3

SRS пише сегменти локально, їх потрібно синхронізувати в S3 для CDN:

# inotifywait: монітуримо директорію з сегментами, пушимо в S3
inotifywait -m -e close_write /var/srs/hls/ --format '%f' |
while read filename; do
  if [[ "$filename" == *.ts || "$filename" == *.m3u8 ]]; then
    # m3u8 з коротким cache — часто змінюється
    # ts з довгим — 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

Або через s3fs / rclone mount — простіше, але менш контрольовано.

Чат у реальному часі

Чат трансляції — обов'язковий елемент. WebSocket через Redis Pub/Sub:

// Node.js: WebSocket-сервер для чату
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;
}

// Підписуємся на канал трансляції
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}`;

  // Слухаємо Redis Pub/Sub для цього стріму
  redisSub.subscribe(channel, (message) => {
    if (ws.readyState === ws.OPEN) {
      ws.send(message);
    }
  });

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

    // Антиспам: rate limit на користувача
    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 повідомлень за 5 секунд — занадто багато
      ws.send(JSON.stringify({ type: 'slowmode', waitMs: 5000 }));
      return;
    }

    // Зберігаємо в Redis Stream (ковзне вікно 1000 повідомлень)
    await redis.xAdd(`stream_chat:${streamId}`, '*', msg as any, {
      TRIM: { strategy: 'MAXLEN', threshold: 1000 }
    });

    // Публікуємо всім підключеним
    await redis.publish(channel, JSON.stringify(msg));
  });

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

Запис трансляцій

VOD (Video on Demand) після закінчення стріму — стандартне очікування:

# Celery task: після завершення стріму конвертуємо в VOD
@app.task
def process_vod(stream_id: int):
    stream = LiveStream.objects.get(id=stream_id)

    # Збираємо TS-сегменти в один файл
    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)

    # Перекодуємо для VOD (оптимізація для веб: 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 на початок для псевдострімінгу
        vod_mp4
    ], check=True)

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

Масштабування: багатосерверний ingest

Один ingest-сервер — SPOF. Для продакшну потрібен кластер з балансуванням:

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

Стримери направляються на найближчий ingest за GeoDNS. Кожен ingest пише в спільне об'єктне сховище або реплікує сегменти синхронно.

Часові рамки

MVP з RTMP-приймом, HLS-доставкою, WebSocket-чатом і записом в VOD: 10–12 тижнів. Додавання транскодування кількох якостей, gift-subscriptions, модерації чату, мобільного плеєра: ще 8–10 тижнів. Масштабування до 10k+ concurrent viewers, балансування ingest, CDN з origin shield: окремий етап архітектурного проектування.