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

"""
A group's content is several named roots, not one directory.

    /                       (the group's virtual root)
    ├── Films/          →   D:\\Media\\Films
    ├── Musique/        →   E:\\Audio            (external drive)
    └── Documents/      →   C:\\Users\\me\\Partage

Every index path carries the root name as its first segment, uniformly — a group
with one root is not a special case, because two path shapes would have to be
kept right forever and one shape only has to be got right once.

Three rules, and each of them is load-bearing rather than tidy:

  * **The name is the directory's basename, derived once and stored.** Never
    recomputed from the path, or renaming a folder on disk silently re-identifies
    every file under it.
  * **No root may contain another.** Otherwise the same bytes are indexed twice
    under two identities, and deleting one leaves the other pointing at nothing.
  * **Availability is per root.** A root whose volume goes away *freezes*: its
    entries stay in the index, marked unavailable. Emptying it would propagate
    deletions for a whole library as though the owner had erased it.

Comparison of names is case-insensitive and NFC-normalized (`meshbay_common.paths`),
because most of these directories live on exFAT or NTFS, where `Films` and `films`
are one directory.
"""

from __future__ import annotations

import asyncio
import logging
import re
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from pathlib import Path

from meshbay_common.paths import fold, portable_name_problem

log = logging.getLogger(__name__)

VALID_KINDS = ("generic", "video", "audio", "photo")

# Filenames and subdirectory names sent by clients. A leading dot is a hidden
# file on every platform, a leading hyphen confuses CLI tools, a leading space
# cannot start one, and a trailing space or dot is refused because it makes two
# different files look identical in a list.
SAFE_UPLOAD_NAME = re.compile(
    r"^[^\W_]"                            # letter or digit — never ‘.’, ‘-’ or space
    r"[\w .\-()\[\]'\u2019,&+#@]{0,127}"  # body: word chars plus mild punctuation
    r"(?<![ .])$",                        # and never ending on a space or a dot
    re.UNICODE)


def _free_name(directory: Path, filename: str) -> str:
    """
    `filename`, or the first "name (n).ext" that is not taken.

    Never returns the name of a file that exists, so an upload cannot replace
    one — the property the per-user quarantine used to provide (C5a).
    """
    if not (directory / filename).exists():
        return filename
    stem, dot, ext = filename.rpartition(".")
    if not dot:
        stem, ext = filename, ""
    for n in range(2, 1000):
        candidate = f"{stem} ({n}){dot}{ext}"
        if not (directory / candidate).exists():
            return candidate
    raise FileExistsError(filename)


def safe_subdir(roots: RootSet, rel: str) -> Path | None:
    """
    Resolve a client-supplied directory inside one of the group's roots, or refuse.

    The path arrives from the wire, so every part is checked: the first segment
    must name a root that is readable right now, each later segment against the
    same allowlist as filenames, and the resolved result against that root's
    directory. `..`, absolute paths, symlinks pointing out, and anything with a
    separator in a segment are all refused here rather than in the caller, so
    there is one place to get it right.

    The virtual root itself — `""` — is deliberately **not** resolvable. It is
    not a directory on anyone's disk: a file cannot be written there and a
    directory cannot be created there, because it belongs to no volume. Callers
    that used to receive the shared root for an empty path now receive None,
    which is the honest answer.

    The quarantine was the fix for C5a; what actually mattered in it — no
    overwrite, a name allowlist, and confinement — is kept by this plus the
    caller's existing checks.
    """
    found = roots.split(rel or "")
    if found is None:
        return None
    root, tail = found
    if not root.available:
        return None
    parts = [seg for seg in tail.split("/") if seg not in ("", ".")]
    if any(seg == ".." or not SAFE_UPLOAD_NAME.match(seg) for seg in parts):
        return None
    try:
        target = (root.path / Path(*parts)).resolve() if parts else root.path.resolve()
        base = root.path.resolve()
    except OSError:
        return None
    if target != base and base not in target.parents:
        return None
    return target


class RootError(ValueError):
    """A root set that cannot be built. The message is shown to the operator."""


