path: root/live/server.js

// live: the coinos app's live streams, sent from the browser and watched as
// HLS (NIP-53 streams: the app publishes the kind 30311 that points here).
//
// The browser records its camera or screen (MediaRecorder, a chunk every
// half second) and sends the chunks over a WebSocket — on 443, through
// Cloudflare like everything else, so no port has to be opened for WebRTC or
// RTMP. ffmpeg makes 2 s HLS segments of it (H.264 720p + AAC). Watchers get
// live.m3u8, the last few segments; index.m3u8 is the whole of it, the
// stream's recording once it ends.
//
// A sending connection that drops (a phone changing networks) can come back
// within a minute and carry on the same stream: each connection is a "run"
// with its own ffmpeg and playlist, and the playlists served here are the
// runs joined with a discontinuity between them.
//
//   GET  /live/ingest?auth=<base64 NIP-98 event for GET /live/ingest>&name=<coinos name>[&resume=<id>]  (WebSocket)
//        → {"id","live","recording"} first, then binary chunks in
//   GET  /live/<id>/live.m3u8      the live window (and the viewer count)
//   GET  /live/<id>/index.m3u8     everything so far; the recording once ended
//   GET  /live/<id>/<run>-<n>.ts   a segment
//   GET  /live/<id>/stats          {state, viewers, started, seconds}
//   GET  /live/ping                (WebSocket) says pong: sockets get through
//
// Who may send: a coinos name's owner (its nostr key or the wallet key that
// manages it), checked with the names registrar. One stream per person,
// MAX_STREAMS at once, MAX_HOURS each.
import { verifyEvent } from 'nostr-tools/pure';
import { mkdirSync, readFileSync, writeFileSync, existsSync, readdirSync, rmSync, statSync } from 'node:fs';
import { randomBytes } from 'node:crypto';

const DATA = process.env.LIVE_DATA || '/data';
const PORT = +(process.env.LIVE_PORT || 8795);
const BASE = (process.env.LIVE_BASE || 'https://v3.coinos.io/live').replace(/\/$/, '');
const NAMES = (process.env.NAMES_URL || 'http://names:8798').replace(/\/$/, '');
const MAX_STREAMS = +(process.env.LIVE_MAX_STREAMS || 3);
const MAX_HOURS = +(process.env.LIVE_MAX_HOURS || 4);
const KEEP_DAYS = +(process.env.LIVE_KEEP_DAYS || 7); // recordings
const CAP_BYTES = +(process.env.LIVE_CAP_GB || 30) * 1e9;
const RESUME_MS = 60_000; // a dropped sender may come back this long
const IDLE_MS = 20_000; // a sender with nothing for this long is gone
const WINDOW = 8; // segments in the live playlist
const SEG = 2; // seconds a segment

mkdirSync(DATA, { recursive: true });
const streams = new Map(); // id -> stream
const log = (...a) => console.log(new Date().toISOString(), ...a);
const idOk = (id) => /^[0-9a-f]{16}$/.test(id || '');

// ---------------------------------------------------------------- streams
function metaPath(id) { return `${DATA}/${id}/stream.json`; }
function save(s) {
  const { ws, ff, viewers, timer, ...keep } = s;
  try { writeFileSync(metaPath(s.id), JSON.stringify(keep)); } catch {}
}
function load(id) {
  if (streams.has(id)) return streams.get(id);
  if (!idOk(id) || !existsSync(metaPath(id))) return null;
  try {
    const s = JSON.parse(readFileSync(metaPath(id), 'utf8'));
    s.viewers = new Map();
    if (s.state !== 'ended') { s.state = 'ended'; s.ended = s.ended || Date.now(); save(s); } // a restart ended it
    streams.set(id, s);
    return s;
  } catch { return null; }
}
const live = () => [...streams.values()].filter((s) => s.state !== 'ended');

