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

"""Tests for indexer and GroupIndex."""

import asyncio
import os
import time
import pytest
from pathlib import Path
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey

from meshbay_common.crypto import generate_gek
from meshbay_node.indexer import DirectoryIndexer, GroupIndex, IndexCache
import meshbay_node.indexer.indexer as indexer_mod
from conftest import one_root
from meshbay_node.keystore import NodeKeys


@pytest.fixture
def sk_node():
    return Ed25519PrivateKey.generate()

@pytest.fixture
def gek():
    return generate_gek()

@pytest.fixture
def shared_dir(tmp_path):
    d = tmp_path / "shared"
    d.mkdir()
    (d / "video.mkv").write_bytes(os.urandom(1024))
    (d / "music.mp3").write_bytes(os.urandom(512))
    (d / "readme.md").write_bytes(b"# Hello MeshBay")
    subdir = d / "docs"
    subdir.mkdir()
    (subdir / "manual.pdf").write_bytes(os.urandom(2048))
    return d


# ── GroupIndex tests ──────────────────────────────────────────────────────────

def test_group_index_serialize_deserialize_private(sk_node, gek, shared_dir):
    idx = GroupIndex(group_id="grp-001", sk_node=sk_node, gek=gek)
    from meshbay_common.protocol import IndexEntry
    idx.add_entry(IndexEntry(
        id="abc123", name="video.mkv", path="", size=1024,
        type="video", added_at=int(time.time()), duration=120))

    wire = idx.serialize()
    recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek)

    assert recovered.group_id == "grp-001"
    assert recovered.count == 1
    assert recovered.entries[0].name == "video.mkv"
    assert recovered.entries[0].type == "video"


def test_group_index_serialize_deserialize_public(sk_node):
    idx = GroupIndex(group_id="pub-001", sk_node=sk_node, gek=None)
    from meshbay_common.protocol import IndexEntry
    idx.add_entry(IndexEntry(
        id="xyz789", name="readme.txt", path="", size=42,
        type="document", added_at=int(time.time())))

    wire = idx.serialize()
    recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=None)
    assert recovered.count == 1
    assert recovered.entries[0].id == "xyz789"


def test_group_index_wrong_gek_rejected(sk_node, gek):
    idx = GroupIndex(group_id="grp-002", sk_node=sk_node, gek=gek)
    from meshbay_common.protocol import IndexEntry
    idx.add_entry(IndexEntry(id="a", name="f.mp3", path="", size=1,
                             type="audio", added_at=0))
    wire = idx.serialize()

    wrong_gek = generate_gek()
    with pytest.raises(Exception):   # InvalidTag from AEAD
        GroupIndex.deserialize(wire, sk_node=sk_node, gek=wrong_gek)


def test_group_index_tampered_rejected(sk_node, gek):
    idx = GroupIndex(group_id="grp-003", sk_node=sk_node, gek=gek)
    from meshbay_common.protocol import IndexEntry
    idx.add_entry(IndexEntry(id="b", name="f.mp4", path="", size=1,
                             type="video", added_at=0))
    wire = bytearray(idx.serialize())
    wire[-5] ^= 0xFF   # flip bytes at the end
    with pytest.raises(Exception):
        GroupIndex.deserialize(bytes(wire), sk_node=sk_node, gek=gek)


def test_group_index_diff(sk_node, gek):
    from meshbay_common.protocol import IndexEntry
    v1 = GroupIndex(group_id="g", sk_node=sk_node, gek=gek, version=1)
    v1.add_entry(IndexEntry(id="aaa", name="a.mp4", path="", size=1,
                            type="video", added_at=0))
    v1.add_entry(IndexEntry(id="bbb", name="b.mp3", path="", size=1,
                            type="audio", added_at=0))

    v2 = GroupIndex(group_id="g", sk_node=sk_node, gek=gek, version=2)
    v2.add_entry(IndexEntry(id="aaa", name="a.mp4", path="", size=1,
                            type="video", added_at=0))
    v2.add_entry(IndexEntry(id="ccc", name="c.mkv", path="", size=1,
                            type="video", added_at=0))

    delta = v2.diff(v1)
    assert delta.base_version == 1
    assert delta.version == 2
    assert len(delta.additions) == 1
    assert delta.additions[0].id == "ccc"
    assert "bbb" in delta.deletions


# ── DirectoryIndexer tests ────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_initial_scan(shared_dir, sk_node, gek):
    indexer = DirectoryIndexer(
        roots=one_root(shared_dir),
        group_id="scan-test",
        sk_node=sk_node,
        gek=gek,
    )
    await indexer.initial_scan()

    entries = indexer.index.entries
    names   = {e.name for e in entries}

    assert "video.mkv"  in names
    assert "music.mp3"  in names
    assert "readme.md"  in names
    assert "manual.pdf" in names
    assert indexer.index.count == 4