@dataclass
class Root:
    """One named directory inside a group."""

    name: str
    path: Path
    kind: str = "generic"
    writable: bool = False
    removable: bool = False
    ejected: bool = False
    available: bool = True

    @property
    def folded(self) -> str:
        return fold(self.name)

    def is_live(self) -> bool:
        """Readable right now. The question `available` caches."""
        try:
            return self.path.is_dir()
        except OSError:
            return False


def derive_name(path: Path) -> str:
    """
    The name a directory gets when it is added: its basename.

    A path that has no usable basename — a drive root such as `E:\\`, or `/` —
    has nothing to derive from, and the operator has to supply a name.
    """
    name = path.name or ""
    if not name:
        raise RootError(
            f"{path} has no directory name to use — give the root an explicit "
            f"name (a drive or filesystem root cannot supply one)")
    return name


@dataclass
class RootSet:
    """
    The roots of one group, and the only place a virtual path is resolved.

    `resolve()` is the single entry point for turning something that arrived
    over the wire into a path on disk. Callers must not join paths themselves —
    that is how a traversal gets in through the one site nobody reviewed.
    """

    roots: list[Root] = field(default_factory=list)

    # Roots this set ejected by itself — a removable device that went away
    # without the operator clicking Eject. Drained by the indexer, which is
    # the only caller holding a roster to write the state to. Without that
    # the flag is lost on the next restart, and the surprise unplug looks
    # like a deletion all over again on the pass after it.
    auto_ejected: list[str] = field(default_factory=list)

    # The thread every blocking filesystem call on these roots is made from.
    #
    # **Why it exists at all.** A root that has spun down, or that lives on a
    # network mount, answers its first syscall in seconds rather than
    # microseconds. Made from the event loop that is the whole node: no other
    # group is served, no stream is fed, no chat message is delivered and the
    # hub socket is not read, for as long as the platter takes to come back.
    # A client's connection attempt times out and has to be made again, which
    # is what this was found by. It is `AV9`'s lesson with the disk in the
    # place of the mail server.
    #
    # **Why one worker and not a pool.** The same reason the indexer's executor
    # has one: two interleaved reads on a spinning drive seek-thrash against
    # each other rather than go faster, measured there on a USB disk. One
    # worker also keeps every read of these roots in the order it was asked
    # for, which costs nothing — the protocol addresses a chunk by index, so
    # no caller depends on that order — and leaves no way for two threads to
    # be inside the same file at once.
    #
    # **Why per root set and not one for the node.** A node serves several
    # groups, and their roots are not all on the same volume. A single worker
    # would put the sleeping USB drive of one group in front of the SSD of
    # another, which is the symptom this removes, one level down.
    #
    # Created on first use, so a RootSet that never reads anything — most of
    # them, in tests — never starts a thread. Not compared and not printed: it
    # is machinery, not part of what a root set *is*, and `daemon.py` compares
    # root sets to decide whether a reload changed anything.
    _io: ThreadPoolExecutor | None = field(
        default=None, init=False, repr=False, compare=False)

    @property
    def io_executor(self) -> ThreadPoolExecutor:
        if self._io is None:
            self._io = ThreadPoolExecutor(max_workers=1, thread_name_prefix="rootio")
        return self._io

    def close_io(self) -> None:
        """Stop the disk thread. Safe to call twice, and on a set that never read."""
        if self._io is not None:
            self._io.shutdown(wait=False)
            self._io = None

    # ── Construction ─────────────────────────────────────────────────────────

    @classmethod
    def build(cls, specs: list[dict]) -> RootSet:
        """
        Build from configuration, refusing anything ambiguous.

        `specs` are dicts with `path`, and optionally `name`, `kind`, `writable`,
        `removable`. Raises RootError with a message meant for an operator reading
        a log.
        """
        roots: list[Root] = []
        by_folded: dict[str, Root] = {}

        for spec in specs:
            raw = str(spec.get("path", "")).strip()
            if not raw:
                raise RootError("a root has no path")
            path = Path(raw).expanduser()
            try:
                path = path.resolve()
            except OSError as e:
                raise RootError(f"{raw}: {e}") from e

            name = str(spec.get("name") or "").strip() or derive_name(path)

            problem = portable_name_problem(name)
            if problem:
                raise RootError(
                    f"root name {name!r} ({problem}) — every member sees this as a "
                    f"folder name, including on Windows. Give the root an explicit "
                    f"name in the config")

            clash = by_folded.get(fold(name))
            if clash:
                raise RootError(
                    f"two roots would both be called {name!r}: {clash.path} and "
                    f"{path}. Names are compared without regard to case. Give one "
                    f"of them an explicit name")

            kind = str(spec.get("kind") or "generic").strip().lower()
            if kind not in VALID_KINDS:
                log.warning("root %r: unknown kind %r — using 'generic'", name, kind)
                kind = "generic"

            # Backward compat: old configs use `upload` instead of `writable`
            writable = bool(spec.get("writable", spec.get("upload", False)))
            # `ejected` is runtime state, not configuration — it reaches here
            # only from the roster, restored at startup so a drive ejected
            # before a restart does not come back on its own.
            root = Root(name=name, path=path, kind=kind,
                        writable=writable,
                        removable=bool(spec.get("removable", False)),
                        ejected=bool(spec.get("ejected", False)),
                        available=not bool(spec.get("ejected", False)))
            _refuse_nesting(root, roots)
            roots.append(root)
            by_folded[root.folded] = root

        return cls(roots=roots)

    # ── Lookup ───────────────────────────────────────────────────────────────

    def by_name(self, name: str) -> Root | None:
        target = fold(name)
        for root in self.roots:
            if root.folded == target:
                return root
        return None

    @property
    def writable_roots(self) -> list[Root]:
        return [r for r in self.roots if r.writable]

    @property
    def names(self) -> list[str]:
        return [r.name for r in self.roots]

    def __bool__(self) -> bool:
        return bool(self.roots)

    def __len__(self) -> int:
        return len(self.roots)

    def __iter__(self):
        return iter(self.roots)

    # ── Resolution ───────────────────────────────────────────────────────────

    def split(self, virtual: str) -> tuple[Root, str] | None:
        """
        `"Films/2024"` → (the Films root, `"2024"`). None if no such root.

        Does not touch the filesystem, so it is safe to call on a root whose
        volume is absent.
        """
        rel = (virtual or "").strip().strip("/")
        if not rel:
            return None
        head, _, tail = rel.partition("/")
        root = self.by_name(head)
        if root is None:
            return None
        return root, tail

    def resolve(self, virtual: str, *, require_available: bool = True) -> Path | None:
        """
        A virtual path from the wire → a real path inside its root, or None.

        Refuses `..`, absolute segments, and anything whose resolved form escapes
        the root — symlinks included, which is why this resolves before
        comparing rather than checking the string.
        """
        found = self.split(virtual)
        if found is None:
            return None
        root, tail = found
        if require_available and not root.available:
            return None

        parts = [seg for seg in tail.split("/") if seg not in ("", ".")]
        if any(seg == ".." for seg in parts):
            return None
        try:
            target = (root.path / Path(*parts)).resolve() if parts else root.path.resolve()
            base = root.path.resolve()
        except OSError:
            return None
        if target != base and base not in target.parents:
            return None
        return target

    def virtual_of(self, absolute: Path) -> str | None:
        """A real path anywhere in this group → `"Films/2024"`, or None."""
        for root in self.roots:
            virtual = self.virtual_path(root, absolute)
            if virtual is not None:
                return virtual
        return None

    def virtual_path(self, root: Root, absolute: Path) -> str | None:
        """The inverse of `resolve`: a real path → `"Films/2024"`."""
        try:
            rel = absolute.resolve().relative_to(root.path.resolve())
        except (ValueError, OSError):
            return None
        return root.name if str(rel) == "." else f"{root.name}/{rel.as_posix()}"

    # ── Availability ─────────────────────────────────────────────────────────

    def refresh_availability(self) -> list[tuple[Root, bool]]:
        """
        Re-read which roots are readable. Returns the ones that changed.

        Called periodically and after a filesystem event that looks like a
        disappearance. A change here never edits the index: a root going away
        freezes its entries, and a root coming back triggers a rescan.

        An ejected root stays unavailable regardless of `is_live()` — the
        operator must explicitly plug it back. A removable root whose path
        disappears without an eject is auto-ejected as a safety net.
        """
        changed: list[tuple[Root, bool]] = []
        for root in self.roots:
            if root.ejected:
                if root.available:
                    root.available = False
                    changed.append((root, False))
                continue
            live = root.is_live()
            if not live and root.removable:
                root.ejected = True
                # Recorded for the caller to persist. A flag that only lives
                # in memory would be forgotten on the next restart, and the
                # rescan that followed would read an empty mount point as an
                # erased library — the exact outcome eject exists to prevent.
                self.auto_ejected.append(root.name)
                log.warning("Root %r auto-ejected (device disappeared): %s",
                            root.name, root.path)
            if live != root.available:
                root.available = live
                changed.append((root, live))
                log.warning("Root %r is now %s (%s)", root.name,
                            "available" if live else "UNAVAILABLE", root.path)
        return changed

    def describe(self, *, with_paths: bool = False) -> list[dict]:
        """
        Per-root state for the index payload and the admin UI.

        Deliberately no paths by default: this is what every member receives.
        `with_paths=True` is the operator's own view, over a channel that is
        already theirs alone (loopback + run token).
        """
        out = []
        for r in self.roots:
            d: dict = {"name": r.name, "kind": r.kind,
                        "available": r.available,
                        "writable": r.writable,
                        "removable": r.removable,
                        "ejected": r.ejected}
            # `with_paths` is for the operator's *own* channels only — the
            # loopback API and the CLI reading it, both of which already
            # require being on this machine with the run token. A member is
            # told what exists and whether it is readable, never where on the
            # operator's disk it lives, and the index payload every member
            # receives must keep calling this without the flag.
            if with_paths:
                d["path"] = str(r.path)
            out.append(d)
        return out


