path: root/packages/meshbay-node/tests/test_root_ops_reach_the_live_set.py

"""
Adding or removing a root has to reach the running node, not only node.toml.

Two front doors do this — the loopback API and a signed MNP op — and `ops.py`
exists so they behave identically. They did not. The loopback path fired the
daemon's `reload_fn`, which re-reads node.toml and builds a fresh `RootSet`;
the MNP path instead re-pointed the indexer at `groups_ctx[gid]["roots"]`, the
very object the op had just been asked about. `DirectoryIndexer.retarget`
decides what to scan by diffing the names it holds against the ones it is
given, so a set compared against itself scans nothing and drops nothing.

A directory added from a browser therefore reached node.toml and was invisible
everywhere else until a restart — and adding it again was refused as colliding
with itself, which is the only reason anyone found out. One removed would have
kept serving its files.

**The obvious repair is wrong in the other direction**, and was committed once
before this file said so: making the op edit the live set in place puts the new
root on *both* sides of retarget's comparison. The table would show it and it
would stay permanently empty. So the ops leave that object alone, the MNP path
reloads like the loopback one always did, and the tests below check the files —
`describe()` agreeing proves nothing about whether anything was scanned.
"""

from dataclasses import asdict
from pathlib import Path
from types import SimpleNamespace

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from meshbay_node import ops
from meshbay_node.config import GroupConfig, NodeConfig, RootSpec
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.indexer.indexer import DirectoryIndexer
from meshbay_node.roots import RootSet
from meshbay_node.roster import Roster

pytestmark = pytest.mark.asyncio

GROUP = "g" * 32


async def _state(tmp_path: Path) -> tuple[dict, Roster]:
    """A node hosting one group with two roots, as node.toml and as live state."""
    for name in ("one", "two"):
        (tmp_path / name).mkdir()

    cfg = GroupConfig(id=GROUP, name="plop", roots=[
        RootSpec(path=str(tmp_path / "one"), name="one"),
        RootSpec(path=str(tmp_path / "two"), name="two"),
    ])
    node_cfg = NodeConfig.__new__(NodeConfig)
    node_cfg.groups = [cfg]

    conf = tmp_path / "node.toml"
    conf.write_text(
        f'[[groups]]\nid = "{GROUP}"\nname = "plop"\n\n'
        f'  [[groups.roots]]\n  path = "{(tmp_path / "one").as_posix()}"\n'
        f'  name = "one"\n\n'
        f'  [[groups.roots]]\n  path = "{(tmp_path / "two").as_posix()}"\n'
        f'  name = "two"\n')

    live = RootSet.build([asdict(r) for r in cfg.roots])
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()
    state = {
        "config": node_cfg,
        "config_path": str(conf),
        "groups_ctx": {GROUP: {"index": index, "roots": live}},
        "roster": roster,
        "node_user_id": "operator",
    }
    return state, roster


def _live(state) -> RootSet:
    return state["groups_ctx"][GROUP]["roots"]


async def _indexer(state) -> DirectoryIndexer:
    idx = DirectoryIndexer(roots=_live(state), group_id=GROUP,
                           sk_node=Ed25519PrivateKey.generate(), gek=None)
    await idx.initial_scan()
    return idx


def _rebuilt(state) -> RootSet:
    """What a reload produces: a fresh set from the config the op just wrote."""
    return RootSet.build([asdict(r) for r in state["config"].groups[0].roots])


# ── What the op writes ───────────────────────────────────────────────────────

async def test_adding_a_root_reaches_node_toml_and_the_ack(tmp_path):
    state, roster = await _state(tmp_path)
    (tmp_path / "uploads").mkdir()
    try:
        result = await ops.add_root(state, GROUP, str(tmp_path / "uploads"))

        assert [r["name"] for r in result["roots"]] == ["one", "two", "uploads"]
        assert [r.name for r in state["config"].groups[0].roots] == [
            "one", "two", "uploads"]
        assert "uploads" in Path(state["config_path"]).read_text()
    finally:
        await roster.close()


async def test_removing_a_root_reaches_node_toml_and_the_ack(tmp_path):
    state, roster = await _state(tmp_path)
    try:
        result = await ops.remove_root(state, GROUP, "two")
        assert [r["name"] for r in result["roots"]] == ["one"]
        assert Path(state["config_path"]).read_text().count(
            "[[groups.roots]]") == 1
    finally:
        await roster.close()


