diff options
Diffstat (limited to 'live/server.js')
| -rw-r--r-- | live/server.js | 290 |
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); +} |