diff options
| -rw-r--r-- | README.md | 1 | ||||
| -rw-r--r-- | compose.yml | 21 | ||||
| -rwxr-xr-x | deploy/post-receive | 2 | ||||
| -rw-r--r-- | live/server.js | 290 | ||||
| -rw-r--r-- | tools/live-ingest-test.js | 102 |
5 files changed, 415 insertions, 1 deletions
@@ -13,6 +13,7 @@ and store its encrypted sync data. | `nwcpush/` | **NWC push notifier.** Watches relays for Nostr Wallet Connect requests addressed to registered wallet-service keys and wakes the device with a Web Push, so a closed wallet can answer. | (internal, behind the app) | | `blossom/` | **Blossom server for sync envelopes.** Stores only the wallet's encrypted cross-device state blobs, which can outgrow a relay's event size. Refuses anything else. | `blossom.coinos.io` | | `feed/` | **Public feed snapshot** (`/api/feed`) and server-rendered note and profile pages. | `v3.coinos.io` | +| `live/` | **Live streams** sent from the app: the browser's recording over a WebSocket, out as HLS (a live window and the recording), for NIP-53 streams. Senders own a coinos name. | `v3.coinos.io/live/` | | `vmedia/` | **Media helper** for the feed: video posters, H.264 fallbacks and resized pictures. | `v3.coinos.io/vmedia/` | | `bridge/` | Ark to Lightning swap bridge. **Retired**: the app uses the Ark server's Lightning. `bridge/ln.js` (a minimal CLN RPC client) is still used by `names/`. | — | diff --git a/compose.yml b/compose.yml index df0d2d1..d48ab2f 100644 --- a/compose.yml +++ b/compose.yml @@ -5,6 +5,7 @@ # $STATE/names/config.json $STATE/names/data/ (names.db) # $STATE/nwcpush/config.json $STATE/nwcpush/data/ # $STATE/blossom/data/ +# $STATE/live/data/ (streams and their recordings) services: names: image: oven/bun:1.3.14 @@ -57,6 +58,26 @@ services: networks: [net] restart: unless-stopped + # live streams sent from the app (live/server.js). ffmpeg comes with the + # vmedia image (vmedia/Dockerfile: `docker build -t vmedia vmedia/` on the host). + live: + image: vmedia + container_name: live + working_dir: /app + command: bun live/server.js + environment: + LIVE_DATA: /data + LIVE_PORT: "8795" + LIVE_BASE: https://v3.coinos.io/live + NAMES_URL: http://names:8798 + volumes: + - .:/app:ro + - ${STATE}/live/data:/data + cpus: 4 + mem_limit: 2g + networks: [net] + restart: unless-stopped + networks: net: external: true diff --git a/deploy/post-receive b/deploy/post-receive index b6b0b29..8334b4b 100755 --- a/deploy/post-receive +++ b/deploy/post-receive @@ -11,7 +11,7 @@ LOG=/home/adam/coinosv3-server-deploy.log MIRROR=/home/adam/git/coinosv3-server.git [email protected]:npub1w27mc4aa6m0ufe3xs5z3m6qyr52gc0rglept7vqlwx4xeaf72tasnqz70r/coinosv3-server.git BARE=$(pwd) -SERVICES="names nwcpush blossom" +SERVICES="names nwcpush blossom live" while read -r oldrev newrev ref; do [ "$ref" = refs/heads/prod ] || continue 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); +} diff --git a/tools/live-ingest-test.js b/tools/live-ingest-test.js new file mode 100644 index 0000000..2d57d74 --- /dev/null +++ b/tools/live-ingest-test.js @@ -0,0 +1,102 @@ +// live/: a coinos name's owner sends a recording over the WebSocket and it +// comes out as HLS — a live window and the whole of it — with the viewers +// counted; a sender that drops comes back to the same stream (a +// discontinuity between); the end closes the playlist; nobody else may send. +// Run: bun tools/live-ingest-test.js (needs ffmpeg) +import { generateSecretKey, getPublicKey, finalizeEvent } from 'nostr-tools/pure'; +import { mkdtempSync, rmSync, readFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); +let ok = true; +const check = (n, c, d = '') => { console.log(` ${c ? '✓' : '✗'} ${n}${d ? ' — ' + d : ''}`); if (!c) ok = false; }; + +const dir = mkdtempSync(tmpdir() + '/live-ingest-'); +// what a browser's MediaRecorder sends: WebM, VP8 + Opus, a portrait camera +Bun.spawnSync(['ffmpeg', '-loglevel', 'error', '-y', '-f', 'lavfi', '-i', 'testsrc=size=720x1280:rate=30', '-f', 'lavfi', '-i', 'sine=frequency=440', + '-t', '12', '-c:v', 'libvpx', '-deadline', 'realtime', '-b:v', '1M', '-c:a', 'libopus', '-f', 'webm', dir + '/cam.webm']); +const webm = readFileSync(dir + '/cam.webm'); + +const SK = generateSecretKey(), PK = getPublicKey(SK), OTHER = generateSecretKey(); +const names = Bun.serve({ port: 0, hostname: '127.0.0.1', fetch: (req) => { + const n = new URL(req.url).pathname.split('/').pop(); + return Response.json(n === 'alice' ? { taken: true, pubkey: PK, manager: null } : { taken: false }); +} }); +process.env.LIVE_DATA = dir + '/data'; +process.env.NAMES_URL = names.url.origin; +process.env.LIVE_BASE = 'http://x/live'; +const live = await import('../live/server.js'); +const srv = Bun.serve({ port: 0, hostname: '127.0.0.1', fetch: live.handle, websocket: live.websocket }); +const HTTP = `http://127.0.0.1:${srv.port}/live`, WS = `ws://127.0.0.1:${srv.port}/live/ingest`; +const auth = (sk, method = 'GET') => Buffer.from(JSON.stringify(finalizeEvent({ kind: 27235, created_at: Math.floor(Date.now() / 1000), content: '', + tags: [['u', 'https://v3.coinos.io/live/ingest'], ['method', method]] }, sk))).toString('base64'); +// open a sender; resolves with its first message (or the refusal) +const connect = (q) => new Promise((resolve) => { + const ws = new WebSocket(`${WS}?${q}`); + ws.binaryType = 'arraybuffer'; + ws.onmessage = (e) => resolve({ ws, hello: JSON.parse(e.data) }); + ws.onerror = () => resolve({ ws, error: true }); + ws.onclose = (e) => resolve({ ws, closed: e.code }); +}); +// the recording, in half-second chunks at the pace it was made +async function send(ws, from = 0, to = webm.length) { + const step = Math.ceil(webm.length / 24); + for (let i = from; i < to; i += step) { ws.send(webm.subarray(i, Math.min(to, i + step))); await sleep(500); } +} +const text = async (u) => { const r = await fetch(u); return { status: r.status, body: await r.text() }; }; +const segs = (m3u8) => m3u8.split('\n').filter((l) => /\.ts$/.test(l)); + +try { + console.log('[who may send]'); + check('a stranger to the name is refused', (await connect(`auth=${auth(OTHER)}&name=alice`)).error === true); + check('...as is a name nobody owns', (await connect(`auth=${auth(SK)}&name=bob`)).error === true); + check('...and a forged auth', (await connect(`auth=${auth(SK).slice(0, -8)}AAAAAAAA&name=alice`)).error === true); + + console.log('\n[sending]'); + const a = await connect(`auth=${auth(SK)}&name=alice`); + check('the owner is let in and told where the stream is', !!a.hello?.id && /\/live\/[0-9a-f]{16}\/live\.m3u8$/.test(a.hello.live) && /index\.m3u8$/.test(a.hello.recording), JSON.stringify(a.hello)); + const id = a.hello.id; + check('...and can\'t start a second one at once', (await connect(`auth=${auth(SK)}&name=alice`)).error === true); + await send(a.ws, 0, Math.floor(webm.length * 0.6)); + await sleep(2500); + const l1 = await text(`${HTTP}/${id}/live.m3u8`); + check('the live playlist has segments, and no end', l1.status === 200 && segs(l1.body).length >= 2 && !/ENDLIST/.test(l1.body), segs(l1.body).length + ' segments'); + const seg = await fetch(`${HTTP}/${id}/${segs(l1.body)[0]}`); + const b = new Uint8Array(await seg.arrayBuffer()); + check('a segment is MPEG-TS', seg.status === 200 && b[0] === 0x47, seg.status + ' ' + b.length + ' bytes'); + const probe = JSON.parse(Bun.spawnSync(['ffprobe', '-v', 'error', '-show_streams', '-of', 'json', `${dir}/data/${id}/${segs(l1.body)[0]}`]).stdout.toString()).streams; + const v = probe.find((x) => x.codec_type === 'video'), au = probe.find((x) => x.codec_type === 'audio'); + check('...H.264 at most 1280 on its long side, AAC sound', v?.codec_name === 'h264' && Math.max(v.width, v.height) <= 1280 && v.height > v.width && au?.codec_name === 'aac', v && `${v.codec_name} ${v.width}x${v.height} ${au?.codec_name}`); + const st = await (await fetch(`${HTTP}/${id}/stats`)).json(); + check('the watcher who fetched the playlist is counted', st.viewers === 1 && st.state === 'live', JSON.stringify(st)); + + console.log('\n[a sender that drops comes back]'); + a.ws.close(); + await sleep(500); + check('the stream waits for it', (await (await fetch(`${HTTP}/${id}/stats`)).json()).state === 'waiting'); + const b2 = await connect(`auth=${auth(SK)}&name=alice&resume=${id}`); + check('...it carries on the same stream', b2.hello?.id === id && b2.hello.resumed === true, JSON.stringify(b2.hello)); + await send(b2.ws, 0, webm.length); // a new recording starts from its own header + await sleep(2500); + const full = await text(`${HTTP}/${id}/index.m3u8`); + check('the whole playlist joins the two with a discontinuity', /#EXT-X-DISCONTINUITY\n/.test(full.body) && segs(full.body).some((x) => x.startsWith('0-')) && segs(full.body).some((x) => x.startsWith('1-')), segs(full.body).length + ' segments'); + check('...as an EVENT playlist while live', /PLAYLIST-TYPE:EVENT/.test(full.body) && !/ENDLIST/.test(full.body)); + + console.log('\n[the end]'); + b2.ws.send('end'); + await sleep(1500); + const done = await text(`${HTTP}/${id}/index.m3u8`); + check('ending closes the playlist: the recording', /ENDLIST/.test(done.body) && /PLAYLIST-TYPE:VOD/.test(done.body)); + check('...the live one ends too', /ENDLIST/.test((await text(`${HTTP}/${id}/live.m3u8`)).body)); + const total = segs(done.body).length * 2; + check('...about the length sent (12 s + 7 s)', total >= 14 && total <= 24, total + ' s'); + check('a stream that never was: 404', (await fetch(`${HTTP}/0123456789abcdef/live.m3u8`)).status === 404); + const c = await connect(`auth=${auth(SK)}&name=alice`); + check('once ended, the owner can go live again', !!c.hello?.id && c.hello.id !== id); + c.ws.send('end'); +} catch (e) { console.error(e); ok = false; } finally { + srv.stop(true); names.stop(true); + await sleep(300); + rmSync(dir, { recursive: true, force: true }); +} +console.log(ok ? '\n✅ live: sent in, watched as HLS' : '\n❌ failed'); +process.exit(ok ? 0 : 1); |