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

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

import asyncio
import os
import threading
import time
from pathlib import Path

import meshbay_node.indexer.indexer as indexer_mod
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.crypto import generate_gek
from meshbay_node.indexer import DirectoryIndexer, GroupIndex, IndexCache

from conftest import one_root


@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))
    # Above MIN_AUDIO_SIZE_BYTES: these tests use this file as a generic
    # "here's an audio entry" stand-in, not as a test of the tiny-file gate
    # itself (see test_is_indexable_size_* / test_initial_scan_skips_a_
    # corrupt_tiny_audio_file below for that).
    (d / "music.mp3").write_bytes(os.urandom(indexer_mod.MIN_AUDIO_SIZE_BYTES + 1024))
    (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"


def test_type_detection_covers_wma_and_musepack():
    """
    Both formats are real audio a real library can contain, even though
    neither one plays natively in a browser <audio> element — that gap is
    a separate, later concern (transcoding); being correctly typed and
    tagged is not the same thing as being playable.
    """
    assert indexer_mod._detect_type(Path("track.wma")) == "audio"
    assert indexer_mod._detect_type(Path("track.mpc")) == "audio"


def test_is_indexable_size_rejects_tiny_audio(tmp_path):
    assert indexer_mod._is_indexable_size(Path("track.mp3"), 1256) is False
    assert indexer_mod._is_indexable_size(
        Path("track.mp3"), indexer_mod.MIN_AUDIO_SIZE_BYTES) is True
    assert indexer_mod._is_indexable_size(
        Path("track.mp3"), indexer_mod.MIN_AUDIO_SIZE_BYTES - 1) is False


def test_is_indexable_size_does_not_apply_to_other_types():
    """A tiny document/image is still worth indexing — the rule exists
    because a truncated *audio* file is a known corruption signature, not
    because small files in general are suspect."""
    assert indexer_mod._is_indexable_size(Path("readme.txt"), 10) is True
    assert indexer_mod._is_indexable_size(Path("icon.png"), 10) is True


@pytest.mark.asyncio
async def test_initial_scan_skips_a_corrupt_tiny_audio_file(tmp_path, sk_node, gek):
    d = tmp_path / "shared"
    d.mkdir()
    (d / "good.mp3").write_bytes(os.urandom(indexer_mod.MIN_AUDIO_SIZE_BYTES + 1))
    (d / "corrupt.mp3").write_bytes(os.urandom(1256))

    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 "good.mp3" in names
    assert "corrupt.mp3" not in names


@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_second_group_sharing_the_same_folder_hashes_nothing(
        shared_dir, sk_node, gek, index_cache):
    """
    The scenario the cache was made node-wide for (2026-08-25): an operator
    shares the same physical folder into a second group. `IndexCache` is
    keyed purely by absolute path, with no notion of group_id at all — a
    *different* group_id scanning the same folder through the same shared
    cache instance must hit exactly as hard as a same-group restart does
    (the test right above this one). Before this cache was shared node-wide,
    each group got its own on-disk cache file and this scan would have
    rehashed every byte again.
    """
    first = DirectoryIndexer(roots=one_root(shared_dir), group_id="group-a",
                             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="group-b",
                                  sk_node=sk_node, gek=gek, cache=index_cache)
        await second.initial_scan()
    finally:
        indexer_mod._scan_file = real_scan_file

    assert calls == [], (
        f"a second group scanning the same folder must not rehash it, got {calls}")
    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_scanning_flag_is_true_while_the_walk_is_still_running(tmp_path, sk_node, gek):
    """
    Regression for the actual bug this was found by (2026-08-25): `scanning`
    used to flip on only *after* the walk finished, so a consumer polling
    IndexProgress — index-status; the Create Group wizard's own progress
    poll, which gives up after a short grace period if it never observes
    `scanning: true` — read "not scanning" for however long a large/slow
    root's discovery phase took, even though the node was already doing real
    work. Confirmed against production logs: a wizard step waiting on this
    flag gave up exactly at its grace-period deadline for a root whose walk
    was still running.
    """
    d = tmp_path / "shared"
    d.mkdir()
    (d / "f.bin").write_bytes(b"x")

    real_walk = indexer_mod._walk_root
    walk_started = threading.Event()
    release_walk = threading.Event()

    def slow_walk(root):
        walk_started.set()
        release_walk.wait(timeout=5)
        return real_walk(root)

    indexer_mod._walk_root = slow_walk
    indexer = DirectoryIndexer(roots=one_root(d), group_id="g", sk_node=sk_node, gek=gek)
    try:
        scan_task = asyncio.create_task(indexer.initial_scan())
        # Off-loop wait for the walk to actually start — busy-polling the
        # event loop itself here would defeat the point of the test.
        await asyncio.get_event_loop().run_in_executor(None, walk_started.wait, 5)

        assert indexer.progress.scanning is True, (
            "scanning must be True the moment the walk starts, not only "
            "once it (and the sizing pass) finish")
    finally:
        release_walk.set()
        indexer_mod._walk_root = real_walk
        await scan_task

    assert indexer.progress.scanning is False


@pytest.mark.asyncio
async def test_scanning_flag_is_true_while_sizing_files_is_still_running(
        tmp_path, sk_node, gek):
    """
    Same regression, one phase later: sizing (stat()-ing every walked file)
    used to be a synchronous loop straight on the asyncio event loop thread —
    for a root with many thousands of files (a real personal library, not a
    hypothetical) that blocked the entire daemon, and did so before
    `scanning` was ever set. Now off-loop (_size_files) and `scanning` is
    already true throughout, same principle as the walk phase above.
    """
    d = tmp_path / "shared"
    d.mkdir()
    (d / "f.bin").write_bytes(b"x")

    real_size = indexer_mod._size_files
    sizing_started = threading.Event()
    release_sizing = threading.Event()

    def slow_size(files):
        sizing_started.set()
        release_sizing.wait(timeout=5)
        return real_size(files)

    indexer_mod._size_files = slow_size
    indexer = DirectoryIndexer(roots=one_root(d), group_id="g", sk_node=sk_node, gek=gek)
    try:
        scan_task = asyncio.create_task(indexer.initial_scan())
        await asyncio.get_event_loop().run_in_executor(None, sizing_started.wait, 5)

        assert indexer.progress.scanning is True
    finally:
        release_sizing.set()
        indexer_mod._size_files = real_size
        await scan_task

    assert indexer.progress.scanning is False


@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


# ── Indexing v2 — partial-read hashing ────────────────────────────────────────


def test_small_file_gets_hash_version_1(tmp_path):
    from meshbay_node.indexer.indexer import _scan_file
    d = tmp_path / "root"
    d.mkdir()
    f = d / "small.mp4"
    f.write_bytes(os.urandom(1024))

    entry = _scan_file(one_root(d).roots[0], f)
    assert entry is not None
    assert entry.hash_version == 1


def test_large_file_gets_hash_version_2(tmp_path):
    from meshbay_node.indexer.indexer import _PARTIAL_THRESHOLD, _scan_file
    d = tmp_path / "root"
    d.mkdir()
    f = d / "big.mkv"
    size = _PARTIAL_THRESHOLD + 1
    f.write_bytes(os.urandom(size))

    entry = _scan_file(one_root(d).roots[0], f)
    assert entry is not None
    assert entry.hash_version == 2
    assert entry.size == size


def test_file_at_threshold_gets_hash_version_1(tmp_path):
    from meshbay_node.indexer.indexer import _PARTIAL_THRESHOLD, _scan_file
    d = tmp_path / "root"
    d.mkdir()
    f = d / "exact.mkv"
    f.write_bytes(os.urandom(_PARTIAL_THRESHOLD))

    entry = _scan_file(one_root(d).roots[0], f)
    assert entry is not None
    assert entry.hash_version == 1


def test_partial_hash_differs_from_full_hash(tmp_path):
    """For a file above the threshold, the partial hash must differ from what
    a full-file blake3 would produce (they read different bytes)."""
    import blake3 as b3
    from meshbay_node.indexer.indexer import _PARTIAL_THRESHOLD, _scan_file
    d = tmp_path / "root"
    d.mkdir()
    f = d / "big.mkv"
    content = os.urandom(_PARTIAL_THRESHOLD + 1024 * 1024)
    f.write_bytes(content)

    entry = _scan_file(one_root(d).roots[0], f)
    full_hash = b3.blake3(content).hexdigest()

    assert entry.id != full_hash
    assert entry.hash_version == 2


def test_partial_hash_is_deterministic(tmp_path):
    from meshbay_node.indexer.indexer import _PARTIAL_THRESHOLD, _scan_file
    d = tmp_path / "root"
    d.mkdir()
    f = d / "big.mkv"
    f.write_bytes(os.urandom(_PARTIAL_THRESHOLD + 1))

    e1 = _scan_file(one_root(d).roots[0], f)
    e2 = _scan_file(one_root(d).roots[0], f)
    assert e1.id == e2.id


def test_group_index_roundtrip_preserves_hash_version(sk_node, gek):
    from meshbay_common.protocol import IndexEntry
    idx = GroupIndex(group_id="hv-test", sk_node=sk_node, gek=gek)
    idx.add_entry(IndexEntry(
        id="aaa", name="small.mp4", path="root", size=1024,
        type="video", added_at=100, hash_version=1))
    idx.add_entry(IndexEntry(
        id="bbb", name="big.mkv", path="root", size=50_000_000,
        type="video", added_at=200, hash_version=2))

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

    by_id = {e.id: e for e in recovered.entries}
    assert by_id["aaa"].hash_version == 1
    assert by_id["bbb"].hash_version == 2


def test_deserialize_without_hash_version_defaults_to_1(sk_node, gek):
    """Entries serialized by old code (no hash_version field) must deserialize
    as hash_version=1."""
    from meshbay_common.protocol import IndexEntry
    idx = GroupIndex(group_id="compat", sk_node=sk_node, gek=gek)
    idx.add_entry(IndexEntry(
        id="old", name="f.mp4", path="root", size=1024,
        type="video", added_at=100))
    wire = idx.serialize()
    recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek)
    assert recovered.entries[0].hash_version == 1


def test_index_entry_wire_includes_hash_version():
    from meshbay_common.protocol import IndexEntry, index_entry_wire
    e = IndexEntry(id="x", name="f.mp4", path="root", size=1,
                   type="video", added_at=0, hash_version=2)
    w = index_entry_wire(e)
    assert w["hash_version"] == 2


@pytest.mark.asyncio
async def test_cache_aware_scan_uses_hash_version(tmp_path, sk_node, gek):
    from meshbay_node.indexer.indexer import _PARTIAL_THRESHOLD
    d = tmp_path / "root"
    d.mkdir()
    small = d / "small.mp4"
    small.write_bytes(os.urandom(1024))
    big = d / "big.mkv"
    big.write_bytes(os.urandom(_PARTIAL_THRESHOLD + 1))

    cache = IndexCache(db_path=tmp_path / "cache.db")
    await cache.open()

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

    by_name = {e.name: e for e in indexer.index.entries}
    assert by_name["small.mp4"].hash_version == 1
    assert by_name["big.mkv"].hash_version == 2

    hit_small = await cache.lookup(
        str(small), small.stat().st_size, small.stat().st_mtime,
        hash_version=1)
    assert hit_small is not None

    hit_big = await cache.lookup(
        str(big), big.stat().st_size, big.stat().st_mtime,
        hash_version=2)
    assert hit_big is not None

    await cache.close()