// A run's playlist as ffmpeg wrote it: [{dur, uri}]
function runSegments(s, run) {
  let text = '';
  try { text = readFileSync(`${DATA}/${s.id}/run${run}.m3u8`, 'utf8'); } catch { return []; }
  const out = [];
  const lines = text.split('\n');
  for (let i = 0; i < lines.length; i++) {
    const m = /^#EXTINF:([\d.]+)/.exec(lines[i]);
    if (m && lines[i + 1] && !lines[i + 1].startsWith('#')) out.push({ dur: +m[1], uri: lines[i + 1].trim(), run });
  }
  return out;
}
function allSegments(s) {
  const out = [];
  for (let r = 0; r < s.runs; r++) out.push(...runSegments(s, r));
  return out;
}
// The playlist: the runs joined, a discontinuity where one ends and the next begins.
function playlist(s, window) {
  const segs = allSegments(s);
  const from = window ? Math.max(0, segs.length - WINDOW) : 0;
  let discBefore = 0;
  for (let i = 1; i <= from && i < segs.length; i++) if (segs[i].run !== segs[i - 1].run) discBefore++;
  const shown = segs.slice(from);
  const target = Math.max(SEG, ...shown.map((x) => Math.ceil(x.dur)));
  const ended = s.state === 'ended';
  const out = ['#EXTM3U', '#EXT-X-VERSION:3', '#EXT-X-TARGETDURATION:' + target, '#EXT-X-MEDIA-SEQUENCE:' + from,
    '#EXT-X-DISCONTINUITY-SEQUENCE:' + discBefore, '#EXT-X-INDEPENDENT-SEGMENTS'];
  if (!window) out.push('#EXT-X-PLAYLIST-TYPE:' + (ended ? 'VOD' : 'EVENT'));
  shown.forEach((x, i) => {
    if (i > 0 && x.run !== shown[i - 1].run) out.push('#EXT-X-DISCONTINUITY');
    out.push('#EXTINF:' + x.dur.toFixed(3) + ',', x.uri);
  });
  if (ended) out.push('#EXT-X-ENDLIST');
  return out.join('\n') + '\n';
}

// ffmpeg for one run: whatever the browser recorded (WebM or MP4, VP8/VP9/H.264,
// Opus/AAC) in on stdin; H.264 at most 1280 on the long side, 30 fps,
// keyframes every segment, AAC; 2 s segments, every one kept.
function ffmpegArgs(dir, run) {
  return ['-hide_banner', '-loglevel', 'error', '-fflags', '+genpts', '-i', 'pipe:0',
    '-map', '0:v:0', '-map', '0:a:0?',
    '-vf', "scale=w='if(gt(iw,ih),min(1280,iw),-2)':h='if(gt(iw,ih),-2,min(1280,ih))',fps=30,format=yuv420p",
    '-c:v', 'libx264', '-preset', 'veryfast', '-tune', 'zerolatency', '-threads', '2',
    '-b:v', '2500k', '-maxrate', '2500k', '-bufsize', '5000k',
    '-g', String(SEG * 30), '-keyint_min', String(SEG * 30), '-sc_threshold', '0',
    '-force_key_frames', `expr:gte(t,n_forced*${SEG})`,
    '-c:a', 'aac', '-b:a', '128k', '-ar', '48000', '-ac', '2',
    '-f', 'hls', '-hls_time', String(SEG), '-hls_list_size', '0', '-hls_playlist_type', 'event',
    '-hls_flags', 'independent_segments+temp_file',
    '-hls_segment_filename', `${dir}/${run}-%05d.ts`, `${dir}/run${run}.m3u8`];
}

function startRun(s, ws) {
  const run = s.runs++;
  const dir = `${DATA}/${s.id}`;
  const ff = Bun.spawn(['ffmpeg', ...ffmpegArgs(dir, run)], { stdin: 'pipe', stdout: 'ignore', stderr: 'pipe' });
  s.ff = ff; s.ws = ws; s.state = 'live'; s.lastData = Date.now();
  s.dropped = null;
  save(s);
  (async () => {
    const err = await new Response(ff.stderr).text().catch(() => '');
    const code = await ff.exited;
    if (code && err.trim()) log('ffmpeg', s.id, 'run', run, 'exit', code, err.trim().slice(-400));
  })();
  log('run', s.id, run, 'for', s.name);
}
function endRun(s) {
  const ff = s.ff;
  s.ff = null; s.ws = null;
  if (ff) { try { ff.stdin.end(); } catch {} setTimeout(() => { try { ff.kill(); } catch {} }, 8000); }
}
// the sender went away: the stream waits a minute for it, then it's over
function dropped(s) {
  endRun(s);
  if (s.state === 'ended') return;
  s.state = 'waiting';
  s.dropped = Date.now();
  save(s);
}
function end(s, why) {
  endRun(s);
  if (s.state === 'ended') return;
  s.state = 'ended';
  s.ended = Date.now();
  save(s);
  log('end', s.id, why);
}

