refactor(node): move per-app enrichment out of NodeDaemon

The _enrich_*/_reenrich_* methods and _on_enriched become
EnrichmentMixin in meshbay_node/enrichment.py, with the two directory
helpers only they use. APP_DIR_KEYS stays on NodeDaemon.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 4 insertions, 291 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index a81aced..6d808ae 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -31,7 +31,7 @@ import os
 import signal
 import sys
 import time
-from dataclasses import asdict, replace
+from dataclasses import asdict
 from pathlib import Path
 
 import uvicorn
@@ -47,6 +47,7 @@ from meshbay_node.bundle_store import BundleStore
 from meshbay_node.chat.store import ChatStore
 from meshbay_node.cli.dispatch import run, start
 from meshbay_node.config import DEFAULT_CONFIG_PATH, Config, load_config
+from meshbay_node.enrichment import EnrichmentMixin
 from meshbay_node.hub_client import HubClient, HubConfig
 from meshbay_node.indexer import DirectoryIndexer, GroupIndex, IndexCache
 from meshbay_node.indexer.enrich import Enricher
@@ -56,7 +57,7 @@ from meshbay_node.keystore import load_or_create_keystore
 from meshbay_node.media_cache import MediaCache
 from meshbay_node.musicbrainz import MusicBrainzClient
 from meshbay_node.platform import chmod_private
