"""
Videos-app enrichment (ffprobe/thumbnailing/TMDB, mediacenter.md §5.2/§10)
only ever runs for a group that has a video_root configured, and only for
files under it — see daemon.py's _enrich_new_video_entries. Burning TMDB's
rate limit and the node's CPU on an operator's whole shared index before
they have chosen which folder is actually their media library would be
real, ongoing cost for files never meant to be in the Videos app at all.

Setting or changing the root (ops.set_video_root) fires a one-off sweep
(_enrich_video_root_now) of whatever it already contains: the ordinary
per-broadcast path only ever looks at files new since the last broadcast,
so anything already sitting in a folder before it became the video_root
would otherwise never be picked up.
"""

import asyncio

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from meshbay_common.crypto import generate_gek
from meshbay_node import ops
from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig
from meshbay_node.daemon import NodeDaemon
from meshbay_node.indexer import DirectoryIndexer
from meshbay_node.indexer.enrich import Enricher
from meshbay_node.media_cache import MediaCache
from meshbay_node.roster import Roster

from conftest import one_root

pytestmark = pytest.mark.asyncio


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


async def _make_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=29014,
        )],
        keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
        data_dir=tmp_path / "data",
    )
    daemon = NodeDaemon(config)
    daemon._broadcast_coalesce_secs = 0.01  # real value would make these tests wait 0.5s
    daemon._media_cache = MediaCache(db_path=tmp_path / "media_cache.db")
    await daemon._media_cache.open()
    daemon._enricher = Enricher(daemon._media_cache)
    daemon._roster = Roster(db_path=tmp_path / "roster.db")
    await daemon._roster.open()
    return daemon


async def _teardown(daemon):
    await daemon._media_cache.close()
    await daemon._roster.close()


async def test_no_video_root_means_no_enrichment_at_all(tmp_path):
    group_id = "a" * 32
    shared = tmp_path / "shared"
    shared.mkdir()
    (shared / "movie.mkv").write_bytes(b"not a real video, just needs to be indexed as one")

    daemon = await _make_daemon(tmp_path, shared, group_id)
    try:
        indexer = DirectoryIndexer(
            roots=one_root(shared), group_id=group_id,
            sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
        await indexer.initial_scan()
        await daemon._on_index_change(indexer)
        await asyncio.sleep(0.05)

        assert not daemon._enriched_attempted, (
            "a group with no video_root configured must not enrich anything, "
            "not even fall back to the whole index")
    finally:
        await _teardown(daemon)


async def test_only_entries_under_the_configured_root_are_enriched(tmp_path):
    group_id = "a" * 32
    shared = tmp_path / "shared"
    shared.mkdir()
    (shared / "Movies").mkdir()
    # Distinct content: the index dedupes by content hash, and two files with
    # the same bytes would otherwise collapse into a single entry.
    (shared / "Movies" / "in-root.mkv").write_bytes(b"in-root content")
    (shared / "outside.mkv").write_bytes(b"outside content")

    daemon = await _make_daemon(tmp_path, shared, group_id)
    try:
        await daemon._roster.set_video_root(group_id, "shared/Movies", set_by="op")
        indexer = DirectoryIndexer(
            roots=one_root(shared), group_id=group_id,
            sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
        await indexer.initial_scan()
        await daemon._on_index_change(indexer)
        await asyncio.sleep(0.05)

        by_name = {e.name: e for e in indexer.index.entries}
        assert (group_id, by_name["in-root.mkv"].id) in daemon._enriched_attempted
        assert (group_id, by_name["outside.mkv"].id) not in daemon._enriched_attempted, (
            "a file outside the configured video_root must never be enriched")
    finally:
        await _teardown(daemon)


async def test_setting_the_video_root_sweeps_what_it_already_contains(tmp_path):
    group_id = "a" * 32
    shared = tmp_path / "shared"
    shared.mkdir()
    (shared / "Movies").mkdir()
    (shared / "Movies" / "already-there.mkv").write_bytes(b"x")

    daemon = await _make_daemon(tmp_path, shared, group_id)
    try:
        indexer = DirectoryIndexer(
            roots=one_root(shared), group_id=group_id,
            sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
        await indexer.initial_scan()
        daemon._state["indexers"][group_id] = indexer

        # Broadcast once with nothing configured — nothing should be scheduled.
        await daemon._on_index_change(indexer)
        await asyncio.sleep(0.05)
        assert not daemon._enriched_attempted

        # Now the operator points video_root at the folder that already held
        # this file all along.
        state = {
            "roster": daemon._roster,
            "groups_ctx": {group_id: {}},
            "enrich_video_root_fn": daemon._enrich_video_root_now,
        }
        await ops.set_video_root(state, group_id, "shared/Movies")
        await asyncio.sleep(0.05)  # let the fire-and-forget sweep actually run

        entry = next(iter(indexer.index.entries))
        assert (group_id, entry.id) in daemon._enriched_attempted, (
            "a file already sitting in the newly-chosen root must be picked "
            "up by the sweep, not wait for some unrelated future change")
    finally:
        await _teardown(daemon)