// the stream's tending: a sender gone quiet, one gone for good, the time cap
setInterval(() => {
  const now = Date.now();
  for (const s of live()) {
    if (s.state === 'live' && now - s.lastData > IDLE_MS) { try { s.ws && s.ws.close(4000, 'idle'); } catch {} dropped(s); }
    if (s.state === 'waiting' && now - s.dropped > RESUME_MS) end(s, 'sender gone');
    if (now - s.started > MAX_HOURS * 3600_000) { try { s.ws && s.ws.close(4001, 'time'); } catch {} end(s, 'time cap'); }
  }
}, 2000).unref?.();

// recordings past their keep, and the oldest ended ones past the cap
function trim() {
  let dirs = [];
  try { dirs = readdirSync(DATA).filter(idOk); } catch { return; }
  const info = dirs.map((id) => {
    let bytes = 0, at = 0;
    try { for (const f of readdirSync(`${DATA}/${id}`)) { const st = statSync(`${DATA}/${id}/${f}`); bytes += st.size; at = Math.max(at, st.mtimeMs); } } catch {}
    return { id, bytes, at, live: streams.get(id) && streams.get(id).state !== 'ended' };
  }).sort((a, b) => a.at - b.at);
  let total = info.reduce((n, x) => n + x.bytes, 0);
  for (const x of info) {
    if (x.live) continue;
    if (Date.now() - x.at > KEEP_DAYS * 86400_000 || total > CAP_BYTES) {
      rmSync(`${DATA}/${x.id}`, { recursive: true, force: true });
      streams.delete(x.id);
      total -= x.bytes;
    }
  }
}
setInterval(trim, 10 * 60_000).unref?.();

// ------------------------------------------------------------------ auth
async function ownsName(name, pubkey) {
  if (!/^[a-z0-9._-]{1,30}$/.test(name || '')) return false;
  try {
    const r = await fetch(`${NAMES}/name/${name}`, { signal: AbortSignal.timeout(5000) });
    const j = await r.json();
    return !!j.taken && (j.pubkey === pubkey || j.manager === pubkey);
  } catch { return false; }
}
function checkAuth(url) {
  let evt;
  try { evt = JSON.parse(Buffer.from(url.searchParams.get('auth') || '', 'base64').toString('utf8')); } catch { return { error: 'bad auth' }; }
  if (!evt || evt.kind !== 27235) return { error: 'auth must be kind 27235' };
  if (Math.abs(Date.now() / 1000 - (evt.created_at || 0)) > 60) return { error: 'auth expired' };
  if (!verifyEvent(evt)) return { error: 'bad signature' };
  const tag = (k) => (evt.tags.find((x) => x[0] === k) || [])[1] || '';
  if (tag('method').toUpperCase() !== 'GET') return { error: 'method mismatch' };
  let u; try { u = new URL(tag('u')); } catch { return { error: 'bad url tag' }; }
  if (u.pathname !== '/live/ingest') return { error: 'url mismatch' };
  return { pubkey: evt.pubkey };
}

// ----------------------------------------------------------------- serving
const ipOf = (req, server) => req.headers.get('cf-connecting-ip') || (req.headers.get('x-forwarded-for') || '').split(',')[0].trim() || server.requestIP(req)?.address || '';
const viewersOf = (s) => {
  const now = Date.now();
  for (const [ip, at] of s.viewers) if (now - at > 20_000) s.viewers.delete(ip);
  return s.viewers.size;
};
const json = (o, status = 200) => new Response(JSON.stringify(o), { status, headers: { 'content-type': 'application/json', 'access-control-allow-origin': '*', 'cache-control': 'no-store' } });
const hlsHeaders = { 'content-type': 'application/vnd.apple.mpegurl', 'access-control-allow-origin': '*', 'cache-control': 'no-store' };

