"""
A disk that stops answering, and the scan that used to wait on it for ever.

A read on a failing drive, a share that dropped or a file a sync client has not
fetched does not fail, it does not return. The hash ran on the indexer's one
thread with nothing to bound it, so one such file stopped the scan of its group
where it stood: every root behind it unread, the enrichment that follows a scan
never started, a pause unable to take effect (it is looked at between two
files), and nothing in the log to say which file, since a file is named there
only when it fails.

What is timed is silence, not work: a listing of a million entries and a file
on a slow link both take as long as they take. The files below "never answer"
by waiting on an event the test sets when it ends.
"""

import asyncio
import os
import threading
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.indexer import indexer as indexer_mod
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.indexer.indexer import NO_ANSWER, DirectoryIndexer
from meshbay_node.roots import RootSet
from meshbay_node.roster import Roster

pytestmark = pytest.mark.asyncio

GID = "g" * 32


@pytest.fixture(autouse=True)
def _short_patience(monkeypatch):
    """Tenths of a second where the node waits a minute."""
    monkeypatch.setattr(DirectoryIndexer, "STALL_SECS", 0.3)
    monkeypatch.setattr(DirectoryIndexer, "SLOW_SECS", 0.1)
    monkeypatch.setattr(DirectoryIndexer, "DISK_POLL_SECS", 0.02)


@pytest.fixture
def never():
    """What a read that never returns waits on. Set at the end, so the threads
    the tests leave behind do end."""
    event = threading.Event()
    yield event
    event.set()


def _folder(tmp_path: Path, name: str, files: list[str]) -> Path:
    d = tmp_path / name
    d.mkdir()
    for f in files:
        (d / f).write_bytes(f"{name}/{f}".encode())
    return d


def _indexer(roots: RootSet, **kw) -> DirectoryIndexer:
    return DirectoryIndexer(roots=roots, group_id=GID,
                            sk_node=Ed25519PrivateKey.generate(), gek=None, **kw)


def _names(idx: DirectoryIndexer) -> set[str]:
    return {e.name for e in idx.index.entries}


def _silent_files(monkeypatch, never: threading.Event, silent=lambda name: False) -> list[str]:
    """Reads of the files `silent` names never return. Every read is recorded."""
    read: list[str] = []
    real = indexer_mod._scan_file

    def scan(root, path):
        read.append(path.name)
        if silent(path.name):
            never.wait()
            return None
        return real(root, path)

    monkeypatch.setattr(indexer_mod, "_scan_file", scan)
    return read


# ── One file ─────────────────────────────────────────────────────────────────

async def test_a_read_that_never_returns_costs_its_file_not_the_scan(
        tmp_path, monkeypatch, never):
    films = _folder(tmp_path, "Films", ["a.mkv", "b.mkv", "stuck.mkv", "c.mkv", "d.mkv"])
    roots = RootSet.build([{"path": str(films)}])
    idx = _indexer(roots)
    _silent_files(monkeypatch, never, lambda name: name == "stuck.mkv")

    await asyncio.wait_for(idx.initial_scan(), timeout=10)

    assert _names(idx) == {"a.mkv", "b.mkv", "c.mkv", "d.mkv"}
    (unread,) = idx.unread.values()
    assert (Path(unread.path).name, unread.root, unread.reason) == (
        "stuck.mkv", "Films", NO_ANSWER)
    assert roots.roots[0].paused is False, "one file is not the root's fault"
    assert idx.progress.scanning is False
    assert idx.progress.files_done == 5, "the file given up on is done with, too"


