"""
Where downloads are written.

The module is mostly browser plumbing — a directory handle from a picker, kept
in IndexedDB — but two pieces decide behaviour and can be checked here: that
automatic is the default, and that writing into the same folder repeatedly does
not quietly replace what is already there. The second one is the whole risk of
the automatic mode: a Save As dialog warns you about a collision, and a folder
you never look at does not.
"""

import json
import shutil
import subprocess
from pathlib import Path

import pytest

STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
DOWNLOADS = STATIC / "downloads.js"

pytestmark = pytest.mark.skipif(
    shutil.which("node") is None or not DOWNLOADS.exists(),
    reason="node or the SPA sources are not available")


def _run(body, tmp_path):
    module = tmp_path / "downloads.mjs"
    module.write_text(DOWNLOADS.read_text())
    script = tmp_path / "case.mjs"
    script.write_text(
        # A localStorage good enough for a preference, so the module can be
        # imported outside a browser at all.
        "const store = new Map();\n"
        "globalThis.localStorage = {\n"
        "  getItem: k => (store.has(k) ? store.get(k) : null),\n"
        "  setItem: (k, v) => store.set(k, String(v)),\n"
        "};\n"
        f"const M = await import('{module.as_posix()}');\n"
        "const out = [];\n"
        "const say = (...a) => out.push(...a);\n"
        f"{body}\n"
        "console.log(JSON.stringify(out));\n")
    proc = subprocess.run(["node", str(script)], capture_output=True, text=True)
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


def test_saving_automatically_is_the_default(tmp_path):
    """
    Grouped downloads are the reason this setting exists: twelve files must not
    mean twelve dialogs unless someone asked for that.
    """
    assert _run("say(M.getMode());", tmp_path) == ["auto"]


def test_the_choice_is_remembered_and_nothing_else_is_accepted(tmp_path):
    result = _run("""
M.setMode('ask');   say(M.getMode());
M.setMode('auto');  say(M.getMode());
M.setMode('nonsense'); say(M.getMode());
""", tmp_path)
    assert result == ["ask", "auto", "auto"]


def test_a_second_copy_does_not_replace_the_first(tmp_path):
    result = _run("""
const taken = new Set(['clip.mp4', 'clip (2).mp4', 'notes']);
const exists = async n => taken.has(n);
say(await M.freeName('clip.mp4', exists));
say(await M.freeName('other.mp4', exists));
say(await M.freeName('notes', exists));
say(await M.freeName('archive.tar.gz', exists));
""", tmp_path)
    assert result == [
        "clip (3).mp4",      # (2) was taken as well
        "other.mp4",         # free: left alone
        "notes (2)",         # no extension to keep
        "archive.tar.gz",    # free, and the double extension is not mangled
    ]


def test_the_suffix_goes_before_the_extension(tmp_path):
    """
    "clip.mp4 (2)" would stop being a video as far as the operating system is
    concerned, which is how a download folder ends up full of files nothing opens.
    """
    result = _run("""
say(await M.freeName('clip.mp4', async () => false));
say(await M.freeName('clip.mp4', async n => n === 'clip.mp4'));
""", tmp_path)
    assert result == ["clip.mp4", "clip (2).mp4"]


def test_a_browser_without_the_api_reports_it(tmp_path):
    """`SUPPORTED` decides whether Settings offers a choice or an explanation."""
    assert _run("say(M.SUPPORTED);", tmp_path) == [False]


def test_the_open_action_reads_the_file_back(tmp_path):
    """
    "Open" is the browser being handed the bytes, not a desktop application
    being started — no web page can do the second, and none can show a file
    manager either. It is only offered for a file written into a granted folder,
    since that is the one a page can read back.
    """
    src = DOWNLOADS.read_text()
    target = src[src.index("export async function openTarget"):]
    assert "getFile()" in target and "window.open(" in target
    assert "revokeObjectURL" in target, "the blob URL must not be leaked"


# ── Streaming to disk without the File System Access API ────────────────────

SW = STATIC / "sw.js"


def test_the_worker_only_answers_its_own_urls():
    """
    It is registered at the root scope, so it sees every request the page makes.
    Anything that is not a download of ours has to fall through untouched — a
    service worker that answers more than it should is a cache bug waiting to
    happen.
    """
    src = SW.read_text()
    assert "startsWith(PREFIX)" in src
    assert "self.location.origin" in src, "cross-origin requests must fall through"
    # The API, not the word: the file explains in prose that it caches nothing.
    for api in ("caches.open", "caches.match", "cache.put"):
        assert api not in src, f"this worker must not cache anything ({api})"


def test_the_download_is_announced_as_an_attachment():
    src = SW.read_text()
    assert "Content-Disposition" in src and "attachment" in src
    assert "filename*=UTF-8''" in src, "a name with accents would be mangled"
    assert "Content-Length" in src


