"""
A root that comes back keeps its Videos/Music/Photos metadata.

Reported live: a removable root ejected from the Files app and plugged back
in returned with its files and without its albums. Music showed "no music
found", and it did not come back.

`plug_root` has to re-walk the root — the drive may have changed while it
was away — and `_scan_root` produces bare entries: `_hash_or_cached` fills
id/name/path/size/type and nothing else. Every enrichment field went with
the old object, and the Music tag fields are cached nowhere by design
(`enrich_audio.py` re-reads them so a rename can re-derive the filename
fallback).

Two gates then stopped anything from filling them in again:

* enrichment is scheduled for `delta.additions`, and ejecting broadcasts
  nothing — the last snapshot still held those ids, so the rebuilt entries
  diffed as *updates*;
* `_enrich_new_*_entries` skips anything in `_enriched_attempted`, which is
  only discarded for `delta.deletions` — and dropping and rescanning inside
  one call broadcasts no deletion either.

Only a restart cleared both, an empty snapshot making every entry an
addition. That is why it looked like it might fix itself and never did.

Re-enriching is now the *fallback*, not the fix. An entry's id is its
content hash, so one that comes back under the same id, name and path is
the same bytes in the same place and its enrichment still holds:
`_rescan_root` carries those fields across. Re-deriving them instead meant
tag reads, ffprobe runs and rate-limited lookups — measured at 14 seconds
of empty Music tab on a real library with a cold cache, which to the
operator is indistinguishable from the original bug.

What these assert is therefore the field on the entry, not a call to an
enricher. Counting calls is what made an earlier version of this file pass
while the operator still watched their albums vanish.

The same drop-and-rescan runs in `reconcile()` — "Root %r is back" — so a
USB drive that falls off and returns on its own hits all of this without
anybody touching the UI.
"""

import asyncio
import os

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.crypto import generate_gek
from meshbay_node.config import Config, GroupConfig, HubConfig, KeystoreConfig, NodeConfig
from meshbay_node.daemon import NodeDaemon
from meshbay_node.indexer import DirectoryIndexer
from meshbay_node.indexer.enrich_audio import AudioEnricher
from meshbay_node.media_cache import MediaCache
from meshbay_node.roots import RootSet
from meshbay_node.roster import Roster

pytestmark = pytest.mark.asyncio

# Above indexer.py's MIN_AUDIO_SIZE_BYTES, or nothing would be indexed.
_AUDIO_BYTES = os.urandom(60 * 1024)


def _free_port() -> int:
    import socket
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


class _CountingEnricher:
    """Stands in for an enricher: records who it was asked to enrich."""

    def __init__(self):
        self.spawned: list[str] = []

    def spawn(self, entry, file_path, on_done, boundary=None):
        self.spawned.append(entry.name)


async def _daemon(tmp_path, shared, group_id):
    config = Config(
        hub=HubConfig(url="http://localhost:9999", username="testuser"),
        node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()),
        groups=[GroupConfig(
            id=group_id, name="test-group", shared_dir=str(shared),
            visibility="private", quic_port=_free_port(),
        )],
        keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
        data_dir=tmp_path / "data",
    )
    daemon = NodeDaemon(config)
    daemon._broadcast_coalesce_secs = 0.01
    daemon._media_cache = MediaCache(db_path=tmp_path / "media_cache.db")
    await daemon._media_cache.open()
    daemon._roster = Roster(db_path=tmp_path / "roster.db")
    await daemon._roster.open()
    return daemon


async def _settled(daemon, indexer):
    await daemon._on_index_change(indexer)
    await asyncio.sleep(0.05)