async def test_reconcile_does_not_wait_on_that_file_again_until_it_changes(
        tmp_path, monkeypatch, never):
    """Reconcile indexes every file the index lacks, at every pass. Without a
    memory of this one it would wait out the limit on it each time."""
    films = _folder(tmp_path, "Films", ["a.mkv", "stuck.mkv"])
    idx = _indexer(RootSet.build([{"path": str(films)}]))
    answering = indexer_mod._scan_file
    read = _silent_files(monkeypatch, never, lambda name: name == "stuck.mkv")
    await idx.initial_scan()
    assert read.count("stuck.mkv") == 1

    await idx.reconcile()
    await idx.reconcile()
    assert read.count("stuck.mkv") == 1, "asked the disk again for a file it left unanswered"

    monkeypatch.setattr(indexer_mod, "_scan_file", answering)   # the disk is back...
    await idx.reconcile()
    assert "stuck.mkv" not in _names(idx), "...which nothing has told the node"

    (films / "stuck.mkv").write_bytes(b"written again, and longer")
    await idx.reconcile()
    assert "stuck.mkv" in _names(idx)
    assert idx.unread == {}


async def test_a_file_the_system_refuses_is_listed_with_what_it_said(
        tmp_path, monkeypatch):
    films = _folder(tmp_path, "Films", ["a.mkv", "locked.mkv"])
    idx = _indexer(RootSet.build([{"path": str(films)}]))

    def refuse(path, *args, **kw):
        # What a locked file does: it lists and stats, and refuses to open.
        if Path(path).name == "locked.mkv":
            raise PermissionError(13, "Permission denied")
        return open(path, *args, **kw)

    monkeypatch.setattr(indexer_mod, "open", refuse, raising=False)
    await idx.initial_scan()

    assert _names(idx) == {"a.mkv"}
    (unread,) = idx.unread.values()
    assert (Path(unread.path).name, unread.reason) == ("locked.mkv", "Permission denied")

    # A refusal costs nothing to ask again, and permissions change without the
    # file changing: it is not remembered the way a silence is.
    monkeypatch.delattr(indexer_mod, "open")
    await idx.reconcile()
    assert "locked.mkv" in _names(idx)
    assert idx.unread == {}


async def test_a_file_that_has_gone_is_no_longer_unread(tmp_path, monkeypatch, never):
    films = _folder(tmp_path, "Films", ["a.mkv", "stuck.mkv"])
    idx = _indexer(RootSet.build([{"path": str(films)}]))
    _silent_files(monkeypatch, never, lambda name: name == "stuck.mkv")
    await idx.initial_scan()
    assert len(idx.unread) == 1

    (films / "stuck.mkv").unlink()
    await idx.reconcile()

    assert idx.unread == {}


# ── What counts as silence ───────────────────────────────────────────────────

async def test_a_slow_file_that_keeps_arriving_is_not_given_up_on(tmp_path, monkeypatch):
    """Twice the limit in all, never the limit without a megabyte."""
    films = _folder(tmp_path, "Films", ["slow.mkv"])
    root = RootSet.build([{"path": str(films)}])
    idx = _indexer(root)
    real = indexer_mod._scan_file

    def trickle(root, path):
        for _ in range(6):
            threading.Event().wait(0.1)
            indexer_mod._beat()
        return real(root, path)

    monkeypatch.setattr(indexer_mod, "_scan_file", trickle)
    await idx.initial_scan()

    assert _names(idx) == {"slow.mkv"}
    assert idx._executor.left_behind == 0


async def test_files_waiting_behind_a_slow_one_are_not_timed_for_the_wait(
        tmp_path, monkeypatch):
    """A season dropped into a watched folder is hashed one file at a time on
    the one thread, each event its own task. Timed from when it was asked, the
    last of them would be given up on for having queued."""
    names = [f"e{i}.mkv" for i in range(5)]
    films = _folder(tmp_path, "Films", names)
    roots = RootSet.build([{"path": str(films)}])
    idx = _indexer(roots)
    real = indexer_mod._scan_file

    def unhurried(root, path):
        threading.Event().wait(0.15)
        return real(root, path)

    monkeypatch.setattr(indexer_mod, "_scan_file", unhurried)
    entries = await asyncio.gather(*(
        idx._hash_or_cached(roots.roots[0], films / name) for name in names))

    assert all(entries), "0.75 s of queue against a limit of 0.3 s"
    assert idx.unread == {} and idx._executor.left_behind == 0