export async function handle(req, server) {
  const url = new URL(req.url);
  const path = url.pathname.replace(/^\/live/, '');
  if (path === '/health') return json({ ok: true, live: live().length });
  // a WebSocket that says pong: is the way in (Cloudflare, the proxies) open to sockets?
  if (path === '/ping') return server.upgrade(req, { data: { ping: true } }) ? undefined : json({ error: 'websocket expected' }, 426);
  if (path === '/ingest') {
    const a = checkAuth(url);
    if (a.error) return json({ error: a.error }, 401);
    const name = (url.searchParams.get('name') || '').toLowerCase();
    if (!(await ownsName(name, a.pubkey))) return json({ error: 'only a coinos name\'s owner can go live' }, 403);
    const resume = url.searchParams.get('resume');
    let s = resume ? load(resume) : null;
    if (s && (s.pubkey !== a.pubkey || s.state !== 'waiting')) s = null;
    if (!s) {
      if (live().some((x) => x.pubkey === a.pubkey)) return json({ error: 'you are already live' }, 409);
      if (live().length >= MAX_STREAMS) return json({ error: 'too many streams right now' }, 503);
    }
    const ok = server.upgrade(req, { data: { pubkey: a.pubkey, name, resume: s ? s.id : null } });
    return ok ? undefined : json({ error: 'websocket expected' }, 426);
  }
  const m = /^\/([0-9a-f]{16})\/(live\.m3u8|index\.m3u8|stats|\d+-\d{5}\.ts)$/.exec(path);
  if (!m || req.method !== 'GET') return json({ error: 'not found' }, 404);
  const s = load(m[1]);
  if (!s) return json({ error: 'no such stream' }, 404);
  if (m[2] === 'stats') return json({ state: s.state, viewers: viewersOf(s), started: s.started, seconds: Math.round(((s.ended || Date.now()) - s.started) / 1000) });
  if (m[2].endsWith('.m3u8')) {
    if (s.state !== 'ended') s.viewers.set(ipOf(req, server), Date.now());
    return new Response(playlist(s, m[2] === 'live.m3u8'), { headers: hlsHeaders });
  }
  const f = Bun.file(`${DATA}/${s.id}/${m[2]}`);
  if (!(await f.exists())) return json({ error: 'no such segment' }, 404);
  return new Response(f, { headers: { 'content-type': 'video/mp2t', 'access-control-allow-origin': '*', 'cache-control': 'public, max-age=86400, immutable' } });
}

export const websocket = {
  maxPayloadLength: 8 * 1024 * 1024,
  idleTimeout: 60,
  open(ws) {
    const d = ws.data;
    if (d.ping) { ws.send('pong'); ws.close(1000, 'pong'); return; }
    let s = d.resume ? streams.get(d.resume) : null;
    if (!s) {
      const id = randomBytes(8).toString('hex');
      mkdirSync(`${DATA}/${id}`, { recursive: true });
      s = { id, pubkey: d.pubkey, name: d.name, started: Date.now(), ended: null, runs: 0, state: 'live', viewers: new Map() };
      streams.set(id, s);
      log('start', id, 'for', d.name);
    }
    d.id = s.id;
    startRun(s, ws);
    ws.send(JSON.stringify({ id: s.id, live: `${BASE}/${s.id}/live.m3u8`, recording: `${BASE}/${s.id}/index.m3u8`, resumed: s.runs > 1 }));
  },
  message(ws, msg) {
    const s = streams.get(ws.data.id);
    if (!s || s.ws !== ws || !s.ff) return;
    if (typeof msg === 'string') {
      if (msg === 'end') { end(s, 'sender ended'); try { ws.close(1000, 'ended'); } catch {} }
      return;
    }
    s.lastData = Date.now();
    try { s.ff.stdin.write(msg); s.ff.stdin.flush(); } catch { try { ws.close(4002, 'encoder'); } catch {} dropped(s); }
  },
  close(ws) {
    const s = streams.get(ws.data.id);
    if (s && s.ws === ws) dropped(s);
  },
};

if (import.meta.main) {
  trim();
  Bun.serve({ port: PORT, hostname: '0.0.0.0', fetch: handle, websocket, idleTimeout: 30 });
  log('live on', PORT);
}