summaryrefslogtreecommitdiffstats
path: root/packages/meshbay-node
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-08-24 21:08:34 +0200
committerChristophe Besson <cbesson@gmail.com>2026-08-24 21:08:34 +0200
commitf4ed927fa873133c5fed73fb7dd60e7747fc3dc2 (patch)
treed126c5e505c342ee6b02aadc1e46b1a026c2a027 /packages/meshbay-node
parent75d1f8b93dfa0bffda3a59a6d143b06dcc3ca67f (diff)
downloadmeshbay-f4ed927fa873133c5fed73fb7dd60e7747fc3dc2.tar.gz
feat(music): transcode WMA/Musepack to AAC so they actually play
Tagging and covers for these two formats landed already, but neither one decodes in any mainstream browser's <audio> element at all — a real library scan turned up 273 such files that would show up correctly in the Music app and then simply fail on click. This closes that gap: the node transcodes to AAC/M4A on request (a one-shot whole-file conversion, not live-piped like video's fMP4 segments — an audio file is small enough that streaming it buys nothing), caches the result under its own content hash the same way a TMDB poster or a MusicBrainz cover is cached, and serves it back through the ordinary file_req/chunk path. That path used to assume anything in the media cache was thumbnail-sized (single chunk, always); generalized it to slice a cached blob the same way a real file on disk gets sliced, since a transcoded track can be several MB. New MNP pair (`audio_transcode_req`/`_resp`, version bump to 0.9), shares its concurrency cap with video's transcode pool rather than getting its own — both are real ffmpeg processes on the same node. Every other audio format is untouched: this only fires for .wma/.mpc, the two extensions that need it.
Diffstat (limited to 'packages/meshbay-node')
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py136
-rw-r--r--packages/meshbay-node/tests/test_audio_transcode.py193
2 files changed, 322 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.
diff --git a/packages/meshbay-node/tests/test_audio_transcode.py b/packages/meshbay-node/tests/test_audio_transcode.py
new file mode 100644
index 0000000..725839b
--- /dev/null
+++ b/packages/meshbay-node/tests/test_audio_transcode.py
@@ -0,0 +1,193 @@
+"""
+Tests for the Music app's one exception to "no node-side transcode pool"
+(docs/musicbay.md §2.2): WMA and Musepack tag/cover fine (enrich_audio.py)
+but decode in no mainstream browser's <audio> element at all, so
+`_do_audio_transcode_request` converts to AAC/M4A on request and caches the
+result — served back through the ordinary file_req/chunk path, generalized
+in `_try_serve_thumbnail` to handle more than one chunk.
+"""
+
+import shutil
+import subprocess
+from pathlib import Path
+
+import blake3
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_common.crypto import generate_gek
+from meshbay_common.protocol import MNP, IndexEntry
+from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.media_cache import MediaCache
+from meshbay_node.transport import webrtc_server as wrs
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+from conftest import one_root
+
+_HAVE_FFMPEG = shutil.which("ffmpeg") and shutil.which("ffprobe")
+pytestmark = pytest.mark.asyncio
+
+
+@pytest.fixture
+async def media_cache(tmp_path):
+ c = MediaCache(db_path=tmp_path / "media_cache.db")
+ await c.open()
+ yield c
+ await c.close()
+
+
+def _make_wma_clip(path: Path) -> None:
+ subprocess.run(
+ ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
+ "-f", "lavfi", "-i", "sine=frequency=440:duration=1",
+ "-c:a", "wmav2", "-b:a", "64k", str(path)],
+ check=True, capture_output=True,
+ )
+
+
+def _session(tmp_path: Path, file_path: Path, gek: bytes, media_cache: MediaCache):
+ file_bytes = file_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)
+ index.add_entry(IndexEntry(
+ id=file_id, name=file_path.name, path=file_path.parent.name,
+ size=len(file_bytes), type="audio", added_at=0))
+
+ session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+ session._ctx = {
+ "roots": one_root(file_path.parent),
+ "index": index,
+ "gek": gek,
+ "sk_node": sk_node,
+ "media_cache": media_cache,
+ "max_concurrent_streams": 4,
+ }
+ session._group_id = None
+ session._user_id = "tester"
+ session.sent = []
+ session._send = session.sent.append
+ session._audit = lambda *a, **k: None
+ return session, file_id
+
+
+def _reassemble_file_chunks(sent: list[dict], gek: bytes, hash_hex: str) -> bytes:
+ file_hash = bytes.fromhex(hash_hex)
+ chunks = sorted(
+ (m for m in sent if m.get("type") == "file_chunk" and m.get("file_id") == hash_hex),
+ key=lambda m: m["chunk_index"])
+ out = b""
+ for m in chunks:
+ key = chunk_key_aes(gek, file_hash, m["chunk_index"])
+ out += decrypt_chunk_aes(key, m["nonce"], m["ct"])
+ return out
+
+
+@pytest.mark.skipif(not _HAVE_FFMPEG, reason="ffmpeg/ffprobe not installed")
+async def test_wma_transcodes_to_a_browser_playable_aac_file(tmp_path, media_cache):
+ clip = tmp_path / "clip.wma"
+ _make_wma_clip(clip)
+ gek = generate_gek()
+ session, file_id = _session(tmp_path, clip, gek, media_cache)
+
+ await session._do_audio_transcode_request({"file_id": file_id})
+
+ resp = next(m for m in session.sent if m.get("type") == MNP.AUDIO_TRANSCODE_RESP)
+ assert resp["mime"] == "audio/mp4"
+ assert resp["size"] > 0
+ transcode_hash = resp["hash"]
+
+ # Pull it back exactly the way a client would: file_req/chunk, resolved
+ # against the media cache since this hash is not a real index entry.
+ await session._do_file_request({"file_id": transcode_hash, "chunk_index": 0})
+ out_bytes = _reassemble_file_chunks(session.sent, gek, transcode_hash)
+ assert len(out_bytes) == resp["size"]
+
+ out_path = tmp_path / "out.m4a"
+ out_path.write_bytes(out_bytes)
+ probe = subprocess.run(
+ ["ffprobe", "-v", "error", "-show_entries", "stream=codec_name,codec_type",
+ "-of", "csv=p=0", str(out_path)],
+ check=True, capture_output=True, text=True)
+ rows = [line.split(",") for line in probe.stdout.strip().splitlines()]
+ codecs = {r[1]: r[0] for r in rows}
+ assert codecs.get("audio") == "aac", f"must be AAC, playable in a browser: {codecs}"
+
+
+@pytest.mark.skipif(not _HAVE_FFMPEG, reason="ffmpeg/ffprobe not installed")
+async def test_second_request_reuses_the_cached_transcode(tmp_path, media_cache, monkeypatch):
+ clip = tmp_path / "clip.wma"
+ _make_wma_clip(clip)
+ gek = generate_gek()
+ session, file_id = _session(tmp_path, clip, gek, media_cache)
+
+ calls = {"n": 0}
+ real = wrs._transcode_audio_to_aac
+
+ async def counting(path):
+ calls["n"] += 1
+ return await real(path)
+
+ monkeypatch.setattr(wrs, "_transcode_audio_to_aac", counting)
+
+ await session._do_audio_transcode_request({"file_id": file_id})
+ session.sent.clear()
+ await session._do_audio_transcode_request({"file_id": file_id})
+
+ resp = next(m for m in session.sent if m.get("type") == MNP.AUDIO_TRANSCODE_RESP)
+ assert resp["hash"]
+ assert calls["n"] == 1, "a second request for the same file must not re-run ffmpeg"
+
+
+async def test_missing_file_id_is_an_error(tmp_path, media_cache):
+ gek = generate_gek()
+ sk_node = Ed25519PrivateKey.generate()
+ session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+ session._ctx = {
+ "roots": one_root(tmp_path), "index": GroupIndex(group_id="g" * 32, sk_node=sk_node, gek=gek),
+ "gek": gek, "sk_node": sk_node, "media_cache": media_cache,
+ }
+ session._group_id = None
+ session.sent = []
+ session._send = session.sent.append
+
+ await session._do_audio_transcode_request({"file_id": "nonexistent"})
+
+ assert session.sent[-1]["type"] == "error"
+
+
+async def test_multi_chunk_cached_blob_reassembles_correctly(tmp_path, media_cache):
+ """
+ `_try_serve_thumbnail` used to assume a cached blob never exceeds one
+ chunk (true for a thumbnail, not true for a multi-MB audio transcode) —
+ this exercises the slicing directly, without needing ffmpeg at all: a
+ blob a little over two chunks, fetched chunk by chunk, must reassemble
+ to exactly the original bytes.
+ """
+ import os
+ gek = generate_gek()
+ sk_node = Ed25519PrivateKey.generate()
+ session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+ session._ctx = {
+ "roots": one_root(tmp_path), "index": GroupIndex(group_id="g" * 32, sk_node=sk_node, gek=gek),
+ "gek": gek, "sk_node": sk_node, "media_cache": media_cache,
+ }
+ session._group_id = None
+ session.sent = []
+ session._send = session.sent.append
+
+ blob = os.urandom(int(wrs.CHUNK_SIZE * 2.3))
+ blob_hash = blake3.blake3(blob).hexdigest()
+ await media_cache.put_thumb(blob_hash, "synthetic:test", blob)
+
+ total_chunks = -(-len(blob) // wrs.CHUNK_SIZE)
+ for i in range(total_chunks):
+ await session._do_file_request({"file_id": blob_hash, "chunk_index": i})
+ # One past the end must be a clean miss, not a partial/garbage chunk.
+ await session._do_file_request({"file_id": blob_hash, "chunk_index": total_chunks})
+
+ chunk_msgs = [m for m in session.sent if m.get("type") == "file_chunk"]
+ assert len(chunk_msgs) == total_chunks
+ reassembled = _reassemble_file_chunks(session.sent, gek, blob_hash)
+ assert reassembled == blob