async def test_the_op_does_not_edit_the_live_set_in_place(tmp_path):
    """
    The property that made the original bug, and then made the first repair for
    it wrong in the other direction.

    `retarget` diffs the names it holds against the ones it is handed. Editing
    that same object and passing it back puts a new root on both sides of the
    comparison: nothing is scanned, and the directory shows in the table
    permanently empty. `_reload_config_inner` diffs the same way and would
    likewise conclude nothing had changed.
    """
    state, roster = await _state(tmp_path)
    before = [r.name for r in _live(state)]
    (tmp_path / "uploads").mkdir()
    try:
        await ops.add_root(state, GROUP, str(tmp_path / "uploads"))
        assert [r.name for r in _live(state)] == before, (
            "add_root edited the live RootSet, which is the object retarget "
            "diffs against — the new root would never be scanned")

        await ops.remove_root(state, GROUP, "two")
        assert [r.name for r in _live(state)] == before, (
            "remove_root edited the live RootSet, so retarget cannot tell the "
            "removed root's entries should go")
    finally:
        await roster.close()


# ── What the node then serves ────────────────────────────────────────────────

async def test_a_retarget_from_the_config_scans_the_new_root(tmp_path):
    """
    The half no assertion about `describe()` can reach: the files.

    A root that appears in the table and holds nothing is the same bug one step
    later, and it is what editing the live set in place would produce.
    """
    state, roster = await _state(tmp_path)
    (tmp_path / "one" / "kept.txt").write_bytes(b"kept")
    fresh = tmp_path / "uploads"
    fresh.mkdir()
    (fresh / "new.txt").write_bytes(b"new")

    idx = await _indexer(state)
    assert {e.name for e in idx.index.entries} == {"kept.txt"}
    try:
        await ops.add_root(state, GROUP, str(fresh))
        await idx.retarget(_rebuilt(state))

        assert {e.name for e in idx.index.entries} == {"kept.txt", "new.txt"}, (
            "the added directory was not scanned — it would show in the table "
            "and stay empty")
        assert [r["name"] for r in idx.index.roots] == ["one", "two", "uploads"]
    finally:
        await roster.close()


async def test_handing_retarget_the_edited_set_scans_nothing(tmp_path):
    """
    The failure mode above, demonstrated rather than described — so the reason
    the ops leave the live set alone is checkable instead of asserted in a
    comment. If this ever starts failing, `retarget` has changed and the rule
    in `add_root` can be revisited.
    """
    state, roster = await _state(tmp_path)
    fresh = tmp_path / "uploads"
    fresh.mkdir()
    (fresh / "new.txt").write_bytes(b"new")

    idx = await _indexer(state)
    try:
        await ops.add_root(state, GROUP, str(fresh))
        # What editing in place would have left behind.
        _live(state).roots.append(_rebuilt(state).roots[-1])
        await idx.retarget(_live(state))

        assert {e.name for e in idx.index.entries} == set(), (
            "retarget now scans a root it was handed on both sides of its own "
            "diff — the constraint this file is built on has changed")
    finally:
        await roster.close()


async def test_a_retarget_from_the_config_drops_a_removed_root(tmp_path):
    """The mirror: a removed directory's files must stop being served."""
    state, roster = await _state(tmp_path)
    (tmp_path / "one" / "kept.txt").write_bytes(b"kept")
    (tmp_path / "two" / "going.txt").write_bytes(b"going")

    idx = await _indexer(state)
    assert {e.name for e in idx.index.entries} == {"kept.txt", "going.txt"}
    try:
        await ops.remove_root(state, GROUP, "two")
        await idx.retarget(_rebuilt(state))
        assert {e.name for e in idx.index.entries} == {"kept.txt"}, (
            "the removed directory's files are still being served")
    finally:
        await roster.close()


# ── The invariants around them ───────────────────────────────────────────────

async def test_adding_the_same_directory_twice_is_still_refused(tmp_path):
    """A group with one path under two names indexes every file in it twice."""
    state, roster = await _state(tmp_path)
    (tmp_path / "uploads").mkdir()
    try:
        await ops.add_root(state, GROUP, str(tmp_path / "uploads"))
        with pytest.raises(ops.OpError):
            await ops.add_root(state, GROUP, str(tmp_path / "uploads"))
        assert len(state["config"].groups[0].roots) == 3, (
            "the refused add left something behind")
        assert Path(state["config_path"]).read_text().count(
            "[[groups.roots]]") == 3
    finally:
        await roster.close()


