"""
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