async def _library(tmp_path, group_id, *, enricher=None):
    """A one-track library under <root>/<artist>/<album>, enriched once."""
    library = tmp_path / "music"
    (library / "an artist" / "a record").mkdir(parents=True)
    (library / "an artist" / "a record" / "01 first track.mp3").write_bytes(_AUDIO_BYTES)

    daemon = await _daemon(tmp_path, library, group_id)
    daemon._audio_enricher = enricher or AudioEnricher(daemon._media_cache)
    await daemon._roster.set_app_directories(
        group_id, "music", ["music"], set_by="op")

    roots = RootSet.build([{"path": str(library), "removable": True}])
    indexer = DirectoryIndexer(
        roots=roots, group_id=group_id,
        sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
    await indexer.initial_scan()
    await _settled(daemon, indexer)
    await asyncio.sleep(0.4)  # the enricher runs off the broadcast
    return daemon, indexer, roots, library


def _only(indexer):
    return next(iter(indexer.index.entries))


# ── The operator's own eject and plug ────────────────────────────────────────

async def test_the_albums_are_still_there_after_a_replug(tmp_path):
    """
    No tags are written: `enrich_audio._artist_album_from_ancestors` derives
    artist and album from the folder names when a file has none, which is the
    <library>/<artist>/<album>/<track> layout this was reported against.
    """
    group_id = "a" * 32
    daemon, indexer, _, _ = await _library(tmp_path, group_id)
    try:
        before = _only(indexer)
        assert before.album == "a record" and before.artist == "an artist", (
            f"the first pass never filled the fields: {before}")

        indexer.eject_root("music")
        await indexer.plug_root("music")
        await _settled(daemon, indexer)

        after = _only(indexer)
        assert after.album == "a record" and after.artist == "an artist", (
            "the entry came back from the rescan with no album — this is what "
            "an empty Music tab after a replug looks like on the node")
    finally:
        await daemon._media_cache.close()
        await daemon._roster.close()


async def test_the_fields_survive_without_re_deriving_them(tmp_path):
    """
    Carried across, not recomputed. Re-deriving is correct and far too slow:
    on a real library with a cold metadata cache it left Music empty for 14
    seconds, and an operator who looks in that window sees the bug.
    """
    group_id = "b" * 32
    enricher = _CountingEnricher()
    daemon, indexer, _, _ = await _library(tmp_path, group_id, enricher=enricher)
    try:
        assert enricher.spawned == ["01 first track.mp3"]

        indexer.eject_root("music")
        await indexer.plug_root("music")
        await _settled(daemon, indexer)

        assert enricher.spawned == ["01 first track.mp3"], (
            "an unchanged file was enriched a second time — the whole point "
            "of the content hash is that it did not need to be")
    finally:
        await daemon._media_cache.close()
        await daemon._roster.close()


async def test_who_uploaded_a_file_survives_it_too(tmp_path):
    """
    `uploader_id`/`uploader_pk` are the same shape of field — set once, on an
    entry, readable from nowhere on disk — and they decide who may delete the
    file. Losing them to a replug quietly takes a right away.
    """
    group_id = "c" * 32
    daemon, indexer, _, _ = await _library(tmp_path, group_id)
    try:
        entry = _only(indexer)
        entry.uploader_id = "alice"
        entry.uploader_pk = "a-pinned-key"

        indexer.eject_root("music")
        await indexer.plug_root("music")
        await _settled(daemon, indexer)

        after = _only(indexer)
        assert after.uploader_id == "alice" and after.uploader_pk == "a-pinned-key"
    finally:
        await daemon._media_cache.close()
        await daemon._roster.close()


# ── A drive that leaves and returns on its own ──────────────────────────────

async def test_a_root_that_returns_on_its_own_is_treated_the_same(tmp_path):
    """
    `reconcile()` rescans a root that reappears without anyone asking — a USB
    drive re-mounting. Same drop-and-rescan, so it lost the same fields, with
    no click anywhere to blame it on.
    """
    group_id = "d" * 32
    daemon, indexer, roots, _ = await _library(tmp_path, group_id)
    try:
        assert _only(indexer).album == "a record"

        roots.roots[0].available = False
        await indexer.reconcile()
        await _settled(daemon, indexer)
        await indexer.reconcile()
        await _settled(daemon, indexer)
        await asyncio.sleep(0.4)

        assert _only(indexer).album == "a record", (
            "a drive that fell off and came back left the library with no "
            "metadata")
    finally:
        await daemon._media_cache.close()
        await daemon._roster.close()


# ── What genuinely does have to be re-derived ───────────────────────────────

async def test_a_track_moved_while_the_drive_was_away_is_enriched_again(tmp_path):
    """
    The counter-case, and the reason the carry-over is keyed on name and path
    as well as id. `artist`, `album`, `display_title` and `track_no` all fall
    back to the folder and filename when a file carries no tags, so the same
    bytes under a new name are not the same metadata. Those are the entries
    the daemon still re-enriches, off `rescanned_ids`.
    """
    group_id = "e" * 32
    enricher = _CountingEnricher()
    daemon, indexer, _, library = await _library(
        tmp_path, group_id, enricher=enricher)
    try:
        assert enricher.spawned == ["01 first track.mp3"]

        moved = library / "another artist" / "another record"
        moved.mkdir(parents=True)
        (library / "an artist" / "a record" / "01 first track.mp3").rename(
            moved / "01 first track.mp3")

        indexer.eject_root("music")
        await indexer.plug_root("music")
        await _settled(daemon, indexer)

        assert enricher.spawned == ["01 first track.mp3"] * 2, (
            "the file is under a different artist and album now; carrying the "
            "old ones across would file it under a folder it left")
    finally:
        await daemon._media_cache.close()
        await daemon._roster.close()


async def test_videos_and_photos_are_covered_by_the_same_path(tmp_path):
    """
    Nothing here is specific to Music — Videos and Photos lost their durations,
    titles and thumbnails the same way. Music is simply where it shows up
    loudest: a track with no tags has no album to file it under, so the app
    goes empty rather than merely plain.
    """
    group_id = "f" * 32
    library = tmp_path / "media"
    (library / "films").mkdir(parents=True)
    (library / "films" / "clip.mkv").write_bytes(os.urandom(60 * 1024))
    (library / "album").mkdir(parents=True)
    (library / "album" / "shot.jpg").write_bytes(os.urandom(60 * 1024))

    daemon = await _daemon(tmp_path, library, group_id)
    video, photo = _CountingEnricher(), _CountingEnricher()
    daemon._enricher, daemon._photo_enricher = video, photo
    try:
        await daemon._roster.set_app_directories(
            group_id, "video", ["media/films"], set_by="op")
        await daemon._roster.set_app_directories(
            group_id, "photo", ["media/album"], set_by="op")

        roots = RootSet.build([{"path": str(library), "removable": True}])
        indexer = DirectoryIndexer(
            roots=roots, group_id=group_id,
            sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
        await indexer.initial_scan()
        await _settled(daemon, indexer)
        assert video.spawned == ["clip.mkv"] and photo.spawned == ["shot.jpg"]

        by_name = {e.name: e for e in indexer.index.entries}
        by_name["clip.mkv"].duration = 1234
        by_name["shot.jpg"].thumb_hash = "a-thumbnail"

        indexer.eject_root("media")
        await indexer.plug_root("media")
        await _settled(daemon, indexer)

        back = {e.name: e for e in indexer.index.entries}
        assert back["clip.mkv"].duration == 1234, "the film lost its probe"
        assert back["shot.jpg"].thumb_hash == "a-thumbnail", (
            "the photo lost its thumbnail")
        assert video.spawned == ["clip.mkv"] and photo.spawned == ["shot.jpg"], (
            "unchanged files were probed and thumbnailed all over again")
    finally:
        await daemon._media_cache.close()
        await daemon._roster.close()