"""
One file, one entry: merging the same content announced by several groups.

Reported live: a node hosting two groups that were given the same video
directory — the point of two groups being that different people are invited to
different libraries, and one library may be in several of them. The Search
view, which concatenates every group's index into one list, then showed every
film twice, every episode twice inside a show, every track twice inside an
album. Inside one group this cannot happen: `GroupIndex` is keyed by blake3, so
the same bytes at two paths are already one entry. The duplication is the
Search page's own.

`source-merge.js` merges on the content hash and resolves one source per
*unit* — a film, a whole show, a whole album — rather than per file, so a
season's episodes do not scatter across two nodes. This holds the rules that
decide which source that is, and what survives the merge.

The whole module is executed here rather than a regex-extracted function of it:
it has no imports precisely so that it can be, and a copy of the picking rule
in a test would keep agreeing with the original right up until one of them
changed.

See docs/refactoring-search.md.
"""

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

import pytest

STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
SRC = STATIC / "source-merge.js"

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

IMPORT = re.compile(r"^\s*import\b", re.M)
EXPORT = re.compile(r"^export \{[^}]*\};?\s*$", re.M)


@pytest.fixture(scope="module")
def module_source():
    text = SRC.read_text()
    assert not IMPORT.search(text), (
        "source-merge.js has gained an import. It is executed standalone here, "
        "and the merge is untested from the moment it cannot be — keep the "
        "module free of imports, or this test needs a bundler")
    stripped, n = EXPORT.subn("", text)
    assert n == 1, (
        "source-merge.js no longer ends in a single export statement — the "
        "test can no longer strip it to run the module")
    return stripped


def _run(tmp_path, module_source, body):
    script = tmp_path / "case.js"
    script.write_text(f"{module_source}\n{body}\n")
    out = subprocess.run(
        ["node", str(script)], capture_output=True, text=True, timeout=30)
    assert out.returncode == 0, out.stderr
    return json.loads(out.stdout)


def _entry(file_id, group, **kw):
    e = {
        "id": file_id,
        "name": f"{file_id}.mkv",
        "type": "video",
        "groupId": group,
        "groupName": group.upper(),
        "groupOwner": "someone",
        "_tRef": f"t:{group}",
    }
    e.update(kw)
    return e


def _merge(tmp_path, module_source, units, salt="u1", local=(), down=()):
    body = f"""
      const units = {json.dumps(units)};
      const local = new Set({json.dumps(list(local))});
      const down = new Set({json.dumps(list(down))});
      const merged = mergeUnitEntries(units, {{
        salt: {json.dumps(salt)},
        isLocal: (g) => local.has(g),
        isDown: (g) => down.has(g),
      }});
      console.log(JSON.stringify(merged));
    """
    return _run(tmp_path, module_source, body)


# ── the reported case ────────────────────────────────────────────────────────

def test_one_file_two_groups_becomes_one_entry(tmp_path, module_source):
    """The whole bug: a film shared by two groups was two poster cards."""
    units = [
        {"key": "movie:aaa", "entries": [_entry("aaa", "demo35")]},
        {"key": "movie:aaa", "entries": [_entry("aaa", "media")]},
    ]
    merged = _merge(tmp_path, module_source, units)
    assert len(merged) == 1
    assert merged[0]["id"] == "aaa"
    assert [s["groupId"] for s in merged[0]["_sources"]] == ["demo35", "media"]


def test_a_show_is_one_list_of_episodes(tmp_path, module_source):
    """Two seasons' worth of episodes announced twice, one list out."""
    eps = [f"e{n}" for n in range(6)]
    units = [{
        "key": "show:Some Saga",
        "entries": [_entry(i, g) for g in ("demo35", "media") for i in eps],
    }]
    merged = _merge(tmp_path, module_source, units)
    assert sorted(e["id"] for e in merged) == sorted(eps)


def test_a_lone_source_is_left_alone(tmp_path, module_source):
    """The overwhelmingly common case must come through untouched."""
    units = [{"key": "movie:aaa", "entries": [_entry("aaa", "solo")]}]
    merged = _merge(tmp_path, module_source, units)
    assert len(merged) == 1
    assert merged[0]["groupId"] == "solo"
    assert merged[0]["_tRef"] == "t:solo"
    assert len(merged[0]["_sources"]) == 1


# ── which source ─────────────────────────────────────────────────────────────

def test_the_local_node_always_wins(tmp_path, module_source):
    """
    Whatever the salt, a group on the reader's own node is the source. Every
    salt is tried rather than one: a rule that only holds for the salt the test
    happened to pick is not the rule.
    """
    units = [{"key": "movie:aaa",
              "entries": [_entry("aaa", "remote"), _entry("aaa", "mine")]}]
    for salt in [f"user-{n}" for n in range(20)]:
        merged = _merge(tmp_path, module_source, units, salt=salt, local=["mine"])
        assert merged[0]["groupId"] == "mine", f"salt {salt} escaped the local rule"


