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
//   GET  /live/turn                ICE servers for a call (Cloudflare TURN, a few hours'
//                                  credentials), Authorization: Nostr <NIP-98 for GET /live/turn>
//   GET  /live/allowed?name=       may this key go live? {ok} or {ok:false, reason, until}
//                                  (Authorization: Nostr <NIP-98 for GET /live/allowed>)
//   POST /live/<id>/report         {reason, text}: a viewer reports the stream; the admins
//                                  get a DM (Authorization: Nostr <NIP-98 for POST /live/<id>/report>)
//   POST /live/<id>/remove         {ban}: an admin ends the stream and takes it down
//
// Who may send: a coinos name's owner (its nostr key or the wallet key that
// manages it), checked with the names registrar, once the name is a day old
// (a throwaway name can't stream the minute it's made), and not banned. One
// stream per person, MAX_STREAMS at once, MAX_HOURS each.
//
// A removed stream is not deleted: it moves to $LIVE_DATA/.removed/<id>,
// served to nobody and never trimmed, so it's there to report (Canada's
// mandatory reporting law expects it kept); an admin deletes it by hand.
import { verifyEvent } from 'nostr-tools/pure';
import { SimplePool } from 'nostr-tools/pool';
import { wrapManyEvents } from 'nostr-tools/nip17';
import { naddrEncode, npubEncode } from 'nostr-tools/nip19';
import { mkdirSync, readFileSync, writeFileSync, existsSync, readdirSync, rmSync, statSync, renameSync, appendFileSync } 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 || 90); // recordings (the app lists a month of past streams)
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
// Calls (the app's 1:1 voice and video, NIP-AC): TURN for the connections
// that can't go direct, from Cloudflare's TURN service; the key is in
// $LIVE_CONFIG ({turnKeyId, turnKeyToken}).
const CONFIG = (() => { try { return JSON.parse(readFileSync(process.env.LIVE_CONFIG || '/etc/coinosv3/config.json', 'utf8')); } catch { return {}; } })();
const TURN_TTL = 4 * 3600;
const STUN_ONLY = [{ urls: ['stun:stun.cloudflare.com:3478', 'stun:stun.l.google.com:19302'] }];
const turnAsks = new Map(); // pubkey or ip -> [times]
function turnOk(key, max) {
  const now = Date.now(), list = (turnAsks.get(key) || []).filter((t) => now - t < 3600_000);
  if (list.length >= max) return false;
  list.push(now); turnAsks.set(key, list);
  if (turnAsks.size > 20000) turnAsks.clear();
  return true;
}
async function turnServers() {
  if (!CONFIG.turnKeyId || !CONFIG.turnKeyToken) return STUN_ONLY;
  const r = await fetch(`https://rtc.live.cloudflare.com/v1/turn/keys/${CONFIG.turnKeyId}/credentials/generate-ice-servers`, {
    method: 'POST', signal: AbortSignal.timeout(8000),
    headers: { authorization: 'Bearer ' + CONFIG.turnKeyToken, 'content-type': 'application/json', 'user-agent': 'coinos-live/1' },
    body: JSON.stringify({ ttl: TURN_TTL }),
  });
  if (!r.ok) throw new Error('turn ' + r.status);
  const j = await r.json();
  return Array.isArray(j.iceServers) ? j.iceServers : [j.iceServers].filter(Boolean);
}
// The admins (the remove button, report DMs); the DMs come from config.alertSk
// (the registrar's welcome key) to the admins' inbox relays.
const ADMINS = Array.isArray(CONFIG.admins) ? CONFIG.admins : ['98ae4da926c471c23fd12d1ebdd5839ba82917baa618e184e0c9916d93dcf4f7'];
const ALERT_SK = /^[0-9a-f]{64}$/.test(CONFIG.alertSk || '') ? Uint8Array.from(Buffer.from(CONFIG.alertSk, 'hex')) : null;
const NAME_MIN_AGE = +(process.env.LIVE_NAME_MIN_AGE_H || 24) * 3600_000;
const DM_RELAYS = ['wss://relay.coinos.io', 'wss://relay.damus.io', 'wss://nos.lol'];
const WINDOW = 8; // segments in the live playlist
const SEG = 2; // seconds a segment

