"""
Directory indexer — watches a group's roots and maintains a GroupIndex.

Uses watchdog for filesystem events. On any change the affected file is
re-scanned and the GroupIndex is updated. File metadata (blake3 hash, size,
type) is computed on first scan; hashing runs in a thread pool.

Two properties are worth stating because they are what the code is shaped
around, not incidental:

**A root that goes away freezes; it never empties.** Unmounting a volume either
makes watchdog emit a deletion for every file under it or presents an empty
directory to the next scan. Both would propagate deletions for a whole library
as though the owner had erased it, to every member. So a deletion is acted on
only once the root it belongs to has been confirmed still readable, and a root
that is not is marked unavailable with its entries left exactly where they are.

**Events are not trusted to be complete.** `ReadDirectoryChangesW` drops events
when its buffer overflows under a burst, and inotify on a FUSE mount misses
changes made outside it. Most users are on Windows sharing from exFAT, so both
apply. A periodic reconciliation scan is therefore not a belt-and-braces extra;
it is the only thing that recovers a missed event.
"""

import asyncio
import logging
import time
from collections.abc import Awaitable, Callable
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from dataclasses import field as dataclass_field
from pathlib import Path

import blake3
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.background import spawn
from meshbay_common.paths import find_fold_collisions, fold, long_path
from meshbay_common.protocol import IndexEntry
from watchdog.events import FileSystemEvent, FileSystemEventHandler
from watchdog.observers import Observer

from meshbay_node.indexer.cache import IndexCache
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.roots import Root, RootSet, off_disk

log = logging.getLogger(__name__)

# File types we include in the index (skip hidden files, temp files, etc.)
EXCLUDED_PREFIXES = (".", "~", "#")
EXCLUDED_SUFFIXES = (".tmp", ".part", ".crdownload", ".download")

MEDIA_EXTENSIONS = {
    "video": {".mp4", ".mkv", ".avi", ".mov", ".wmv", ".flv", ".webm", ".m4v"},
    # .wma and .mpc are real audio, tagged the same as anything else here
    # (enrich_audio.py reads both), but neither one has native decode
    # support in any mainstream browser's <audio> element — they show up,
    # get metadata, and fail to play in-browser until/unless server-side
    # transcoding is added (same gap video already has for HEVC).
    "audio": {".mp3", ".flac", ".ogg", ".wav", ".aac", ".m4a", ".opus", ".wma", ".mpc"},
    "image": {".jpg", ".jpeg", ".png", ".gif", ".webp", ".svg", ".bmp", ".tiff"},
    "document": {".pdf", ".epub", ".mobi", ".txt", ".md", ".docx", ".odt"},
    "archive": {".zip", ".tar", ".gz", ".bz2", ".xz", ".7z", ".rar"},
}


def _detect_type(path: Path) -> str:
    suffix = path.suffix.lower()
    for ftype, exts in MEDIA_EXTENSIONS.items():
        if suffix in exts:
            return ftype
    return "other"


def _is_indexable(path: Path) -> bool:
    if not path.is_file():
        return False
    name = path.name
    return (
        not any(name.startswith(p) for p in EXCLUDED_PREFIXES)
        and not any(name.endswith(s) for s in EXCLUDED_SUFFIXES)
    )


# Found live: a 1256-byte ".mp3" with no audio stream at all, just an ID3
# tag — a truncated/corrupted rip, sitting between two good tracks of the
# same album (docs/MESHBAY_DESIGN.md §9.8). A source this small claiming to
# be audio
# is far more likely broken than real, so it is skipped before ever being
# hashed rather than indexed and left to fail at playback time. Scoped to
# audio only — a tiny real file of any other type is still worth indexing.
MIN_AUDIO_SIZE_BYTES = 50 * 1024


def _is_indexable_size(path: Path, size: int) -> bool:
    return not (_detect_type(path) == "audio" and size < MIN_AUDIO_SIZE_BYTES)


_HASH_CHUNK = 8 * 1024 * 1024   # 8 MB streaming hash chunks

_PARTIAL_THRESHOLD = 40 * 1024 * 1024   # files above this use partial-read hashing
_PARTIAL_HEAD      = 20 * 1024 * 1024
_PARTIAL_TAIL      = 20 * 1024 * 1024
_PARTIAL_MID       =  5 * 1024 * 1024


def _feed(hasher, f, nbytes: int) -> None:
    remaining = nbytes
    while remaining > 0:
        chunk = f.read(min(_HASH_CHUNK, remaining))
        if not chunk:
            break
        hasher.update(chunk)
        remaining -= len(chunk)