@pytest.mark.asyncio
async def test_type_detection(shared_dir, sk_node, gek):
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    by_name = {e.name: e.type for e in indexer.index.entries}
    assert by_name["video.mkv"] == "video"
    assert by_name["music.mp3"] == "audio"
    assert by_name["readme.md"] == "document"
    assert by_name["manual.pdf"] == "document"


@pytest.mark.asyncio
async def test_hidden_files_excluded(tmp_path, sk_node, gek):
    d = tmp_path / "dir"
    d.mkdir()
    (d / ".hidden").write_bytes(b"secret")
    (d / "visible.txt").write_bytes(b"visible")
    (d / "file.tmp").write_bytes(b"tmp")

    indexer = DirectoryIndexer(roots=one_root(d), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    names = {e.name for e in indexer.index.entries}
    assert "visible.txt" in names
    assert ".hidden" not in names
    assert "file.tmp" not in names


@pytest.mark.asyncio
async def test_on_change_callback(shared_dir, sk_node, gek):
    changes = []

    async def on_change(idx):
        changes.append(idx.index.count)

    indexer = DirectoryIndexer(
        roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek,
        on_change=on_change)
    await indexer.start()
    await asyncio.sleep(0.1)

    (shared_dir / "newfile.mp4").write_bytes(os.urandom(256))
    await asyncio.sleep(3.0)   # watchdog detect + 2s debounce

    await indexer.stop()
    assert len(changes) >= 1, "on_change should have been called"


@pytest.mark.asyncio
async def test_index_roundtrip_after_scan(shared_dir, sk_node, gek):
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    wire = indexer.index.serialize()
    recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek)
    assert recovered.count == indexer.index.count


# ── Cache-aware scanning ───────────────────────────────────────────────────────

@pytest.fixture
async def index_cache(tmp_path):
    c = IndexCache(db_path=tmp_path / "index_cache.db")
    await c.open()
    yield c
    await c.close()


@pytest.mark.asyncio
async def test_second_scan_with_same_cache_hashes_nothing(
        shared_dir, sk_node, gek, index_cache):
    """
    The whole point of the cache: a "restart" (a fresh DirectoryIndexer, same
    on-disk cache) that finds every file's (size, mtime) unchanged must not
    read a single byte of file content.
    """
    first = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
                             sk_node=sk_node, gek=gek, cache=index_cache)
    await first.initial_scan()
    assert first.index.count == 4

    calls = []
    real_scan_file = indexer_mod._scan_file

    def spy(root, path):
        calls.append(path)
        return real_scan_file(root, path)

    indexer_mod._scan_file = spy
    try:
        second = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
                                  sk_node=sk_node, gek=gek, cache=index_cache)
        await second.initial_scan()
    finally:
        indexer_mod._scan_file = real_scan_file

    assert calls == [], f"expected zero hash calls on a fully-cached rescan, got {calls}"
    assert second.index.count == first.index.count
    assert {e.id for e in second.index.entries} == {e.id for e in first.index.entries}


@pytest.mark.asyncio
async def test_modified_file_is_rehashed(tmp_path, sk_node, gek, index_cache):
    d = tmp_path / "shared"
    d.mkdir()
    f = d / "movie.mkv"
    f.write_bytes(b"original content")

    first = DirectoryIndexer(roots=one_root(d), group_id="g",
                             sk_node=sk_node, gek=gek, cache=index_cache)
    await first.initial_scan()
    old_id = first.index.entries[0].id

    # Change both content and mtime, as any real edit would.
    f.write_bytes(b"a completely different, longer payload")
    os.utime(f, (time.time() + 5, time.time() + 5))

    second = DirectoryIndexer(roots=one_root(d), group_id="g",
                              sk_node=sk_node, gek=gek, cache=index_cache)
    await second.initial_scan()

    assert second.index.count == 1
    assert second.index.entries[0].id != old_id