# ── Three in a row ───────────────────────────────────────────────────────────

async def test_three_in_a_row_pause_the_root_and_the_next_root_is_indexed(
        tmp_path, monkeypatch, never):
    """A minute a file is weeks on a music library, with a thread left behind
    each time. After three with none read in between it is the disk."""
    sick = _folder(tmp_path, "Sick", [f"s{i}.mkv" for i in range(6)])
    well = _folder(tmp_path, "Well", ["w0.mkv", "w1.mkv"])
    roots = RootSet.build([{"path": str(sick)}, {"path": str(well)}])
    told = []

    async def on_change(indexer):
        told.append([r["paused"] for r in indexer.index.roots])

    idx = _indexer(roots, on_change=on_change)
    read = _silent_files(monkeypatch, never, lambda name: name.startswith("s"))

    await asyncio.wait_for(idx.initial_scan(), timeout=10)

    assert sum(name.startswith("s") for name in read) == 3, read
    assert roots.by_name("Sick").paused is True
    assert Path(idx.stalled["Sick"]).parent == sick
    assert _names(idx) == {"w0.mkv", "w1.mkv"}, "the root behind it waited for ever"
    assert told and told[0] == [True, False], "members are told the root is paused"
    assert idx.progress.scanning is False and idx.progress.queued == []


async def test_a_file_read_in_between_starts_the_count_again(tmp_path, monkeypatch, never):
    names = ["s0", "s1", "good", "s2", "s3", "s4"]
    films = _folder(tmp_path, "Films", [n + ".mkv" for n in names])
    roots = RootSet.build([{"path": str(films)}])
    root = roots.roots[0]
    idx = _indexer(roots)
    _silent_files(monkeypatch, never, lambda name: name.startswith("s"))

    for name in names[:5]:
        await idx._hash_or_cached(root, films / f"{name}.mkv")
    assert root.paused is False, "two, one read, two: never three in a row"

    await idx._hash_or_cached(root, films / "s4.mkv")
    assert root.paused is True


async def test_resume_gets_past_the_files_that_gave_no_answer(tmp_path, monkeypatch, never):
    films = _folder(tmp_path, "Films", ["s0.mkv", "s1.mkv", "s2.mkv"])
    roots = RootSet.build([{"path": str(films)}])
    idx = _indexer(roots)
    read = _silent_files(monkeypatch, never, lambda name: name.startswith("s"))
    await idx.initial_scan()
    assert roots.roots[0].paused is True
    (films / "new.mkv").write_bytes(b"dropped in while it was paused")

    await idx.resume_root("Films")
    await asyncio.gather(*idx._scan_tasks)

    assert _names(idx) == {"new.mkv"}
    assert len(read) == 4, f"read again what gave no answer: {read}"
    assert roots.roots[0].paused is False and idx.stalled == {}
    assert len(idx.unread) == 3, "still listed for the operator"


async def test_a_reload_does_not_start_a_stalled_root_again(tmp_path, monkeypatch, never):
    """A reload builds the set from the roster, which holds the operator's
    pauses. This one is the node's own."""
    films = _folder(tmp_path, "Films", ["s0.mkv", "s1.mkv", "s2.mkv"])
    idx = _indexer(RootSet.build([{"path": str(films)}]))
    _silent_files(monkeypatch, never, lambda name: True)
    await idx.initial_scan()

    again = RootSet.build([{"path": str(films)}])
    await idx.retarget(again)

    assert again.roots[0].paused is True


def _state(roots: RootSet, roster: Roster, idx: DirectoryIndexer) -> dict:
    index = GroupIndex(group_id=GID, sk_node=Ed25519PrivateKey.generate())
    return {
        "groups_ctx": {GID: {"index": index, "roots": roots, "gek": None}},
        "indexers": {GID: idx},
        "config": SimpleNamespace(groups=[SimpleNamespace(id=GID, roots=[])]),
        "roster": roster,
        "node_user_id": "operator",
    }


