1 files changed, 102 insertions, 0 deletions
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);