@pytest.mark.asyncio
async def test_scan_interrupted_partway_leaves_only_completed_files_cached(
        tmp_path, sk_node, gek, index_cache):
    """
    A cache row is only ever written after a file is fully hashed (cache.py),
    so a crash mid-scan cannot leave a stale/partial row — the next scan just
    treats the not-yet-cached files as new, and finishes the job.
    """
    d = tmp_path / "shared"
    d.mkdir()
    names = [f"file{i}.bin" for i in range(5)]
    for i, name in enumerate(names):
        (d / name).write_bytes(os.urandom(64) * (i + 1))

    real_scan_file = indexer_mod._scan_file
    hashed_before_crash = []

    def crash_after_three(root, path):
        if len(hashed_before_crash) >= 3:
            raise RuntimeError("simulated crash mid-scan")
        entry = real_scan_file(root, path)
        hashed_before_crash.append(path)
        return entry

    indexer_mod._scan_file = crash_after_three
    try:
        crashing = DirectoryIndexer(roots=one_root(d), group_id="g",
                                    sk_node=sk_node, gek=gek, cache=index_cache)
        with pytest.raises(RuntimeError):
            await crashing.initial_scan()
    finally:
        indexer_mod._scan_file = real_scan_file

    assert len(hashed_before_crash) == 3

    # A normal rescan (same cache) must still end up with all 5 files
    # correctly indexed, hashing only the ones the crash never got to.
    calls = []

    def spy(root, path):
        calls.append(path)
        return real_scan_file(root, path)

    indexer_mod._scan_file = spy
    try:
        resumed = DirectoryIndexer(roots=one_root(d), group_id="g",
                                   sk_node=sk_node, gek=gek, cache=index_cache)
        await resumed.initial_scan()
    finally:
        indexer_mod._scan_file = real_scan_file

    assert resumed.index.count == 5
    assert len(calls) == 2, f"expected only the 2 not-yet-cached files to be hashed, got {len(calls)}"


# ── Progress state ───────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_progress_reflects_bytes_scanned(shared_dir, sk_node, gek):
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
                               sk_node=sk_node, gek=gek)
    assert indexer.progress.scanning is False

    await indexer.initial_scan()

    total_size = sum(f.stat().st_size for f in shared_dir.rglob("*") if f.is_file())
    assert indexer.progress.scanning is False, "must end idle, not stuck scanning"
    assert indexer.progress.scanned_bytes == total_size
    assert indexer.progress.total_bytes == total_size


@pytest.mark.asyncio
async def test_progress_stops_even_when_hashing_raises(tmp_path, sk_node, gek):
    d = tmp_path / "shared"
    d.mkdir()
    (d / "a.bin").write_bytes(os.urandom(64))
    (d / "b.bin").write_bytes(os.urandom(64))

    real_scan_file = indexer_mod._scan_file

    def boom(root, path):
        raise RuntimeError("simulated failure mid-scan")

    indexer = DirectoryIndexer(roots=one_root(d), group_id="g",
                               sk_node=sk_node, gek=gek)
    indexer_mod._scan_file = boom
    try:
        with pytest.raises(RuntimeError):
            await indexer.initial_scan()
    finally:
        indexer_mod._scan_file = real_scan_file

    assert indexer.progress.scanning is False, \
        "an exception mid-scan must not leave the scanning flag stuck on"


@pytest.mark.asyncio
async def test_realtime_watchdog_add_reports_progress_like_a_bulk_scan(tmp_path, sk_node, gek):
    """
    Found live: dropping a whole season into an already-watched folder gave
    no scanning indicator and no progress bar at all — only the initial scan
    and the periodic reconcile backstop ever touched `progress`, never the
    real-time per-file watchdog path (_schedule_update/_debounce/
    _update_entry). Drives that path directly (as _WatchdogHandler would),
    without a real filesystem observer, exactly like test_root_availability.py
    already does for _update_entry alone.
    """
    d = tmp_path / "shared"
    d.mkdir()
    paths = []
    for i in range(3):
        p = d / f"ep{i}.mkv"
        p.write_bytes(os.urandom(1024 * (i + 1)))
        paths.append(p)
    total_size = sum(p.stat().st_size for p in paths)

    indexer = DirectoryIndexer(roots=one_root(d), group_id="g", sk_node=sk_node,
                               gek=gek, debounce_secs=0.01)
    indexer._loop = asyncio.get_running_loop()
    assert indexer.progress.scanning is False

    for p in paths:
        indexer._schedule_update(p)
    # _schedule_update only posts _debounce via call_soon_threadsafe (it is
    # written to be called from the watchdog thread) — give the loop one
    # turn to actually run the three posted calls before asserting on them.
    await asyncio.sleep(0)

    # All three scheduled near-simultaneously (as a burst of watchdog events
    # for one `mv` would arrive) — the indicator must flip on immediately,
    # before any single file has actually finished hashing.
    assert indexer.progress.scanning is True
    assert indexer.progress.total_bytes == total_size
    assert indexer.progress.scanned_bytes == 0

    await asyncio.sleep(0.05)   # past debounce_secs; lets all three fire and finish

    assert indexer.progress.scanning is False, "must end idle, not stuck scanning"
    assert indexer.progress.scanned_bytes == total_size
    assert indexer.progress.total_bytes == total_size
    assert len(indexer.index.entries) == 3