def test_a_length_is_only_promised_when_it_is_known(tmp_path):
    """
    An archive is assembled as it goes and is larger than the files in it.
    Announcing the sum of their sizes would truncate the download at that mark.
    """
    src = SW.read_text()
    assert "if (entry.size > 0)" in src

    # The zip-directory download started in files-app.js (group-page refactor)
    # and was lifted into file-utils.js's downloadDirectory (docs/photos.md
    # §3) so photos-app.js's own "zip this album" button calls the same
    # implementation rather than a second one.
    # Anchored on the call, not on how its result is bound: the assignment
    # became a bare `target = ...` inside a try when _openDownloadTarget gained
    # the ability to refuse an oversized download (test_memory_ceiling.py).
    # What this test is about -- the `0` -- did not move.
    app = (STATIC / "file-utils.js").read_text()
    zip_call = app[app.index("_openDownloadTarget(suggested"):]
    zip_call = zip_call[:zip_call.index(");") + 2]
    assert zip_call.rstrip().endswith(", 0);"), (
        "the zip download announces a Content-Length it will not match")


def test_backpressure_is_real(tmp_path):
    """
    The point of the service worker path is not holding the file. A stream that
    is transferred gives `writer.write()` something to wait on; posting chunks
    to a port would queue them in memory and look identical from here.
    """
    src = DOWNLOADS.read_text()
    fn = src[src.index("export async function openStreamedDownload"):]
    assert "new TransformStream()" in fn
    # The transfer list may carry more than the stream — a reply port rides
    # along now — so this asserts that `readable` is transferred, not the exact
    # shape of the list.
    transfer = fn[fn.index("worker.postMessage("):]
    transfer = transfer[transfer.index("["):transfer.index("]") + 1]
    assert "readable" in transfer, "the readable half must be transferred, not copied"
    assert "writer.write(bytes)" in fn
    assert "return null" in fn, "a browser that cannot transfer streams must say so"


def test_the_streamed_path_gives_up_rather_than_blocking_for_ever():
    """Reported 2026-08-16: a download frozen at exactly one chunk.

    The writable half applies real backpressure, which is the whole point — and
    the trap. If nothing ever reads the readable half, `writer.write()` waits
    for room that never comes, and the transfer stops dead after the stream's
    internal queue fills. Two ways that happens on a phone: the page is not yet
    *controlled* by the worker, so the iframe's request is never handed to its
    fetch handler; or the browser refuses a download started from a hidden
    iframe. Both are silent.

    So the worker confirms that it actually answered, and this path reports
    failure instead of returning a sink nobody drains.
    """
    src = DOWNLOADS.read_text()
    fn = src[src.index("export async function openStreamedDownload"):]
    assert "mbdl-serving" in fn, "the worker has to confirm it served the request"
    assert "Promise.race" in fn, "the confirmation needs a deadline"
    assert "writable.abort" in fn, "give up cleanly so the caller can fall back"

    sw = (DOWNLOADS.parent / "sw.js").read_text()
    assert "mbdl-serving" in sw, "and the worker has to send that confirmation"


def test_an_uncontrolled_page_is_not_treated_as_ready():
    """`registration.active` says a worker exists, not that it will see our fetch."""
    # Anchored on the streaming section rather than on one function: waiting
    # for control moved into `_awaitControl`/`_claimController` when the budget
    # became a parameter, and `serviceWorker()` no longer contains the words.
    # The behaviour itself is executed in test_streamed_download_reliability.py;
    # this stays as the cheap guard on the module's shape.
    src = DOWNLOADS.read_text()
    section = src[src.index("// ── Streaming to disk"):]
    assert "navigator.serviceWorker.controller" in section
    assert "controllerchange" in section, (
        "control can arrive a tick after registration; waiting beats refusing")
    assert "mbdl-claim" in section, (
        "an active-but-uncontrolled page must ask for a claim, not give up")


def test_an_apostrophe_in_a_name_does_not_lose_the_name(tmp_path):
    """
    `encodeURIComponent` leaves `'` alone, and `'` is the delimiter in RFC
    5987's `filename*=<charset>'<lang>'<value>`. One apostrophe made the header
    unparseable, and a browser that cannot parse it names the file after the
    last segment of the URL — which for this worker is a made-up id. The file
    arrived complete and 449 MB of it was called "mtsshk9w-ohqty535".

    Found by downloading three files where exactly one had an apostrophe.
    Nothing in the suite could have: the header was built correctly for every
    name anybody had tested with.

    The real function is lifted out of sw.js and run — a second copy here would
    have the same blind spot as the first.
    """
    src = SW.read_text()
    fn = src[src.index("function contentDisposition"):]
    fn = fn[:fn.index("\n}") + 2]

    script = tmp_path / "case.mjs"
    script.write_text(fn + """
const out = {};
for (const name of ["S03E02. Queen's Landing.mp4", 'Caf\\u00e9 (2019).mkv',
                    'plain.mp4', 'quote".mp4', 'star*.mp4']) {
  out[name] = contentDisposition(name);
}
console.log(JSON.stringify(out));
""")
    proc = subprocess.run(["node", str(script)], capture_output=True, text=True)
    assert proc.returncode == 0, proc.stderr
    out = json.loads(proc.stdout)

    for name, header in out.items():
        starred = header.split("filename*=UTF-8''", 1)[1]
        assert "'" not in starred, (
            f"{name!r}: an apostrophe survived into the starred value, which "
            f"is where RFC 5987 puts its delimiter — the name is lost")
        for forbidden in "()*":
            assert forbidden not in starred, (
                f"{name!r}: {forbidden!r} is not an attr-char and must be "
                f"percent-encoded")
        # The starred value has to decode back to the real name, or the escaping
        # fixed the parse and broke the result.
        from urllib.parse import unquote
        assert unquote(starred) == name

    # The ASCII fallback must not end its own quoted string.
    for name, header in out.items():
        ascii_part = header.split('filename="', 1)[1].split('";', 1)[0]
        assert '"' not in ascii_part and "\\" not in ascii_part


