path: root/packages/meshbay-node/src/meshbay_node/indexer/indexer.py

"""
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 concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from typing import Callable, Awaitable

import blake3
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from watchdog.events import FileSystemEvent, FileSystemEventHandler
from watchdog.observers import Observer

from meshbay_common.paths import fold, find_fold_collisions, long_path
from meshbay_common.protocol import IndexEntry
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.roots import Root, RootSet

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"},
    "audio": {".mp3", ".flac", ".ogg", ".wav", ".aac", ".m4a", ".opus"},
    "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)
    )


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


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 _scan_file(root: Root, file_path: Path) -> IndexEntry | None:
    """Compute IndexEntry for a file. Blocking — run in executor.
    Uses streaming blake3 so arbitrarily large files (ISOs, VM images, etc.)
    don't require loading the whole file into memory."""
    if not _is_indexable(file_path):
        return None
    try:
        stat = file_path.stat()
        hasher = blake3.blake3()
        # long_path is a no-op off Windows; there it is what lets a deep media
        # library past MAX_PATH.
        with open(long_path(file_path), "rb") as f:
            while chunk := f.read(_HASH_CHUNK):
                hasher.update(chunk)
        return IndexEntry(
            id=hasher.hexdigest(),
            # Stored exactly as the filesystem gave it: this is the string that
            # opens the file. Normalization is for comparison only.
            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),
        )
    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.
    RECONCILE_SECS = 60.0

    def __init__(
        self,
        roots: RootSet,
        group_id: str,
        sk_node: Ed25519PrivateKey,
        gek: bytes | None,
        on_change: Callable[["DirectoryIndexer"], 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

        self._index    = GroupIndex(group_id=group_id, sk_node=sk_node, gek=gek)
        self._index.roots = roots.describe()
        self._executor = ThreadPoolExecutor(max_workers=2, 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] = {}

    @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."""
        self.roots.refresh_availability()
        total = 0
        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
            total += await self._scan_root(root)
        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) -> int:
        log.info("Scanning %s (root %r) ...", root.path, root.name)
        loop = asyncio.get_event_loop()
        count = 0
        try:
            files = [p for p in root.path.rglob("*") if p.is_file()]
        except OSError as e:
            log.warning("Cannot scan root %r: %s", root.name, e)
            return 0
        for file_path in files:
            entry = await loop.run_in_executor(
                self._executor, _scan_file, root, file_path)
            if entry:
                self._index.add_entry(entry)
                count += 1
        return count

    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) -> None:
        """Start initial scan + filesystem watcher + reconciler."""
        self._loop = asyncio.get_event_loop()
        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
        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) -> 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 §6.9 exists to
        prevent. Roots that survive keep their entries; new ones are scanned.
        """
        old_names = {r.folded for r in self.roots}
        new_names = {r.folded for r in roots}

        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
        roots.refresh_availability()
        for root in roots:
            if root.folded not in old_names and root.available:
                await self._scan_root(root)

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

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

    async def _reconcile_loop(self) -> None:
        while True:
            try:
                await asyncio.sleep(self.RECONCILE_SECS)
                await self.reconcile()
            except asyncio.CancelledError:
                raise
            except Exception:
                log.exception("Reconcile failed — continuing")

    async def reconcile(self) -> None:
        """
        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.
        """
        changed = self.roots.refresh_availability()
        touched = False

        for root, available in changed:
            if available:
                log.info("Root %r is back — rescanning", root.name)
                self._drop_root_entries(root)
                await self._scan_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)

    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.
        """
        loop = asyncio.get_event_loop()
        changed = False
        for root in self.roots:
            if not root.available:
                continue
            try:
                on_disk = {p.resolve() for p in root.path.rglob("*")
                           if _is_indexable(p)}
            except OSError as e:
                log.warning("Cannot reconcile root %r: %s", root.name, e)
                continue

            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

            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

            for added in on_disk - set(known):
                entry = await loop.run_in_executor(
                    self._executor, _scan_file, root, added)
                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
        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]

    def _drop_root_entries(self, root: Root) -> None:
        for entry in self._entries_under(root):
            self._index.remove_entry(entry.id)

    @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 _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()

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

    _DEBOUNCE_SECS = 2.0

    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()

        def fire() -> None:
            self._pending_timers.pop(key, None)
            asyncio.ensure_future(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:
        root = self._root_for(file_path)
        if root is None:
            return

        if deleted and not 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:
            loop = asyncio.get_event_loop()
            entry = await loop.run_in_executor(
                self._executor, _scan_file, root, file_path)
            if entry:
                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)


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))