live: live streams sent from the app — the browser's recording over a WebSocket (on 443, through Cloudflare: no ports to open), ffmpeg to 2 s HLS (H.264 720p, AAC), a live window and the whole stream (its recording), viewers counted, a dropped sender resuming the same stream, NIP-98 + coinos-name owners only, 3 at once, 4 h each, recordings kept a week.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_013ZDf1cyCQLkfjftYhymZAg

1 files changed, 290 insertions, 0 deletions

diff --git a/live/server.js b/live/server.js
new file mode 100644
index 0000000..0e04aca
--- /dev/null
+++ b/live/server.js
@@ -0,0 +1,290 @@
+// 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}
+//
+// 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 });
+  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;
+    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);
+}