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, 298 insertions, 0 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/enrichment.py b/packages/meshbay-node/src/meshbay_node/enrichment.py
new file mode 100644
index 0000000..66b69b9
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/enrichment.py
@@ -0,0 +1,298 @@
+"""What the node does with a file once it is indexed, per app: the TMDB match and
+thumbnail for Videos, tags and cover for Music, thumbnail and EXIF for Photos.
+NodeDaemon runs it; the enrichers themselves are in indexer/."""
+
+from dataclasses import replace
+
+from meshbay_node.indexer import DirectoryIndexer, GroupIndex
+from meshbay_node.roots import entry_abs_path
+
+
+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
+
+
+class EnrichmentMixin:
+    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)