diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/daemon.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/daemon.py | 209 |
1 files changed, 196 insertions, 13 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index af85340..bf6bd64 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -26,7 +26,7 @@ Usage: """ import asyncio -from dataclasses import asdict +from dataclasses import asdict, replace import base64 import json import logging @@ -38,14 +38,17 @@ from pathlib import Path import uvicorn from meshbay_common import MNP_VERSION -from meshbay_common.protocol import MNP +from meshbay_common.protocol import MNP, index_entry_wire from meshbay_node.audit import AuditStore from meshbay_node.bundle_store import BundleStore from meshbay_node.chat.store import ChatStore from meshbay_node.config import Config, DEFAULT_CONFIG_PATH, load_config, write_example_config -from meshbay_node.roots import RootSet, RootError +from meshbay_node.roots import RootSet, RootError, entry_abs_path from meshbay_node.hub_client import HubClient, HubConfig from meshbay_node.indexer import DirectoryIndexer, IndexCache, GroupIndex +from meshbay_node.indexer.enrich import Enricher +from meshbay_node.media_cache import MediaCache +from meshbay_node.tmdb import TmdbClient from meshbay_node.keystore import create_keystore, load_keystore, load_or_create_keystore from meshbay_node.roster import Roster from meshbay_node.transport import ( @@ -62,6 +65,12 @@ if WEBRTC_AVAILABLE: log = logging.getLogger(__name__) +def _under_video_root(path: str, video_root: str) -> bool: + """Mirrors video-app.js's underVideoRoot: same folder, or a descendant.""" + path = path or "" + return path == video_root or path.startswith(video_root + "/") + + # ── Argon2id calibration ────────────────────────────────────────────────────── def calibrate_argon2(target_ms: int = 500) -> None: @@ -132,6 +141,15 @@ class NodeDaemon: self._last_broadcast_snapshot: dict[str, tuple] = {} self._audit_store: AuditStore | None = None self._bundle_store: BundleStore | None = None + self._media_cache: MediaCache | None = None + self._enricher: Enricher | None = None + self._tmdb_client: TmdbClient | None = None + # A file id attempted at most once per daemon run, success or + # failure — a persistently unprobeable file (corrupt, still being + # written) does not get re-queued on every coalesced broadcast. A + # restart retries everything, matching the "disposable, rebuildable" + # stance the rest of this cache takes (docs/mediacenter.md §1/§2). + self._enriched_attempted: set[str] = set() self._roster: Roster | None = None self._indexers: list[DirectoryIndexer] = [] self._tasks: list[asyncio.Task] = [] @@ -312,6 +330,10 @@ class NodeDaemon: # by the signed operation that changes it. "enabled_apps": await self._roster.enabled_apps( group_cfg.id) if self._roster else list(Roster.DEFAULT_APPS), + # Which folder is the Videos app's entry point for this + # group — "" means the whole group index. + "video_root": await self._roster.video_root( + group_cfg.id) if self._roster else "", } if not groups_ctx: @@ -333,6 +355,26 @@ class NodeDaemon: await self._audit_store.open() log.info("Audit store opened: %s", audit_db) + # 6b. Media cache (Videos app — TMDB metadata + thumbnails). + # Node-wide like audit.db, not per-group: a thumbnail is the same + # bytes regardless of which group happens to share the file + # (docs/mediacenter.md §2/§5.5). + media_cache_db = data_dir / "media_cache.db" + self._media_cache = MediaCache(db_path=media_cache_db) + await self._media_cache.open() + self._enricher = Enricher(self._media_cache) + self._tmdb_client = TmdbClient(roster=self._roster) + # Read once at load, like member_upload/enabled_apps — kept + # current in place by ops.set_tmdb_config (the signed op), + # exposed to every group's handshake ack via `daemon_state` + # (already wired to self._webrtc._ctx below) since this is + # node-wide, not per-group. + tmdb_enabled, tmdb_token, tmdb_language = await self._roster.tmdb_config() + self._state["tmdb_enabled"] = tmdb_enabled + self._state["tmdb_token_customized"] = bool(tmdb_token) + self._state["tmdb_language"] = tmdb_language or "" + log.info("Media cache opened: %s", media_cache_db) + # 5. Denylist denylist = self._denylist @@ -359,6 +401,8 @@ class NodeDaemon: self._webrtc._ctx["node_user_id"] = session.user_id self._webrtc._ctx["audit_store"] = self._audit_store self._webrtc._ctx["bundle_store"] = self._bundle_store + self._webrtc._ctx["media_cache"] = self._media_cache + self._webrtc._ctx["tmdb_client"] = self._tmdb_client self._webrtc._ctx["sk_x25519_raw"] = sk_x_raw self._webrtc._ctx["pk_x25519_raw"] = pk_x_raw self._webrtc._ctx["pk_x25519_b64"] = keys.pk_x25519_b64 @@ -468,6 +512,7 @@ class NodeDaemon: self._state["quic_server"] = self._quic_server self._state["hub"] = hub self._state["reload_fn"] = self._reload_config + self._state["enrich_video_root_fn"] = self._enrich_video_root_now # Rotating a key has to reach every transport holding a copy of it, # and clearing the denylist has to reach the one the handshake # consults — so both are published rather than reachable only @@ -494,6 +539,16 @@ class NodeDaemon: hashes = [e.id for e in gctx["index"].entries] if hashes: await self._register_swarm(hashes, endpoint) + # initial_scan() itself never calls on_change (it predates + # the concept — every existing caller only cared about the + # scan finishing, not about notifying anyone) — but Videos + # app enrichment (duration/thumb_hash/display_title/...) + # hangs entirely off that callback (_broadcast_index_change). + # Without this, every file already on disk at startup — the + # common case, an existing library — would never get + # enriched at all; only a file added later, while the node + # is already running, would trigger it via the watchdog. + await self._on_index_change(indexer) for idx, group_cfg in zip(self._indexers, self._config.groups): gctx = groups_ctx.get(group_cfg.id) @@ -663,6 +718,9 @@ class NodeDaemon: "enabled_apps": ( await self._roster.enabled_apps(group_cfg.id) if self._roster else list(Roster.DEFAULT_APPS)), + "video_root": ( + await self._roster.video_root(group_cfg.id) + if self._roster else ""), "chat_store": store, } groups_ctx[group_cfg.id] = new_ctx @@ -880,6 +938,7 @@ class NodeDaemon: prev = self._last_broadcast_snapshot.get(group_id) delta = None + previous = None if prev is not None: prev_version, prev_entries = prev previous = GroupIndex._snapshot( @@ -887,6 +946,34 @@ class NodeDaemon: delta = idx.diff(previous) self._last_broadcast_snapshot[group_id] = (idx.version, idx.entries_by_id()) + # Videos app (docs/mediacenter.md §5.2): schedule async technical + # probe + title parse + thumbnail generation for every newly-seen + # video entry under the group's configured video_root. Never blocks + # this broadcast — enrichment fields arrive later as their own + # INDEX_DELTA update (_on_enriched below). + new_entries = delta.additions if delta is not None else list(idx.entries) + asyncio.ensure_future(self._enrich_new_video_entries(indexer, new_entries)) + + # A rename/move changes the very filename (or season folder) that + # §3.3/§3.4's title-parse read display_title/season/episode from, + # but leaves the file's content — and so its id and everything + # ffprobe/thumbnailing already found — untouched. Only entries + # whose name or path actually differ from the last broadcast get a + # fresh pass; an update that is enrichment's own field-fill + # (duration/thumb_hash/... landing via _on_enriched below) leaves + # name/path alone and must not re-trigger itself forever. + if delta is not None and delta.updates and previous is not None: + asyncio.ensure_future( + self._reenrich_renamed_video_entries(indexer, delta.updates, previous)) + + # Videos app: a file that leaves the index also loses its thumbnail + # and file->tmdb mapping — the "real deletion obligation" docs/ + # mediacenter.md §2/§8 calls out explicitly rather than leaving + # implicit. tmdb_meta rows are left alone (§2: shared across files). + if delta is not None and delta.deletions and self._media_cache: + for file_id in delta.deletions: + asyncio.ensure_future(self._media_cache.prune_file(file_id)) + # 11.5 — Push to connected WebRTC peers in this group if self._webrtc: if delta is not None: @@ -896,12 +983,9 @@ class NodeDaemon: "group_id": idx.group_id, "base_version": delta.base_version, "version": delta.version, - "additions": [ - {"id": e.id, "name": e.name, "path": e.path, - "size": e.size, "type": e.type, "added_at": e.added_at} - for e in delta.additions - ], + "additions": [index_entry_wire(e) for e in delta.additions], "deletions": delta.deletions, + "updates": [index_entry_wire(e) for e in delta.updates], } else: msg = { @@ -909,11 +993,7 @@ class NodeDaemon: "v": MNP_VERSION, "group_id": idx.group_id, "version": idx.version, - "entries": [ - {"id": e.id, "name": e.name, "path": e.path, - "size": e.size, "type": e.type, "added_at": e.added_at} - for e in idx.entries - ], + "entries": [index_entry_wire(e) for e in idx.entries], } pushed = 0 for session in list(self._webrtc._sessions.values()): @@ -942,6 +1022,103 @@ class NodeDaemon: endpoint = f"webrtc:{self._config.node.quic_port}" asyncio.ensure_future(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 + (mediacenter.md §5.2/§10), 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 from ops.set_video_root) + 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_root = await self._roster.video_root(indexer.group_id) + if not video_root: + return + for entry in entries: + if entry.type != "video" or entry.id in self._enriched_attempted: + continue + if not _under_video_root(entry.path, video_root): + continue + file_path = entry_abs_path(indexer.roots, entry) + if not file_path or not file_path.exists(): + continue + self._enriched_attempted.add(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 (ops.set_video_root). + + 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(entry.id) + await self._enrich_new_video_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: @@ -983,6 +1160,12 @@ class NodeDaemon: if self._bundle_store: await self._bundle_store.close() + if self._tmdb_client: + await self._tmdb_client.close() + + if self._media_cache: + await self._media_cache.close() + if self._roster: await self._roster.close() |