@pytest.mark.asyncio
async def test_realtime_watchdog_rapid_rewrite_does_not_double_count(tmp_path, sk_node, gek):
    """A file rewritten during its own debounce window (on_modified firing
    again before the first timer fires) must count its size once, not once
    per event — the old timer is cancelled, and its accounting must transfer
    to whichever fire() actually runs rather than being counted twice."""
    d = tmp_path / "shared"
    d.mkdir()
    p = d / "ep0.mkv"
    p.write_bytes(os.urandom(2048))

    indexer = DirectoryIndexer(roots=one_root(d), group_id="g", sk_node=sk_node,
                               gek=gek, debounce_secs=0.05)
    indexer._loop = asyncio.get_running_loop()

    indexer._schedule_update(p)
    indexer._schedule_update(p)   # re-triggered before the first timer fires
    indexer._schedule_update(p)
    await asyncio.sleep(0)   # let the three posted _debounce calls actually run

    assert indexer.progress.total_bytes == p.stat().st_size, \
        "one file re-triggered must count its size once, not three times"

    await asyncio.sleep(0.1)

    assert indexer.progress.scanning is False
    assert indexer.progress.scanned_bytes == p.stat().st_size


# ── Off-loop directory walks, reconcile backoff ─────────────────────────────

@pytest.mark.asyncio
async def test_walk_root_does_not_stall_the_event_loop(tmp_path, sk_node, gek):
    d = tmp_path / "shared"
    d.mkdir()
    (d / "f.bin").write_bytes(b"x")

    real_walk = indexer_mod._walk_root

    def slow_walk(root):
        time.sleep(0.2)
        return real_walk(root)

    indexer_mod._walk_root = slow_walk
    ticks = 0

    async def ticker():
        nonlocal ticks
        while True:
            await asyncio.sleep(0.01)
            ticks += 1

    ticker_task = asyncio.create_task(ticker())
    try:
        indexer = DirectoryIndexer(roots=one_root(d), group_id="g",
                                   sk_node=sk_node, gek=gek)
        await indexer.initial_scan()
    finally:
        indexer_mod._walk_root = real_walk
        ticker_task.cancel()
        try:
            await ticker_task
        except asyncio.CancelledError:
            pass

    assert ticks >= 5, (
        "the event loop must keep running other tasks while a directory "
        f"walk is in progress in the executor — only {ticks} ticks happened "
        "during a 0.2s walk")


@pytest.mark.asyncio
async def test_reconcile_backoff_grows_with_no_changes_then_caps(shared_dir, sk_node, gek):
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
                               sk_node=sk_node, gek=gek, reconcile_secs=0.01)
    await indexer.initial_scan()
    assert indexer._reconcile_delay == 0.01

    task = asyncio.create_task(indexer._reconcile_loop())
    try:
        await asyncio.sleep(0.2)
        assert indexer._reconcile_delay > 0.01, \
            "several no-change ticks must have grown the delay"
    finally:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass

    # A low, instance-only cap so the clamp is observable without waiting
    # through dozens of real doublings up to the real 7200s ceiling.
    indexer.RECONCILE_BACKOFF_CAP = 0.05
    indexer._reconcile_delay = 0.04
    task = asyncio.create_task(indexer._reconcile_loop())
    try:
        await asyncio.sleep(0.15)
        assert indexer._reconcile_delay <= 0.05, "delay must never exceed the cap"
    finally:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass


@pytest.mark.asyncio
async def test_note_activity_resets_backoff(shared_dir, sk_node, gek):
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
                               sk_node=sk_node, gek=gek, reconcile_secs=10.0)
    await indexer.initial_scan()
    indexer._reconcile_delay = 5000.0  # simulate a long-idle backoff

    indexer.note_activity()

    assert indexer._reconcile_delay == 10.0


@pytest.mark.asyncio
async def test_reconcile_backoff_resets_when_something_actually_changes(
        shared_dir, sk_node, gek):
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
                               sk_node=sk_node, gek=gek, reconcile_secs=0.02)
    await indexer.initial_scan()
    indexer._reconcile_delay = 5.0  # pretend it had already backed off a lot

    # A fake reconcile() rather than a real filesystem change: the real
    # sweep's timing (disk I/O, the executor round trip) would race against
    # this test's own sleeps. What matters here is only _reconcile_loop's
    # reaction to "something changed", not reconcile()'s own detection logic
    # — that is covered separately (test_root_availability.py).
    reconciled_once = asyncio.Event()

    async def fake_reconcile():
        reconciled_once.set()
        return True

    indexer._reconcile_delay = 0.01
    indexer.reconcile = fake_reconcile

    task = asyncio.create_task(indexer._reconcile_loop())
    try:
        await asyncio.wait_for(reconciled_once.wait(), timeout=2.0)
        assert indexer._reconcile_delay == 0.02, \
            "a real change must reset the delay back to the base interval"
    finally:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass