"""
Reconciling N copies of a playlist when nodes are ON and OFF.

This is the one part of the playlist design that can be properly tested, so
`playlist-merge.js` is kept free of imports and the whole module is executed
here — a copy of the merge rules in a test would keep agreeing with the
original right up until one of them changed.

The stated fear is well founded for a single blob under last-writer-wins: node
A is off while an edit is made, node B is off while the next one is, and one
edit disappears with nothing to show for it. Four rules remove it, and each one
has its own case below. The tombstone rule has two, because dropping it does
not break anything that looks broken: a node rehomed after three weeks quietly
resurrects every deleted playlist, and it reads as a sync working correctly
right up until it doesn't.

See docs/playlists.md §5, §6.
"""

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 / "playlist-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)
# The export list spans several lines here; source-merge.js's fits on one.
EXPORT = re.compile(r"^export \{[^}]*\};?\s*$", re.M | re.S)


@pytest.fixture(scope="module")
def module_source():
    text = SRC.read_text()
    assert not IMPORT.search(text), (
        "playlist-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, (
        "playlist-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 _eval(tmp_path, module_source, expr):
    return _run(tmp_path, module_source,
                f"console.log(JSON.stringify({expr}));")


def _entry(name, rev, device, *, body_rev=0, count=0, deleted=False):
    return {"name": name, "rev": rev, "device": device,
            "body_rev": body_rev, "count": count, "deleted": deleted,
            # Carried for display and never read by the merge. Deliberately
            # set backwards from `rev` in several cases below: a merge that
            # reads the clock gives the wrong answer for all of them.
            "updated_at": 1_000_000 - rev}


def _manifest(**playlists):
    return {"v": 1, "playlists": playlists}


def _merge(tmp_path, src, a, b):
    return _eval(tmp_path, src,
                 f"mergeManifests({json.dumps(a)}, {json.dumps(b)})")


# ── the unit is one playlist, not the collection ─────────────────────────────

def test_two_playlists_edited_on_two_devices_both_survive(tmp_path, module_source):
    """The overwhelmingly common case for one person with two devices, and the
    whole reason the blob is a map rather than a list."""
    a = _manifest(evening=_entry("Evening", 4, "dev-a"),
                  drive=_entry("Drive", 1, "dev-a"))
    b = _manifest(evening=_entry("Evening", 1, "dev-b"),
                  drive=_entry("Drive", 7, "dev-b"))
    out = _merge(tmp_path, module_source, a, b)["playlists"]
    assert out["evening"]["rev"] == 4 and out["evening"]["device"] == "dev-a"
    assert out["drive"]["rev"] == 7 and out["drive"]["device"] == "dev-b"


def test_a_playlist_only_one_side_has_is_kept(tmp_path, module_source):
    """A node that has never seen a playlist must not delete it by omission —
    which is the same rule as the tombstone one, from the other end."""
    a = _manifest(evening=_entry("Evening", 2, "dev-a"))
    b = _manifest(drive=_entry("Drive", 1, "dev-b"))
    out = _merge(tmp_path, module_source, a, b)["playlists"]
    assert set(out) == {"evening", "drive"}


# ── revisions, never the wall clock ──────────────────────────────────────────

def test_the_higher_revision_wins_even_with_an_older_timestamp(
        tmp_path, module_source):
    """`updated_at` is set backwards from `rev` throughout this file. A merge
    that reads the clock fails here, and clocks across devices are exactly as
    trustworthy as this test assumes."""
    a = _manifest(x=_entry("new name", 9, "dev-a"))
    b = _manifest(x=_entry("old name", 2, "dev-b"))
    out = _merge(tmp_path, module_source, a, b)["playlists"]["x"]
    assert out["name"] == "new name"
    assert a["playlists"]["x"]["updated_at"] < b["playlists"]["x"]["updated_at"], (
        "the fixture must actually have the older clock on the winning side")


def test_a_revision_tie_is_broken_the_same_way_on_every_device(
        tmp_path, module_source):
    """Two devices, no conversation between them, one answer. Merged in both
    orders because a tie-break that depends on argument order is not one."""
    a = _manifest(x=_entry("from A", 5, "dev-a"))
    b = _manifest(x=_entry("from B", 5, "dev-b"))
    forward = _merge(tmp_path, module_source, a, b)["playlists"]["x"]
    backward = _merge(tmp_path, module_source, b, a)["playlists"]["x"]
    assert forward == backward
    assert forward["device"] == "dev-a"


# ── a deletion is a tombstone, never an absence ──────────────────────────────

def test_a_node_rehomed_after_three_weeks_does_not_resurrect_a_deletion(
        tmp_path, module_source):
    """The single most likely defect in the whole design.

    The client deleted "Evening" (rev 5, tombstoned). A node that went offline
    at rev 4 comes back still holding it, alive. Absence on that node must not
    win, and the tombstone must not be dropped just because the other side has
    a live entry.
    """
    local = _manifest(evening=_entry("Evening", 5, "dev-a", deleted=True))
    stale_node = _manifest(evening=_entry("Evening", 4, "dev-a"))
    out = _merge(tmp_path, module_source, local, stale_node)["playlists"]
    assert out["evening"]["deleted"] is True, "the deleted playlist came back"


def test_a_deletion_loses_to_a_later_edit(tmp_path, module_source):
    """A tombstone is not special-cased into always winning: it is one more
    revision. Someone who deletes a playlist and then, from another device that
    had not seen the deletion, renames it at a higher revision, gets the
    rename — which is last-writer-wins doing exactly what it says."""
    deleted = _manifest(x=_entry("X", 5, "dev-a", deleted=True))
    later = _manifest(x=_entry("X renamed", 6, "dev-b"))
    out = _merge(tmp_path, module_source, deleted, later)["playlists"]["x"]
    assert out["deleted"] is False and out["name"] == "X renamed"


def test_a_tombstone_is_not_shown_to_the_reader(tmp_path, module_source):
    m = _manifest(gone=_entry("Gone", 3, "dev-a", deleted=True),
                  kept=_entry("Kept", 1, "dev-a"))
    live = _eval(tmp_path, module_source, f"livePlaylists({json.dumps(m)})")
    assert [p["id"] for p in live] == ["kept"]


# ── the two counters ─────────────────────────────────────────────────────────

def test_a_rename_and_a_track_added_elsewhere_both_survive(
        tmp_path, module_source):
    """Why there are two counters rather than one.

    Device A renames the playlist; device B adds a track to it. Neither edit
    touches the other, but with a single revision counter both write n+1 and
    one of them is lost. `rev` carries the name, `body_rev` is a watermark for
    the tracks, and they move independently.
    """
    renamed = _manifest(x=_entry("New name", 8, "dev-a", body_rev=3, count=10))
    tracked = _manifest(x=_entry("Old name", 7, "dev-b", body_rev=4, count=11))
    out = _merge(tmp_path, module_source, renamed, tracked)["playlists"]["x"]
    assert out["name"] == "New name", "the rename was lost"
    assert out["body_rev"] == 4 and out["count"] == 11, "the added track was lost"


def test_the_body_watermark_never_goes_backwards(tmp_path, module_source):
    """A stale node reporting body_rev 2 against a local 9 must not lower it —
    the local copy is one of the inputs, and that is what stops a node that
    serves an old copy from being able to undo anything."""
    local = _manifest(x=_entry("X", 4, "dev-a", body_rev=9, count=40))
    stale = _manifest(x=_entry("X", 4, "dev-a", body_rev=2, count=5))
    out = _merge(tmp_path, module_source, local, stale)["playlists"]["x"]
    assert out["body_rev"] == 9 and out["count"] == 40


# ── the client is the authority ──────────────────────────────────────────────

def test_the_offline_node_dance_loses_nothing(tmp_path, module_source):
    """The scenario the fear is actually about, played out.

    Device 1 renames Evening while node B is off; device 2 renames Drive while
    node A is off. Each node holds one of the two edits. The client folds its
    own copy together with both and keeps both edits — which it can only do
    because its own copy is one of the inputs.
    """
    body = """
      const base = {v:1, playlists: {
        evening: {name:'Evening', rev:1, device:'dev-a', body_rev:1, count:2, deleted:false},
        drive:   {name:'Drive',   rev:1, device:'dev-a', body_rev:1, count:2, deleted:false}}};
      const nodeA = JSON.parse(JSON.stringify(base));
      nodeA.playlists.evening = {...nodeA.playlists.evening, name:'Soirée', rev:2, device:'dev-1'};
      const nodeB = JSON.parse(JSON.stringify(base));
      nodeB.playlists.drive = {...nodeB.playlists.drive, name:'Route', rev:2, device:'dev-2'};
      const merged = mergeAllManifests([base, nodeA, nodeB]);
      console.log(JSON.stringify({
        evening: merged.playlists.evening.name,
        drive: merged.playlists.drive.name}));
    """
    out = _run(tmp_path, module_source, body)
    assert out == {"evening": "Soirée", "drive": "Route"}


def test_folding_in_any_order_gives_the_same_answer(tmp_path, module_source):
    """Nodes answer in whatever order they answer in; the merged state cannot
    depend on that."""
    body = """
      const c = [
        {v:1, playlists:{x:{name:'a', rev:1, device:'d1', body_rev:1, count:1, deleted:false}}},
        {v:1, playlists:{x:{name:'b', rev:3, device:'d2', body_rev:5, count:9, deleted:false}}},
        {v:1, playlists:{x:{name:'c', rev:2, device:'d3', body_rev:2, count:4, deleted:false}}},
      ];
      const one = mergeAllManifests(c);
      const two = mergeAllManifests([c[2], c[0], c[1]]);
      const three = mergeAllManifests([c[1], c[2], c[0]]);
      console.log(JSON.stringify([one, two, three]));
    """
    one, two, three = _run(tmp_path, module_source, body)
    assert one == two == three
    assert one["playlists"]["x"]["name"] == "b"


# ── bodies ───────────────────────────────────────────────────────────────────

def test_a_body_merge_takes_the_higher_revision(tmp_path, module_source):
    a = {"v": 1, "id": "x", "rev": 4, "device": "d1", "tracks": [{"id": "t1"}]}
    b = {"v": 1, "id": "x", "rev": 6, "device": "d2",
         "tracks": [{"id": "t1"}, {"id": "t2"}]}
    out = _eval(tmp_path, module_source,
                f"mergeBodies({json.dumps(a)}, {json.dumps(b)})")
    assert out["rev"] == 6 and len(out["tracks"]) == 2


def test_a_body_a_node_has_never_seen_is_not_an_absence(tmp_path, module_source):
    a = {"v": 1, "id": "x", "rev": 3, "device": "d1", "tracks": [{"id": "t1"}]}
    assert _eval(tmp_path, module_source,
                 f"mergeBodies({json.dumps(a)}, null)")["rev"] == 3
    assert _eval(tmp_path, module_source,
                 f"mergeBodies(null, {json.dumps(a)})")["rev"] == 3


# ── what is stored, and what must not be ─────────────────────────────────────

def test_a_stored_track_carries_the_fields_the_player_cannot_work_without(
        tmp_path, module_source):
    """`s` and `n` are the two the first draft of the design left out. The
    player computes its chunk count from the size and reads the name both for
    the MIME type and to decide whether the file needs converting first, so
    without them a playlist entry cannot be downloaded at all."""
    entry = {"id": "abc", "name": "03 - A Track.flac", "size": 41238711,
             "path": "Artist/Album", "type": "audio", "duration": 214,
             "display_title": "A Track", "artist": "Artist", "album": "Album",
             "track_no": 3, "hash_version": 2}
    stored = _eval(tmp_path, module_source,
                   f"toStored({json.dumps(entry)}, 'g1')")
    assert stored["s"] == 41238711
    assert stored["n"] == "03 - A Track.flac"
    assert stored["g"] == "g1" and stored["hv"] == 2


def test_live_objects_are_never_stored(tmp_path, module_source):
    """The entries the views hand over carry a transport and a CryptoKey,
    attached by the Search page's merge. Storing one is at best unserialisable
    and at worst a dead connection read back a week later."""
    body = """
      const entry = { id:'abc', name:'x.flac', size:1, path:'p', type:'audio',
                      groupId:'g9', _tRef:{live:'transport'}, _gRef:{key:true},
                      _origPath:'elsewhere', _sources:[{groupId:'g9'}] };
      const stored = toStored(entry, 'g1');
      console.log(JSON.stringify({ keys: Object.keys(stored).sort(),
                                   g: stored.g, json: JSON.stringify(stored) }));
    """
    out = _run(tmp_path, module_source, body)
    assert out["keys"] == sorted(["id", "g", "hv", "n", "s", "p", "t", "a", "b", "d", "tn"])
    assert out["g"] == "g9", "the entry's own group must win over the caller's"
    for leak in ("_tRef", "_gRef", "_origPath", "_sources", "transport"):
        assert leak not in out["json"]


def test_a_stored_track_round_trips_to_what_the_player_consumes(
        tmp_path, module_source):
    body = """
      const entry = { id:'abc', name:'03 - A Track.flac', size:99, path:'A/B',
                      type:'audio', duration:214, display_title:'A Track',
                      artist:'Artist', album:'Album', track_no:3,
                      hash_version:1, groupId:'g1' };
      console.log(JSON.stringify(fromStored(toStored(entry, 'g1'))));
    """
    out = _run(tmp_path, module_source, body)
    for field in ("id", "name", "size", "path", "duration", "display_title",
                  "artist", "album", "track_no", "hash_version"):
        assert out[field] == {**{"display_title": "A Track"},
                              **{"id": "abc", "name": "03 - A Track.flac",
                                 "size": 99, "path": "A/B", "duration": 214,
                                 "artist": "Artist", "album": "Album",
                                 "track_no": 3, "hash_version": 1}}[field]
    assert out["groupId"] == "g1"


# ── repair ───────────────────────────────────────────────────────────────────

def test_a_moved_file_is_found_by_its_hash_and_its_path_rewritten(
        tmp_path, module_source):
    """Content addressing survives a move."""
    body = """
      const index = indexTracks([
        {id:'t1', type:'audio', path:'New/Place', name:'a.flac'}]);
      const { tracks, repaired } = repairTracks(
        [{id:'t1', g:'g1', p:'Old/Place', n:'a.flac'}], 'g1', index);
      console.log(JSON.stringify({tracks, repaired}));
    """
    out = _run(tmp_path, module_source, body)
    assert out["repaired"] == 1
    assert out["tracks"][0]["p"] == "New/Place"


def test_a_re_encoded_file_is_found_by_its_path_and_its_hash_rewritten(
        tmp_path, module_source):
    """A path survives a re-encode. Either field can repair the other, which is
    the whole reason both are stored."""
    body = """
      const index = indexTracks([
        {id:'t1-new', type:'audio', path:'A/B', name:'a.flac'}]);
      const { tracks, repaired } = repairTracks(
        [{id:'t1-old', g:'g1', p:'A/B', n:'a.flac'}], 'g1', index);
      console.log(JSON.stringify({tracks, repaired}));
    """
    out = _run(tmp_path, module_source, body)
    assert out["repaired"] == 1
    assert out["tracks"][0]["id"] == "t1-new"


def test_entries_from_other_groups_are_left_alone(tmp_path, module_source):
    """"Not in the index I happen to be holding" is not evidence that a track
    is gone — it is evidence about a different group."""
    body = """
      const index = indexTracks([{id:'x', type:'audio', path:'A', name:'a.flac'}]);
      const { tracks, repaired } = repairTracks(
        [{id:'t1', g:'g2', p:'A', n:'a.flac'}], 'g1', index);
      console.log(JSON.stringify({tracks, repaired}));
    """
    out = _run(tmp_path, module_source, body)
    assert out["repaired"] == 0
    assert out["tracks"][0]["id"] == "t1"


def test_an_entry_that_is_simply_gone_is_left_as_it_is(tmp_path, module_source):
    """Greyed at play time, not deleted from the playlist: a file that is
    missing today may be a disk that is unplugged today."""
    body = """
      const index = indexTracks([{id:'other', type:'audio', path:'Z', name:'z.flac'}]);
      const { tracks, repaired } = repairTracks(
        [{id:'t1', g:'g1', p:'A', n:'a.flac'}], 'g1', index);
      console.log(JSON.stringify({tracks, repaired}));
    """
    out = _run(tmp_path, module_source, body)
    assert out["repaired"] == 0 and out["tracks"][0]["id"] == "t1"


# ── favourites ───────────────────────────────────────────────────────────────

def test_favourites_is_first_whatever_it_is_called(tmp_path, module_source):
    """The menus promise it first, and the reserved id is what is stored — the
    localised name is never written, or an account that switches language
    grows a second favourites list."""
    m = _manifest(zzz=_entry("Zzz", 1, "d"),
                  aaa=_entry("Aaa", 1, "d"),
                  favorites=_entry("Favoris", 1, "d"))
    live = _eval(tmp_path, module_source, f"livePlaylists({json.dumps(m)})")
    assert [p["id"] for p in live] == ["favorites", "aaa", "zzz"]


def test_the_reserved_id_is_a_constant_not_a_literal(tmp_path, module_source):
    assert _eval(tmp_path, module_source, "FAVORITES_ID") == "favorites"
    assert _eval(tmp_path, module_source, "bodyKind('favorites')") == "playlist:favorites"
    assert _eval(tmp_path, module_source, "MANIFEST_KIND") == "playlists"