diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 136 |
1 files changed, 129 insertions, 7 deletions
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 5da1b7e..1629801 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -29,6 +29,7 @@ import hmac import logging import os import struct +import tempfile import time from pathlib import Path from typing import Any @@ -137,6 +138,18 @@ PRE_HANDSHAKE_MAX_MSG = 64 * 1024 # so the operator sets `max_concurrent_streams` under [node] in node.toml. This # value applies when they have said nothing. MAX_CONCURRENT_TRANSCODES = 8 +# Extensions no mainstream browser's <audio> element decodes natively, no +# matter how well-tagged (enrich_audio.py's problem) — the Music app's own +# analogue of BROWSER_INCOMPATIBLE_VIDEO_CODECS above, keyed by extension +# rather than a probed codec name since these two are a red flag on their +# own, not something that varies by how the file happens to be encoded +# inside. +BROWSER_INCOMPATIBLE_AUDIO_EXTS = frozenset({".wma", ".mpc"}) +# A whole audio file is small enough to transcode in one shot rather than +# live-piped like video's fMP4 segments — a few seconds of ffmpeg at most, +# bounded generously so one slow/huge outlier can't pin a transcode slot +# (shared with video, MAX_CONCURRENT_TRANSCODES above) indefinitely. +AUDIO_TRANSCODE_TIMEOUT_SECS = 120 # Bundle fetches are served in the pre-proof window (C4). Bounded and audited # until the native client removes remote keypair bundles entirely. MAX_PRE_PROOF_FETCHES = 4 @@ -399,6 +412,8 @@ class WebRTCPeerSession: self._do_musicbrainz_enabled(msg) elif mtype == MNP.MUSIC_META_REQ: self._spawn(self._do_music_meta_request(msg)) + elif mtype == MNP.AUDIO_TRANSCODE_REQ: + self._spawn(self._do_audio_transcode_request(msg)) elif mtype == MNP.MEMBER_UNPIN: self._do_member_unpin(msg) elif mtype == MNP.GEK_ROTATE: @@ -2439,18 +2454,26 @@ class WebRTCPeerSession: """ docs/mediacenter.md §5.3: a thumbnail is served through the same chunked file_req path as a real file, resolved against the media - cache instead of the index when the id doesn't match a file. Always - a single chunk in practice (a thumbnail-sized JPEG never approaches - CHUNK_SIZE) — a request for any chunk beyond 0 is just a miss. + cache instead of the index when the id doesn't match a file. + Sliced by `chunk_index` like a real file's chunks, not just handed + back whole: a thumbnail/poster/cover never approached CHUNK_SIZE so + this used to be equivalent to "only chunk 0 exists", but an audio + transcode result (docs/musicbay.md, the WMA/Musepack exception) is + cached in the same media_cache blob store and can be several MB — + genuinely multi-chunk, same as a file read straight off disk. """ media_cache = self._ctx.get("media_cache") - if media_cache is None or chunk_index != 0: + if media_cache is None: return None - jpeg = await media_cache.get_thumb(thumb_hash) - if jpeg is None: + blob = await media_cache.get_thumb(thumb_hash) + if blob is None: return None + start = chunk_index * CHUNK_SIZE + if start > len(blob) or (start == len(blob) and chunk_index != 0): + return None + piece = blob[start:start + CHUNK_SIZE] return _encrypt_chunk_bytes( - self._ctx["sk_node"], gek, jpeg, 0, + self._ctx["sk_node"], gek, piece, chunk_index, bytes.fromhex(thumb_hash), thumb_hash, ) @@ -2556,6 +2579,64 @@ class WebRTCPeerSession: await media_cache.put_thumb(thumb_hash, synthetic_id, content) return thumb_hash + async def _do_audio_transcode_request(self, msg: dict) -> None: + """ + docs/musicbay.md's one exception to "no node-side transcode pool": + WMA and Musepack tag/cover fine (enrich_audio.py) but decode in no + mainstream browser's <audio> element at all. Transcoded to AAC/M4A + once and cached under its own content hash — same "computed once, + reused forever" shape as `_fetch_and_cache_poster`/`_cover` above, + served back to the client through the ordinary file_req/chunk path + (`_try_serve_thumbnail`, generalized to multi-chunk for this) rather + than a new download mechanism. + """ + 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 = entry_abs_path(ctx["roots"], entry) + if not file_path.exists(): + self._send({"type": "error", "detail": "File not on disk"}) + return + + media_cache = self._ctx.get("media_cache") + if media_cache is None: + self._send({"type": "error", "detail": "Transcoding unavailable"}) + return + + synthetic_id = f"audio_transcode:{entry.id}" + 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: + self._send({"type": MNP.AUDIO_TRANSCODE_RESP, "v": MNP_VERSION, + "file_id": file_id, "hash": cached_hash, + "size": len(blob), "mime": "audio/mp4"}) + return + # Cached hash but the blob itself was pruned: fall through and + # transcode again below, same as a cold cache. + + sem = self._transcode_semaphore() + if sem.locked() and sem._value <= 0: + self._send({"type": "error", "detail": "Server busy, retry shortly"}) + return + async with sem: + try: + blob = await _transcode_audio_to_aac(file_path) + except Exception as e: + log.warning("Audio transcode failed for %s: %s", entry.id[:12], e) + self._send({"type": "error", "detail": f"Transcode failed: {e}"}) + return + + transcode_hash = blake3.blake3(blob).hexdigest() + await media_cache.put_thumb(transcode_hash, synthetic_id, blob) + self._audit("audio_transcode", entry.name) + self._send({"type": MNP.AUDIO_TRANSCODE_RESP, "v": MNP_VERSION, + "file_id": file_id, "hash": transcode_hash, + "size": len(blob), "mime": "audio/mp4"}) + async def _do_music_meta_request(self, msg: dict) -> None: """ docs/musicbay.md §4.3: MusicBrainz metadata for one path, resolved @@ -4018,6 +4099,47 @@ def _read_and_encrypt( return _encrypt_chunk_bytes(sk_node, gek, plaintext, chunk_index, file_hash, file_id) +async def _transcode_audio_to_aac(file_path: Path) -> bytes: + """ + One-shot, whole-file transcode to AAC in an M4A container — no live + piping, no seeking, unlike `_stream_video_inner`'s fMP4 segments: a + WMA/Musepack source here is a few MB at most, so there is nothing to + gain from streaming it and a real cost to the added complexity + (fragmented output needs `-movflags empty_moov` and its own + client-side reassembly). A plain temp file lets ffmpeg write a normal, + fully-seekable M4A container instead. `-vn` drops any attached-picture + "video" stream some taggers embed as cover art — without it, ffmpeg's + mp4 muxer has been seen treating that picture as a video track to + encode, which is not what this is for; cover art still comes from the + ordinary embedded/sibling-file path (enrich_audio.py), never from here. + """ + fd, tmp_name = tempfile.mkstemp(suffix=".m4a") + os.close(fd) + tmp_path = Path(tmp_name) + try: + proc = await asyncio.create_subprocess_exec( + "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", + "-i", str(file_path), + "-vn", "-c:a", "aac", "-ac", "2", "-b:a", "192k", + "-f", "ipod", str(tmp_path), + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.PIPE, + ) + try: + _, stderr = await asyncio.wait_for( + proc.communicate(), timeout=AUDIO_TRANSCODE_TIMEOUT_SECS) + except asyncio.TimeoutError: + proc.kill() + await proc.wait() + raise RuntimeError(f"ffmpeg timed out after {AUDIO_TRANSCODE_TIMEOUT_SECS}s") + if proc.returncode != 0: + raise RuntimeError( + f"ffmpeg exited {proc.returncode}: {stderr.decode(errors='replace')[:300]}") + return tmp_path.read_bytes() + finally: + tmp_path.unlink(missing_ok=True) + + class WebRTCTransport: """ Manages WebRTC peer connections for browser clients. |