"""
The Music app's index-time enrichment survives a restart.

`video_meta`, `photo_meta` and `thumbs` were all durable; the audio tags were
not, so every node start re-read the tags of every audio file it serves, and
until that pass landed the index it served carried no artist on any track. A
client that asked in that window got an index the Music app cannot group — a
toolbar over a blank page, for as long as the page stayed open.

What is cached is what the file's *bytes* said. The filename and folder
fallbacks on top of it are not, and these tests are mostly about that line:
a cache that also remembered the derived fields would hand a renamed file the
old name's answer, which is the fault `_reenrich_renamed_audio_entries` exists
to prevent.
"""
import asyncio
import subprocess
from pathlib import Path

import pytest
from meshbay_common.protocol import IndexEntry
from meshbay_node.indexer import enrich_audio
from meshbay_node.indexer.enrich_audio import AudioEnricher
from meshbay_node.media_cache import MediaCache

_HAVE_FFMPEG = (
    subprocess.run(["which", "ffmpeg"], capture_output=True).returncode == 0)

pytestmark = pytest.mark.skipif(not _HAVE_FFMPEG, reason="ffmpeg not installed")


def _make_clip(path: Path, *, title=None, artist=None, album=None, track=None) -> None:
    subprocess.run(
        ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
         "-f", "lavfi", "-i", "sine=frequency=440:duration=1",
         "-c:a", "libmp3lame", "-b:a", "64k",
         *(["-metadata", f"title={title}"] if title else []),
         *(["-metadata", f"artist={artist}"] if artist else []),
         *(["-metadata", f"album={album}"] if album else []),
         *(["-metadata", f"track={track}"] if track else []),
         str(path)],
        check=True, capture_output=True,
    )


@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()


async def _enrich(enricher, entry, clip, root_path=None):
    done = asyncio.get_event_loop().create_future()

    async def on_done(file_id, fields):
        done.set_result(fields)

    enricher.spawn(entry, clip, on_done, root_path)
    return await asyncio.wait_for(done, timeout=30)


@pytest.fixture
def counted_reads(monkeypatch):
    """How many times the audio file itself was opened and parsed."""
    calls = []
    real = enrich_audio._read_tags_and_cover

    def counting(path, skip_cover=False):
        calls.append(Path(path).name)
        return real(path, skip_cover=skip_cover)

    monkeypatch.setattr(enrich_audio, "_read_tags_and_cover", counting)
    return calls


@pytest.mark.asyncio
async def test_a_second_pass_over_the_same_content_does_not_open_the_file(
        tmp_path, media_cache, counted_reads):
    """The restart case, in one process: same bytes, no second read."""
    clip = tmp_path / "01 - a track.mp3"
    _make_clip(clip, title="A Title", artist="An Act", album="A Record", track=2)
    entry = IndexEntry(id="content1", name=clip.name, path=clip.name,
                       size=clip.stat().st_size, type="audio", added_at=0)

    first = await _enrich(AudioEnricher(media_cache), entry, clip)
    assert len(counted_reads) == 1

    # A different enricher, as a restarted daemon would build — only the cache
    # on disk is shared.
    second = await _enrich(AudioEnricher(media_cache), entry, clip)
    assert len(counted_reads) == 1, "the file was opened again despite a cache hit"
    assert second == first, "a cache hit must produce the fields the read produced"


@pytest.mark.asyncio
async def test_the_cached_answer_is_the_tags_and_the_duration(tmp_path, media_cache):
    clip = tmp_path / "01 - a track.mp3"
    _make_clip(clip, title="A Title", artist="An Act", album="A Record", track=2)
    entry = IndexEntry(id="content2", name=clip.name, path=clip.name,
                       size=clip.stat().st_size, type="audio", added_at=0)
    await _enrich(AudioEnricher(media_cache), entry, clip)

    row = await media_cache.get_audio_meta("content2")
    assert row["title"] == "A Title"
    assert row["artist"] == "An Act"
    assert row["album"] == "A Record"
    assert row["track_no"] == 2
    assert row["duration"] == 1
    assert row["cover_seen"] is True


@pytest.mark.asyncio
async def test_a_file_that_says_nothing_is_remembered_as_saying_nothing(
        tmp_path, media_cache, counted_reads):
    """
    The commonest row in a real library, and the one worth caching most: an
    untagged file costs exactly the same open as a tagged one to learn nothing
    from.
    """
    folder = tmp_path / "Some Act" / "Some Record"
    folder.mkdir(parents=True)
    clip = folder / "05 - Filename Title.mp3"
    _make_clip(clip)
    entry = IndexEntry(id="content3", name=clip.name,
                       path=str(clip.relative_to(tmp_path)),
                       size=clip.stat().st_size, type="audio", added_at=0)

    first = await _enrich(AudioEnricher(media_cache), entry, clip, tmp_path)
    assert await media_cache.get_audio_meta("content3") is not None

    second = await _enrich(AudioEnricher(media_cache), entry, clip, tmp_path)
    assert len(counted_reads) == 1
    # Still resolved, and from the folder rather than from any tag.
    assert second["artist"] == "Some Act"
    assert second["album"] == "Some Record"
    assert second == first


