diff options
| author | Christophe Besson <cbesson@gmail.com> | 2026-08-24 15:57:41 +0200 |
|---|---|---|
| committer | Christophe Besson <cbesson@gmail.com> | 2026-08-24 15:57:41 +0200 |
| commit | c5585beab3d6adefaa2ef9444946dd3816960a7c (patch) | |
| tree | a321e540c2db0458716d526e4a045fca123937b2 /packages/meshbay-node | |
| parent | 317f09328ed8bf20148b707470c9b0fe82e59575 (diff) | |
| download | meshbay-c5585beab3d6adefaa2ef9444946dd3816960a7c.tar.gz | |
fix(node,hub): HEVC transcode fallback, live-add progress, per-group TMDB toggle
Three bugs found live testing the Videos app against a real HEVC/EAC3 show,
plus a design change requested afterward:
- Streaming always did "-c:v copy", which faithfully reports a source's real
hev1 codec string but is unplayable in a browser with no HEVC decoder
(most Chrome/Linux builds). The node now transcodes to H264 whenever the
probed codec is browser-incompatible (media_probe.py's new
BROWSER_INCOMPATIBLE_VIDEO_CODECS), with a `transcode_incompatible_video`
node.toml opt-out for operators who know their viewers already decode it.
- Dropping a whole season into an already-watched folder gave no scanning
indicator and no progress bar: IndexProgress was only ever updated by the
two bulk scan paths, never by the real-time per-file watchdog path
(_schedule_update/_debounce/_update_entry). That path now accounts a
"burst" the same way, without double-counting a file rewritten mid-debounce.
- A stray literal "0" rendered in the video detail modal when there was no
TMDB match (`meta.confidence` is 0, and `0 && x` renders "0" in JSX/htm,
not nothing) — `confident` is now a real boolean.
- Whether TMDB is used at all moves from a node-wide setting to per-group
(OP_TMDB_ENABLED/tmdb_enabled/tmdb_enabled_ack, scoped like OP_VIDEO_ROOT):
an operator running a real media-library group alongside test/demo groups
on one node wants outbound TMDB traffic for the one that needs it, not all
of them. The custom API token and query language stay node-wide, one
shared credential/cache (tmdb_config/OP_TMDB_CONFIG, unchanged reasoning).
MNP_VERSION 0.6 -> 0.7, additive.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LAmyXtc6dAADsH23ydXQpY
Diffstat (limited to 'packages/meshbay-node')
16 files changed, 750 insertions, 202 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py index 8372e5c..e7f3116 100644 --- a/packages/meshbay-node/src/meshbay_node/config.py +++ b/packages/meshbay-node/src/meshbay_node/config.py @@ -46,6 +46,11 @@ device_request_ttl_minutes = 60 # lower it on a Pi. max_concurrent_streams = 8 +# HEVC sources have no browser decoder on most platforms, so streaming one is +# transcoded to H264 rather than the usual free copy — real CPU per viewer. +# Set to false only if every viewer's client is known to decode HEVC itself. +transcode_incompatible_video = true + # Browser and native clients reach this node over WebRTC DataChannel via hub # signaling — no inbound port to open. QUIC is the optional direct path. @@ -120,6 +125,14 @@ class NodeConfig: # the node answers "server busy" — see MAX_CONCURRENT_TRANSCODES in # transport/webrtc_server.py for what one costs. max_concurrent_streams: int = 8 + # HEVC (and any future codec in media_probe.py's + # BROWSER_INCOMPATIBLE_VIDEO_CODECS) has no decoder in most browsers, so + # streaming it needs a real re-encode to H264 rather than the usual free + # copy. On by default since the alternative is a hard "codec not + # supported" error; set to false if this node's viewers are all clients + # that already decode the source codec directly, since transcoding costs + # real CPU per concurrent viewer, unlike the copy path. + transcode_incompatible_video: bool = True @dataclass @@ -275,6 +288,8 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config: cfg.node.max_concurrent_streams = _positive( nd.get("max_concurrent_streams", cfg.node.max_concurrent_streams), cfg.node.max_concurrent_streams, "max_concurrent_streams") + cfg.node.transcode_incompatible_video = bool( + nd.get("transcode_incompatible_video", cfg.node.transcode_incompatible_video)) # Multi-group: [[groups]] array if "groups" in raw: diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index bf6bd64..7691467 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -334,6 +334,12 @@ class NodeDaemon: # group — "" means the whole group index. "video_root": await self._roster.video_root( group_cfg.id) if self._roster else "", + # Whether TMDB lookups run for this group at all — + # per-group (2026-08-24, used to be node-wide), same + # "read once, kept current in place by the signed op" + # shape as video_root above. + "tmdb_enabled": await self._roster.tmdb_enabled( + group_cfg.id) if self._roster else True, } if not groups_ctx: @@ -364,13 +370,14 @@ class NodeDaemon: 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 + # Token/language only — read once at load, 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 these stay node-wide, one + # shared credential/cache. Whether TMDB is used at all is now + # per-group instead — see each group's own "tmdb_enabled" in + # groups_ctx above. + tmdb_token, tmdb_language = await self._roster.tmdb_config() self._state["tmdb_token_customized"] = bool(tmdb_token) self._state["tmdb_language"] = tmdb_language or "" log.info("Media cache opened: %s", media_cache_db) @@ -390,6 +397,7 @@ class NodeDaemon: groups=groups_ctx, denylist=denylist, max_concurrent_streams=self._config.node.max_concurrent_streams, + transcode_incompatible_video=self._config.node.transcode_incompatible_video, ) # No global chat_store here: each group's store lives in # groups_ctx[gid]["chat_store"] and is resolved per session via @@ -721,6 +729,9 @@ class NodeDaemon: "video_root": ( await self._roster.video_root(group_cfg.id) if self._roster else ""), + "tmdb_enabled": ( + await self._roster.tmdb_enabled(group_cfg.id) + if self._roster else True), "chat_store": store, } groups_ctx[group_cfg.id] = new_ctx diff --git a/packages/meshbay-node/src/meshbay_node/indexer/enrich.py b/packages/meshbay-node/src/meshbay_node/indexer/enrich.py index 784b2a3..4dddc27 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/enrich.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/enrich.py @@ -131,7 +131,7 @@ class Enricher: fields: dict = {} duration: float | None = None try: - _codec, duration, _has_audio, width, height = await asyncio.wait_for( + _codec, duration, _has_audio, width, height, _raw = await asyncio.wait_for( probe_video(str(file_path)), timeout=PROBE_TIMEOUT_SECS) fields["duration"] = int(duration) if duration else None fields["width"] = width diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index b479f7e..b643046 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -226,6 +226,18 @@ class DirectoryIndexer: self._reconciler: asyncio.Task | None = None self._pending_timers: dict[str, asyncio.TimerHandle] = {} self.progress = IndexProgress() + # Real-time watchdog adds (_debounce/_update_entry below) previously + # never touched `progress` at all — a whole season dropped into an + # already-watched folder gave the operator no scanning indicator and + # no progress bar, files just appeared one at a time with no feedback + # (found live). `_burst_inflight` counts files currently scheduled or + # being hashed in the current burst; `scanning` only drops back to + # False once it reaches zero *and* no more timers are pending — a + # debounce timer firing is not the same as the hash it schedules + # having finished, and the whole point of a progress indicator is to + # stay up for exactly as long as the slow part (hashing) is running. + self._burst_inflight = 0 + self._burst_sizes: dict[str, int] = {} @property def index(self) -> GroupIndex: @@ -637,6 +649,28 @@ class DirectoryIndexer: if old: old.cancel() + # Progress accounting for the real-time path — see `_burst_inflight`'s + # docstring in __init__. Only on this key's *first* appearance in the + # current burst: a rapid re-trigger of the same path (several writes + # debounced together) cancels the old timer above and must not also + # double-count its size, so `_burst_sizes` (keyed the same as + # `_pending_timers`) is the source of truth for "already counted", + # not "does a timer currently exist for it" — a cancelled timer's + # bookkeeping still has to reach the `fire()` that actually runs. + if not deleted and key not in self._burst_sizes: + try: + size = file_path.stat().st_size + except OSError: + size = 0 + if not self.progress.scanning: + self.progress.scanning = True + self.progress.scanned_bytes = 0 + self.progress.total_bytes = 0 + self.progress.total_bytes += size + self.progress.current_dir = file_path.parent.name + self._burst_sizes[key] = size + self._burst_inflight += 1 + def fire() -> None: self._pending_timers.pop(key, None) asyncio.ensure_future(self._update_entry(file_path, deleted)) @@ -654,41 +688,55 @@ class DirectoryIndexer: self._index.remove_entry(entry.id) async def _update_entry(self, file_path: Path, deleted: bool) -> None: - root = self._root_for(file_path) - if root is None: - return + try: + root = self._root_for(file_path) + if root is None: + return - if deleted and not root.is_live(): - # The volume went away rather than the file. Freeze: mark the root - # and touch nothing. Every other event for this root will arrive - # here too and be dropped the same way, which is the intent — one - # unplugged drive must not empty a library. - if root.available: - root.available = False - self._index.roots = self.roots.describe() - log.warning("Root %r disappeared — ignoring deletion events and " - "freezing %d entries", root.name, - len(self._entries_under(root))) - self._index.version = int(time.time()) - if self.on_change: - await self.on_change(self) - return + if deleted and not root.is_live(): + # The volume went away rather than the file. Freeze: mark the + # root and touch nothing. Every other event for this root + # will arrive here too and be dropped the same way, which is + # the intent — one unplugged drive must not empty a library. + if root.available: + root.available = False + self._index.roots = self.roots.describe() + log.warning("Root %r disappeared — ignoring deletion events and " + "freezing %d entries", root.name, + len(self._entries_under(root))) + self._index.version = int(time.time()) + if self.on_change: + await self.on_change(self) + return - if not root.available: - return + if not root.available: + return - self._remove_by_path(root, file_path) + self._remove_by_path(root, file_path) - if not deleted: - entry = await self._hash_or_cached(root, file_path) - if entry: - self._index.add_entry(entry) - log.debug("Indexed: %s (%s, %d bytes)", - file_path.name, entry.id[:8], entry.size) + if not deleted: + entry = await self._hash_or_cached(root, file_path) + if entry: + self._index.add_entry(entry) + log.debug("Indexed: %s (%s, %d bytes)", + file_path.name, entry.id[:8], entry.size) - self._index.version = int(time.time()) - if self.on_change: - await self.on_change(self) + self._index.version = int(time.time()) + if self.on_change: + await self.on_change(self) + finally: + # Mirror image of the accounting in _debounce, run whichever way + # this method exits (including the several early returns above) — + # otherwise a frozen/unavailable root's files would leave + # `scanning` stuck True forever, the exact bug this is fixing but + # in the other direction. + size = self._burst_sizes.pop(str(file_path), None) + if size is not None: + self.progress.scanned_bytes += size + self._burst_inflight -= 1 + if self._burst_inflight <= 0 and not self._pending_timers: + self.progress.scanning = False + self.progress.current_dir = "" class _WatchdogHandler(FileSystemEventHandler): diff --git a/packages/meshbay-node/src/meshbay_node/media_probe.py b/packages/meshbay-node/src/meshbay_node/media_probe.py index 8ad883d..6dd3baa 100644 --- a/packages/meshbay-node/src/meshbay_node/media_probe.py +++ b/packages/meshbay-node/src/meshbay_node/media_probe.py @@ -11,11 +11,22 @@ import json _H264_PROFILES = {"Baseline": "42", "Main": "4d", "High": "64", "High 10": "6e"} +# Source video codecs whose MSE codec string is real but which no mainstream +# browser can actually decode via MediaSource on most desktop platforms — HEVC +# has no royalty-free decoder in Chrome/Firefox on Linux (and is spotty even +# on platforms with one). Found live: a real HEVC/EAC3 WEB-DL reported +# "Codec not supported for streaming: hev1.1.6.L93.B0,mp4a.40.2" from +# MediaSource.isTypeSupported, even though ffprobe/VLC play it fine. VP9/AV1 +# are not in this set — those decode natively in every mainstream browser. +BROWSER_INCOMPATIBLE_VIDEO_CODECS = frozenset({"hevc"}) -async def probe_video(path: str) -> tuple[str | None, float, bool, int | None, int | None]: + +async def probe_video( + path: str, +) -> tuple[str | None, float, bool, int | None, int | None, str | None]: """ Probe video file with ffprobe, return (MSE codec string, duration, - has_audio, width, height). + has_audio, width, height, raw video codec name). The audio half of the codec string is always "mp4a.40.2" (AAC-LC) or absent — never the source's real audio codec — because the streaming @@ -24,8 +35,13 @@ async def probe_video(path: str) -> tuple[str | None, float, bool, int | None, i that (AC-3, E-AC-3, DTS, ...) is at best silently unplayable and at worst, for E-AC-3 at least, makes ffmpeg itself refuse to write the fragmented MP4 header ("Cannot write moov atom before EAC3 packets - parsed" — reproduced against a real 5.1 E-AC-3 WEB-DL). Video stays - whatever it actually is: it is always copied, never transcoded. + parsed" — reproduced against a real 5.1 E-AC-3 WEB-DL). Video is copied + whenever the browser can decode it directly; the raw codec name is + returned alongside the MSE string so the caller can decide whether this + source needs a real re-encode instead (BROWSER_INCOMPATIBLE_VIDEO_CODECS + above) — the MSE string alone can't drive that decision, since it still + faithfully reports "hev1..." for a source this pipeline cannot actually + deliver copied. width/height come from the same ffprobe call (one extra `-show_entries` field, no second process spawn) — resolution is deliberately never @@ -43,12 +59,14 @@ async def probe_video(path: str) -> tuple[str | None, float, bool, int | None, i duration = float(info.get("format", {}).get("duration", 0)) v_codec = "" + raw_codec_name: str | None = None has_audio = False width: int | None = None height: int | None = None for s in info.get("streams", []): if s.get("codec_type") == "video" and not v_codec: cn = s.get("codec_name", "") + raw_codec_name = cn or None if cn == "h264": p = _H264_PROFILES.get(s.get("profile", "High"), "64") lvl = int(s.get("level", 40)) @@ -65,6 +83,6 @@ async def probe_video(path: str) -> tuple[str | None, float, bool, int | None, i has_audio = True if not v_codec: - return None, duration, has_audio, width, height + return None, duration, has_audio, width, height, raw_codec_name codec = f"{v_codec},mp4a.40.2" if has_audio else v_codec - return codec, duration, has_audio, width, height + return codec, duration, has_audio, width, height, raw_codec_name diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py index 6848e3a..a232ba2 100644 --- a/packages/meshbay-node/src/meshbay_node/ops.py +++ b/packages/meshbay-node/src/meshbay_node/ops.py @@ -733,23 +733,23 @@ async def set_enabled_apps(state: dict, group_id: str, apps: list[str]) -> dict: # ── TMDB config (Videos app) ───────────────────────────────────────────────── -async def set_tmdb_config(state: dict, enabled: bool, token: str | None = None, +async def set_tmdb_config(state: dict, token: str | None = None, language: str | None = None) -> dict: """ - Whether the node calls TMDB at all, whether it uses a custom API token - instead of the shipped default, and in what language it queries TMDB - (docs/mediacenter.md §5.5). + Whether the node uses a custom API token instead of the shipped default, + and in what language it queries TMDB (docs/mediacenter.md §5.5). Node-wide (roster.py group_settings, group_id="") rather than per-group - like set_member_upload/set_enabled_apps: TMDB is one operator's budget, - one credential and one shared cache, not a per-group or per-viewer - concern. `token=""` explicitly clears a previously-set custom token - (reverts to the shipped default); `token=None` leaves whatever was - there unchanged. Same discipline for `language`. + like set_member_upload/set_enabled_apps: the token and the shared-cache + language are one operator's budget and one credential, not a per-group + or per-viewer concern. Whether TMDB is used *at all* is the per-group + decision set_tmdb_enabled below makes instead. `token=""` explicitly + clears a previously-set custom token (reverts to the shipped default); + `token=None` leaves whatever was there unchanged. Same discipline for + `language`. """ roster = _roster(state) - await roster.set_tmdb_config(enabled, token, language, set_by=state.get("node_user_id", "")) - state["tmdb_enabled"] = enabled + await roster.set_tmdb_config(token, language, set_by=state.get("node_user_id", "")) # `token=None` means "leave whatever was there" (§ set_tmdb_config's own # docstring) — so the customized flag only changes when a value (a real # token, or "" to clear one) was actually given. @@ -757,15 +757,30 @@ async def set_tmdb_config(state: dict, enabled: bool, token: str | None = None, state["tmdb_token_customized"] = bool(token) if language is not None: state["tmdb_language"] = language - log.info("TMDB config: enabled=%s custom_token=%s language=%s", - enabled, bool(token), language or state.get("tmdb_language", "")) + log.info("TMDB config: custom_token=%s language=%s", + bool(token), language or state.get("tmdb_language", "")) return { - "enabled": enabled, "token_customized": state.get("tmdb_token_customized", False), "language": state.get("tmdb_language", ""), } +async def set_tmdb_enabled(state: dict, group_id: str, enabled: bool) -> dict: + """ + Whether TMDB lookups run for this group at all (docs/mediacenter.md + §5.5) — per-group, unlike set_tmdb_config above: an operator running a + real media library alongside test/demo groups on one node wants + outbound TMDB traffic (and API quota) spent for the one that needs it, + not all of them just because one process serves both. + """ + roster = _roster(state) + ctx = _group_ctx(state, group_id) + await roster.set_tmdb_enabled(group_id, enabled, set_by=state.get("node_user_id", "")) + ctx["tmdb_enabled"] = enabled + log.info("TMDB enabled for group %s: %s", group_id[:8], enabled) + return {"enabled": enabled, "group_id": group_id} + + async def set_video_root(state: dict, group_id: str, path: str) -> dict: """ Which folder (possibly a subfolder of a shared root) is the Videos app's diff --git a/packages/meshbay-node/src/meshbay_node/roster.py b/packages/meshbay-node/src/meshbay_node/roster.py index 701462f..b4efb7c 100644 --- a/packages/meshbay-node/src/meshbay_node/roster.py +++ b/packages/meshbay-node/src/meshbay_node/roster.py @@ -599,13 +599,20 @@ class Roster: json.dumps(sorted(apps)), set_by) return apps - # TMDB is one operator's budget and credential, not a per-group concern - # (docs/mediacenter.md §5.5) — stored under the group_id="" sentinel, - # the same precedent as `roster.get_member("", user_id)` authorizing the - # operator node-wide (desktop-client-v1.md §6.3). Unset means "on, using - # the shipped default token" — the same "absent means the old behaviour" - # discipline member_upload/enabled_apps already follow. - SETTING_TMDB_ENABLED = "tmdb_enabled" + # The TMDB credential and query language are one operator's budget, not a + # per-group concern (docs/mediacenter.md §5.5) — stored under the + # group_id="" sentinel, the same precedent as `roster.get_member("", + # user_id)` authorizing the operator node-wide (desktop-client-v1.md + # §6.3). Unset means "the shipped default token, TMDB's own default + # language" — the same "absent means the old behaviour" discipline + # member_upload/enabled_apps already follow. + # + # Whether TMDB is used *at all*, though, is per-group (moved off the + # node-wide sentinel below, 2026-08-24): an operator running a real media + # library alongside test/demo groups wants outbound TMDB traffic for the + # one group that needs it, not all of them just because one node process + # serves both. See SETTING_TMDB_ENABLED's own per-group methods further + # down, next to video_root. SETTING_TMDB_TOKEN = "tmdb_api_token" # A TMDB language tag (e.g. "fr-FR") — one for the whole node, same # reasoning as the token: one shared cache, not a per-viewer request. @@ -614,18 +621,14 @@ class Roster: SETTING_TMDB_LANGUAGE = "tmdb_language" NODE_WIDE_GROUP_ID = "" - async def tmdb_config(self) -> tuple[bool, str | None, str | None]: - """Returns (enabled, custom_token_or_None, language_or_None).""" - enabled = (await self.get_setting( - self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_ENABLED, "1")) != "0" + async def tmdb_config(self) -> tuple[str | None, str | None]: + """Returns (custom_token_or_None, language_or_None).""" token = await self.get_setting(self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_TOKEN) language = await self.get_setting(self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_LANGUAGE) - return enabled, (token or None), (language or None) + return (token or None), (language or None) - async def set_tmdb_config(self, enabled: bool, token: str | None = None, + async def set_tmdb_config(self, token: str | None = None, language: str | None = None, set_by: str = "") -> None: - await self.set_setting(self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_ENABLED, - "1" if enabled else "0", set_by) if token is not None: await self.set_setting(self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_TOKEN, token, set_by) @@ -634,8 +637,9 @@ class Roster: language, set_by) # Which folder is the Videos app's entry point for this group — per-group - # (unlike tmdb_config above), since different groups share different - # trees. Empty/unset means the whole group index, exactly as today. + # (unlike the token/language above), since different groups share + # different trees. Empty/unset means the whole group index, exactly as + # today. SETTING_VIDEO_ROOT = "video_root" async def video_root(self, group_id: str) -> str: @@ -645,6 +649,22 @@ class Roster: await self.set_setting(group_id, self.SETTING_VIDEO_ROOT, path or "", set_by) return path or "" + # Whether TMDB lookups run for this group at all — per-group, unlike the + # token/language above: one node process can share a real media library + # group and several test/demo groups, and outbound TMDB traffic (and API + # quota) for the demo groups is not something turning it on for the real + # one should imply. Unset means on, same "absent means the old + # behaviour" discipline as everything else here — a node that predates + # this setting keeps working exactly as before for every group. + SETTING_TMDB_ENABLED = "tmdb_enabled" + + async def tmdb_enabled(self, group_id: str) -> bool: + return (await self.get_setting(group_id, self.SETTING_TMDB_ENABLED, "1")) != "0" + + async def set_tmdb_enabled(self, group_id: str, enabled: bool, set_by: str = "") -> None: + await self.set_setting(group_id, self.SETTING_TMDB_ENABLED, + "1" if enabled else "0", set_by) + # How often the indexer's reconciliation backstop runs, and how long it # waits after the last change on a file before hashing it. Unset means # the indexer's own defaults — an existing group's behaviour must not diff --git a/packages/meshbay-node/src/meshbay_node/tmdb.py b/packages/meshbay-node/src/meshbay_node/tmdb.py index 448a2b9..a530660 100644 --- a/packages/meshbay-node/src/meshbay_node/tmdb.py +++ b/packages/meshbay-node/src/meshbay_node/tmdb.py @@ -20,6 +20,12 @@ Results also come back in whatever language the operator configured node, same reasoning as the token: one shared cache, not a per-viewer request. Omitted entirely when unset, which lets TMDB fall back to its own default (English) rather than this client guessing one. + +Whether TMDB is used *at all* is a **per-group** decision (roster.py's +`tmdb_enabled(group_id)`, moved off the node-wide sentinel 2026-08-24) — +this client has no group in scope, so that check happens once, in +webrtc_server.py, before any of this client's methods are ever called for a +given request. This client only resolves the shared credential/language. """ import difflib @@ -80,17 +86,24 @@ class TmdbClient: await self._client.aclose() async def _resolve(self) -> tuple[bool, str | None, str | None]: - """Returns (enabled, token, language). token/language are None when unset.""" + """ + Returns (has_token, token, language). token/language are None when + unset. Whether TMDB is used *at all* is a per-group decision made by + the caller (roster.tmdb_enabled(group_id), checked in + webrtc_server.py before any of this client's methods are called) — + this client only knows the node-wide credential/language, and has no + group to check against. + """ if self._roster is not None: - enabled, custom_token, language = await self._roster.tmdb_config() + custom_token, language = await self._roster.tmdb_config() else: - enabled, custom_token, language = True, None, None + custom_token, language = None, None token = custom_token or os.environ.get(_DEFAULT_TOKEN_ENV) or None - return enabled and bool(token), token, language + return bool(token), token, language async def _get(self, path: str, params: dict) -> dict | None: - enabled, token, language = await self._resolve() - if not enabled: + has_token, token, language = await self._resolve() + if not has_token: return None if language and "language" not in params: params = {**params, "language": language} diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index f4d1dee..c35e3ea 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -66,6 +66,7 @@ from meshbay_common.adminop import ( OP_APPS_ENABLED, OP_SET_SCAN_SETTINGS, OP_TMDB_CONFIG, + OP_TMDB_ENABLED, OP_VIDEO_ROOT, OP_TMDB_OVERRIDE, OP_ROOT_ADD, @@ -96,7 +97,10 @@ from meshbay_node import ops # this module still refers to it as `_probe_video`. The implementation lives # in media_probe.py so the indexer package (imported just above) can call it # too, for index-time enrichment, without a circular import. -from meshbay_node.media_probe import probe_video as _probe_video +from meshbay_node.media_probe import ( + BROWSER_INCOMPATIBLE_VIDEO_CODECS, + probe_video as _probe_video, +) from meshbay_node.roots import ( RootSet, entry_abs_path, SAFE_UPLOAD_NAME, safe_subdir, _free_name, ) @@ -375,6 +379,8 @@ class WebRTCPeerSession: self._do_set_scan_settings(msg) elif mtype == MNP.TMDB_CONFIG: self._do_tmdb_config(msg) + elif mtype == MNP.TMDB_ENABLED: + self._do_tmdb_enabled(msg) elif mtype == MNP.VIDEO_ROOT: self._do_video_root(msg) elif mtype == MNP.MEDIA_META_REQ: @@ -638,13 +644,14 @@ class WebRTCPeerSession: # Which folder the Videos app treats as its entry point for # this group — "" means the whole group index. "video_root": self._group_ctx().get("video_root") or "", - # Node-wide (not per-group), same "read once, kept current by - # the signed op" shape — surfaced here rather than only via - # tmdb_config_ack so a client that connects after the operator - # already configured it does not have to wait for a live change - # to find out (docs/mediacenter.md §5.5). - "tmdb_enabled": bool( - self._ctx.get("daemon_state", {}).get("tmdb_enabled", True)), + # Per-group (2026-08-24 — used to be node-wide), same "read once, + # kept current in place by the signed op" shape as video_root + # above — surfaced here rather than only via tmdb_enabled_ack so + # a client that connects after the operator already configured + # it does not have to wait for a live change to find out. + "tmdb_enabled": bool(self._group_ctx().get("tmdb_enabled", True)), + # Token/language stay node-wide (one shared credential/cache) — + # via daemon_state, kept current by tmdb_config_ack. "tmdb_token_customized": bool( self._ctx.get("daemon_state", {}).get("tmdb_token_customized", False)), "tmdb_language": str( @@ -1664,19 +1671,15 @@ class WebRTCPeerSession: def _do_tmdb_config(self, msg: dict) -> None: """ - Turn TMDB lookups on/off node-wide, optionally set (or clear) a - custom API token, and optionally set the language TMDB is queried - in (e.g. "fr-FR") — one for the whole node, same reasoning as the - token: one shared cache, not a per-viewer request. Signed like the - rest: this turns on outbound third-party network traffic the node - did not have before the Videos app (docs/mediacenter.md §5.5, §8) - — an unsigned toggle would let any member turn on egress the - operator never agreed to. + Optionally set (or clear) a custom TMDB API token, and optionally set + the language TMDB is queried in (e.g. "fr-FR") — one for the whole + node, since both are one operator's shared credential/cache, not a + per-group concern (see _do_tmdb_enabled for the per-group on/off + switch). Signed like the rest: this changes outbound third-party + network traffic the node did not have before the Videos app + (docs/mediacenter.md §5.5, §8) — an unsigned change would let any + member alter egress the operator never agreed to. """ - enabled = msg.get("enabled") - if not isinstance(enabled, bool): - self._send({"type": "error", "detail": "Missing or invalid 'enabled'"}) - return token = msg.get("token") if token is not None and not isinstance(token, str): self._send({"type": "error", "detail": "Invalid 'token'"}) @@ -1693,11 +1696,10 @@ class WebRTCPeerSession: # in plaintext). The actual token travels only in `payload`, which # is node-side context, never re-sent or re-verified from the wire. # The language is not a secret, so it travels in the subject itself. - subject = (f"enabled={enabled},custom_token={'yes' if token else 'no'}," - f"language={language or 'default'}") + subject = f"custom_token={'yes' if token else 'no'},language={language or 'default'}" self._issue_admin_challenge( OP_TMDB_CONFIG, subject, - payload={"enabled": enabled, "token": token, "language": language}, + payload={"token": token, "language": language}, group_id="") async def _admin_exec_tmdb_config( @@ -1710,17 +1712,18 @@ class WebRTCPeerSession: p = pending.get("payload") or {} try: result = await self._run_op( - ops.set_tmdb_config, p.get("enabled", True), p.get("token"), p.get("language")) + ops.set_tmdb_config, p.get("token"), p.get("language")) except ops.OpError as e: self._send({"type": "error", "detail": e.message}) return self._audit("tmdb_config", pending["subject"]) # Node-wide setting: every connected peer in every group is told, not - # just this group's peers (unlike apps_enabled/member_upload). + # just this group's peers (unlike apps_enabled/member_upload/the + # per-group tmdb_enabled below). notice = { "type": MNP.TMDB_CONFIG_ACK, "v": MNP_VERSION, - "enabled": result["enabled"], "token_customized": result["token_customized"], + "token_customized": result["token_customized"], "language": result["language"], } for gctx in self._ctx.get("groups", {}).values(): @@ -1730,6 +1733,44 @@ class WebRTCPeerSession: except Exception: pass + def _do_tmdb_enabled(self, msg: dict) -> None: + """ + Whether TMDB lookups run for this group at all. Per-group, unlike + tmdb_config's token/language — see ops.set_tmdb_enabled. Signed like + video_root: it decides whether this group's members' Videos tab ever + makes outbound TMDB traffic. + """ + enabled = msg.get("enabled") + if not isinstance(enabled, bool): + self._send({"type": "error", "detail": "Missing or invalid 'enabled'"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + self._issue_admin_challenge(OP_TMDB_ENABLED, str(enabled)) + + async def _admin_exec_tmdb_enabled( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + enabled = pending["subject"] == "True" + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"tmdb_enabled:{pending['subject']}") + return + try: + await self._run_op(ops.set_tmdb_enabled, self._group_id or "", enabled) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + self._audit("tmdb_enabled", pending["subject"]) + + notice = {"type": MNP.TMDB_ENABLED_ACK, "v": MNP_VERSION, "enabled": enabled} + for uid, session in list(self._peer_registry().items()): + try: + session._send(notice) + except Exception: + pass + def _do_video_root(self, msg: dict) -> None: """ Which folder (possibly a subfolder of a shared root) the Videos app @@ -2405,7 +2446,11 @@ class WebRTCPeerSession: media_cache = self._ctx.get("media_cache") tmdb_client = self._ctx.get("tmdb_client") - if media_cache is None or tmdb_client is None: + # Per-group, not node-wide (docs/mediacenter.md §5.5, 2026-08-24): + # treated exactly like "no client configured" — same silent, no-error + # degradation, since a member's Videos tab already has to handle "no + # TMDB match" as the ordinary case. + if media_cache is None or tmdb_client is None or not ctx.get("tmdb_enabled", True): self._send({"type": MNP.MEDIA_META_RESP, "v": MNP_VERSION, "path": path, "confidence": 0}) return @@ -2476,7 +2521,10 @@ class WebRTCPeerSession: return media_cache = self._ctx.get("media_cache") tmdb_client = self._ctx.get("tmdb_client") - if media_cache is None or tmdb_client is None: + # Per-group, not node-wide (docs/mediacenter.md §5.5, 2026-08-24) — + # same silent zero-confidence degradation as "no client configured". + if (media_cache is None or tmdb_client is None + or not self._group_ctx().get("tmdb_enabled", True)): self._send({"type": MNP.SEASON_META_RESP, "v": MNP_VERSION, "tmdb_id": tmdb_id, "season": season, "confidence": 0}) return @@ -2524,7 +2572,12 @@ class WebRTCPeerSession: return media_cache = self._ctx.get("media_cache") tmdb_client = self._ctx.get("tmdb_client") - if media_cache is None or tmdb_client is None: + # Per-group, not node-wide (docs/mediacenter.md §5.5, 2026-08-24) — + # same silent empty-results degradation as "no client configured": + # a member with TMDB off for this group sees the same "type it in + # yourself" affordance either way, never an error. + if (media_cache is None or tmdb_client is None + or not self._group_ctx().get("tmdb_enabled", True)): self._send({"type": MNP.TMDB_SEARCH_RESP, "v": MNP_VERSION, "query": query, "media_type": media_type, "results": []}) return @@ -3202,6 +3255,9 @@ class WebRTCPeerSession: elif pending["op"] == OP_TMDB_CONFIG: self._spawn( self._admin_exec_tmdb_config(pending, transcript, sig_bytes)) + elif pending["op"] == OP_TMDB_ENABLED: + self._spawn( + self._admin_exec_tmdb_enabled(pending, transcript, sig_bytes)) elif pending["op"] == OP_VIDEO_ROOT: self._spawn( self._admin_exec_video_root(pending, transcript, sig_bytes)) @@ -3479,7 +3535,8 @@ class WebRTCPeerSession: file_hash = bytes.fromhex(entry.id) try: - codec_str, duration, has_audio, _width, _height = await _probe_video(str(file_path)) + codec_str, duration, has_audio, _width, _height, raw_video_codec = \ + await _probe_video(str(file_path)) except Exception as e: self._send({"type": "error", "detail": f"Probe failed: {e}"}) return @@ -3508,16 +3565,42 @@ class WebRTCPeerSession: # what every streaming player does, and why the client is told the # value used rather than left to assume its own. seek_args = ["-ss", f"{start:.3f}"] if start > 0 else [] - # Video is always copied — re-encoding it is the expensive thing this - # pipeline exists to avoid, and H264/HEVC/VP9/AV1 already decode fine - # in-browser. Audio is always transcoded to AAC, never copied — see + # Video is copied whenever the browser can decode it directly — + # re-encoding it is the expensive thing this pipeline exists to avoid, + # and H264/VP9/AV1 already decode fine in-browser. HEVC is the one + # exception (BROWSER_INCOMPATIBLE_VIDEO_CODECS, media_probe.py): found + # live, a real HEVC/EAC3 WEB-DL reported "Codec not supported for + # streaming" from MediaSource.isTypeSupported even though ffprobe/VLC + # play it fine — Chrome has no HEVC decoder on most non-Apple + # platforms. The operator can turn this fallback off (node.toml + # transcode_incompatible_video = false) for a client fleet they know + # already decodes HEVC, since it is real CPU cost, unlike the copy + # path. Audio is always transcoded to AAC, never copied — see # _probe_video for why "copy" there is not an option, not even for a # codec that sounds close enough (plain AC-3 has the same in-browser # decode problem as E-AC-3, just without ffmpeg also refusing to mux # it). Transcoding audio is cheap; it does not change the cost model # the transcode-slot semaphore is sized around. + transcode_video = ( + raw_video_codec in BROWSER_INCOMPATIBLE_VIDEO_CODECS + and self._ctx.get("transcode_incompatible_video", True) + ) map_args = ["-map", "0:v:0"] - codec_args = ["-c:v", "copy"] + if transcode_video: + # -pix_fmt yuv420p: a 10-bit or 4:4:4 HEVC source (common for HDR + # WEB-DLs) fails "-profile:v high" outright otherwise — libx264's + # High profile is 8-bit 4:2:0 only. Downsampling loses nothing a + # browser could show anyway (MSE/HTML5 video has no HDR path). + codec_args = ["-c:v", "libx264", "-pix_fmt", "yuv420p", + "-profile:v", "high", "-level", "4.1", + "-preset", "veryfast", "-crf", "21"] + # Must match "-profile:v high -level 4.1" byte-for-byte (avc1.<profile + # hex><constraint><level hex>) — the client checks this string with + # MediaSource.isTypeSupported before trusting a single byte of the + # stream, so a mismatch here fails exactly the check this exists to pass. + codec_str = "avc1.640029,mp4a.40.2" if has_audio else "avc1.640029" + else: + codec_args = ["-c:v", "copy"] if has_audio: map_args += ["-map", "0:a:0"] # Downmixed to stereo: a WEB-DL's 5.1 track becomes 6-channel AAC @@ -3750,6 +3833,7 @@ class WebRTCTransport: denylist: Any | None = None, stun_servers: list[str] | None = None, max_concurrent_streams: int | None = None, + transcode_incompatible_video: bool = True, ): self._ctx: dict[str, Any] = { "sk_node": sk_node, @@ -3761,6 +3845,9 @@ class WebRTCTransport: # None means "the operator said nothing" — the default applies. It # is read once, when the first stream builds the semaphore. "max_concurrent_streams": max_concurrent_streams, + # Operator opt-out (node.toml) for the HEVC-etc. transcode + # fallback in _stream_video_inner — real CPU cost, unlike copy. + "transcode_incompatible_video": transcode_incompatible_video, } if groups: self._ctx["groups"] = groups diff --git a/packages/meshbay-node/tests/test_indexer.py b/packages/meshbay-node/tests/test_indexer.py index 68ccc4c..3eac39e 100644 --- a/packages/meshbay-node/tests/test_indexer.py +++ b/packages/meshbay-node/tests/test_indexer.py @@ -356,6 +356,82 @@ async def test_progress_stops_even_when_hashing_raises(tmp_path, sk_node, gek): "an exception mid-scan must not leave the scanning flag stuck on" +@pytest.mark.asyncio +async def test_realtime_watchdog_add_reports_progress_like_a_bulk_scan(tmp_path, sk_node, gek): + """ + Found live: dropping a whole season into an already-watched folder gave + no scanning indicator and no progress bar at all — only the initial scan + and the periodic reconcile backstop ever touched `progress`, never the + real-time per-file watchdog path (_schedule_update/_debounce/ + _update_entry). Drives that path directly (as _WatchdogHandler would), + without a real filesystem observer, exactly like test_root_availability.py + already does for _update_entry alone. + """ + d = tmp_path / "shared" + d.mkdir() + paths = [] + for i in range(3): + p = d / f"ep{i}.mkv" + p.write_bytes(os.urandom(1024 * (i + 1))) + paths.append(p) + total_size = sum(p.stat().st_size for p in paths) + + indexer = DirectoryIndexer(roots=one_root(d), group_id="g", sk_node=sk_node, + gek=gek, debounce_secs=0.01) + indexer._loop = asyncio.get_running_loop() + assert indexer.progress.scanning is False + + for p in paths: + indexer._schedule_update(p) + # _schedule_update only posts _debounce via call_soon_threadsafe (it is + # written to be called from the watchdog thread) — give the loop one + # turn to actually run the three posted calls before asserting on them. + await asyncio.sleep(0) + + # All three scheduled near-simultaneously (as a burst of watchdog events + # for one `mv` would arrive) — the indicator must flip on immediately, + # before any single file has actually finished hashing. + assert indexer.progress.scanning is True + assert indexer.progress.total_bytes == total_size + assert indexer.progress.scanned_bytes == 0 + + await asyncio.sleep(0.05) # past debounce_secs; lets all three fire and finish + + assert indexer.progress.scanning is False, "must end idle, not stuck scanning" + assert indexer.progress.scanned_bytes == total_size + assert indexer.progress.total_bytes == total_size + assert len(indexer.index.entries) == 3 + + +@pytest.mark.asyncio +async def test_realtime_watchdog_rapid_rewrite_does_not_double_count(tmp_path, sk_node, gek): + """A file rewritten during its own debounce window (on_modified firing + again before the first timer fires) must count its size once, not once + per event — the old timer is cancelled, and its accounting must transfer + to whichever fire() actually runs rather than being counted twice.""" + d = tmp_path / "shared" + d.mkdir() + p = d / "ep0.mkv" + p.write_bytes(os.urandom(2048)) + + indexer = DirectoryIndexer(roots=one_root(d), group_id="g", sk_node=sk_node, + gek=gek, debounce_secs=0.05) + indexer._loop = asyncio.get_running_loop() + + indexer._schedule_update(p) + indexer._schedule_update(p) # re-triggered before the first timer fires + indexer._schedule_update(p) + await asyncio.sleep(0) # let the three posted _debounce calls actually run + + assert indexer.progress.total_bytes == p.stat().st_size, \ + "one file re-triggered must count its size once, not three times" + + await asyncio.sleep(0.1) + + assert indexer.progress.scanning is False + assert indexer.progress.scanned_bytes == p.stat().st_size + + # ── Off-loop directory walks, reconcile backoff ───────────────────────────── @pytest.mark.asyncio diff --git a/packages/meshbay-node/tests/test_season_and_search_requests.py b/packages/meshbay-node/tests/test_season_and_search_requests.py index 8ba55fb..856797e 100644 --- a/packages/meshbay-node/tests/test_season_and_search_requests.py +++ b/packages/meshbay-node/tests/test_season_and_search_requests.py @@ -20,6 +20,7 @@ pytestmark = pytest.mark.asyncio def _session(media_cache=None, tmdb_client=None) -> WebRTCPeerSession: session = WebRTCPeerSession.__new__(WebRTCPeerSession) session._ctx = {"media_cache": media_cache, "tmdb_client": tmdb_client} + session._group_id = None session.sent = [] session._send = session.sent.append return session diff --git a/packages/meshbay-node/tests/test_stream_audio_transcode.py b/packages/meshbay-node/tests/test_stream_audio_transcode.py index dde0df4..aaf1595 100644 --- a/packages/meshbay-node/tests/test_stream_audio_transcode.py +++ b/packages/meshbay-node/tests/test_stream_audio_transcode.py @@ -149,7 +149,7 @@ async def test_probe_video_reports_aac_regardless_of_source_audio_codec(tmp_path clip = tmp_path / "clip.mkv" _make_clip(clip, acodec="eac3", channels=6) - codec, duration, has_audio, width, height = await _probe_video(str(clip)) + codec, duration, has_audio, width, height, raw_codec = await _probe_video(str(clip)) assert has_audio is True assert duration > 0 @@ -157,6 +157,7 @@ async def test_probe_video_reports_aac_regardless_of_source_audio_codec(tmp_path assert "eac3" not in codec and "ec-3" not in codec assert "mp4a.40.2" in codec assert (width, height) == (320, 240) + assert raw_codec == "h264" async def test_probe_video_handles_no_audio_track(tmp_path): @@ -168,9 +169,10 @@ async def test_probe_video_handles_no_audio_track(tmp_path): check=True, capture_output=True, ) - codec, duration, has_audio, width, height = await _probe_video(str(clip)) + codec, duration, has_audio, width, height, raw_codec = await _probe_video(str(clip)) assert has_audio is False assert codec is not None and "," not in codec, \ "no audio track must not produce a dangling ',' or a fake audio codec" assert (width, height) == (320, 240) + assert raw_codec == "h264" diff --git a/packages/meshbay-node/tests/test_stream_hevc_transcode.py b/packages/meshbay-node/tests/test_stream_hevc_transcode.py new file mode 100644 index 0000000..b3f0474 --- /dev/null +++ b/packages/meshbay-node/tests/test_stream_hevc_transcode.py @@ -0,0 +1,154 @@ +""" +HEVC video is transcoded to H264 for streaming, never copied — unlike the +codecs media_probe.py's BROWSER_INCOMPATIBLE_VIDEO_CODECS excludes. + +Found live: a real HEVC/EAC3 WEB-DL streamed fine over MNP (ffprobe/VLC play +it) but the browser reported "Codec not supported for streaming: +hev1.1.6.L93.B0,mp4a.40.2" from MediaSource.isTypeSupported — Chrome has no +HEVC decoder on most non-Apple platforms. "-c:v copy" on an incompatible +codec is not a mux failure the way EAC3 audio is (test_stream_audio_ +transcode.py); ffmpeg happily remuxes it, and the browser is the one that +then refuses it, silently, at playback rather than at stream_init. + +These tests spawn real ffmpeg/ffprobe against small synthetic files (lavfi +test sources, ~1s), the same style as test_stream_audio_transcode.py. +""" + +import shutil +import subprocess +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from meshbay_common.crypto import generate_gek +from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.transport.webrtc_server import WebRTCPeerSession, _probe_video + +from conftest import one_root + +_HAVE_FFMPEG = shutil.which("ffmpeg") and shutil.which("ffprobe") +_HAVE_HEVC_ENCODER = _HAVE_FFMPEG and b"libx265" in subprocess.run( + ["ffmpeg", "-hide_banner", "-encoders"], capture_output=True).stdout +pytestmark = [ + pytest.mark.asyncio, + pytest.mark.skipif(not _HAVE_HEVC_ENCODER, reason="ffmpeg/libx265 not installed"), +] + + +def _make_hevc_clip(path: Path) -> None: + """~1s of HEVC video + AAC audio — a minimal stand-in for a real HEVC WEB-DL.""" + subprocess.run( + ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y", + "-f", "lavfi", "-i", "testsrc=size=320x240:rate=25:duration=1", + "-f", "lavfi", "-i", "sine=frequency=440:duration=1:sample_rate=48000", + "-c:v", "libx265", "-preset", "ultrafast", "-c:a", "aac", + str(path)], + check=True, capture_output=True, + ) + + +def _session(video_path: Path, gek: bytes, *, transcode_incompatible_video: bool = True): + import blake3 + file_bytes = video_path.read_bytes() + file_id = blake3.blake3(file_bytes).hexdigest() + + sk_node = Ed25519PrivateKey.generate() + index = GroupIndex(group_id="g" * 32, sk_node=sk_node, gek=gek) + from meshbay_common.protocol import IndexEntry + index.add_entry(IndexEntry( + id=file_id, name=video_path.name, path=video_path.parent.name, + size=len(file_bytes), type="video", added_at=0)) + + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = { + "roots": one_root(video_path.parent), + "index": index, + "gek": gek, + "sk_node": sk_node, + "max_concurrent_streams": 4, + "transcode_incompatible_video": transcode_incompatible_video, + } + session._group_id = None + session._user_id = "tester" + session._stream_stopped = False + session._stream_keepalives = 0 + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session, file_id + + +def _reassemble(sent: list[dict], gek: bytes, file_id: str) -> bytes: + file_hash = bytes.fromhex(file_id) + segments = sorted( + (m for m in sent if m.get("type") == "stream_data"), + key=lambda m: m["segment_index"]) + out = b"" + for m in segments: + key = chunk_key_aes(gek, file_hash, m["segment_index"]) + out += decrypt_chunk_aes(key, m["nonce"], m["ct"]) + return out + + +def _output_video_codec(path: Path) -> str: + probe = subprocess.run( + ["ffprobe", "-v", "error", "-select_streams", "v:0", + "-show_entries", "stream=codec_name", "-of", "csv=p=0", str(path)], + check=True, capture_output=True, text=True) + return probe.stdout.strip() + + +async def test_hevc_video_is_transcoded_to_h264_by_default(tmp_path): + clip = tmp_path / "clip.mkv" + _make_hevc_clip(clip) + gek = generate_gek() + session, file_id = _session(clip, gek) + + await session._stream_video_inner({"file_id": file_id, "start": 0, "credits": 0}) + + errors = [m for m in session.sent if m.get("type") == "error"] + assert not errors, f"streaming must not fail: {errors}" + + init = next(m for m in session.sent if m.get("type") == "stream_init") + assert init["codec"].startswith("avc1."), ( + "the reported codec must be the transcoded H264 string, never the " + f"source's hev1 string a browser cannot decode: {init['codec']}") + + remuxed = _reassemble(session.sent, gek, file_id) + out_path = tmp_path / "out.mp4" + out_path.write_bytes(remuxed) + assert _output_video_codec(out_path) == "h264", \ + "the bytes on the wire must actually be H264, not just the reported label" + + +async def test_hevc_transcode_can_be_disabled_by_the_operator(tmp_path): + clip = tmp_path / "clip.mkv" + _make_hevc_clip(clip) + gek = generate_gek() + session, file_id = _session(clip, gek, transcode_incompatible_video=False) + + await session._stream_video_inner({"file_id": file_id, "start": 0, "credits": 0}) + + assert not [m for m in session.sent if m.get("type") == "error"] + init = next(m for m in session.sent if m.get("type") == "stream_init") + assert init["codec"].startswith("hev1."), ( + "with the fallback disabled, the source is copied as-is and the " + f"original HEVC codec string must be reported unchanged: {init['codec']}") + + remuxed = _reassemble(session.sent, gek, file_id) + out_path = tmp_path / "out.mp4" + out_path.write_bytes(remuxed) + assert _output_video_codec(out_path) == "hevc", \ + "with the fallback disabled, the wire bytes must still be copied HEVC" + + +async def test_probe_video_reports_raw_codec_name_for_hevc(tmp_path): + clip = tmp_path / "clip.mkv" + _make_hevc_clip(clip) + + codec, duration, has_audio, width, height, raw_codec = await _probe_video(str(clip)) + + assert raw_codec == "hevc" + assert codec is not None and codec.startswith("hev1.") diff --git a/packages/meshbay-node/tests/test_tmdb.py b/packages/meshbay-node/tests/test_tmdb.py index 4500feb..fcbc2a2 100644 --- a/packages/meshbay-node/tests/test_tmdb.py +++ b/packages/meshbay-node/tests/test_tmdb.py @@ -7,14 +7,12 @@ from meshbay_node.tmdb import TmdbClient class FakeRoster: - def __init__(self, enabled: bool = True, token: str | None = "fake-token", - language: str | None = None): - self._enabled = enabled + def __init__(self, token: str | None = "fake-token", language: str | None = None): self._token = token self._language = language async def tmdb_config(self): - return self._enabled, self._token, self._language + return self._token, self._language def _handler(response_map): @@ -106,25 +104,6 @@ async def test_no_results_returns_none_and_zero_confidence(): @pytest.mark.asyncio -async def test_disabled_via_roster_setting_makes_no_request(): - calls = [] - - def handle(request: httpx.Request) -> httpx.Response: - calls.append(request) - return httpx.Response(200, json={"results": []}) - - client = TmdbClient( - roster=FakeRoster(enabled=False), - transport=httpx.MockTransport(handle), - ) - result, ratio = await client.search_movie("Anything") - - assert result is None - assert calls == [] # confirms the disabled check short-circuits before any request - await client.close() - - -@pytest.mark.asyncio async def test_no_token_resolvable_makes_no_request(monkeypatch): monkeypatch.delenv("MESHBAY_TMDB_DEFAULT_TOKEN", raising=False) calls = [] @@ -134,7 +113,7 @@ async def test_no_token_resolvable_makes_no_request(monkeypatch): return httpx.Response(200, json={"results": []}) client = TmdbClient( - roster=FakeRoster(enabled=True, token=None), + roster=FakeRoster(token=None), transport=httpx.MockTransport(handle), ) result, ratio = await client.search_movie("Anything") diff --git a/packages/meshbay-node/tests/test_tmdb_config_policy.py b/packages/meshbay-node/tests/test_tmdb_config_policy.py index 29ef54c..c1781e3 100644 --- a/packages/meshbay-node/tests/test_tmdb_config_policy.py +++ b/packages/meshbay-node/tests/test_tmdb_config_policy.py @@ -1,9 +1,14 @@ """ -The operator decides whether the node calls TMDB at all, and whether it uses -a custom API token — docs/mediacenter.md §5.5. Same shape as -test_apps_enabled_policy.py/test_scan_settings_policy.py: a signed operator -instruction, node-wide (group_id="") rather than per-group, stored via -roster.py's group_settings table. +The operator's custom TMDB API token and query language — docs/mediacenter.md +§5.5. Same shape as test_apps_enabled_policy.py/test_scan_settings_policy.py: +a signed operator instruction, node-wide (group_id="") rather than per-group, +stored via roster.py's group_settings table. + +Whether TMDB is used *at all* used to live in this same op — moved to its +own per-group op (test_tmdb_enabled_policy.py, 2026-08-24): a real +media-library group and a test/demo group on the same node need not share +that decision, while the token and query language stay one operator's +shared credential/cache. Specific to this one: the subject signed/audited must never contain the token itself (it would end up in the audit log in plaintext) — only whether @@ -55,51 +60,39 @@ def _fake_challenge(issued: list): # ── Refused before a challenge is even issued ─────────────────────────────── -async def test_missing_enabled_is_refused(tmp_path): +async def test_non_string_token_is_refused(tmp_path): session = _session(tmp_path, "op", operator="op") session._has_admin_authority = lambda: True issued = [] session._issue_admin_challenge = _fake_challenge(issued) - session._do_tmdb_config({}) + session._do_tmdb_config({"token": 12345}) assert not issued assert [m for m in session.sent if m.get("type") == "error"] -async def test_non_bool_enabled_is_refused(tmp_path): - session = _session(tmp_path, "op", operator="op") - session._has_admin_authority = lambda: True - issued = [] - session._issue_admin_challenge = _fake_challenge(issued) +async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path): + session = _session(tmp_path, "member-1", operator="the-operator") + session._has_admin_authority = lambda: False - session._do_tmdb_config({"enabled": "yes"}) + session._do_tmdb_config({"token": "x"}) - assert not issued assert [m for m in session.sent if m.get("type") == "error"] -async def test_non_string_token_is_refused(tmp_path): +async def test_non_string_language_is_refused(tmp_path): session = _session(tmp_path, "op", operator="op") session._has_admin_authority = lambda: True issued = [] session._issue_admin_challenge = _fake_challenge(issued) - session._do_tmdb_config({"enabled": True, "token": 12345}) + session._do_tmdb_config({"language": 42}) assert not issued assert [m for m in session.sent if m.get("type") == "error"] -async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path): - session = _session(tmp_path, "member-1", operator="the-operator") - session._has_admin_authority = lambda: False - - session._do_tmdb_config({"enabled": False}) - - assert [m for m in session.sent if m.get("type") == "error"] - - # ── Who may change it, and what gets signed ───────────────────────────────── async def test_changing_it_needs_a_signature(tmp_path): @@ -108,7 +101,7 @@ async def test_changing_it_needs_a_signature(tmp_path): issued = [] session._issue_admin_challenge = _fake_challenge(issued) - session._do_tmdb_config({"enabled": True}) + session._do_tmdb_config({}) assert len(issued) == 1 op, subject, payload, group_id = issued[0] @@ -125,23 +118,23 @@ async def test_the_token_itself_never_appears_in_the_signed_subject(tmp_path): session._issue_admin_challenge = _fake_challenge(issued) secret = "sk-super-secret-tmdb-token" - session._do_tmdb_config({"enabled": True, "token": secret}) + session._do_tmdb_config({"token": secret}) _, subject, payload, _ = issued[0] assert secret not in subject assert payload["token"] == secret, "the real value still has to reach the exec step somehow" -async def test_subject_reflects_enabled_and_whether_a_token_was_supplied(tmp_path): +async def test_subject_reflects_whether_a_token_was_supplied(tmp_path): session = _session(tmp_path, "op", operator="op") session._has_admin_authority = lambda: True issued = [] session._issue_admin_challenge = _fake_challenge(issued) - session._do_tmdb_config({"enabled": False, "token": "x"}) + session._do_tmdb_config({"token": "x"}) _, subject, _, _ = issued[0] - assert subject == "enabled=False,custom_token=yes,language=default" + assert subject == "custom_token=yes,language=default" async def test_subject_says_no_custom_token_when_none_given(tmp_path): @@ -150,10 +143,10 @@ async def test_subject_says_no_custom_token_when_none_given(tmp_path): issued = [] session._issue_admin_challenge = _fake_challenge(issued) - session._do_tmdb_config({"enabled": True}) + session._do_tmdb_config({}) _, subject, _, _ = issued[0] - assert subject == "enabled=True,custom_token=no,language=default" + assert subject == "custom_token=no,language=default" async def test_subject_reflects_a_configured_language(tmp_path): @@ -162,44 +155,32 @@ async def test_subject_reflects_a_configured_language(tmp_path): issued = [] session._issue_admin_challenge = _fake_challenge(issued) - session._do_tmdb_config({"enabled": True, "language": "fr-FR"}) + session._do_tmdb_config({"language": "fr-FR"}) _, subject, payload, _ = issued[0] - assert subject == "enabled=True,custom_token=no,language=fr-FR" + assert subject == "custom_token=no,language=fr-FR" assert payload["language"] == "fr-FR" -async def test_non_string_language_is_refused(tmp_path): - session = _session(tmp_path, "op", operator="op") - session._has_admin_authority = lambda: True - issued = [] - session._issue_admin_challenge = _fake_challenge(issued) - - session._do_tmdb_config({"enabled": True, "language": 42}) - - assert not issued - assert [m for m in session.sent if m.get("type") == "error"] - - # ── Where it is stored ────────────────────────────────────────────────────── async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path): roster = Roster(db_path=tmp_path / "roster.db") await roster.open() try: - enabled, token, language = await roster.tmdb_config() - assert (enabled, token, language) == (True, None, None), ( - "absent must mean on, with the shipped default token, TMDB's own default language") - await roster.set_tmdb_config(True, "my-custom-token", "fr-FR", set_by="op") - enabled, token, language = await roster.tmdb_config() - assert (enabled, token, language) == (True, "my-custom-token", "fr-FR") + token, language = await roster.tmdb_config() + assert (token, language) == (None, None), ( + "absent must mean the shipped default token, TMDB's own default language") + await roster.set_tmdb_config("my-custom-token", "fr-FR", set_by="op") + token, language = await roster.tmdb_config() + assert (token, language) == ("my-custom-token", "fr-FR") finally: await roster.close() reopened = Roster(db_path=tmp_path / "roster.db") await reopened.open() try: - assert await reopened.tmdb_config() == (True, "my-custom-token", "fr-FR") + assert await reopened.tmdb_config() == ("my-custom-token", "fr-FR") finally: await reopened.close() @@ -208,11 +189,11 @@ async def test_clearing_the_token_reverts_to_the_default(tmp_path): roster = Roster(db_path=tmp_path / "roster.db") await roster.open() try: - await roster.set_tmdb_config(True, "a-token", set_by="op") - assert (await roster.tmdb_config())[1] == "a-token" + await roster.set_tmdb_config("a-token", set_by="op") + assert (await roster.tmdb_config())[0] == "a-token" - await roster.set_tmdb_config(True, "", set_by="op") - enabled, token, language = await roster.tmdb_config() + await roster.set_tmdb_config("", set_by="op") + token, language = await roster.tmdb_config() assert token is None, "an explicit empty string clears the custom token" finally: await roster.close() @@ -222,9 +203,9 @@ async def test_omitting_the_token_leaves_it_unchanged(tmp_path): roster = Roster(db_path=tmp_path / "roster.db") await roster.open() try: - await roster.set_tmdb_config(True, "a-token", set_by="op") - await roster.set_tmdb_config(False, None, set_by="op") - enabled, token, language = await roster.tmdb_config() - assert (enabled, token) == (False, "a-token") + await roster.set_tmdb_config("a-token", set_by="op") + await roster.set_tmdb_config(set_by="op") + token, language = await roster.tmdb_config() + assert token == "a-token" finally: await roster.close() diff --git a/packages/meshbay-node/tests/test_tmdb_enabled_policy.py b/packages/meshbay-node/tests/test_tmdb_enabled_policy.py new file mode 100644 index 0000000..8e945ad --- /dev/null +++ b/packages/meshbay-node/tests/test_tmdb_enabled_policy.py @@ -0,0 +1,128 @@ +""" +Whether TMDB lookups run *at all* for a group — docs/mediacenter.md §5.5. +Per-group (2026-08-24 — used to be node-wide, folded into tmdb_config): a +real media-library group and a test/demo group on the same node need not +share the decision to spend TMDB quota and make outbound requests. Same +shape as test_video_root_policy.py: a signed operator instruction, scoped to +self._group_id (not passed explicitly on the wire), stored via roster.py's +group_settings table under the real group_id. + +The custom API token and query language stay node-wide — see +test_tmdb_config_policy.py for those. +""" + +from pathlib import Path + +import pytest + +from meshbay_common.adminop import OP_TMDB_ENABLED +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.roster import Roster +from meshbay_node.transport.webrtc_server import WebRTCPeerSession +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from conftest import one_root + +pytestmark = pytest.mark.asyncio + + +def _session(tmp_path: Path, user_id: str, *, operator: str | None = None) -> WebRTCPeerSession: + shared_root = tmp_path / "shared" + shared_root.mkdir(exist_ok=True) + index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate()) + ctx = { + "roots": one_root(shared_root), + "index": index, + "sk_node": index.sk_node, + "node_user_id": operator, + } + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = ctx + session._group_id = "g" * 32 + session._user_id = user_id + session._pk_user = "" + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session + + +# ── Refused before a challenge is even issued ─────────────────────────────── + +async def test_missing_enabled_is_refused(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_tmdb_enabled({}) + + assert not issued + assert [m for m in session.sent if m.get("type") == "error"] + + +async def test_non_bool_enabled_is_refused(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_tmdb_enabled({"enabled": "yes"}) + + assert not issued + assert [m for m in session.sent if m.get("type") == "error"] + + +async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path): + session = _session(tmp_path, "member-1", operator="the-operator") + session._has_admin_authority = lambda: False + + session._do_tmdb_enabled({"enabled": False}) + + assert [m for m in session.sent if m.get("type") == "error"] + + +# ── Accepted cases ─────────────────────────────────────────────────────────── + +async def test_a_valid_request_is_signed_against_this_groups_id(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_tmdb_enabled({"enabled": True}) + + assert issued == [(OP_TMDB_ENABLED, "True")] + + +async def test_disabling_is_signed_too(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_tmdb_enabled({"enabled": False}) + + assert issued == [(OP_TMDB_ENABLED, "False")] + + +# ── Where it is stored ────────────────────────────────────────────────────── + +async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path): + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + assert await roster.tmdb_enabled("g1") is True, "absent must mean on" + await roster.set_tmdb_enabled("g1", False, set_by="op") + assert await roster.tmdb_enabled("g1") is False + finally: + await roster.close() + + reopened = Roster(db_path=tmp_path / "roster.db") + await reopened.open() + try: + assert await reopened.tmdb_enabled("g1") is False + assert await reopened.tmdb_enabled("g2") is True, \ + "one group's setting must not answer for another" + finally: + await reopened.close() |