// 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, writeFileSync, existsSync, readdirSync } 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(), ADMIN = generateSecretKey();
const NEWBIE = 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, created: Date.now() - 3 * 86400_000 }
    : n === 'newbie' ? { taken: true, pubkey: getPublicKey(NEWBIE), created: Date.now() - 3600_000 } : { taken: false });
} });
writeFileSync(dir + '/config.json', JSON.stringify({ admins: [getPublicKey(ADMIN)] }));
process.env.LIVE_CONFIG = dir + '/config.json';
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);
  const nip98 = (sk, path, method = 'GET') => 'Nostr ' + Buffer.from(JSON.stringify(finalizeEvent({ kind: 27235, created_at: Math.floor(Date.now() / 1000), content: '',
    tags: [['u', 'https://v3.coinos.io' + path], ['method', method]] }, sk))).toString('base64');
  check('...and a name claimed an hour ago (a day first)', (await connect(`auth=${auth(NEWBIE)}&name=newbie`)).error === true);
  const al = await (await fetch(`${HTTP}/allowed?name=newbie`, { headers: { authorization: nip98(NEWBIE, '/live/allowed') } })).json();
  check('...and asked, the server says when it can', al.ok === false && al.reason === 'new-name' && al.until > Date.now() + 22 * 3600_000, JSON.stringify(al));
  check('a day-old name\'s owner may', (await (await fetch(`${HTTP}/allowed?name=alice`, { headers: { authorization: nip98(SK, '/live/allowed') } })).json()).ok === true);

  console.log('\n[TURN for calls]');
  const turnAuth = (sk, path = '/live/turn') => 'Nostr ' + Buffer.from(JSON.stringify(finalizeEvent({ kind: 27235, created_at: Math.floor(Date.now() / 1000), content: '',
    tags: [['u', 'https://v3.coinos.io' + path], ['method', 'GET']] }, sk))).toString('base64');
  check('ICE servers need a signed ask', (await fetch(`${HTTP}/turn`)).status === 401);
  check('...one signed for this address', (await fetch(`${HTTP}/turn`, { headers: { authorization: turnAuth(OTHER, '/live/ingest') } })).status === 401);
  const ice = await (await fetch(`${HTTP}/turn`, { headers: { authorization: turnAuth(OTHER) } })).json();
  check('...anyone signed gets them (STUN alone with no TURN key configured)', Array.isArray(ice.iceServers) && /^stun:/.test(ice.iceServers[0].urls[0]), JSON.stringify(ice));

  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));
  // WebRTC (Cloudflare's SFU) needs its key in the config: here there's none, and the app keeps to HLS
  check('with no SFU key, no WebRTC: stats say so, and watch/publish answer 501', st.rtc === null
    && (await fetch(`${HTTP}/${id}/rtc/watch`, { method: 'POST' })).status === 501 && (await fetch(`${HTTP}/${id}/rtc/publish`, { method: 'POST', body: '{}' })).status === 501);

  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[reports and taking down]');
  const post = (sk, path, body) => fetch(`http://127.0.0.1:${srv.port}${path}`, { method: 'POST', headers: { authorization: nip98(sk, path, 'POST'), 'content-type': 'application/json' }, body: JSON.stringify(body) });
  check('a viewer reports the stream', (await post(OTHER, `/live/${id}/report`, { reason: 'sexual', text: 'look' })).ok);
  const reps = readFileSync(dir + '/data/.reports.jsonl', 'utf8').trim().split('\n').map((l) => JSON.parse(l));
  check('...kept, with who and why', reps.length === 1 && reps[0].stream === id && reps[0].reason === 'sexual' && reps[0].by === getPublicKey(OTHER));
  check('...unsigned, refused', (await fetch(`${HTTP}/${id}/report`, { method: 'POST', body: '{}' })).status === 401);
  check('only an admin can take it down', (await post(OTHER, `/live/${id}/remove`, { ban: true })).status === 403);
  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 = Math.round([...done.body.matchAll(/#EXTINF:([\d.]+)/g)].reduce((n, m) => n + +m[1], 0));
  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');
  await sleep(500);

  console.log('\n[the streamer deletes a recording]');
  const cid = c.hello.id;
  check('someone else can\'t delete it', (await post(OTHER, `/live/${cid}/delete`, {})).status === 403);
  const del = await (await post(SK, `/live/${cid}/delete`, {})).json();
  check('its streamer can', del.deleted === true, JSON.stringify(del));
  check('...its playlist is gone (410), said to be deleted', (await fetch(`${HTTP}/${cid}/index.m3u8`)).status === 410 && (await (await fetch(`${HTTP}/${cid}/stats`)).json()).state === 'deleted');
  check('...its files with it (only a note of it left)', readdirSync(`${dir}/data/${cid}`).join() === 'stream.json');
  check('...and the owner can still go live', (await (await fetch(`${HTTP}/allowed?name=alice`, { headers: { authorization: nip98(SK, '/live/allowed') } })).json()).ok === true);
  const kept = await (await post(SK, `/live/${id}/delete`, {})).json();
  check('a reported one, deleted by its streamer, is gone from view but kept aside for reporting', kept.deleted === true
    && (await fetch(`${HTTP}/${id}/index.m3u8`)).status === 410 && existsSync(`${dir}/data/.removed/${id}`) && !existsSync(`${dir}/data/${id}`));
  const c0 = await connect(`auth=${auth(SK)}&name=alice`);
  const tid = c0.hello.id;
  await send(c0.ws, 0, Math.floor(webm.length * 0.3));
  await sleep(1500);
  const rm = await (await post(ADMIN, `/live/${tid}/remove`, { ban: true })).json();
  check('the admin takes a stream down, banning its sender', rm.removed === true && rm.banned === true, JSON.stringify(rm));
  check('...its playlists are gone (410), its segments too', (await fetch(`${HTTP}/${tid}/live.m3u8`)).status === 410 && (await fetch(`${HTTP}/${tid}/0-00000.ts`)).status === 410);
  check('...kept aside for reporting, not deleted', existsSync(`${dir}/data/.removed/${tid}`) && !existsSync(`${dir}/data/${tid}`));
  check('...and its sender can\'t go live again', (await connect(`auth=${auth(SK)}&name=alice`)).error === true
    && (await (await fetch(`${HTTP}/allowed?name=alice`, { headers: { authorization: nip98(SK, '/live/allowed') } })).json()).reason === 'banned');
} 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);