path: root/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/music.py

"""What the node does for the Music app: tags and cover art, and a transcode
for the formats no browser plays."""

import logging
from pathlib import Path

import blake3
from meshbay_common import MNP_VERSION
from meshbay_common.adminop import OP_MUSICBRAINZ_ENABLED
from meshbay_common.protocol import MNP

from meshbay_node import ops
from meshbay_node.roots import off_disk
from meshbay_node.transport.webrtc.disk import _locate
from meshbay_node.transport.webrtc.media_tools import _transcode_audio_to_aac

log = logging.getLogger("meshbay_node.transport.webrtc_server")


# 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 media_probe's BROWSER_INCOMPATIBLE_VIDEO_CODECS, 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"})


class MusicMixin:
    @staticmethod
    async def _fetch_and_cache_cover(media_cache, musicbrainz_client,
                                     mbid: str | None) -> str | None:
        """
        Music app equivalent of `_fetch_and_cache_poster` — a release's
        Cover Art Archive image, fetched once per mbid and cached under its
        own blake3, addressed the same synthetic-id trick
        (`musicbrainz:{mbid}`) so a second track of the same album never
        re-downloads it. Most releases have no scan at all; that's a normal
        outcome (None), not an error.
        """
        if not mbid:
            return None
        synthetic_id = f"musicbrainz:{mbid}"
        cached_hash = await media_cache.get_thumb_hash_by_file_id(synthetic_id)
        if cached_hash is not None:
            return cached_hash
        content = await musicbrainz_client.fetch_cover_art(mbid)
        if content is None:
            return None
        thumb_hash = blake3.blake3(content).hexdigest()
        await media_cache.put_thumb(thumb_hash, synthetic_id, content)
        return thumb_hash

    async def _do_audio_transcode_request(self, msg: dict) -> None:
        """
        docs/MESHBAY_DESIGN.md §9.8'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`,
        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

        # The gate `BROWSER_INCOMPATIBLE_AUDIO_EXTS` exists for, applied where it
        # costs something. Nothing on the node read it: the player asks for these
        # two extensions and no others, and `music-player.js` described itself as
        # "kept in sync with the node's" constant — so the whole restriction lived
        # in the caller, and a member's own message is not the caller.
        #
        # What that let through: this converts a *whole file* and holds a
        # transcode slot shared with video streaming while it runs. Pointed at a
        # two-hour film it spends minutes of the operator's CPU and a slot every
        # other viewer is queued behind. `AUDIO_TRANSCODE_MAX_BYTES` catches the
        # result, after the work; only this catches the work.
        if Path(entry.name).suffix.lower() not in BROWSER_INCOMPATIBLE_AUDIO_EXTS:
            self._send({
                "type": "error",
                "detail": "This file does not need transcoding — play it directly.",
                "code": "transcode_not_applicable",
            })
            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": "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/MESHBAY_DESIGN.md §9.8: MusicBrainz metadata for one track, resolved
        from the group's index by its content id. Album-level (release), the
        direct analogue of Videos' show-level TMDB caching: one search per
        (artist, album) pair serves cover art and canonical naming to every
        track of the same release, keyed off the `artist`/`album` fields
        enrich_audio.py already populated at index time (from tags, or the
        filename-parse fallback) — never re-parsed here.

        Keyed by `file_id` (the entry's own content hash), not `path`: found
        live (2026-08-25) — `IndexEntry.path` is the *folder* a file is in
        (indexer.py's `_virtual_dir`), so any two tracks in the same folder
        (routinely true — an album is one folder, many tracks) shared the
        same `.path`, and looking a track up by it silently resolved to
        whichever entry happened to be first in the index. Three unrelated
        albums showed the same wrong cover before this fix, all sharing one
        folder with the track that legitimately matched it.
        """
        file_id = msg.get("file_id")
        log.debug("music_meta_req file_id=%r", file_id)
        if not isinstance(file_id, str) or not file_id:
            self._send({"type": "error", "detail": "Missing file_id"})
            return
        ctx = self._group_ctx()
        entry = ctx["index"].get_entry(file_id)
        if not entry:
            self._send({"type": "error", "detail": "File not found"})
            return

        media_cache = self._ctx.get("media_cache")
        musicbrainz_client = self._ctx.get("musicbrainz_client")
        # Same silent, no-error degradation as _do_media_meta_request: no
        # client configured, MusicBrainz off for this group, or nothing to
        # search with (no artist/album — an untagged, unparseable file) all
        # look identical to the caller, which already has to handle "no
        # match" as the ordinary case in flat mode.
        if (media_cache is None or musicbrainz_client is None
                or not ctx.get("musicbrainz_enabled", True)
                or not entry.artist or not entry.album):
            self._send({"type": MNP.MUSIC_META_RESP, "v": MNP_VERSION,
                       "file_id": file_id, "confidence": 0})
            return

        mbid = await media_cache.get_file_mbid(entry.id)
        meta = await media_cache.get_mbid_meta(mbid) if mbid else None

        if meta is None:
            result, ratio = await musicbrainz_client.search_release(entry.artist, entry.album)
            if result is None or ratio < 0.6:
                self._send({"type": MNP.MUSIC_META_RESP, "v": MNP_VERSION,
                           "file_id": file_id, "confidence": 0})
                return
            mbid = result.get("id")
            artist_credit = result.get("artist-credit") or []
            meta = {
                "artist": artist_credit[0].get("name") if artist_credit else entry.artist,
                "album": result.get("title"),
                "release_date": result.get("date"),
                "confidence": ratio,
            }
            await media_cache.set_file_mbid(entry.id, mbid)
            await media_cache.set_mbid_meta(mbid, meta)

        cover_thumb_hash = await self._fetch_and_cache_cover(
            media_cache, musicbrainz_client, mbid)
        log.debug("music_meta_req file_id=%r: replying mbid=%s cover=%s",
                  file_id, mbid, cover_thumb_hash)

        self._send({
            "type": MNP.MUSIC_META_RESP, "v": MNP_VERSION, "file_id": file_id,
            "mbid": mbid,
            "artist": meta.get("artist"),
            "album": meta.get("album"),
            "title": entry.display_title,
            "release_date": meta.get("release_date"),
            "cover_thumb_hash": cover_thumb_hash,
            "confidence": meta.get("confidence", 1.0),
        })

    def _do_musicbrainz_enabled(self, msg: dict) -> None:
        """
        Whether MusicBrainz lookups run for this group at all. Per-group
        from the start (docs/MESHBAY_DESIGN.md §9.8) — signed like
        tmdb_enabled: it decides whether this group's members'
        Music tab ever makes outbound MusicBrainz 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_MUSICBRAINZ_ENABLED, str(enabled))

    async def _admin_exec_musicbrainz_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"musicbrainz_enabled:{pending['subject']}")
            return
        try:
            await self._run_op(ops.set_musicbrainz_enabled, self._group_id or "", enabled)
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
            return
        self._audit("musicbrainz_enabled", pending["subject"])

        notice = {"type": MNP.MUSICBRAINZ_ENABLED_ACK, "v": MNP_VERSION, "enabled": enabled}
        for uid, session in list(self._peer_registry().items()):
            try:
                session._send(notice)
            except Exception:
                pass