mkdirSync(DATA, { recursive: true });
const REMOVED = `${DATA}/.removed`;
mkdirSync(REMOVED, { recursive: true });
const BANNED_FILE = `${DATA}/.banned.json`;
const banned = (() => { try { return JSON.parse(readFileSync(BANNED_FILE, 'utf8')); } catch { return {}; } })(); // pubkey -> {at, by, stream}
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(`${REMOVED}/${id}`)) return { id, state: 'removed', viewers: new Map() };
  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
// May this key go live under this name? {ok} or {ok:false, reason[, until]}
async function mayStream(name, pubkey) {
  if (banned[pubkey]) return { ok: false, reason: 'banned' };
  if (!/^[a-z0-9._-]{1,30}$/.test(name || '')) return { ok: false, reason: 'no-name' };
  let j;
  try { j = await (await fetch(`${NAMES}/name/${name}`, { signal: AbortSignal.timeout(5000) })).json(); } catch { return { ok: false, reason: 'no-name' }; }
  if (!j.taken || (j.pubkey !== pubkey && j.manager !== pubkey)) return { ok: false, reason: 'no-name' };
  if (j.created && Date.now() - j.created < NAME_MIN_AGE) return { ok: false, reason: 'new-name', until: j.created + NAME_MIN_AGE };
  return { ok: true };
}
const REFUSALS = { banned: 'you can\'t go live here', 'no-name': 'only a coinos name\'s owner can go live', 'new-name': 'a name can go live a day after it\'s claimed' };
function checkAuth(url, req, path = '/live/ingest', method = 'GET') {
  let evt;
  const header = /^Nostr\s+(.+)$/i.exec((req && req.headers.get('authorization')) || '');
  try { evt = JSON.parse(Buffer.from((header && header[1]) || 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() !== method) return { error: 'method mismatch' };
  let u; try { u = new URL(tag('u')); } catch { return { error: 'bad url tag' }; }
  if (u.pathname !== path) 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 === '/turn' && req.method === 'GET') {
    const a = checkAuth(url, req, '/live/turn');
    if (a.error) return json({ error: a.error }, 401);
    if (!turnOk('pk:' + a.pubkey, 30) || !turnOk('ip:' + ipOf(req, server), 60)) return json({ error: 'rate limited' }, 429);
    try { return json({ iceServers: await turnServers(), ttl: TURN_TTL }); }
    catch (e) { log('turn', e.message); return json({ iceServers: STUN_ONLY, ttl: 600 }); }
  }
  // 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, req);
    if (a.error) return json({ error: a.error }, 401);
    const name = (url.searchParams.get('name') || '').toLowerCase();
    const may = await mayStream(name, a.pubkey);
    if (!may.ok) return json({ error: REFUSALS[may.reason], reason: may.reason, until: may.until }, 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);
  }
  if (path === '/allowed' && req.method === 'GET') {
    const a = checkAuth(url, req, '/live/allowed');
    if (a.error) return json({ error: a.error }, 401);
    return json(await mayStream((url.searchParams.get('name') || '').toLowerCase(), a.pubkey));
  }
  const act = /^\/([0-9a-f]{16})\/(report|remove)$/.exec(path);
  if (act && req.method === 'OPTIONS') return new Response(null, { headers: { 'access-control-allow-origin': '*', 'access-control-allow-headers': 'authorization, content-type', 'access-control-allow-methods': 'POST' } });
  if (act && req.method === 'POST') {
    const a = checkAuth(url, req, `/live/${act[1]}/${act[2]}`, 'POST');
    if (a.error) return json({ error: a.error }, 401);
    const s = load(act[1]);
    if (!s) return json({ error: 'no such stream' }, 404);
    const body = await req.json().catch(() => ({}));
    return act[2] === 'report' ? report(s, a.pubkey, body, ipOf(req, server)) : remove(s, a.pubkey, body);
  }
  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 (s.state === 'removed') return json({ error: 'removed', state: 'removed' }, 410);
  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' } });
}