async def test_a_second_different_root_still_lands(tmp_path):
    state, roster = await _state(tmp_path)
    (tmp_path / "uploads").mkdir()
    (tmp_path / "incoming").mkdir()
    try:
        await ops.add_root(state, GROUP, str(tmp_path / "uploads"))
        result = await ops.add_root(state, GROUP, str(tmp_path / "incoming"),
                                    writable=True)
        assert [r["name"] for r in result["roots"]] == [
            "one", "two", "uploads", "incoming"]
        assert result["roots"][-1]["writable"] is True
    finally:
        await roster.close()


async def test_updating_flags_may_edit_the_live_set(tmp_path):
    """
    The exception, and why it is one: `writable` and `removable` change nothing
    about which files exist, so there is nothing for retarget to scan or drop.
    Editing in place is what makes the flag true for the upload handler on the
    very next request, which is synchronous and reads the live set.
    """
    state, roster = await _state(tmp_path)
    state["config"] = SimpleNamespace(groups=state["config"].groups)
    try:
        result = await ops.update_root(state, GROUP, "two",
                                       writable=True, removable=True)
        live = _live(state).by_name("two")
        assert live.writable is True and live.removable is True
        assert result["roots"] == _live(state).describe()
    finally:
        await roster.close()


async def test_the_file_on_disk_and_the_config_in_memory_agree(tmp_path):
    """
    A reload re-reads the file, so a config edited in memory but not on disk is
    undone by the next restart — and one written to disk but not in memory
    makes the *next* op validate against a stale picture.
    """
    state, roster = await _state(tmp_path)
    (tmp_path / "uploads").mkdir()
    try:
        await ops.add_root(state, GROUP, str(tmp_path / "uploads"),
                           writable=True)
        await ops.remove_root(state, GROUP, "one")

        import tomllib
        on_disk = tomllib.loads(Path(state["config_path"]).read_text())
        disk_paths = [str(r["path"]) for r in on_disk["groups"][0]["roots"]]
        memory_paths = [Path(r.path).as_posix()
                        for r in state["config"].groups[0].roots]
        assert disk_paths == memory_paths
        assert Path(state["config_path"]).read_text().count(
            "[[groups.roots]]") == 2
    finally:
        await roster.close()


# ── The seam that was actually broken ────────────────────────────────────────

async def test_the_mnp_path_reloads_like_the_loopback_one(tmp_path):
    """
    The two front doors, doing the same thing.

    `ui/app.py` has always fired the daemon's `reload_fn` after a root op.
    `_retarget_indexer` did not — it re-pointed the indexer at the live set
    instead, which is the object the ops leave alone, so nothing happened at
    all. That divergence *is* the bug: the loopback path worked, the MNP path
    did not, and it survived until an operator added a directory from a
    browser.

    Not awaited: a reload rescans, and a new library is minutes. The ack
    already carries the set the node is moving to.
    """
    from meshbay_node.transport.webrtc_server import WebRTCPeerSession

    state, roster = await _state(tmp_path)
    reloaded: list[bool] = []

    async def fake_reload():
        reloaded.append(True)

    state["reload_fn"] = fake_reload
    spawned = []

    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = {"daemon_state": state}
    session._spawn = lambda coro: spawned.append(coro)
    try:
        await session._retarget_indexer(GROUP)
        assert spawned, "the MNP path did not ask the daemon to reload"
        await spawned[0]
        assert reloaded == [True]
    finally:
        await roster.close()


async def test_without_a_daemon_it_still_retargets(tmp_path):
    """
    A context assembled by hand — a harness, or a test — has no `reload_fn`.
    Falling through to a direct retarget keeps those working, and is correct
    precisely because the ops no longer edit the set being passed.
    """
    from meshbay_node.transport.webrtc_server import WebRTCPeerSession

    state, roster = await _state(tmp_path)
    fresh = tmp_path / "uploads"
    fresh.mkdir()
    (fresh / "new.txt").write_bytes(b"new")
    idx = await _indexer(state)
    state["indexers"] = {GROUP: idx}

    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = {"daemon_state": state}
    try:
        await ops.add_root(state, GROUP, str(fresh))
        # What a reload would have installed, done by hand here.
        state["groups_ctx"][GROUP]["roots"] = _rebuilt(state)
        await session._retarget_indexer(GROUP)
        assert {e.name for e in idx.index.entries} == {"new.txt"}
    finally:
        await roster.close()