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 logging
import re
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"
    upload: bool = False
    direct: bool = False
    # Runtime, not configuration: set by the indexer when the directory can no
    # longer be read, and cleared when it comes back.
    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)

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

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

        `specs` are dicts with `path`, and optionally `name`, `kind`, `upload`.
        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"

            root = Root(name=name, path=path, kind=kind,
                        upload=bool(spec.get("upload", False)),
                        direct=bool(spec.get("direct", False)))
            _refuse_nesting(root, roots)
            roots.append(root)
            by_folded[root.folded] = root

        cls._settle_upload_root(roots)
        return cls(roots=roots)

    @staticmethod
    def _settle_upload_root(roots: list[Root]) -> None:
        """
        Exactly one root receives uploads, and the operator picks it.

        Not guessed when several are marked, because "uploads went somewhere
        else" is discovered weeks later. With none marked and a single root, the
        answer is not ambiguous, so it is taken.
        """
        marked = [r for r in roots if r.upload]
        if len(marked) > 1:
            names = ", ".join(r.name for r in marked)
            raise RootError(
                f"several roots are marked upload = true ({names}) — exactly one "
                f"receives uploads")
        if not marked and len(roots) == 1:
            roots[0].upload = True

    # ── 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 upload_root(self) -> Root | None:
        for root in self.roots:
            if root.upload:
                return root
        return None

    @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.
        """
        changed: list[tuple[Root, bool]] = []
        for root in self.roots:
            live = root.is_live()
            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) -> list[dict]:
        """Per-root state for the index payload and the admin UI."""
        out = []
        for r in self.roots:
            d: dict = {"name": r.name, "kind": r.kind,
                        "available": r.available, "upload": r.upload}
            if r.direct:
                d["direct"] = True
            out.append(d)
        return out


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


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