@pytest.mark.asyncio
async def test_a_renamed_file_re_derives_what_the_name_decides(
        tmp_path, media_cache, counted_reads):
    """
    The line the cache must not cross. Same content, so the tags are reused —
    and `display_title`/`track_no`/artist/album still come from the new name
    and the new folder, which is what `_reenrich_renamed_audio_entries` asks
    for when it discards the attempt.
    """
    first_folder = tmp_path / "Old Act" / "Old Record"
    first_folder.mkdir(parents=True)
    clip = first_folder / "05 - Old Name.mp3"
    _make_clip(clip)                       # no tags at all: the name decides
    entry = IndexEntry(id="content4", name=clip.name,
                       path=str(clip.relative_to(tmp_path)),
                       size=clip.stat().st_size, type="audio", added_at=0)
    before = await _enrich(AudioEnricher(media_cache), entry, clip, tmp_path)
    assert before["display_title"] == "Old Name"
    assert before["artist"] == "Old Act"

    moved_folder = tmp_path / "New Act" / "New Record"
    moved_folder.mkdir(parents=True)
    moved = moved_folder / "07 - New Name.mp3"
    clip.rename(moved)
    renamed = IndexEntry(id="content4", name=moved.name,
                         path=str(moved.relative_to(tmp_path)),
                         size=moved.stat().st_size, type="audio", added_at=0)

    after = await _enrich(AudioEnricher(media_cache), renamed, moved, tmp_path)
    assert len(counted_reads) == 1, "the bytes did not change; the file need not be reopened"
    assert after["display_title"] == "New Name", "the cache answered for the old name"
    assert after["track_no"] == 7
    assert after["artist"] == "New Act"
    assert after["album"] == "New Record"


@pytest.mark.asyncio
async def test_a_cover_dropped_in_afterwards_is_still_found(tmp_path, media_cache):
    """
    The sibling scan reads the *folder*, so it stays live on top of the cache.
    Caching it would have made "no cover" permanent for content that later got
    one — and the scan costs 0.8s across a 6000-file library, measured.
    """
    folder = tmp_path / "An Act" / "A Record"
    folder.mkdir(parents=True)
    clip = folder / "01 - a track.mp3"
    _make_clip(clip)
    entry = IndexEntry(id="content5", name=clip.name,
                       path=str(clip.relative_to(tmp_path)),
                       size=clip.stat().st_size, type="audio", added_at=0)

    first = await _enrich(AudioEnricher(media_cache), entry, clip, tmp_path)
    assert first.get("thumb_hash") is None

    (folder / "cover.jpg").write_bytes(b"\xff\xd8\xff\xe0 not really a jpeg")
    second = await _enrich(AudioEnricher(media_cache), entry, clip, tmp_path)
    assert second.get("thumb_hash"), "the cache made a missing cover permanent"


@pytest.mark.asyncio
async def test_an_unreadable_file_is_not_remembered_as_empty(tmp_path, media_cache):
    """
    A drive that did not answer is not a file that carries no tags. Writing
    "says nothing" for it would make one bad read permanent.
    """
    clip = tmp_path / "01 - a track.mp3"
    _make_clip(clip, title="A Title", artist="An Act")
    entry = IndexEntry(id="content6", name=clip.name, path=clip.name,
                       size=clip.stat().st_size, type="audio", added_at=0)

    def boom(path, skip_cover=False):
        raise OSError("the drive said no")

    real = enrich_audio._read_tags_and_cover
    enrich_audio._read_tags_and_cover = boom
    try:
        await _enrich(AudioEnricher(media_cache), entry, clip)
    finally:
        enrich_audio._read_tags_and_cover = real
    assert await media_cache.get_audio_meta("content6") is None

    fields = await _enrich(AudioEnricher(media_cache), entry, clip)
    assert fields["artist"] == "An Act", "the failed read was cached and never retried"


@pytest.mark.asyncio
async def test_a_file_leaving_the_index_takes_its_row_with_it(tmp_path, media_cache):
    clip = tmp_path / "01 - a track.mp3"
    _make_clip(clip, artist="An Act")
    entry = IndexEntry(id="content7", name=clip.name, path=clip.name,
                       size=clip.stat().st_size, type="audio", added_at=0)
    await _enrich(AudioEnricher(media_cache), entry, clip)
    assert await media_cache.get_audio_meta("content7") is not None

    await media_cache.prune_file("content7")
    assert await media_cache.get_audio_meta("content7") is None