-from meshbay_node.roots import RootError, RootSet, entry_abs_path, off_disk
+from meshbay_node.roots import RootError, RootSet, off_disk
 from meshbay_node.roster import Roster
 from meshbay_node.tmdb import TmdbClient
 from meshbay_node.transport import (
@@ -74,38 +75,6 @@ if WEBRTC_AVAILABLE:
 log = logging.getLogger(__name__)
 
 
-def _under_any_directory(path: str, directories: list[str]) -> bool:
-    """
-    Whether an entry's folder is one of an app's directories, or inside one.
-
-    Mirrors `underAnyDirectory` in the SPA's app modules. One helper for every
-    app since they all take a list: Videos and Music used to take a single
-    folder and had a function each saying the same thing, which is how the two
-    came to differ in what they did with a trailing slash.
-    """
-    path = path or ""
-    return any(path == d or path.startswith(d + "/") for d in directories)
-
-
-def _owning_directory(path: str, directories: list[str]) -> str | None:
-    """
-    Which of an app's directories an entry belongs to — the deepest match.
-
-    Deepest, because directories may nest: with both `Media` and
-    `Media/Albums` configured, a file under the second belongs to the second.
-    Taking the first match instead would measure it against a boundary one
-    level too shallow, which for Music is the difference between reading a
-    folder as an artist and reading it as a release.
-    """
-    path = path or ""
-    best: str | None = None
-    for d in directories:
-        if path == d or path.startswith(d + "/"):
-            if best is None or len(d) > len(best):
-                best = d
-    return best
-
-
 # ── Hub WS sender bridge ─────────────────────────────────────────────────────
 
 class _WsSender:
@@ -125,7 +94,7 @@ def _root_shape(roots) -> set[tuple]:
     return {(r.name, str(r.path), r.writable, r.removable) for r in roots}
 
 
-class NodeDaemon:
+class NodeDaemon(EnrichmentMixin):
     def __init__(self, config: Config, config_path: Path = DEFAULT_CONFIG_PATH):
         self._config = config
         self._config_path = config_path
@@ -1447,262 +1416,6 @@ class NodeDaemon:
                 endpoint = f"webrtc:{self._config.node.quic_port}"
                 spawn(self._register_swarm(hashes, endpoint))
 
-    async def _enrich_new_video_entries(self, indexer: DirectoryIndexer, entries: list) -> None:
-        """
-        Videos app: fire (never await further) enrichment for unattempted
-        video entries under the group's configured video_root.
-
-        A group with no video_root set yet does not enrich anything — TMDB
-        lookups and ffmpeg thumbnailing are real, ongoing per-file cost
-        (docs/MESHBAY_DESIGN.md §6.5), and running them over an operator's
-        whole shared index before they have chosen which folder is actually their
-        media library would burn both TMDB's rate limit and the node's CPU
-        on files that were never meant to be in the Videos app at all. Once a
-        root is set, `_enrich_video_root_now` (called when it changes)
-        separately sweeps whatever it already contains — this path alone only
-        ever sees entries new since the last broadcast.
-        """
-        if not self._enricher or not self._roster:
-            return
-        video_dirs = await self._roster.app_directories(indexer.group_id, "video")
-        if not video_dirs:
-            return
-        for entry in entries:
-            if entry.type != "video" or (indexer.group_id, entry.id) in self._enriched_attempted:
-                continue
-            if not _under_any_directory(entry.path, video_dirs):
-                continue
-            file_path = entry_abs_path(indexer.roots, entry)
-            if not file_path or not file_path.exists():
-                continue
-            self._enriched_attempted.add((indexer.group_id, entry.id))
-
-            async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None:
-                await self._on_enriched(_indexer, file_id, fields)
-
-            self._enricher.spawn(entry, file_path, on_done)
-
-    async def _enrich_video_root_now(self, group_id: str) -> None:
-        """
-        Videos app: sweep a group's existing index for enrichment right
-        after its video root is set or changed.
-
-        The ordinary path above only ever looks at entries new since the
-        last broadcast, so a folder that already had files sitting in it
-        before it became the video_root would otherwise never get enriched
-        at all — nothing else re-visits already-indexed entries once they
-        have been broadcast once.
-        """
-        indexer = self._state.get("indexers", {}).get(group_id)
-        if not indexer:
-            return
-        await self._enrich_new_video_entries(indexer, list(indexer.index.entries))
-
-    async def _reenrich_renamed_video_entries(
-        self, indexer: DirectoryIndexer, updates: list, previous: GroupIndex,
-    ) -> None:
-        """
-        Videos app: found live — a French-named episode file, renamed by
-        the operator to match its English-named siblings, kept showing as
-        its own separate poster-grid card (and its own row in Flat list)
-        indefinitely, because `_enriched_attempted` — there specifically to
-        stop enrichment's own field-fill from re-triggering itself forever
-        (see the caller) — also silently blocked the *new* filename from
-        ever being title-parsed at all. `entry.id in self._enriched_attempted`
-        is the same content, so simply discarding it here and re-running
-        the ordinary enrichment path is enough: a fresh ffprobe/thumbnail
-        for an unchanged file is redundant work, not a correctness issue,
-        and renames are rare enough that the redundancy is not worth a
-        separate "title-parse only" code path.
-        """
-        for entry in updates:
-            if entry.type != "video":
-                continue
-            old = previous.get_entry(entry.id)
-            if old is None or (old.name == entry.name and old.path == entry.path):
-                continue
-            self._enriched_attempted.discard((indexer.group_id, entry.id))
-            # The rename re-derives the title (the whole point of this
-            # method), which can change the correct TMDB match — but
-            # file_tmdb is keyed by content hash, unchanged by a rename, so
-            # nothing else would ever dislodge the old name's match. A
-            # manual "Fix match" correction is kept (clear_file_tmdb skips
-            # anything in media_cache.tmdb_override).
-            if self._media_cache is not None:
-                await self._media_cache.clear_file_tmdb(entry.id)
-        await self._enrich_new_video_entries(indexer, updates)
-
-    async def _enrich_new_audio_entries(self, indexer: DirectoryIndexer, entries: list) -> None:
-        """
-        Music app (docs/MESHBAY_DESIGN.md §9.8): fire (never await further)
-        tag/cover enrichment for unattempted audio entries under the
-        group's configured audio_root — same gate as
-        `_enrich_new_video_entries` above (the original "no root, whole
-        shared tree" call turned out wrong against a real messy library:
-        everything under every shared folder got mixed together with no way
-        to scope it down). `_enriched_attempted` is shared with the video
-        path — content-addressed ids never collide across the two.
-        """
-        if not self._audio_enricher or not self._roster:
-            return
-        audio_dirs = await self._roster.app_directories(indexer.group_id, "music")
-        if not audio_dirs:
-            return
-        # Resolved once per directory, not once per file: a library is
-        # thousands of entries and this is a filesystem call each time.
-        boundaries = {d: indexer.roots.resolve(d, require_available=False)
-                      for d in audio_dirs}
-        for entry in entries:
-            if entry.type != "audio" or (indexer.group_id, entry.id) in self._enriched_attempted:
-                continue
-            owner = _owning_directory(entry.path, audio_dirs)
-            if owner is None:
-                continue
-            file_path = entry_abs_path(indexer.roots, entry)
-            if not file_path or not file_path.exists():
-                continue
-            self._enriched_attempted.add((indexer.group_id, entry.id))
-
-            async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None:
-                await self._on_enriched(_indexer, file_id, fields)
-
-            # The boundary is *the configured directory this file is under*,
-            # not the shared root it lives in — so the ancestor walk
-            # (enrich_audio._artist_album_from_ancestors) treats a flat
-            # top-level folder right under the configured Music directory as
-            # ambiguous (artist-or-release, docs/MESHBAY_DESIGN.md §9.8),
-            # rather than one level too shallow when that directory is
-            # itself a subfolder. With several configured, each file is
-            # measured against its own:
-            # a single shared boundary would be wrong for all but one of them.
-            self._audio_enricher.spawn(entry, file_path, on_done,
-                                       boundaries.get(owner))
-
-    async def _enrich_audio_root_now(self, group_id: str) -> None:
-        """
-        Music app: sweep a group's existing index right after its
-        audio root is set or changed. Mirrors
-        `_enrich_video_root_now` exactly — the ordinary path above only
-        ever looks at entries new since the last broadcast, so a folder
-        that already had files in it before it became the audio_root would
-        otherwise never get enriched at all.
-        """
-        indexer = self._state.get("indexers", {}).get(group_id)
-        if not indexer:
-            return
-        await self._enrich_new_audio_entries(indexer, list(indexer.index.entries))
-
-    async def _reenrich_renamed_audio_entries(
-        self, indexer: DirectoryIndexer, updates: list, previous: GroupIndex,
-    ) -> None:
-        """
-        Music app equivalent of `_reenrich_renamed_video_entries` — a rename
-        can change the filename-parse fallback (title/track_no) even though
-        embedded tags, when present, are unaffected. Re-running the whole
-        pass on a rename is redundant work for a tagged file and a real fix
-        for an untagged one, and renames are rare enough not to need a
-        cheaper, tags-only special case.
-        """
-        for entry in updates:
-            if entry.type != "audio":
-                continue
-            old = previous.get_entry(entry.id)
-            if old is None or (old.name == entry.name and old.path == entry.path):
-                continue
-            self._enriched_attempted.discard((indexer.group_id, entry.id))
-        await self._enrich_new_audio_entries(indexer, updates)
-
-    async def _enrich_new_photo_entries(self, indexer: DirectoryIndexer, entries: list) -> None:
-        """
-        Photos app (docs/MESHBAY_DESIGN.md §9.9): fire (never await further)
-        thumbnail/EXIF enrichment for unattempted image entries under any of the
-        group's configured photo_roots. Same gate shape as
-        `_enrich_new_video_entries`/`_enrich_new_audio_entries` — no root
-        configured yet means no work, since thumbnailing every image in a
-        whole shared tree before the operator has chosen which folders are
-        actually photo albums would burn CPU on files never meant to be in
-        the Photos app at all. `_enriched_attempted` is shared with the
-        video/audio paths — content-addressed ids never collide across them.
-        """
-        if not self._photo_enricher or not self._roster:
-            return
-        photo_dirs = await self._roster.app_directories(indexer.group_id, "photo")
-        if not photo_dirs:
-            return
-        for entry in entries:
-            if entry.type != "image" or (indexer.group_id, entry.id) in self._enriched_attempted:
-                continue
-            if not _under_any_directory(entry.path, photo_dirs):
-                continue
-            file_path = entry_abs_path(indexer.roots, entry)
-            if not file_path or not file_path.exists():
-                continue
-            self._enriched_attempted.add((indexer.group_id, entry.id))
-
-            async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None:
-                await self._on_enriched(_indexer, file_id, fields)
-
-            self._photo_enricher.spawn(entry, file_path, on_done)
-
-    async def _enrich_photo_roots_now(self, group_id: str) -> None:
-        """
-        Photos app: sweep a group's existing index right after its
-        photo roots change. Mirrors
-        `_enrich_video_root_now`/`_enrich_audio_root_now` — the ordinary
-        path above only ever looks at entries new since the last broadcast,
-        so a folder that already had photos in it before it was added to
-        photo_roots would otherwise never get enriched at all. Also covers
-        a root being *removed*: nothing un-enriches on removal (the cache
-        entry is harmless, just unused — the media cache is disposable and
-        tied to the index, docs/MESHBAY_DESIGN.md §6.5), so re-sweeping the
-        new set is enough.
-        """
-        indexer = self._state.get("indexers", {}).get(group_id)
-        if not indexer:
-            return
-        await self._enrich_new_photo_entries(indexer, list(indexer.index.entries))
-
-    async def _reenrich_renamed_photo_entries(
-        self, indexer: DirectoryIndexer, updates: list, previous: GroupIndex,
-    ) -> None:
-        """
-        Photos app equivalent of `_reenrich_renamed_video_entries` — a
-        rename changes nothing about the image's own bytes (thumbnail, EXIF
-        fields are content-derived, not name-derived), so this exists only
-        for consistency/symmetry with Videos/Music and to catch the case of
-        a file moving *into* a newly-covered photo_roots subtree via a
-        rename rather than a fresh add. Re-running enrichment on an
-        unchanged file is redundant work, not a correctness issue.
-        """
-        for entry in updates:
-            if entry.type != "image":
-                continue
-            old = previous.get_entry(entry.id)
-            if old is None or (old.name == entry.name and old.path == entry.path):
-                continue
-            self._enriched_attempted.discard((indexer.group_id, entry.id))
-        await self._enrich_new_photo_entries(indexer, updates)
-
-    async def _on_enriched(self, indexer: DirectoryIndexer, file_id: str, fields: dict) -> None:
-        """
-        Merge enrichment fields into the live index and re-trigger a
-        broadcast so they reach connected clients as an INDEX_DELTA update
-        (GroupIndex.diff's `updates`, not `additions` — same id, new fields).
-
-        Builds a *new* IndexEntry via dataclasses.replace rather than
-        mutating the existing one in place: the diff mechanism compares
-        against a shallow snapshot of entry *references*, so an in-place
-        mutation would silently also change what "previous" looks like,
-        and the change would never show up as a diff (see group_index.py's
-        diff() docstring).
-        """
-        idx = indexer.index
-        entry = idx.get_entry(file_id)
-        if entry is None:
-            return   # removed from the index while enrichment was in flight
-        idx.add_entry(replace(entry, **fields))
-        await self._on_index_change(indexer)
-
     def _drop_group_sessions(self, group_id: str) -> None:
         """Close live sessions for a revoked group (H4)."""
         if not self._webrtc or not group_id: