"""
An indexer has one `progress`, and a whole-root walk owns it while it runs.

Found while planning a progress band for the operator (2026-09-14), by reading
who writes `progress`. Two writers had no business doing it during a scan:

- A watchdog burst — a file dropped into a folder while a 900 GB root was being
  hashed — added its size to the scan's total, and cleared `scanning` when its
  own hash finished. Anything showing progress would have gone blank with hours
  of hashing left.
- A plug rescan took no lock, so it walked its root beside an added root's
  scan: both reset the same counters, and both read the drive in turn.
"""

import asyncio
import os
from pathlib import Path

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_node.indexer.indexer import DirectoryIndexer
from meshbay_node.roots import RootSet

pytestmark = pytest.mark.asyncio

GROUP = "g" * 32


def _set(*specs) -> RootSet:
    return RootSet.build([s if isinstance(s, dict) else {"path": str(s), "name": s.name}
                          for s in specs])


def _names(idx) -> list[str]:
    return sorted(e.name for e in idx.index.entries)


async def _until(predicate, timeout: float = 3.0) -> bool:
    deadline = asyncio.get_running_loop().time() + timeout
    while not predicate():
        if asyncio.get_running_loop().time() > deadline:
            return False
        await asyncio.sleep(0.02)
    return True


class _Held(DirectoryIndexer):
    """Stops hashing under `hold_under` once `hold_after` of its files are done."""
    hold_under: Path | None = None
    hold_after = 0

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.gate = asyncio.Event()
        self.at_gate = asyncio.Event()
        self.rescanned: list[str] = []
        self._seen = 0

    async def _hash_or_cached(self, root, file_path):
        if self.hold_under is not None and file_path.is_relative_to(self.hold_under):
            if self._seen == self.hold_after:
                self.at_gate.set()
                await self.gate.wait()
            self._seen += 1
        return await super()._hash_or_cached(root, file_path)

    async def _rescan_root(self, root):
        self.rescanned.append(root.name)
        return await super()._rescan_root(root)


def _tree(tmp_path: Path) -> tuple[Path, Path, Path]:
    one, two, big = tmp_path / "one", tmp_path / "two", tmp_path / "big"
    for d in (one, two, big):
        d.mkdir()
    (one / "a.txt").write_bytes(b"first root")
    (two / "b.txt").write_bytes(b"removable root")
    (big / "x.bin").write_bytes(os.urandom(3000))
    (big / "y.bin").write_bytes(os.urandom(5000))
    return one, two, big


async def test_a_burst_during_a_scan_leaves_the_scan_on_screen(tmp_path):
    one, _, big = _tree(tmp_path)
    idx = _Held(roots=_set(one), group_id=GROUP, sk_node=Ed25519PrivateKey.generate(),
                gek=None, debounce_secs=0.01)
    await idx.initial_scan()
    idx._loop = asyncio.get_running_loop()
    idx.hold_under, idx.hold_after = big, 1
    try:
        await idx.retarget(_set(one, big), wait=False)
        await asyncio.wait_for(idx.at_gate.wait(), 5)
        total, scanned = idx.progress.total_bytes, idx.progress.scanned_bytes
        assert total == 8000 and scanned in (3000, 5000)

        dropped = one / "dropped.txt"
        dropped.write_bytes(os.urandom(700))
        idx._schedule_update(dropped)
        assert await _until(lambda: "dropped.txt" in _names(idx))
        assert await _until(lambda: idx._burst_inflight == 0)

        assert idx.progress.scanning is True, (
            "the dropped file's hash finishing turned the scan's progress off")
        assert (idx.progress.total_bytes, idx.progress.scanned_bytes) == (total, scanned), (
            "the dropped file was counted into the scan under way")

        idx.gate.set()
        await asyncio.wait_for(asyncio.gather(*list(idx._scan_tasks)), 5)
        assert await _until(lambda: not idx.progress.scanning)
        assert _names(idx) == ["a.txt", "dropped.txt", "x.bin", "y.bin"]
    finally:
        idx.gate.set()
        await idx.stop()


def _with_two_ejected(one: Path, two: Path, *more: Path) -> RootSet:
    return _set(one, {"path": str(two), "name": "two", "ejected": True}, *more)


async def test_a_plug_rescan_waits_for_the_scan_under_way(tmp_path):
    one, two, big = _tree(tmp_path)
    idx = _Held(roots=_set(one, two), group_id=GROUP,
                sk_node=Ed25519PrivateKey.generate(), gek=None)
    await idx.initial_scan()
    idx.eject_root("two")
    idx.hold_under = big
    try:
        await idx.retarget(_with_two_ejected(one, two, big), wait=False)
        await asyncio.wait_for(idx.at_gate.wait(), 5)

        plug = asyncio.create_task(idx.plug_root("two"))
        await asyncio.sleep(0.1)
        assert idx.rescanned == [], "the plug walked its root beside the scan under way"

        idx.gate.set()
        await asyncio.wait_for(plug, 5)
        assert idx.rescanned == ["two"]
        assert _names(idx) == ["a.txt", "b.txt", "x.bin", "y.bin"]
    finally:
        idx.gate.set()
        await idx.stop()


async def test_a_root_ejected_again_while_its_plug_waited_is_not_emptied(tmp_path):
    one, two, big = _tree(tmp_path)
    idx = _Held(roots=_set(one, two), group_id=GROUP,
                sk_node=Ed25519PrivateKey.generate(), gek=None)
    await idx.initial_scan()
    idx.eject_root("two")
    idx.hold_under = big
    try:
        await idx.retarget(_with_two_ejected(one, two, big), wait=False)
        await asyncio.wait_for(idx.at_gate.wait(), 5)

        plug = asyncio.create_task(idx.plug_root("two"))
        await asyncio.sleep(0.1)
        idx.eject_root("two")
        idx.gate.set()
        await asyncio.wait_for(plug, 5)

        assert idx.rescanned == []
        assert "b.txt" in _names(idx), "a root ejected while it waited was emptied"
    finally:
        idx.gate.set()
        await idx.stop()