// ----------------------------------------------------------- moderation
const pool = new SimplePool();
async function inboxOf(pk) {
  try {
    const evs = await pool.querySync(DM_RELAYS, { kinds: [10050], authors: [pk] }, { maxWait: 4000 });
    const ev = evs.sort((a, b) => b.created_at - a.created_at)[0];
    const r = ev ? ev.tags.filter((t) => t[0] === 'relay' && /^wss:\/\//.test(t[1] || '')).map((t) => t[1]) : [];
    return r.slice(0, 4);
  } catch { return []; }
}
// a DM to each admin (NIP-17, the recipient's copy only)
async function alertAdmins(text) {
  if (!ALERT_SK) { log('alert (no alertSk):', text); return; }
  for (const pk of ADMINS) {
    try {
      const [, wrap] = wrapManyEvents(ALERT_SK, [{ publicKey: pk }], text);
      const relays = [...new Set([...(await inboxOf(pk)), ...DM_RELAYS])];
      await Promise.any(pool.publish(relays, wrap)).catch(() => log('alert to', pk.slice(0, 12), 'not delivered'));
    } catch (e) { log('alert failed', e.message); }
  }
}
const streamLink = (s) => 'https://v3.coinos.io/' + naddrEncode({ kind: 30311, pubkey: s.pubkey, identifier: s.id });
const reportsBy = new Map(); // reporter or ip -> [times]
function report(s, pubkey, body, ip) {
  if (s.state === 'removed') return json({ ok: true });
  const now = Date.now();
  for (const k of ['pk:' + pubkey, 'ip:' + ip]) {
    const list = (reportsBy.get(k) || []).filter((t) => now - t < 3600_000);
    if (list.length >= 10) return json({ error: 'rate limited' }, 429);
    list.push(now); reportsBy.set(k, list);
  }
  const reason = ['csam', 'sexual', 'violence', 'illegal', 'spam', 'other'].includes(body.reason) ? body.reason : 'other';
  const text = String(body.text || '').slice(0, 500);
  const rec = { at: now, stream: s.id, host: s.pubkey, name: s.name, by: pubkey, ip, reason, text };
  try { appendFileSync(`${DATA}/.reports.jsonl`, JSON.stringify(rec) + '\n'); } catch {}
  s.reports = (s.reports || 0) + 1;
  if (streams.has(s.id)) save(s);
  log('report', s.id, reason, 'by', pubkey.slice(0, 12));
  alertAdmins(`Live stream reported (${reason}): ${s.name} — ${npubEncode(s.pubkey)}\n${text ? '"' + text + '"\n' : ''}`
    + `${s.reports} report${s.reports === 1 ? '' : 's'} so far. Open it to take it down: ${streamLink(s)}`);
  return json({ ok: true });
}
function remove(s, by, body) {
  if (!ADMINS.includes(by)) return json({ error: 'admins only' }, 403);
  if (s.state === 'removed') return json({ ok: true, removed: true });
  if (s.ws) { try { s.ws.close(4003, 'removed'); } catch {} }
  end(s, 'removed by ' + by.slice(0, 12));
  s.removed = { at: Date.now(), by };
  save(s);
  try { renameSync(`${DATA}/${s.id}`, `${REMOVED}/${s.id}`); } catch (e) { log('remove: move failed', e.message); }
  streams.set(s.id, { id: s.id, state: 'removed', viewers: new Map() });
  if (body && body.ban && s.pubkey) {
    banned[s.pubkey] = { at: Date.now(), by, stream: s.id };
    try { writeFileSync(BANNED_FILE, JSON.stringify(banned)); } catch {}
  }
  log('removed', s.id, body && body.ban ? '(and banned ' + s.pubkey.slice(0, 12) + ')' : '');
  return json({ ok: true, removed: true, banned: !!(body && body.ban) });
}

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