async def test_the_operators_resume_reaches_a_root_the_node_paused(
        tmp_path, monkeypatch, never):
    """Paused on the indexer's set and nowhere in the roster; and after a
    reload the set the operations look at may be another object."""
    films = _folder(tmp_path, "Films", ["s0.mkv", "s1.mkv", "s2.mkv"])
    idx = _indexer(RootSet.build([{"path": str(films)}]))
    _silent_files(monkeypatch, never, lambda name: True)
    await idx.initial_scan()
    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()
    try:
        seen_by_ops = RootSet.build([{"path": str(films)}])
        out = await ops.resume_root(_state(seen_by_ops, roster, idx), GID, "Films")
        await asyncio.gather(*idx._scan_tasks)

        assert out["status"] == "resumed"
        assert idx.roots.roots[0].paused is False and idx.stalled == {}
    finally:
        await roster.close()


# ── The listing ──────────────────────────────────────────────────────────────

async def test_a_listing_that_stops_answering_pauses_the_root(tmp_path, monkeypatch, never):
    sick = _folder(tmp_path, "Sick", ["a.mkv"])
    well = _folder(tmp_path, "Well", ["w.mkv"])
    roots = RootSet.build([{"path": str(sick)}, {"path": str(well)}])
    idx = _indexer(roots)
    real = indexer_mod._walk_root

    def walk(root):
        if root.name == "Sick":
            indexer_mod._beat(sick / "deep" / "down")
            never.wait()
        return real(root)

    monkeypatch.setattr(indexer_mod, "_walk_root", walk)
    await asyncio.wait_for(idx.initial_scan(), timeout=10)

    assert roots.by_name("Sick").paused is True
    assert idx.stalled == {"Sick": str(sick / "deep" / "down")}, "where it went silent"
    assert _names(idx) == {"w.mkv"}


async def test_a_rescan_that_gets_no_listing_keeps_what_the_root_held(
        tmp_path, monkeypatch, never):
    """A rescan drops the root's entries before it walks. A walk that learns
    nothing must not leave the root empty for everyone."""
    films = _folder(tmp_path, "Films", ["a.mkv", "b.mkv"])
    roots = RootSet.build([{"path": str(films)}])
    idx = _indexer(roots)
    await idx.initial_scan()
    for entry in idx.index.entries:
        entry.display_title = "Kept"

    monkeypatch.setattr(indexer_mod, "_walk_root", lambda root: never.wait())
    await asyncio.wait_for(idx._rescan_root(roots.roots[0]), timeout=10)

    assert _names(idx) == {"a.mkv", "b.mkv"}
    assert all(e.display_title == "Kept" for e in idx.index.entries)
    assert idx.drain_rescanned_ids() == set(), "nothing was rebuilt, nothing to enrich again"


async def test_a_reconcile_that_gets_no_listing_pauses_and_removes_nothing(
        tmp_path, monkeypatch, never):
    films = _folder(tmp_path, "Films", ["a.mkv"])
    roots = RootSet.build([{"path": str(films)}])
    idx = _indexer(roots)
    await idx.initial_scan()

    monkeypatch.setattr(indexer_mod, "_walk_root", lambda root: never.wait())
    await asyncio.wait_for(idx.reconcile(), timeout=10)

    assert roots.roots[0].paused is True
    assert _names(idx) == {"a.mkv"}


def _listed_before(root_path: Path) -> set[Path]:
    """The walk this replaced."""
    return {p for p in root_path.rglob("*") if p.is_file()}


async def test_the_walk_lists_what_it_always_listed(tmp_path):
    root = _folder(tmp_path, "Root", ["top.mkv", ".hidden", "x.part"])
    (root / "a" / "b" / "c").mkdir(parents=True)
    (root / "a" / "one.mkv").write_bytes(b"1")
    (root / "a" / "b" / "c" / "deep.mkv").write_bytes(b"2")
    (root / "empty").mkdir()
    outside = _folder(tmp_path, "Outside", ["o.mkv"])
    try:
        os.symlink(outside / "o.mkv", root / "link_to_file.mkv")
        os.symlink(outside, root / "link_to_dir", target_is_directory=True)
    except OSError:
        pass                    # no right to make links here: the rest still holds
    listed = indexer_mod._walk_root(RootSet.build([{"path": str(root)}]).roots[0])

    assert set(listed) == _listed_before(root)
    assert len(listed) == len(set(listed))