# What a transport answers when `entry_abs_path` gives None: the entry is in the
# index but its root is ejected, unplugged, or not in the set being served.
ROOT_NOT_SERVED = "File not available: its folder is not readable right now"


def entry_abs_path(roots: RootSet, entry) -> Path | None:
    """
    Where an index entry actually lives, or None if its root is gone.

    The one place an entry becomes a path. `entry.path` starts with a root name,
    so a group whose drive is unplugged answers None here rather than opening
    something unrelated that happens to share a relative path with another root.
    """
    parent = roots.resolve(entry.path)
    return (parent / entry.name) if parent else None


async def off_disk(roots: RootSet, fn, *args):
    """
    Run one blocking filesystem call on the thread that serves `roots`.

    Every syscall against a group's content goes through here, `resolve()` and
    `exists()` included: a `stat` is what *wakes* a sleeping disk, so a check
    left on the event loop pays the spin-up in full and the read that follows
    it finds the disk already awake. Offloading only the read would move the
    stall, not remove it.

    `fn` must not touch anything the loop also touches — it runs on another
    thread. Reading and encrypting a chunk qualifies; updating a session's
    state does not.
    """
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(roots.io_executor, fn, *args)


def _refuse_nesting(new: Root, existing: list[Root]) -> None:
    """
    No root may contain another, compared case-insensitively.

    `D:\\Media` and `D:\\Media\\Films` together would index the same bytes twice
    under two identities. On NTFS and exFAT `d:\\media` is the same directory as
    `D:\\Media`, so a string comparison that respects case would miss it.
    """
    new_parts = [fold(p) for p in new.path.parts]
    for other in existing:
        other_parts = [fold(p) for p in other.path.parts]
        if new_parts == other_parts:
            raise RootError(
                f"roots {new.name!r} and {other.name!r} are the same directory "
                f"({new.path})")
        shorter, longer, inner, outer = (
            (other_parts, new_parts, new, other)
            if len(other_parts) < len(new_parts)
            else (new_parts, other_parts, other, new))
        if longer[:len(shorter)] == shorter:
            raise RootError(
                f"root {inner.name!r} ({inner.path}) is inside root "
                f"{outer.name!r} ({outer.path}) — its files would be indexed "
                f"twice under two names")