"""
No download above the ceiling is ever collected in the page.

`pipelinedDownload` with no `writable` allocates `new Array(totalChunks)` and
keeps every decrypted chunk, so whatever `_openDownloadTarget` returns `null`
for is a file held whole in RAM. That floor had no upper bound: the
`!window.showSaveFilePicker` branch returned `null` at any size, so on a browser
without the File System Access API a 20 GB film went to memory whenever the
service-worker path did not answer — which happens for ordinary reasons. The
symptom was the tab dying, with nothing in the source to lead back here.

The real `_openDownloadTarget` is lifted out of `file-utils.js` **as text** and
executed against stubbed browsers, on the rule this repo already follows for the
video player: model the environment, never the code under test. A test that
transcribed the decision tree would agree with a broken version of it by
construction.

`test_no_unguarded_memory_floor` is the one that outlives today's branches: it
reads the function and fails if a `return null` appears in it that does not go
through the guard — which is what a fourth fallback added in a hurry would look
like.
"""

import json
import re
import shutil
import subprocess
from pathlib import Path

import pytest

STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
FILE_UTILS = STATIC / "file-utils.js"

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

CEILING = 100 * 1024 * 1024
GB = 1024 * 1024 * 1024


def _lift(name, source):
    """The text of one top-level declaration, from its opening line to the
    column-0 brace that closes it. Nothing is re-typed into this test."""
    start = source.index(name)
    end = source.index("\n}\n", start) + len("\n}\n")
    return source[start:end]


@pytest.fixture(scope="module")
def target_fn():
    """The ceiling, its error and the real function — read, never re-typed."""
    src = FILE_UTILS.read_text()
    ceiling = re.search(r"^const MEMORY_CEILING = .*?;$", src, re.M)
    assert ceiling, "MEMORY_CEILING is gone from file-utils.js"
    # The test's own CEILING constant must agree with the source's, or every
    # boundary case below is asserting against a number nothing uses.
    assert str(CEILING) in ceiling.group(0).replace(" ", "") or \
        eval(ceiling.group(0).split("=")[1].strip(" ;")) == CEILING
    return "\n".join([
        ceiling.group(0),
        _lift("class TooLargeForMemoryError", src),
        _lift("async function _openDownloadTarget", src),
    ])


def _run(target_fn, tmp_path, *, size, native=False, granted=False,
         streamed=False, picker=False, mode="auto", batched=False):
    """Drive the real function against one browser shape."""
    script = tmp_path / "case.mjs"
    script.write_text(f"""
// Stubs for everything the lifted function reaches. `formatSize` and `t` only
// build the message; the assertions are about which branch was taken.
//
// stdout carries the outcome and nothing else, so the function's own logging
// goes to stderr -- where it is still shown when a case fails.
console.info = (...a) => console.error(...a);
const formatSize = (n) => `${{n}} B`;
const t = (key, vars) => key + ' ' + JSON.stringify(vars);
const platform = {{
  capabilities: {{ nativeSave: {json.dumps(native)} }},
  nativeSave: async () => ({{ name: 'n', writable: {{}} }}),
  bridgeMessage: (e) => String(e),
}};
const downloads = {{
  BLOB_LIMIT: 512 * 1024 * 1024,
  // Called by the refusal to name why the streamed path declined -- absent
  // from this stub, the error constructor threw TypeError and the test saw the
  // wrong failure entirely.
  lastStreamFailure: () => 'stubbed: no streamed target in this harness',
  getMode: () => {json.dumps(mode)},
  openTarget: async () => ({json.dumps(granted)} ? {{ name: 'g', writable: {{}} }} : null),
  openStreamedDownload: async () =>
    ({json.dumps(streamed)} ? {{ name: 's', writable: {{}} }} : null),
}};
globalThis.window = {{}};
if ({json.dumps(picker)}) {{
  window.showSaveFilePicker = async () => {{
    if ({json.dumps(picker)} === 'no-gesture') {{
      const e = new Error("Failed to execute 'showSaveFilePicker' on 'Window': "
                          + "Must be handling a user gesture to show a file picker.");
      e.name = 'SecurityError';
      throw e;
    }}
    return {{ name: 'p', createWritable: async () => ({{}}) }};
  }};
}}

{target_fn}

let outcome;
try {{
  const r = await _openDownloadTarget('film.mkv', {size}, {{}}, {size},
                                     {{ batched: {json.dumps(batched)} }});
  outcome = r === null ? {{ kind: 'memory' }}
    : r === false ? {{ kind: 'cancelled' }}
    : {{ kind: 'stream', name: r.name }};
}} catch (err) {{
  outcome = {{ kind: 'refused', name: err.name, message: err.message }};
}}
console.log(JSON.stringify(outcome));
""")
    proc = subprocess.run(["node", str(script)], capture_output=True, text=True)
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