def test_a_sink_that_stops_consuming_fails_instead_of_hanging(tmp_path):
    """
    `writable.write()` was the one await on the download path with no bound.

    Every other one reports itself: `_sendAndWait` logs a Response timeout,
    `_fetchChunkResilient` retries and throws. A sink that stops consuming — a
    service-worker stream the browser has stopped reading — leaves `write()`
    pending for ever. It never rejects, so there is no error, no log and no
    failed transfer: the progress bar stops, the console stays empty, and the
    node is healthy throughout.

    That combination is what made it unfindable: three separate measurements
    cleared the node, the transport and the worker, because none of them was
    wrong. Bounding it does not fix whatever stopped the sink — it turns an
    unexplainable freeze into a failed transfer that names itself.
    """
    src = (STATIC / "file-utils.js").read_text()
    fn = src[src.index("async function _writeOrStall"):]
    fn = fn[:fn.index("\n}\n") + 2]

    script = tmp_path / "case.mjs"
    script.write_text("""
const t = (key, vars) => key + ' ' + JSON.stringify(vars);
const WRITE_STALL_MS = 300;      // the real value is 60s; the shape is the test
""" + fn + """
const out = {};
// A sink that never resolves — the frozen download, exactly.
const dead = { write: () => new Promise(() => {}) };
const t0 = Date.now();
try {
  await _writeOrStall(dead, new Uint8Array(4), 41);
  out.threw = null;
} catch (e) { out.threw = e.message; }
out.ms = Date.now() - t0;

// And a working sink is not slowed down or wrapped in anything.
const live = { write: async () => {} };
await _writeOrStall(live, new Uint8Array(4), 0);
out.liveOk = true;
console.log(JSON.stringify(out));
""")
    proc = subprocess.run(["node", str(script)], capture_output=True, text=True)
    assert proc.returncode == 0, proc.stderr
    out = json.loads(proc.stdout)
    assert out["threw"], "a dead sink hung for ever instead of failing"
    assert "group.download_write_stalled" in out["threw"], (
        "the failure must name itself in the transfers panel")
    assert "41" in out["threw"], "and say which chunk it stopped at"
    assert out["ms"] < 3000
    assert out["liveOk"] is True


def test_the_worker_is_kept_alive_while_it_streams():
    """
    A service worker with no event for ~30 s is terminated — Firefox does it,
    and `respondWith(new Response(stream))` does not extend its life while the
    response is still being written. The reader vanishes mid-file, the page's
    next `write()` never resolves and never rejects: the progress bar stops,
    the console stays empty, and the node looks healthy throughout.

    Measured in real Firefox 154 on 2026-09-08, writing 1 MB every 2 s:
    without the ping it stalled at 17 MB after 59 s; with it, 40 MB in 80 s,
    complete. The first version of that probe wrote 450 MB in two seconds and
    passed — fast enough to hide the bug entirely, which is why the pacing
    matters and is written down here.

    Source-reading, because the behaviour needs a browser and a minute of wall
    clock. What it protects is that the ping exists at all, is cleared on both
    exits, and is answered by the worker.
    """
    dl = DOWNLOADS.read_text()
    sw = SW.read_text()

    assert "SW_KEEPALIVE_MS" in dl and "mbdl-ping" in dl, (
        "nothing keeps the worker alive; downloads longer than ~30 s will "
        "stall on Firefox with no error anywhere")
    fn = dl[dl.index("async function _attemptStreamedDownload"):]
    interval = fn[fn.index("setInterval"):]
    assert "mbdl-ping" in interval[:200]

    # Cleared on both ways out, or a finished download leaves a timer pinging a
    # worker for the life of the page.
    for exit_path in ("close:", "abort:"):
        block = fn[fn.index(exit_path):]
        assert "clearInterval(keepAlive)" in block[:220], (
            f"the keep-alive is not cleared in {exit_path} — it outlives the "
            f"download")

    # And the worker has to answer it: a message it ignores still counts as an
    # event, but the reply is what tells the page it is talking to the worker
    # that holds its stream.
    assert "mbdl-ping" in sw and "mbdl-pong" in sw