async def test_a_directory_that_cannot_be_listed_costs_only_itself(tmp_path, monkeypatch):
    """One refusal used to be silent, and any other error ended the walk of the
    whole root: "Cannot scan root", and nothing indexed on it at all."""
    root = _folder(tmp_path, "Root", ["top.mkv"])
    for sub in ("ok", "broken"):
        (root / sub).mkdir()
        (root / sub / f"{sub}.mkv").write_bytes(sub.encode())
    real = os.scandir

    def scandir(path):
        if Path(path).name == "broken":
            raise OSError(5, "Input/output error")
        return real(path)

    monkeypatch.setattr(indexer_mod.os, "scandir", scandir)
    idx = _indexer(RootSet.build([{"path": str(root)}]))
    await idx.initial_scan()

    assert _names(idx) == {"top.mkv", "ok.mkv"}


# ── Said, and survivable ─────────────────────────────────────────────────────

async def test_the_operator_is_told_which_file_the_disk_is_silent_on(
        tmp_path, monkeypatch, never):
    monkeypatch.setattr(DirectoryIndexer, "STALL_SECS", 30)
    films = _folder(tmp_path, "Films", ["stuck.mkv"])
    idx = _indexer(RootSet.build([{"path": str(films)}]))
    _silent_files(monkeypatch, never, lambda name: True)
    assert idx.disk_wait() is None

    scan = asyncio.create_task(idx.initial_scan())
    try:
        for _ in range(200):
            if idx.disk_wait():
                break
            await asyncio.sleep(0.02)
        path, seconds = idx.disk_wait()
        assert Path(path) == films / "stuck.mkv" and seconds >= 0.1
    finally:
        never.set()
        await asyncio.wait_for(scan, timeout=10)
    assert idx.disk_wait() is None


async def test_the_node_can_stop_with_a_read_still_stuck(tmp_path, monkeypatch, never):
    """The interpreter joins the threads of a pool on its way out. One read
    that never returned kept the node from stopping at all."""
    films = _folder(tmp_path, "Films", ["stuck.mkv"])
    idx = _indexer(RootSet.build([{"path": str(films)}]))
    _silent_files(monkeypatch, never, lambda name: True)
    await idx.initial_scan()
    assert idx._executor.left_behind == 1

    await asyncio.wait_for(idx.stop(), timeout=5)

    stuck = [t for t in threading.enumerate() if t.name.startswith("indexer_")]
    assert stuck and all(t.daemon for t in stuck)


async def test_a_scan_asks_the_disk_nothing_from_the_event_loop(tmp_path, monkeypatch):
    """The hash was off the loop and the stat before it was not. On a disk that
    stops answering that stat is where the loop would wait, and with it every
    connection of the node: no limit on the indexer's thread helps a call made
    somewhere else."""
    films = _folder(tmp_path, "Films", ["a.mkv", "b.mkv"])
    (films / "sub").mkdir()
    (films / "sub" / "c.mkv").write_bytes(b"c")
    idx = _indexer(RootSet.build([{"path": str(films)}]))
    asked: set[str] = set()
    real = Path.stat

    def stat(self, **kw):
        if films in self.parents:
            asked.add(threading.current_thread().name.split("_")[0])
        return real(self, **kw)

    monkeypatch.setattr(Path, "stat", stat)
    await idx.initial_scan()
    (films / "new.mkv").write_bytes(b"missed by the watcher")
    await idx.reconcile()

    assert _names(idx) == {"a.mkv", "b.mkv", "c.mkv", "new.mkv"}
    assert asked == {"indexer"}, f"a stat ran on: {asked}"