# ── The hole this was written for ───────────────────────────────────────────

def test_a_film_is_refused_rather_than_collected_in_memory(target_fn, tmp_path):
    """Firefox/Safari shape: no picker, no granted folder, the worker did not
    answer. This returned null — 20 GB into a tab."""
    out = _run(target_fn, tmp_path, size=20 * GB)
    assert out["kind"] == "refused", out
    assert out["name"] == "TooLargeForMemoryError"


def test_the_refusal_says_how_big_and_what_the_limit_is(target_fn, tmp_path):
    out = _run(target_fn, tmp_path, size=20 * GB)
    assert "download.too_large_for_memory" in out["message"]
    assert str(20 * GB) in out["message"]
    assert str(CEILING) in out["message"]


def test_the_same_browser_in_ask_mode_is_refused_too(target_fn, tmp_path):
    """'ask' skips the service-worker block entirely, so it reached the
    unguarded branch without even trying to stream."""
    out = _run(target_fn, tmp_path, size=20 * GB, mode="ask")
    assert out["kind"] == "refused", out


# ── What must keep working ──────────────────────────────────────────────────

def test_something_small_still_uses_the_memory_floor(target_fn, tmp_path):
    out = _run(target_fn, tmp_path, size=4 * 1024 * 1024)
    assert out["kind"] == "memory", out


def test_the_boundary_is_the_ceiling_itself(target_fn, tmp_path):
    assert _run(target_fn, tmp_path, size=CEILING)["kind"] == "memory"
    assert _run(target_fn, tmp_path, size=CEILING + 1)["kind"] == "refused"


def test_a_granted_folder_streams_whatever_the_size(target_fn, tmp_path):
    out = _run(target_fn, tmp_path, size=20 * GB, granted=True)
    assert out == {"kind": "stream", "name": "g"}


def test_the_service_worker_streams_whatever_the_size(target_fn, tmp_path):
    out = _run(target_fn, tmp_path, size=20 * GB, streamed=True)
    assert out == {"kind": "stream", "name": "s"}


def test_the_desktop_app_streams_whatever_the_size(target_fn, tmp_path):
    out = _run(target_fn, tmp_path, size=20 * GB, native=True)
    assert out == {"kind": "stream", "name": "n"}


def test_a_browser_with_a_picker_is_offered_one_instead_of_being_refused(
        target_fn, tmp_path):
    """Chrome/Edge: the file is large, nothing streamed yet, but Save As does.
    A refusal here would be this fix breaking a path that was never broken."""
    out = _run(target_fn, tmp_path, size=20 * GB, picker=True)
    assert out == {"kind": "stream", "name": "p"}


# ── One dialog per gesture, not one per file ────────────────────────────────

def test_the_first_of_a_batch_still_asks_where_to_save(target_fn, tmp_path):
    """The preference is not being taken away. Someone who asked to choose the
    folder chooses it, for the download they actually clicked."""
    out = _run(target_fn, tmp_path, size=20 * GB, mode="ask", picker=True,
               streamed=True)
    assert out == {"kind": "stream", "name": "p"}