def test_the_pick_is_stable_and_spread(tmp_path, module_source):
    """
    Stable for one reader — a source that changed between renders would tear
    down the connection under a film that is playing — and different across
    readers, which is what "random" was for.
    """
    groups = [f"g{n}" for n in range(8)]
    units = [{"key": "movie:aaa", "entries": [_entry("aaa", g) for g in groups]}]

    first = _merge(tmp_path, module_source, units, salt="u1")[0]["groupId"]
    again = _merge(tmp_path, module_source, units, salt="u1")[0]["groupId"]
    assert first == again

    picks = {_merge(tmp_path, module_source, units, salt=f"u{n}")[0]["groupId"]
             for n in range(40)}
    assert len(picks) > 1, "every reader lands on the same source — nothing is spread"


def test_one_reader_is_spread_across_sources(tmp_path, module_source):
    """
    The spread is across units as well as across readers. Hashing the salt
    alone would be just as stable and just as fair between readers, and would
    point one reader's entire library at one node.
    """
    groups = ("aa", "bb", "cc", "dd")
    units = [{"key": f"movie:{n}", "entries": [_entry(f"f{n}", g) for g in groups]}
             for n in range(40)]
    merged = _merge(tmp_path, module_source, units, salt="u1")
    assert len({e["groupId"] for e in merged}) > 1


def test_source_order_does_not_decide(tmp_path, module_source):
    """
    The pick sorts by group id first. Without that it would follow whichever
    group the index fetch happened to answer first, which is a race.
    """
    groups = [_entry("aaa", g) for g in ("zulu", "alpha", "mike")]
    fwd = _merge(tmp_path, module_source, [{"key": "m", "entries": groups}])
    rev = _merge(tmp_path, module_source, [{"key": "m", "entries": groups[::-1]}])
    assert fwd[0]["groupId"] == rev[0]["groupId"]


def test_every_file_of_a_unit_shares_its_source(tmp_path, module_source):
    """
    The reason the source is picked per unit at all: a season split across two
    nodes would open two connections and two metadata lookups for one show.

    The episodes deliberately do NOT all have the same sources. With every
    episode in every group, picking per file and picking per unit give the same
    answer — the same key over the same set — and a test built that way passes
    against a per-file implementation, which is how the first version of this
    one got written. `e3` is missing from `cc`, so a per-file pick resolves it
    over `{aa, bb}` while its siblings resolve over `{aa, bb, cc}`, and the two
    rules come apart.
    """
    full = ("aa", "bb", "cc")
    units = [{
        "key": "show:Some Saga",
        "entries": ([_entry(i, g) for g in full for i in ("e1", "e2")]
                    + [_entry("e3", g) for g in ("aa", "bb")]),
    }]
    exercised = 0
    for salt in [f"u{n}" for n in range(40)]:
        by_id = {e["id"]: e for e in
                 _merge(tmp_path, module_source, units, salt=salt)}
        assert by_id["e1"]["groupId"] == by_id["e2"]["groupId"]
        chosen = by_id["e1"]["groupId"]
        if chosen == "cc":
            continue        # e3 genuinely does not have it; it falls back
        exercised += 1
        assert by_id["e3"]["groupId"] == chosen, (
            f"salt {salt}: e3 has {chosen} and went to "
            f"{by_id['e3']['groupId']} instead — the source is being picked "
            "per file, not per unit")
    assert exercised > 5, (
        "no salt put the unit on a source e3 also has — this test proved "
        "nothing about per-unit picking")


def test_a_file_the_unit_source_lacks_falls_back_together(tmp_path, module_source):
    """
    An episode only one group holds still plays — and when several episodes are
    missing from the chosen source they all land on the same fallback rather
    than one node apiece.
    """
    shared = [_entry(i, g) for g in ("aa", "bb") for i in ("e1", "e2")]
    # e8/e9 exist in bb and cc only, whichever of the three the unit picked
    extra = [_entry(i, g) for g in ("bb", "cc") for i in ("e8", "e9")]
    merged = _merge(tmp_path, module_source,
                    [{"key": "show:Some Saga", "entries": shared + extra}])
    by_id = {e["id"]: e for e in merged}
    assert sorted(by_id) == ["e1", "e2", "e8", "e9"]
    assert by_id["e8"]["groupId"] == by_id["e9"]["groupId"]
    assert by_id["e8"]["groupId"] in ("bb", "cc")


# ── failover ─────────────────────────────────────────────────────────────────

def test_a_down_source_is_skipped(tmp_path, module_source):
    units = [{"key": "movie:aaa",
              "entries": [_entry("aaa", g) for g in ("aa", "bb", "cc")]}]
    for salt in [f"u{n}" for n in range(20)]:
        merged = _merge(tmp_path, module_source, units, salt=salt, down=["aa", "bb"])
        assert merged[0]["groupId"] == "cc"