def _partial_hash(file_path: Path, size: int) -> str:
    hasher = blake3.blake3()
    with open(long_path(file_path), "rb") as f:
        _feed(hasher, f, _PARTIAL_HEAD)
        f.seek(size - _PARTIAL_TAIL)
        _feed(hasher, f, _PARTIAL_TAIL)
        f.seek(size // 2)
        _feed(hasher, f, _PARTIAL_MID)
    return hasher.hexdigest()


@dataclass
class IndexProgress:
    """
    A snapshot of "is this indexer mid-scan right now", for the status shown
    to the operator (Create Group wizard, adding a directory) and pushed to
    connected members (a presence dot, never anything more specific — see
    daemon.py/webrtc_server.py). Reset per _scan_root() call rather than
    accumulated across a group's roots: the consumers that matter always
    watch exactly one root being scanned.

    Mutated only from the asyncio loop thread (the hashing itself runs in an
    executor thread, but never touches this), so no lock is needed.
    """
    scanning:      bool = False
    scanned_bytes: int  = 0
    total_bytes:   int  = 0
    # Basename only, deliberately not the full path — enough to show progress
    # without broadcasting the operator's directory structure.
    current_dir:   str  = ""
    # What the operator's progress band names. `root` and `queued` are root
    # names, so they stay on the loopback API like `current_dir`; members get
    # `root_pos`, a position in the roots table they already opened from the
    # sealed index, and the length of the queue.
    root:          str  = ""
    root_pos:      int  = -1
    # "scan" (a root walked for the first time), "rescan" (one that came back),
    # "reconcile" (files the watcher missed), "watch" (a burst of events), or
    # "" when idle.
    kind:          str  = ""
    files_done:    int  = 0
    files_total:   int  = 0
    # Roots waiting for the scan lock, in the order they will be walked.
    queued:        list[str] = dataclass_field(default_factory=list)


def _virtual_dir(root: Root, file_path: Path) -> str:
    """
    The directory a file appears in, as members see it: `"Films/2024"`.

    The root name is the first segment for every root, including the only one of
    a single-root group — one path shape has to be got right once, two have to
    be kept right forever.
    """
    rel = file_path.parent.relative_to(root.path)
    return root.name if str(rel) == "." else f"{root.name}/{rel.as_posix()}"


def _walk_root(root: Root) -> list[Path]:
    """
    Blocking directory walk — always run via an executor, never awaited
    directly in the asyncio loop. A tree of tens of thousands of files (or
    one on a slow network share) can take seconds; run inline, that stalls
    every other thing the daemon is doing — WebRTC sessions, chat, the admin
    UI — for as long as it takes.
    """
    return [p for p in root.path.rglob("*") if p.is_file()]


def _size_files(files: list[Path]) -> list[tuple[Path, int]]:
    """
    Blocking: stat() every file from an already-completed walk — run via an
    executor for the same reason `_walk_root` is (its own docstring above).
    Previously a plain 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, every WebRTC session and
    the admin UI included, for as long as the stat() calls took — and did so *before*
    `_scan_root` had even set `progress.scanning`, so a consumer polling it
    saw "not scanning" the whole time real, blocking work was happening.
    """
    sized: list[tuple[Path, int]] = []
    for p in files:
        try:
            sized.append((p, p.stat().st_size))
        except OSError:
            continue
    return sized


def _scan_file(root: Root, file_path: Path) -> IndexEntry | None:
    """Compute IndexEntry for a file. Blocking — run in executor.
    Files <= 40 MB are hashed in full (hash_version 1). Files > 40 MB use a
    45 MB partial read — first 20 MB, last 20 MB, 5 MB at 50% — for
    hash_version 2."""
    if not _is_indexable(file_path):
        return None
    try:
        stat = file_path.stat()
        if not _is_indexable_size(file_path, stat.st_size):
            return None
        if stat.st_size > _PARTIAL_THRESHOLD:
            hex_hash = _partial_hash(file_path, stat.st_size)
            hv = 2
        else:
            hasher = blake3.blake3()
            with open(long_path(file_path), "rb") as f:
                while chunk := f.read(_HASH_CHUNK):
                    hasher.update(chunk)
            hex_hash = hasher.hexdigest()
            hv = 1
        return IndexEntry(
            id=hex_hash,
            name=file_path.name,
            path=_virtual_dir(root, file_path),
            size=stat.st_size,
            type=_detect_type(file_path),
            added_at=int(stat.st_mtime),
            hash_version=hv,
        )
    except (OSError, PermissionError, ValueError) as e:
        log.warning("Cannot index %s: %s", file_path, e)
        return None


class DirectoryIndexer:
    """
    Watches a group's roots and keeps a GroupIndex up to date.

    Usage:
        indexer = DirectoryIndexer(
            roots=RootSet.build([{"path": "/home/user/shared", "upload": True}]),
            group_id="my-group",
            sk_node=sk,
            gek=gek_bytes,
            on_change=async_callback,
        )
        await indexer.start()
        # ... later
        await indexer.stop()
    """

    # How often to re-check which roots are readable and reconcile the index
    # against what is actually on disk. Not a poll for changes — a backstop for
    # the events the OS did not deliver, and the way a re-plugged drive is
    # noticed. Watchdog already covers the common case in real time, so this
    # does not need to run often to do its job; it backs off further still
    # (see _reconcile_loop) when nothing has changed for a while, and a
    # per-group operator setting (roster.py SETTING_RECONCILE_INTERVAL) can
    # override the starting point.
    DEFAULT_RECONCILE_SECS = 600.0    # 10 min
    RECONCILE_BACKOFF_CAP  = 7200.0   # 2 h — never sleeps longer than this
    # How long to wait after the last event on a given path before acting on
    # it — several writes to the same file in quick succession (a slow copy
    # in several passes) collapse into one hash instead of one per write.
    DEFAULT_DEBOUNCE_SECS = 2.0

    def __init__(
        self,
        roots: RootSet,
        group_id: str,
        sk_node: Ed25519PrivateKey,
        gek: bytes | None,
        on_change: Callable[["DirectoryIndexer"], Awaitable[None]] | None = None,
        cache: IndexCache | None = None,
        reconcile_secs: float = DEFAULT_RECONCILE_SECS,
        debounce_secs: float = DEFAULT_DEBOUNCE_SECS,
        on_root_ejected: Callable[[str, bool], Awaitable[None]] | None = None,
    ):
        self.roots    = roots
        self.group_id = group_id
        self.sk_node  = sk_node
        self.gek      = gek
        self.on_change = on_change
        # Called with (root_name, ejected) whenever this indexer changes a
        # root's ejected state by itself — the surprise-unplug safety net.
        # The daemon writes it to the roster, so a restart does not undo it.
        self.on_root_ejected = on_root_ejected
        self.reconcile_secs = reconcile_secs
        self.debounce_secs = debounce_secs
        # Current backoff delay — starts at reconcile_secs, doubles on every
        # tick that finds nothing changed (up to RECONCILE_BACKOFF_CAP), and
        # resets the moment something real happens (a change, or a peer
        # connecting — see note_activity()).
        self._reconcile_delay = reconcile_secs
        # Path -> (size, mtime, hash) accelerator, so a restart does not have
        # to re-read a file it already hashed last time (see cache.py). None
        # in tests that do not care about it — every hash is then a miss.
        self._cache = cache

        self._index    = GroupIndex(group_id=group_id, sk_node=sk_node, gek=gek)
        self._index.roots = roots.describe()
        # One worker, deliberately, not a real pool: hashing two files at once
        # buys nothing here and can cost a lot. It only offloads the blocking
        # read+hash off the asyncio loop; it was never used for concurrency —
        # every call site awaits one run_in_executor before starting the next
        # (see _hash_or_cached below) — and measured on a spinning USB drive,
        # two interleaved multi-GB reads would seek-thrash against each other
        # rather than go faster. Left at 1 so the number does not promise a
        # concurrency this code never provided.
        self._executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="indexer")
        self._observer: Observer | None = None
        self._loop: asyncio.AbstractEventLoop | None = None
        self._reconciler: asyncio.Task | None = None
        self._pending_timers: dict[str, asyncio.TimerHandle] = {}
        self.progress = IndexProgress()
        # Real-time watchdog adds (_debounce/_update_entry below) previously
        # never touched `progress` at all — a whole season dropped into an
        # already-watched folder gave the operator no scanning indicator and
        # no progress bar, files just appeared one at a time with no feedback
        # (found live). `_burst_inflight` counts files currently scheduled or
        # being hashed in the current burst; `scanning` only drops back to
        # False once it reaches zero *and* no more timers are pending — a
        # debounce timer firing is not the same as the hash it schedules
        # having finished, and the whole point of a progress indicator is to
        # stay up for exactly as long as the slow part (hashing) is running.
        self._burst_inflight = 0
        self._burst_sizes: dict[str, int] = {}
        # A burst keeps its own counters and `progress` shows them only while
        # no whole-root job runs. They used to write `progress` directly, and a
        # file dropped into a folder during a 900 GB scan added its size to the
        # scan's total, then cleared `scanning` when its own hash finished —
        # the bar went away with hours of hashing left.
        self._job_running = False
        self._burst_scanned = 0
        self._burst_total = 0
        self._burst_files_done = 0
        self._burst_files_total = 0
        self._burst_dir = ""
        # Ids whose entry this indexer threw away and rebuilt from disk, since
        # the last time a consumer drained this. A rebuilt entry carries only
        # what `_hash_or_cached` fills in — every enrichment field the Videos,
        # Music and Photos apps put there is gone — but its id is the file's
        # content hash, so a diff against the last broadcast sees no addition
        # and no deletion and nothing downstream can tell the fields were
        # wiped. See `_drop_root_entries`.
        self.rescanned_ids: set[str] = set()
        # Held by every whole-root walk — the initial scan, a scan of a root a
        # retarget added, and reconcile. Reconcile compares disk against the
        # index, so while a scan is part-way through a root every file it has
        # not reached yet looks like a missed event: on a 900 GB drive that was
        # thousands of "appeared" lines, the same root hashed twice over the
        # one executor thread, and `progress` rewritten under the scan's feet.
        self._scan_lock = asyncio.Lock()
        self._scan_tasks: set[asyncio.Task] = set()

    @property
    def index(self) -> GroupIndex:
        return self._index

    # ── Root lookup ───────────────────────────────────────────────────────────

    def _root_for(self, file_path: Path) -> Root | None:
        """Which root a real path belongs to, longest match first."""
        try:
            resolved = file_path.resolve()
        except OSError:
            resolved = file_path
        best: Root | None = None
        for root in self.roots:
            try:
                resolved.relative_to(root.path)
            except ValueError:
                continue
            if best is None or len(root.path.parts) > len(best.path.parts):
                best = root
        return best

    # ── Initial scan ──────────────────────────────────────────────────────────

    async def initial_scan(self) -> None:
        """Scan every available root. Run once at startup."""
        async with self._scan_lock:
            await self._initial_scan()

    async def _initial_scan(self) -> None:
        await off_disk(self.roots, self.roots.refresh_availability)
        total = 0
        waiting = [r.name for r in self.roots if r.available]
        self._queue(waiting)
        try:
            for root in self.roots:
                if not root.available:
                    log.warning("Root %r is not readable at startup (%s) — its files "
                                "are not indexed yet and will appear when it returns",
                                root.name, root.path)
                    continue
                waiting.remove(root.name)
                self._unqueue([root.name])
                total += await self._scan_root(root)
        finally:
            self._unqueue(waiting)
        self._index.version = int(time.time())
        self._index.roots = self.roots.describe()
        self._report_collisions()
        log.info("Initial scan complete: %d files across %d root(s)",
                 total, len(self.roots))

    async def _scan_root(self, root: Root, *, kind: str = "scan") -> int:
        log.info("Scanning %s (root %r) ...", root.path, root.name)
        count = 0
        loop = asyncio.get_event_loop()

        # `scanning` flips on here, before the walk — not after it and the
        # sizing pass, which for a large or slow (network/USB) root can
        # themselves take a long time despite running off-loop. A consumer
        # polling IndexProgress (index-status; the Create Group wizard's
        # own progress bar) must see "scanning" the moment real work starts,
        # not only once the file list and its total size are known. Found
        # live (2026-08-25): the wizard's post-creation "add extra roots"
        # step gave up waiting after a short grace period because the flag
        # had not yet turned on, even though the node was already several
        # seconds into walking and sizing a large root — total_bytes is
        # unknown at this point, so it starts at 0 and is corrected below.
        self._begin_job(root.name, root=root, kind=kind)
        try:
            try:
                files = await loop.run_in_executor(self._executor, _walk_root, root)
            except OSError as e:
                log.warning("Cannot scan root %r: %s", root.name, e)
                return 0

            # Sizes up front, off the same listing that already walked the
            # tree — the progress bar's denominator, not a second pass over
            # the disk. Off-loop for the same reason the walk itself is
            # (_size_files's own docstring: this used to be a synchronous
            # loop right here, blocking the whole daemon for a large root).
            sized = await loop.run_in_executor(self._executor, _size_files, files)

            self.progress.total_bytes = sum(size for _, size in sized)
            self.progress.files_total = len(sized)
            self.progress.current_dir = ""
            for file_path, size in sized:
                self.progress.current_dir = file_path.parent.name
                entry = await self._hash_or_cached(root, file_path)
                self.progress.scanned_bytes += size
                self.progress.files_done += 1
                if entry:
                    self._index.add_entry(entry)
                    count += 1
        finally:
            # Must run even if a hash/IO error propagates out of the loop
            # above — an indexing state that never turns back off is worse
            # than the scan itself failing.
            self._end_job()
        return count

    def _begin_job(self, current_dir: str, total_bytes: int = 0, *,
                   root: Root, kind: str, files_total: int = 0) -> None:
        """`progress` now describes a whole-root walk, whatever a burst is doing."""
        self._job_running = True
        p = self.progress
        p.scanning = True
        p.scanned_bytes = 0
        p.total_bytes = total_bytes
        p.files_done = 0
        p.files_total = files_total
        p.current_dir = current_dir
        p.kind = kind
        p.root = root.name
        p.root_pos = next((i for i, r in enumerate(self.roots)
                           if r.folded == root.folded), -1)

    def _end_job(self) -> None:
        self._job_running = False
        p = self.progress
        p.scanning = False
        p.current_dir = ""
        p.kind = ""
        p.root = ""
        p.root_pos = -1
        if self._burst_inflight > 0:
            # A burst that started during the walk is still hashing.
            self._show_burst()

    def _show_burst(self) -> None:
        if self._job_running:
            return
        p = self.progress
        if self._burst_inflight > 0:
            p.scanning = True
            p.current_dir = self._burst_dir
            p.kind = "watch"
            p.root = ""
            p.root_pos = -1
        elif not self._pending_timers:
            p.scanning = False
            p.current_dir = ""
            p.kind = ""
        p.scanned_bytes = self._burst_scanned
        p.total_bytes = self._burst_total
        p.files_done = self._burst_files_done
        p.files_total = self._burst_files_total

    def _queue(self, names: list[str]) -> None:
        self.progress.queued.extend(names)

    def _unqueue(self, names: list[str]) -> None:
        for name in names:
            if name in self.progress.queued:
                self.progress.queued.remove(name)

    async def _hash_or_cached(self, root: Root, file_path: Path) -> IndexEntry | None:
        """
        Cache-aware replacement for a bare _scan_file() call: skips the
        content read entirely when this path's (size, mtime, hash_version)
        still match what was hashed last time.
        """
        if not _is_indexable(file_path):
            return None
        try:
            st = file_path.stat()
        except OSError:
            return None
        if not _is_indexable_size(file_path, st.st_size):
            return None

        expected_hv = 2 if st.st_size > _PARTIAL_THRESHOLD else 1

        if self._cache is not None:
            cached = await self._cache.lookup(
                str(file_path), st.st_size, st.st_mtime, expected_hv)
            if cached is not None:
                return await self._attribute(IndexEntry(
                    id=cached.hash,
                    name=file_path.name,
                    path=_virtual_dir(root, file_path),
                    size=st.st_size,
                    type=cached.type,
                    added_at=cached.added_at,
                    hash_version=cached.hash_version,
                ), file_path, st)

        loop = asyncio.get_event_loop()
        entry = await loop.run_in_executor(self._executor, _scan_file, root, file_path)
        if entry and self._cache is not None:
            await self._cache.put(str(file_path), st.st_size, st.st_mtime,
                                   entry.id, entry.type, entry.added_at,
                                   entry.hash_version)
        return await self._attribute(entry, file_path, st)

    async def _attribute(self, entry: IndexEntry | None, file_path: Path,
                         st) -> IndexEntry | None:
        """Stamp an entry with whoever sent the file, if a member did.

        Here, rather than beside each `add_entry`, because this is the one
        funnel every entry passes through: the initial scan, the watchdog,
        reconcile and a replug all build theirs from `_hash_or_cached`.

        The attribution used to be written at the end of the *upload* instead,
        by walking the index for an entry that by construction did not exist
        yet — the watchdog has not fired, and the `.part` the file was until the
        rename is not indexable. It matched nothing, silently, so every uploaded
        file was owned by nobody and `file_delete` refused everyone but the
        operator, where MESHBAY_DESIGN.md §5.4 grants it to any non-revoked
        device of the uploading account.
        """
        if entry is None or self._cache is None:
            return entry
        who = await self._cache.uploader(str(file_path), st.st_size, st.st_mtime)
        if who is not None:
            entry.uploader_id, entry.uploader_pk = who
        return entry

    async def record_upload(self, file_path: Path, user_id: str,
                            pk_ed25519: str) -> None:
        """Remember who sent this file, for the entry that does not exist yet.

        Called by the transport once the last chunk has landed and the file is
        at its final name. Durable rather than in-memory: the index is rebuilt
        from disk at every start, and an owner the node forgets on restart is an
        owner who cannot delete their own file tomorrow.
        """
        if self._cache is None or not user_id:
            return
        try:
            st = file_path.stat()
        except OSError:
            # Gone between the rename and here. Nothing to attribute, and
            # nothing for anyone to delete either.
            return
        await self._cache.record_upload(
            str(file_path), st.st_size, st.st_mtime, user_id, pk_ed25519 or "")

    def _report_collisions(self) -> None:
        """
        Names that are the same file on a case-insensitive filesystem.

        Reported, never resolved: on ext4 both files exist and only the operator
        knows which was meant. Left silent, the pair reaches somebody on Windows
        who can save one of them.
        """
        by_dir: dict[str, list[str]] = {}
        for entry in self._index.entries:
            by_dir.setdefault(fold(entry.path), []).append(entry.name)
        for folded_dir, names in by_dir.items():
            for _, clashing in find_fold_collisions(names).items():
                log.warning(
                    "Names that differ only by case or accent form in %s: %s — "
                    "these are one file on NTFS or exFAT, and a member on Windows "
                    "can only keep one of them",
                    folded_dir or "/", ", ".join(sorted(clashing)))

    # ── Watchdog integration ──────────────────────────────────────────────────

    async def start(self, *, defer_scan: bool = False) -> None:
        """Start initial scan + filesystem watcher + reconciler.

        With ``defer_scan=True`` the watcher and reconciler start
        immediately but the initial scan is skipped — call
        :meth:`initial_scan` yourself when ready.
        """
        self._loop = asyncio.get_event_loop()
        if not defer_scan:
            await self.initial_scan()
        self._start_observer()
        self._reconciler = asyncio.create_task(self._reconcile_loop())

    def _start_observer(self) -> None:
        handler = _WatchdogHandler(self)
        self._observer = Observer()
        watched = 0
        for root in self.roots:
            if not root.available:
                continue
            try:
                self._observer.schedule(handler, str(root.path), recursive=True)
                watched += 1
            except OSError as e:
                log.warning("Cannot watch root %r: %s", root.name, e)
        self._observer.start()
        log.info("Watching %d root(s) for changes", watched)

    async def stop(self) -> None:
        """Stop the filesystem watcher and the reconciler."""
        if self._reconciler:
            self._reconciler.cancel()
            try:
                await self._reconciler
            except asyncio.CancelledError:
                pass
            self._reconciler = None
        scans = list(self._scan_tasks)
        for task in scans:
            task.cancel()
        await asyncio.gather(*scans, return_exceptions=True)
        for handle in self._pending_timers.values():
            handle.cancel()
        self._pending_timers.clear()
        if self._observer:
            self._observer.stop()
            self._observer.join()
            self._observer = None
        self._executor.shutdown(wait=False)
        log.info("Indexer stopped")

    async def retarget(self, roots: RootSet, *, wait: bool = True) -> None:
        """
        Point this indexer at a new set of roots, without a restart (14.8).

        Entries under a root that is gone from the config are dropped — the
        operator removed it deliberately, which is not the same event as a
        volume disappearing, and conflating the two is what
        docs/MESHBAY_DESIGN.md §6.2 exists to prevent. Roots that survive
        keep their entries; new ones are scanned.

        The set takes effect before anything is scanned: the roots table, the
        watcher and `self.roots` all move at once. With ``wait=False`` the scan
        of the added roots runs in the background and this returns as soon as
        the set is in place — the daemon's reload must not sit on its lock for
        the hours a large drive takes to hash.
        """
        old_names = {r.folded for r in self.roots}
        new_names = {r.folded for r in roots}
        self.progress.queued[:] = [n for n in self.progress.queued if fold(n) in new_names]

        for root in self.roots:
            if root.folded not in new_names:
                dropped = self._entries_under(root)
                log.info("Root %r removed from the config — dropping %d entries",
                         root.name, len(dropped))
                for entry in dropped:
                    self._index.remove_entry(entry.id)

        self.roots = roots
        await off_disk(roots, roots.refresh_availability)
        added = [r for r in roots if r.folded not in old_names and r.available]

        self._index.roots = roots.describe()
        self._index.version = int(time.time())
        self._restart_observer()

        if not added:
            if self.on_change:
                await self.on_change(self)
            return

        self._queue([r.name for r in added])
        task = asyncio.create_task(self._scan_added_roots(added))
        self._scan_tasks.add(task)
        task.add_done_callback(self._scan_tasks.discard)
        if wait:
            await task
        elif self.on_change:
            # The table now; the files when the scan ends.
            await self.on_change(self)

    def _holds(self, root: Root) -> bool:
        return any(r.folded == root.folded and r.path == root.path for r in self.roots)

    async def _scan_added_roots(self, added: list[Root]) -> None:
        waiting = [r.name for r in added]
        try:
            async with self._scan_lock:
                for root in added:
                    waiting.remove(root.name)
                    self._unqueue([root.name])
                    # A later retarget may have removed it while this waited.
                    if not self._holds(root):
                        continue
                    count = await self._scan_root(root)
                    if not self._holds(root):
                        # Removed while being scanned: that retarget's drop ran
                        # before these entries existed.
                        for entry in self._entries_under(root):
                            self._index.remove_entry(entry.id)
                        continue
                    log.info("Scan complete: %d files in root %r", count, root.name)
                self._index.version = int(time.time())
            if self.on_change:
                await self.on_change(self)
        except asyncio.CancelledError:
            raise
        except Exception:
            log.exception("Scanning the added root(s) failed")
        finally:
            self._unqueue(waiting)

    # ── Reconciliation ────────────────────────────────────────────────────────

    async def _reconcile_loop(self) -> None:
        while True:
            try:
                await asyncio.sleep(self._reconcile_delay)
                if self._scan_lock.locked() or self._scan_tasks:
                    # A scan is walking a root right now. Skipped without
                    # backing off: the next tick after it ends is the useful one.
                    continue
                changed = await self.reconcile()
                if changed:
                    self._reconcile_delay = self.reconcile_secs
                else:
                    self._reconcile_delay = min(
                        self._reconcile_delay * 2, self.RECONCILE_BACKOFF_CAP)
            except asyncio.CancelledError:
                raise
            except Exception:
                log.exception("Reconcile failed — continuing")

    def note_activity(self) -> None:
        """
        Called when something makes a prompt reconcile worth having again —
        today, a peer completing the handshake for this group
        (webrtc_server.py). Someone is looking, so the backstop should be at
        its normal cadence rather than however far backoff had stretched it.
        """
        self._reconcile_delay = self.reconcile_secs

    async def reconcile(self) -> bool:
        """
        Re-check availability, and rescan roots that came back.

        The only place a root's entries are dropped: when the root is readable
        and the files are genuinely gone. A root that is not readable is left
        untouched, which is the whole point.

        Returns whether anything actually changed — _reconcile_loop uses this
        to back off when a pass finds nothing to do, rather than running at
        the same cadence forever regardless of how quiet the root is.
        """
        async with self._scan_lock:
            return await self._reconcile()

    async def _reconcile(self) -> bool:
        changed = await off_disk(self.roots, self.roots.refresh_availability)
        touched = False

        # Drained before the loop below, because persisting the flag is what
        # makes the safety net survive a restart — and a restart is exactly
        # what an operator does after noticing a drive fell off.
        while self.roots.auto_ejected:
            name = self.roots.auto_ejected.pop(0)
            if self.on_root_ejected:
                try:
                    await self.on_root_ejected(name, True)
                except Exception:
                    log.exception("Could not persist the auto-eject of root %r", name)

        for root, available in changed:
            if available:
                log.info("Root %r is back — rescanning", root.name)
                await self._rescan_root(root)
                touched = True
            else:
                # Frozen: entries stay, marked unavailable to members through
                # the roots table in the index payload.
                log.warning("Root %r went away — %d entries frozen, not deleted",
                            root.name, len(self._entries_under(root)))
                touched = True

        if changed:
            self._restart_observer()

        if await self._sweep_available_roots():
            touched = True

        if touched:
            self._index.roots = self.roots.describe()
            self._index.version = int(time.time())
            if self.on_change:
                await self.on_change(self)

        return touched

    async def _sweep_available_roots(self) -> bool:
        """
        Catch what the watcher missed: files gone, and files never announced.

        Only touches roots that are readable right now — a root whose volume is
        absent has nothing to compare against, and comparing anyway is exactly
        the mistake this module exists to avoid.
        """
        changed = False
        loop = asyncio.get_event_loop()
        for root in self.roots:
            if not root.available:
                continue
            try:
                on_disk, known = await loop.run_in_executor(
                    self._executor, self._sweep_scan_root, root)
            except OSError as e:
                log.warning("Cannot reconcile root %r: %s", root.name, e)
                continue

            for missing in set(known) - on_disk:
                # Duplicate content is handled without a special case here: the
                # entry goes, and the add loop below re-indexes the surviving
                # copy under its own path, because the id is then absent. A
                # dedicated "find the survivor" lookup was written first and
                # deleted — it rehashed every file under the root on any single
                # deletion, and a test proved it changed nothing.
                self._index.remove_entry(known[missing])
                log.info("Reconcile: %s is gone", missing)
                changed = True

            added_paths = on_disk - set(known)
            if not added_paths:
                continue

            added_sized: list[tuple[Path, int]] = []
            for p in added_paths:
                try:
                    added_sized.append((p, p.stat().st_size))
                except OSError:
                    added_sized.append((p, 0))

            self._begin_job("", sum(size for _, size in added_sized), root=root,
                            kind="reconcile", files_total=len(added_sized))
            try:
                for added, size in added_sized:
                    self.progress.current_dir = added.parent.name
                    entry = await self._hash_or_cached(root, added)
                    self.progress.scanned_bytes += size
                    self.progress.files_done += 1
                    if not entry:
                        continue
                    # The index is keyed by **content hash**, so two identical
                    # files at two paths are one entry and the path comparison
                    # above cannot see the second. Adding it anyway rewrites
                    # that entry's path every cycle, bumps the version, and
                    # pushes an index update to every connected peer once a
                    # minute — for ever. Measured on a live node: `clip.mp4`
                    # present at the root and in uploads/ with the same bytes.
                    if self._index.get_entry(entry.id) is not None:
                        log.debug("Reconcile: %s duplicates content already "
                                  "indexed as %s — leaving the index alone",
                                  added, entry.id[:8])
                        continue
                    self._index.add_entry(entry)
                    log.info("Reconcile: %s appeared (missed event)", added)
                    changed = True
            finally:
                self._end_job()
        return changed

    def _entries_under(self, root: Root) -> list[IndexEntry]:
        prefix = fold(root.name)
        return [e for e in self._index.entries
                if fold(e.path).split("/", 1)[0] == prefix]

    # Everything on an IndexEntry that a scan does not produce. `_scan_root`
    # fills id/name/path/size/type/added_at/hash_version from the file itself;
    # every field below was derived by one of the enrichment passes and is
    # nowhere on disk to be read back.
    _ENRICHED_FIELDS = (
        "duration", "thumb_hash", "width", "height",
        "display_title", "season", "episode",
        "artist", "album", "track_no", "taken_at", "camera",
        "uploader_id", "uploader_pk",
    )

    async def _rescan_root(self, root: Root) -> int:
        """
        Rebuild one root's entries from disk, keeping what the files still say.

        The two callers — `reconcile` when a root reappears, `plug_root` when
        the operator plugs one back in — have to re-walk: the drive may have
        changed while it was away. What they must not do is throw away the
        enrichment. An entry's id is its content hash, so an entry that comes
        back under the same id, name and path is the same bytes in the same
        place, and every field the Videos, Music and Photos passes derived from
        it still holds. Re-deriving them means minutes of tag reads, ffprobe
        runs and rate-limited metadata lookups during which the operator's
        library sits empty — which is exactly what a replug looked like.

        Anything that does *not* match is left bare on purpose: a different id
        is different content, and a different name or path can change the
        filename and folder fallbacks that `display_title`, `track_no`,
        `artist` and `album` fall back to. Those are the entries
        `daemon._broadcast_index_change` re-enriches, off `rescanned_ids`.
        """
        carried = {(e.id, e.name, e.path): e for e in self._entries_under(root)}
        self._drop_root_entries(root)
        count = await self._scan_root(root, kind="rescan")
        for entry in self._entries_under(root):
            old = carried.get((entry.id, entry.name, entry.path))
            if old is None:
                continue
            for field in self._ENRICHED_FIELDS:
                # What the rescan itself established wins. `uploader_id` and
                # `uploader_pk` now come off the durable record (`_attribute`),
                # and the entry being replaced is memory this process happens
                # to still hold — so copying over them would let a stale blank
                # beat the thing that survives a restart. Every other field is
                # None on a freshly scanned entry, so for those this is exactly
                # the carry-over it has always been.
                if getattr(entry, field) is None:
                    setattr(entry, field, getattr(old, field))
            # It came back intact, so it is not one of the entries the daemon
            # needs to enrich again.
            self.rescanned_ids.discard(entry.id)
        return count

    def _drop_root_entries(self, root: Root) -> None:
        """
        Throw away a root's entries. Only ever called to rebuild them.

        The ids are recorded because nothing outside can otherwise tell they
        were rebuilt: no deletion is broadcast (the rescan is immediate) and
        the entries come back under the same content-hash ids, so a diff
        against the last broadcast reports neither an addition nor a deletion.
        `_rescan_root` clears the ones it managed to carry over intact; what is
        left is genuinely new to the apps and is re-enriched by the daemon.
        """
        for entry in self._entries_under(root):
            self.rescanned_ids.add(entry.id)
            self._index.remove_entry(entry.id)

    def drain_rescanned_ids(self) -> set[str]:
        """Take the ids rebuilt since the last call; leave the set empty."""
        drained, self.rescanned_ids = self.rescanned_ids, set()
        return drained

    @staticmethod
    def _entry_path(root: Root, entry: IndexEntry) -> Path | None:
        _, _, tail = entry.path.partition("/")
        try:
            return (root.path / tail / entry.name).resolve() if tail else \
                   (root.path / entry.name).resolve()
        except OSError:
            return None

    def _sweep_scan_root(self, root: Root) -> tuple[set[Path], dict[Path, str]]:
        """
        Blocking: the two disk-touching pieces of one reconcile pass for a
        root, bundled so both run together in the executor rather than in
        the asyncio loop — the tree walk, and resolving the real path of
        every entry already known under this root (one syscall each).
        """
        on_disk = {p.resolve() for p in root.path.rglob("*") if _is_indexable(p)}
        known: dict[Path, str] = {}
        for entry in self._entries_under(root):
            abs_path = self._entry_path(root, entry)
            if abs_path:
                known[abs_path] = entry.id
        return on_disk, known

    def _restart_observer(self) -> None:
        """Re-schedule watches after roots appeared or disappeared."""
        if self._observer:
            self._observer.stop()
            self._observer.join()
            self._observer = None
        self._start_observer()

    def eject_root(self, root_name: str) -> None:
        """Stop watching a root without touching its entries."""
        from meshbay_common.paths import fold
        target = fold(root_name)
        for root in self.roots:
            if fold(root.name) == target:
                root.ejected = True
                root.available = False
                frozen = len(self._entries_under(root))
                log.info("Root %r ejected — %d entries frozen", root.name, frozen)
                break
        self._restart_observer()
        self._index.roots = self.roots.describe()
        self._index.version = int(time.time())

    async def plug_root(self, root_name: str) -> None:
        """Restart watching a previously ejected root and reconcile."""
        from meshbay_common.paths import fold
        target = fold(root_name)
        root = None
        for r in self.roots:
            if fold(r.name) == target:
                root = r
                break
        if root is None:
            return
        root.ejected = False
        root.available = await off_disk(self.roots, root.is_live)
        if not root.available:
            await self._finish_plug(None)
            return
        log.info("Root %r plugged — rescanning", root.name)
        # `_rescan_root` drops the root's entries before it walks the disk, and
        # this is called from an admin op running in the operator's WebRTC
        # session — which cancels everything it started when it closes. So the
        # rescan runs in a task this indexer owns, and a caller that goes away
        # only stops waiting for it instead of leaving the root emptied.
        task = asyncio.create_task(self._finish_plug(root))
        self._scan_tasks.add(task)
        task.add_done_callback(self._scan_tasks.discard)
        await asyncio.shield(task)

    async def _finish_plug(self, root: Root | None) -> None:
        if root is not None:
            # A whole-root walk like any other, so it waits its turn: run beside
            # an added root's scan, the two reset each other's progress and read
            # the same drive in alternation.
            self._queue([root.name])
            waiting = True
            try:
                async with self._scan_lock:
                    self._unqueue([root.name])
                    waiting = False
                    # Ejected or removed again while it waited. `_rescan_root`
                    # drops the entries before it walks, so going ahead would
                    # empty a root that is not there to be read.
                    live = await off_disk(self.roots, root.is_live)
                    if self._holds(root) and not root.ejected and live:
                        await self._rescan_root(root)
            finally:
                if waiting:
                    self._unqueue([root.name])
        self._restart_observer()
        self._index.roots = self.roots.describe()
        self._index.version = int(time.time())
        if self.on_change:
            await self.on_change(self)

    # ── Internal update ───────────────────────────────────────────────────────

    def _schedule_update(self, file_path: Path, deleted: bool = False) -> None:
        """Called from watchdog thread — schedule debounced async update."""
        if not self._loop:
            return
        key = str(file_path)
        self._loop.call_soon_threadsafe(self._debounce, key, file_path, deleted)

    def _debounce(self, key: str, file_path: Path, deleted: bool) -> None:
        old = self._pending_timers.pop(key, None)
        if old:
            old.cancel()

        # Progress accounting for the real-time path — see `_burst_inflight`'s
        # docstring in __init__. Only on this key's *first* appearance in the
        # current burst: a rapid re-trigger of the same path (several writes
        # debounced together) cancels the old timer above and must not also
        # double-count its size, so `_burst_sizes` (keyed the same as
        # `_pending_timers`) is the source of truth for "already counted",
        # not "does a timer currently exist for it" — a cancelled timer's
        # bookkeeping still has to reach the `fire()` that actually runs.
        if not deleted and key not in self._burst_sizes:
            try:
                size = file_path.stat().st_size
            except OSError:
                size = 0
            if self._burst_inflight <= 0:
                self._burst_scanned = 0
                self._burst_total = 0
                self._burst_files_done = 0
                self._burst_files_total = 0
            self._burst_total += size
            self._burst_files_total += 1
            self._burst_dir = file_path.parent.name
            self._burst_sizes[key] = size
            self._burst_inflight += 1
            self._show_burst()

        def fire() -> None:
            self._pending_timers.pop(key, None)
            spawn(self._update_entry(file_path, deleted))

        self._pending_timers[key] = self._loop.call_later(self.debounce_secs, fire)

    def _remove_by_path(self, root: Root, file_path: Path) -> None:
        """Remove any existing entries that point at this file."""
        try:
            resolved = file_path.resolve()
        except OSError:
            resolved = file_path
        for entry in self._entries_under(root):
            if self._entry_path(root, entry) == resolved:
                self._index.remove_entry(entry.id)

    async def _update_entry(self, file_path: Path, deleted: bool) -> None:
        try:
            root = self._root_for(file_path)
            if root is None:
                return

            if deleted and not await off_disk(self.roots, root.is_live):
                # The volume went away rather than the file. Freeze: mark the
                # root and touch nothing. Every other event for this root
                # will arrive here too and be dropped the same way, which is
                # the intent — one unplugged drive must not empty a library.
                if root.available:
                    root.available = False
                    self._index.roots = self.roots.describe()
                    log.warning("Root %r disappeared — ignoring deletion events and "
                                "freezing %d entries", root.name,
                                len(self._entries_under(root)))
                    self._index.version = int(time.time())
                    if self.on_change:
                        await self.on_change(self)
                return

            if not root.available:
                return

            self._remove_by_path(root, file_path)

            if not deleted:
                entry = await self._hash_or_cached(root, file_path)
                if entry:
                    existing = self._index.get_entry(entry.id)
                    if existing is not None:
                        log.debug("Watchdog: %s duplicates content already "
                                  "indexed as %s/%s — leaving the index alone",
                                  file_path.name, existing.path, existing.name)
                    else:
                        self._index.add_entry(entry)
                        log.debug("Indexed: %s (%s, %d bytes)",
                                  file_path.name, entry.id[:8], entry.size)

            self._index.version = int(time.time())
            if self.on_change:
                await self.on_change(self)
        finally:
            # Mirror image of the accounting in _debounce, run whichever way
            # this method exits (including the several early returns above) —
            # otherwise a frozen/unavailable root's files would leave
            # `scanning` stuck True forever, the exact bug this is fixing but
            # in the other direction.
            size = self._burst_sizes.pop(str(file_path), None)
            if size is not None:
                self._burst_scanned += size
                self._burst_files_done += 1
                self._burst_inflight -= 1
                self._show_burst()


class _WatchdogHandler(FileSystemEventHandler):
    def __init__(self, indexer: DirectoryIndexer):
        self._indexer = indexer

    def on_created(self, event: FileSystemEvent):
        if not event.is_directory:
            self._indexer._schedule_update(Path(event.src_path))

    def on_modified(self, event: FileSystemEvent):
        if not event.is_directory:
            self._indexer._schedule_update(Path(event.src_path))

    def on_deleted(self, event: FileSystemEvent):
        if not event.is_directory:
            self._indexer._schedule_update(Path(event.src_path), deleted=True)

    def on_moved(self, event: FileSystemEvent):
        if not event.is_directory:
            self._indexer._schedule_update(Path(event.src_path), deleted=True)
            self._indexer._schedule_update(Path(event.dest_path))