refactor(node): move the subtitle handler out of webrtc_server

SubtitlesMixin in transport/webrtc/apps/subtitles.py; _locate, which it
shares with the files, music and streaming handlers, in webrtc/disk.py.

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

3 files changed, 194 insertions, 0 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/__init__.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/__init__.py
new file mode 100644
index 0000000..a2e212a
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/__init__.py
@@ -0,0 +1 @@
+"""What the node does for one app only: Videos, Music."""
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/subtitles.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/subtitles.py
new file mode 100644
index 0000000..bd2ab52
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/subtitles.py
@@ -0,0 +1,170 @@
+"""Subtitle tracks for the Videos app: extracted once, whole-file, to WebVTT."""
+
+import asyncio
+import logging
+import time
+
+import blake3
+from meshbay_common import MNP_VERSION
+from meshbay_common.protocol import MNP
+
+from meshbay_node.media_probe import probe_video as _probe_video
+from meshbay_node.roots import off_disk
+from meshbay_node.transport.webrtc.disk import _locate
+from meshbay_node.transport.webrtc.media_tools import (
+    _extract_subtitle_to_webvtt,
+    _subtitle_timeout_for,
+)
+
+log = logging.getLogger("meshbay_node.transport.webrtc_server")
+
+
+class SubtitlesMixin:
+    async def _do_subtitle_request(self, msg: dict) -> None:
+        """
+        One embedded subtitle track, extracted whole-file to WebVTT and served
+        back through the ordinary file_req/chunk path — the same indirection
+        as `_do_audio_transcode_request`, and cached the same way, so a
+        film's subtitles are extracted once in the life of the file rather
+        than once per viewing.
+
+        The ordinal is validated against `probe_video`'s *filtered* list and
+        then used as the ffmpeg `-map 0:s:<n>` argument, which is only correct
+        because `SubtitleTrack.ordinal` counts every subtitle stream including
+        the bitmap ones the list omits (see media_probe.py). Checking
+        membership rather than range is what makes that hold: a bitmap
+        ordinal is in range and is not in the list, and extracting it would
+        produce an empty WebVTT — a subtitle track with no subtitles in it,
+        which reports no error anywhere.
+        """
+        ctx = self._group_ctx()
+        file_id = msg.get("file_id", "")
+        entry = ctx["index"].get_entry(file_id)
+        if not entry:
+            self._send({"type": "error", "detail": "File not found"})
+            return
+        file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry)
+        if refusal is not None:
+            self._send({"type": "error", "detail": refusal})
+            return
+
+        media_cache = self._ctx.get("media_cache")
+        if media_cache is None:
+            self._send({"type": "error", "detail": "Subtitles unavailable"})
+            return
+
+        try:
+            ordinal = int(msg.get("track", 0) or 0)
+        except (TypeError, ValueError):
+            ordinal = -1
+
+        # Logged before the cache is consulted, and that ordering is the point:
+        # a cached track used to answer without leaving a line, so the journal
+        # could not say whether a viewer had asked for subtitles at all. That
+        # turned "no request in the log" into evidence it was never entitled
+        # to be — the second time in this feature that a silent success was
+        # read as an absence.
+        log.info("subtitle: req file=%s track=%d size=%.1fMB slots_free=%s",
+                 file_id[:12], ordinal, entry.size / 1e6,
+                 getattr(self._ctx.get("_transcode_sem"), "_value", "?"))
+        t0 = time.monotonic()
+
+        synthetic_id = f"subtitle:{entry.id}:{ordinal}"
+        cached_hash = await media_cache.get_thumb_hash_by_file_id(synthetic_id)
+        if cached_hash is not None:
+            blob = await media_cache.get_thumb(cached_hash)
+            if blob is not None:
+                log.info("subtitle: served file=%s track=%d from cache, %d bytes",
+                         file_id[:12], ordinal, len(blob))
+                self._send({"type": MNP.SUBTITLE_RESP, "v": MNP_VERSION,
+                            "file_id": file_id, "track": ordinal,
+                            "hash": cached_hash, "size": len(blob),
+                            "mime": "text/vtt"})
+                return
+            # Cached hash but the blob was pruned: fall through and extract
+            # again, same as a cold cache.
+
+        inflight = self._ctx.setdefault("_subtitle_inflight", {})
+        running = inflight.get(synthetic_id)
+        if running is not None:
+            log.info("subtitle: file=%s track=%d already extracting, waiting",
+                     file_id[:12], ordinal)
+            # Shielded: this waiter being cancelled must not cancel the
+            # extraction every other waiter is also relying on.
+            answer = await asyncio.shield(running)
+            if answer is None:
+                self._send({"type": "error", "detail": "Subtitle extraction failed"})
+                return
+            subtitle_hash, size = answer
+            log.info("subtitle: served file=%s track=%d from a shared extraction, %d bytes",
+                     file_id[:12], ordinal, size)
+            self._send({"type": MNP.SUBTITLE_RESP, "v": MNP_VERSION,
+                        "file_id": file_id, "track": ordinal,
+                        "hash": subtitle_hash, "size": size, "mime": "text/vtt"})
+            return
+
+        # One extraction per file and track at a time, across every viewer.
+        # The cache is only consulted on the way in, so two clicks seconds
+        # apart both missed it and both ran — seen in the log as two identical
+        # extractions of one 4.3 GB file overlapping, each holding a transcode
+        # slot and reading the whole file. Latecomers wait for the answer the
+        # first one is already producing.
+        #
+        # **Registered before the first `await` below**, never after. The
+        # first version registered it after the probe, and two requests
+        # arriving together both got past the check above while neither had
+        # registered yet — so both extracted, which is the race this exists to
+        # close. The probe and the slot check sit behind it for that reason.
+        fut = asyncio.get_running_loop().create_future()
+        inflight[synthetic_id] = fut
+        try:
+            probe = await _probe_video(str(file_path))
+            if not any(tr.ordinal == ordinal for tr in probe.subtitle_tracks):
+                # Not a range check — see this method's docstring.
+                log.warning("subtitle: file=%s has no text track %d",
+                            file_id[:12], ordinal)
+                self._send({"type": "error", "detail": "No such subtitle track"})
+                return
+
+            sem = self._transcode_semaphore()
+            if sem.locked() and sem._value <= 0:
+                log.info("subtitle: refused, no transcode slot free")
+                self._send({"type": "error", "detail": "Server busy, retry shortly"})
+                return
+
+            budget = _subtitle_timeout_for(entry.size)
+            async with sem:
+                log.info("subtitle: extracting file=%s track=%d (slot taken, up to %.0fs)",
+                         file_id[:12], ordinal, budget)
+                blob = await _extract_subtitle_to_webvtt(file_path, ordinal, budget)
+
+            subtitle_hash = blake3.blake3(blob).hexdigest()
+            await media_cache.put_thumb(subtitle_hash, synthetic_id, blob)
+            self._audit("subtitle_extract", f"{entry.name} [{ordinal}]")
+            log.info("subtitle: extracted file=%s track=%d in %.1fs, %d bytes",
+                     file_id[:12], ordinal, time.monotonic() - t0, len(blob))
+            fut.set_result((subtitle_hash, len(blob)))
+            self._send({"type": MNP.SUBTITLE_RESP, "v": MNP_VERSION,
+                        "file_id": file_id, "track": ordinal,
+                        "hash": subtitle_hash, "size": len(blob),
+                        "mime": "text/vtt"})
+        # BaseException, not Exception: a cancelled task — the peer went away,
+        # the session is being torn down — raises CancelledError, which is not
+        # an Exception and would otherwise leave this handler with no reply
+        # sent and no line in the log. The client is then waiting on something
+        # nothing will ever answer, which is the shape nobody can report.
+        except BaseException as e:
+            log.warning("subtitle: extract failed file=%s track=%d after %.1fs: %r",
+                        file_id[:12], ordinal, time.monotonic() - t0, e)
+            self._send({"type": "error",
+                        "detail": f"Subtitle extraction failed: {e}"})
+            if isinstance(e, asyncio.CancelledError):
+                raise
+        finally:
+            # A result either way, never an exception: a future nobody is
+            # waiting on yet would be an unretrieved-exception warning, and a
+            # cancelled task would otherwise leave every latecomer waiting on
+            # a future that is never resolved at all.
+            if not fut.done():
+                fut.set_result(None)
+            inflight.pop(synthetic_id, None)
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py
new file mode 100644
index 0000000..fd3184c
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py
@@ -0,0 +1,23 @@
+"""Blocking disk work the session hands to `off_disk`."""
+
+from pathlib import Path
+
+from meshbay_node.roots import ROOT_NOT_SERVED, RootSet, entry_abs_path
+
+
+def _locate(roots: RootSet, entry) -> tuple[Path | None, str | None]:
+    """
+    Where an entry is, and whether it is readable — or the refusal to send.
+
+    Both halves are syscalls: `resolve()` walks the path and `exists()` stats
+    it, and a stat is what *wakes* a sleeping disk. Leaving either on the event
+    loop and offloading only the read would move the stall rather than remove
+    it, and the read would then find the disk already awake. Blocking; called
+    through `off_disk`.
+    """
+    path = entry_abs_path(roots, entry)
+    if path is None:
+        return None, ROOT_NOT_SERVED
+    if not path.exists():
+        return None, "File not on disk"
+    return path, None