def test_a_down_local_source_yields_to_a_live_remote(tmp_path, module_source):
    """
    The local preference is a preference, not an override. A local node that is
    not answering must not hold a film hostage while another group serves it.
    """
    units = [{"key": "movie:aaa",
              "entries": [_entry("aaa", "mine"), _entry("aaa", "remote")]}]
    merged = _merge(tmp_path, module_source, units, local=["mine"], down=["mine"])
    assert merged[0]["groupId"] == "remote"


def test_everything_down_still_yields_an_entry(tmp_path, module_source):
    """A tile that fails to load beats a film that vanished from the grid."""
    units = [{"key": "movie:aaa",
              "entries": [_entry("aaa", g) for g in ("aa", "bb")]}]
    merged = _merge(tmp_path, module_source, units, down=["aa", "bb"])
    assert len(merged) == 1
    assert merged[0]["groupId"] in ("aa", "bb")
    assert len(merged[0]["_sources"]) == 2, "a down source is still a source"


# ── what survives the merge ──────────────────────────────────────────────────

def test_fields_come_from_the_chosen_source_only(tmp_path, module_source):
    """
    No field is back-filled from another source. A thumb_hash only one node
    computed is only fetchable over *that* node's connection, so borrowing it
    would produce a poster request the chosen transport cannot answer.
    """
    units = [{"key": "movie:aaa", "entries": [
        _entry("aaa", "mine", thumb_hash=None, display_title="a"),
        _entry("aaa", "remote", thumb_hash="deadbeef", display_title="b"),
    ]}]
    merged = _merge(tmp_path, module_source, units, local=["mine"])
    assert merged[0]["groupId"] == "mine"
    assert merged[0]["thumb_hash"] is None
    assert merged[0]["display_title"] == "a"
    assert merged[0]["_tRef"] == "t:mine"


def test_units_are_kept_apart(tmp_path, module_source):
    """
    Merging is scoped to a unit. Two albums that happen to hold the same photo
    are two albums, and it belongs in both.
    """
    units = [
        {"key": "album:trip", "entries": [_entry("p1", "aa"), _entry("p1", "bb")]},
        {"key": "album:party", "entries": [_entry("p1", "aa")]},
    ]
    merged = _merge(tmp_path, module_source, units)
    assert len(merged) == 2


def test_one_group_announcing_a_file_twice_counts_once(tmp_path, module_source):
    """
    Cannot happen through a well-behaved index (GroupIndex is keyed by id), so
    a duplicate here is a node saying something odd. It must not inflate the
    source count the badge shows.
    """
    units = [{"key": "movie:aaa",
              "entries": [_entry("aaa", "aa"), _entry("aaa", "aa")]}]
    merged = _merge(tmp_path, module_source, units)
    assert len(merged) == 1
    assert len(merged[0]["_sources"]) == 1


def test_empty_input(tmp_path, module_source):
    assert _merge(tmp_path, module_source, []) == []


# ── the badge ────────────────────────────────────────────────────────────────

def test_source_label(tmp_path, module_source):
    body = """
      const one = { groupId: 'g1', groupName: 'G1',
                    _sources: [{ groupId: 'g1', groupName: 'G1' }] };
      const many = { groupId: 'g1', groupName: 'G1',
                     _sources: [{ groupId: 'g1' }, { groupId: 'g2' }] };
      const unmerged = { groupId: 'g9', groupName: 'G9' };
      const bare = { };
      console.log(JSON.stringify(
        [one, many, unmerged, bare].map((e) => sourceLabel(e))));
    """
    one, many, unmerged, bare = _run(tmp_path, module_source, body)
    assert one == {"count": 1, "name": "G1", "groupId": "g1"}
    # Which group was picked is deliberately not shown once there are several.
    assert many == {"count": 2, "name": "", "groupId": None}
    # An entry that never went through the merge is its own single source.
    assert unmerged == {"count": 1, "name": "G9", "groupId": "g9"}
    # The single-group Group page, where entries carry no group at all.
    assert bare == {"count": 0, "name": "", "groupId": None}


def test_source_label_counts_the_unit_not_the_cover(tmp_path, module_source):
    """
    A card stands for a unit; the entry it is drawn from is one file. A show's
    poster entry is chosen for its *thumbnail*, so a show in two groups whose
    cover episode sits in only one of them would have claimed a single source.
    """
    body = """
      const cover = { id: 'e1', groupId: 'aa', groupName: 'AA',
                      _sources: [{ groupId: 'aa', groupName: 'AA' }] };
      const rest  = { id: 'e2', groupId: 'aa', groupName: 'AA',
                      _sources: [{ groupId: 'aa' }, { groupId: 'bb' }] };
      console.log(JSON.stringify(
        [sourceLabel(cover), sourceLabel([cover, rest])]));
    """
    alone, unit = _run(tmp_path, module_source, body)
    assert alone["count"] == 1
    assert unit["count"] == 2