"""
The desktop cast relay's stream, run for real: plain Node, no dependencies.

Three faults, each of which a receiver showed as a cast that hung or froze:

* **The header is everything before the first moof.** The page hands the relay
  its first decrypted chunk, which is a slice of the stream: the ftyp alone,
  the header exactly, or the header and the start of a fragment, depending on
  when the node read ffmpeg's output. Taken as the header, two of the three
  broke the stream, so the same film cast one try in three.
* **A receiver reading at playback speed loses nothing.** The page runs ahead
  of the television by its whole read-ahead; dropped past 8 MB, each lost
  fragment froze the picture for its few seconds.
* **A stop the page does not wait for cannot clear the next start.** The old
  stop cleared its state after the port closed, which was after the new
  start, and the receiver was handed `:null/stream.mp4?t=null`.
"""

import json
import shutil
import subprocess
from pathlib import Path

import pytest

PACKAGES = Path(__file__).resolve().parents[2]
RELAY = PACKAGES / "meshbay-client" / "src" / "cast-relay.js"

SCRIPT = r"""
const http = require('node:http');
const CastRelay = require(process.argv[2]);

const box = (type, size) => {
  const b = Buffer.alloc(size, type.charCodeAt(0));
  b.writeUInt32BE(size, 0); b.write(type, 4, 'latin1');
  return b;
};
const ftyp = box('ftyp', 28);
const moov = box('moov', 1272);
const header = Buffer.concat([ftyp, moov]);
const fragment = (n, size) => Buffer.concat([box('moof', 100 + n), box('mdat', size)]);

// Read a whole response, or only what it holds after `ms`.
const read = (url, ms) => new Promise((resolve, reject) => {
  const parts = [];
  const req = http.get(url, { agent: false }, (res) => {
    res.on('data', (c) => parts.push(c));
    res.on('end', () => resolve(Buffer.concat(parts)));
    if (ms) setTimeout(() => { req.destroy(); resolve(Buffer.concat(parts)); }, ms);
  });
  req.on('error', (e) => (ms ? null : reject(e)));
});
const sleep = (ms) => new Promise((ok) => setTimeout(ok, ms));

// The page's order: start with its first chunk, then push every chunk,
// that one included.
async function cast(relay, chunks, { early } = {}) {
  const started = await relay.start({ initSegment: chunks[0] });
  const body = early ? read(started.url) : null;
  if (early) await sleep(50);
  for (const c of chunks) relay.pushSegment(c);
  const served = early ? null : read(started.url);
  await sleep(50);
  relay.finish();
  return (await (body || served)).toString('hex');
}

(async () => {
  const out = {};
  const frags = [fragment(1, 300000), fragment(2, 800000), fragment(3, 250000)];
  const stream = Buffer.concat([header, ...frags]);
  out.expected = stream.toString('hex');

  const relay = new CastRelay();
  // The three first chunks measured on a real film.
  out.ftypAlone = await cast(relay, [ftyp, moov, ...frags]);
  out.headerExactly = await cast(relay, [header, ...frags]);
  const cut = header.length + 4000;
  out.headerAndFilm = await cast(relay, [stream.subarray(0, cut), stream.subarray(cut)]);
  // A receiver connected before the first moof waits for the whole header.
  out.earlyReceiver = await cast(relay, [ftyp, moov, ...frags], { early: true });
  await relay.stop();

  // A receiver that reads nothing while 40 MB of film is pushed, then reads.
  const started = await relay.start({ initSegment: header });
  let slowRes = null;
  const parts = [];
  const done = new Promise((resolve) => {
    http.get(started.url, { agent: false }, (res) => {
      slowRes = res; res.pause();
      res.on('data', (c) => parts.push(c));
      res.on('end', resolve);
    }).on('error', () => {});
  });
  await sleep(100);
  out.headersBeforeTheHeader = slowRes !== null;
  relay.pushSegment(header);
  const big = [];
  for (let i = 0; i < 8; i++) big.push(fragment(i, 5 * 1024 * 1024));
  for (const f of big) relay.pushSegment(f);
  relay.finish();
  slowRes.resume();
  await done;
  out.slowReceiverGotAll = Buffer.concat(parts).equals(Buffer.concat([header, ...big]));
  out.dropped = relay._dropped;

  // stop() not awaited, then start(), while a receiver still holds the old
  // stream: the new relay keeps its own token, port and header.
  const holder = read(relay.url, 300);
  await sleep(50);
  relay.stop();
  const next = await relay.start({ initSegment: header });
  await holder;
  await sleep(100);
  out.afterRestart = { url: relay.url, returned: next.url, active: relay.active };
  relay.pushSegment(header);
  relay.pushSegment(frags[0]);
  const after = read(relay.url);
  await sleep(50);
  relay.finish();
  out.afterRestartBody = (await after).toString('hex');
  out.afterRestartExpected = Buffer.concat([header, frags[0]]).toString('hex');
  await relay.stop();

  out.trimmed = CastRelay.headerOnly(stream.subarray(0, cut)).toString('hex');
  out.header = header.toString('hex');
  console.log(JSON.stringify(out));
})().catch((e) => { console.error(e); process.exit(1); });
"""


@pytest.fixture(scope="module")
def run(tmp_path_factory):
    if shutil.which("node") is None or not RELAY.exists():
        pytest.skip("node or the desktop client sources are not available")
    script = tmp_path_factory.mktemp("relay") / "stream.cjs"
    script.write_text(SCRIPT, encoding="utf-8")
    proc = subprocess.run(["node", str(script), str(RELAY)],
                          capture_output=True, text=True, encoding="utf-8", timeout=60)
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


@pytest.mark.parametrize("first_chunk", ["ftypAlone", "headerExactly", "headerAndFilm"])
def test_the_header_is_everything_before_the_first_moof(run, first_chunk):
    assert run[first_chunk] == run["expected"]


def test_a_receiver_that_connects_first_waits_for_the_whole_header(run):
    assert run["earlyReceiver"] == run["expected"]


def test_a_waiting_receiver_is_answered_at_once(run):
    assert run["headersBeforeTheHeader"] is True


def test_the_pages_first_chunk_is_trimmed_to_its_header(run):
    assert run["trimmed"] == run["header"]


def test_a_receiver_far_behind_loses_no_fragment(run):
    assert run["slowReceiverGotAll"] is True
    assert run["dropped"] == 0


def test_a_stop_not_waited_for_leaves_the_next_start_alone(run):
    r = run["afterRestart"]
    assert r["active"] is True
    assert r["url"] == r["returned"]
    assert ":null" not in r["url"] and "t=null" not in r["url"]
    assert run["afterRestartBody"] == run["afterRestartExpected"]