def test_the_rest_of_a_batch_stream_instead_of_asking(target_fn, tmp_path):
    """A browser grants one picker per user gesture and selecting four files is
    one gesture. Chrome showed the dialog for the second file anyway and then
    waited for a human, so the third and fourth sat behind it until they timed
    out — reported as three downloads frozen.

    There is no gesture left to spend, so nothing is lost by streaming: the file
    still lands on disk, in the browser's own download folder. Only the choice
    of folder goes, and it was not on offer.
    """
    out = _run(target_fn, tmp_path, size=20 * GB, mode="ask", picker=True,
               streamed=True, batched=True)
    assert out == {"kind": "stream", "name": "s"}


def test_a_batched_download_falls_back_to_the_dialog_rather_than_failing(
        target_fn, tmp_path):
    """When the worker does not answer, asking is better than refusing: a
    dialog that has to be answered is still a download, and the alternative
    here is losing the file. A preference must not cost a capability, and
    neither must the fix for one."""
    out = _run(target_fn, tmp_path, size=20 * GB, mode="ask", picker=True,
               streamed=False, batched=True)
    assert out == {"kind": "stream", "name": "p"}


def test_batching_never_pushes_a_large_file_into_memory(target_fn, tmp_path):
    """Firefox shape — no picker at all. Nothing about the batch flag may reach
    the memory floor above the ceiling."""
    out = _run(target_fn, tmp_path, size=20 * GB, mode="ask", picker=False,
               streamed=False, batched=True)
    assert out["kind"] == "refused", out


# ── The one that outlives today's branches ──────────────────────────────────

def test_no_unguarded_memory_floor(target_fn):
    """Every `return null` in the function goes through the guard.

    A fourth fallback appended to the chain — which is exactly how the third one
    got here — is caught by this even though no case above covers it.
    """
    body = target_fn[target_fn.index("async function _openDownloadTarget"):]
    lines = body.splitlines()
    # The guard's own `return null` is the one legitimate instance, so cut its
    # definition out before looking. Comments go too — the branch that used to
    # be the bug is now described in one, and a test that reads prose is the
    # mistake already recorded in CLAUDE.md for the packaged systemd unit.
    start = next(n for n, l in enumerate(lines) if "const _memoryFloor" in l)
    end = next(n for n in range(start, len(lines)) if lines[n].strip() == "};")
    rest = lines[:start] + lines[end + 1:]
    code = [re.sub(r"//.*$", "", l) for l in rest]
    bare = [l.strip() for l in code if re.search(r"\breturn null\b", l)]
    assert bare == [], (
        "an unguarded in-memory fallback was added to _openDownloadTarget; "
        "return _memoryFloor() instead: " + "; ".join(bare))


def test_the_guard_is_what_the_preview_uses_too(target_fn):
    """`FilePreview` decrypts a whole entry with no writable at all, so it needs
    the same ceiling — and must import it rather than keep a second number."""
    files_app = (STATIC / "files-app.js").read_text()
    assert "MEMORY_CEILING" in files_app
    assert re.search(r"entry\.size\s*>\s*MEMORY_CEILING", files_app), (
        "the preview modal must refuse an oversized entry before fetching it")
    assert not re.search(r"100\s*\*\s*1024\s*\*\s*1024", files_app), (
        "the ceiling is defined once, in file-utils.js")


def test_a_lost_gesture_streams_instead_of_failing(target_fn, tmp_path):
    """
    A browser grants one file picker per user gesture, and downloading three
    files is one gesture — so the second and third throw "Must be handling a
    user gesture". The person sees a failed transfer, with a message from Chrome
    about gestures, for having done something entirely reasonable.

    The streamed path needs no gesture, so it is the right answer rather than a
    consolation: the file lands on disk either way, and the only thing lost is
    the choice of folder, which there was no picker to make anyway.
    """
    out = _run(target_fn, tmp_path, size=20 * GB,
               picker="no-gesture", streamed=True, mode="ask")
    assert out == {"kind": "stream", "name": "s"}, out


def test_a_lost_gesture_with_nothing_to_stream_to_still_refuses(target_fn, tmp_path):
    """And the ceiling still holds underneath: no gesture and no stream is not
    a reason to put twenty gigabytes in the page."""
    out = _run(target_fn, tmp_path, size=20 * GB,
               picker="no-gesture", streamed=False, mode="ask")
    assert out["kind"] == "refused", out