"""
Bug found live, 2026-08-24: an episode file first named in French (its
release folder mixed languages across seasons) was renamed by the operator
to match its English-named siblings — but kept showing as its own
separate poster-grid card, and its own row in Flat list, indefinitely.

`_enriched_attempted` (daemon.py) exists so that enrichment's own
field-fill (duration/thumb_hash/display_title/... landing back via
`_on_enriched`) does not re-trigger itself forever — but it also silently
blocked the *new* filename from ever being title-parsed at all, since the
file's content (and so its id) is unchanged by a rename. A rename/move is
exactly the case `_reenrich_renamed_video_entries` exists to detect: same
id, but `name` or `path` differs from the version last broadcast.
"""

import asyncio

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

from meshbay_common.crypto import generate_gek
from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig
from meshbay_node.daemon import NodeDaemon
from meshbay_node.indexer import DirectoryIndexer

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]


class _StubRoster:
    async def video_root(self, group_id):
        return "shared"


class _SpyEnricher:
    """Records which file ids were actually (re-)scheduled, without
    needing a real ffmpeg/ffprobe pipeline for this test."""

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

    def spawn(self, entry, file_path, on_done):
        self.spawned.append(entry.id)

        async def _noop():
            return None

        return asyncio.ensure_future(_noop())


async def test_a_renamed_file_gets_re_enriched(tmp_path):
    group_id = "a" * 32
    shared = tmp_path / "shared"
    shared.mkdir()
    old_path = shared / "la-guerre-des-mondes-s03e02.mkv"
    old_path.write_bytes(b"not a real video, just needs to be indexed as one")

    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=29016,
        )],
        keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
        data_dir=tmp_path / "data",
    )
    daemon = NodeDaemon(config)
    daemon._broadcast_coalesce_secs = 0.01
    daemon._enricher = _SpyEnricher()
    daemon._roster = _StubRoster()

    indexer = DirectoryIndexer(
        roots=one_root(shared), group_id=group_id,
        sk_node=Ed25519PrivateKey.generate(), gek=generate_gek(),
        # reconcile() below calls this itself — the initial scan doesn't
        # (see test_startup_scan_enrichment.py), so that first broadcast is
        # still triggered manually, matching _bg_scan's real sequence.
        on_change=daemon._on_index_change)
    await indexer.initial_scan()
    await daemon._on_index_change(indexer)
    await asyncio.sleep(0.05)

    entry = next(iter(indexer.index.entries))
    file_id = entry.id
    assert daemon._enricher.spawned == [file_id], (
        "the file must be scheduled for enrichment once, under its original name")

    new_path = shared / "war-of-the-worlds-s03e02.mkv"
    old_path.rename(new_path)
    changed = await indexer.reconcile()
    assert changed, "the rename must actually be picked up by reconcile()"
    # The re-broadcast is a fire-and-forget task chained behind the
    # coalescing timer, itself scheduling another fire-and-forget task —
    # poll rather than guess a single sleep long enough for both hops.
    for _ in range(30):
        if len(daemon._enricher.spawned) >= 2:
            break
        await asyncio.sleep(0.02)

    renamed_entry = indexer.index.get_entry(file_id)
    assert renamed_entry is not None
    assert renamed_entry.name == "war-of-the-worlds-s03e02.mkv"
    assert daemon._enricher.spawned == [file_id, file_id], (
        "a rename must re-schedule enrichment for the same file id — "
        "_enriched_attempted must not permanently block the new filename "
        "from ever being title-parsed")


async def test_an_unrelated_update_does_not_re_trigger_enrichment(tmp_path):
    """
    The other half of the same fix: an update whose name/path did *not*
    change (the ordinary case — enrichment's own field-fill, or the
    reconcile sweep confirming a file unmodified) must not re-schedule
    enrichment. Without this, `_on_enriched` merging a file's own results
    back into the index would count as its own trigger and loop forever.
    """
    group_id = "a" * 32
    shared = tmp_path / "shared"
    shared.mkdir()
    path = shared / "movie.mkv"
    path.write_bytes(b"not a real video, just needs to be indexed as one")

    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=29017,
        )],
        keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
        data_dir=tmp_path / "data",
    )
    daemon = NodeDaemon(config)
    daemon._broadcast_coalesce_secs = 0.01
    daemon._enricher = _SpyEnricher()
    daemon._roster = _StubRoster()

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

    entry = next(iter(indexer.index.entries))
    assert daemon._enricher.spawned == [entry.id]

    # A reconcile pass that finds nothing changed at all — not even a
    # rename — must not re-schedule anything.
    changed = await indexer.reconcile()
    await asyncio.sleep(0.05)
    assert not changed
    assert daemon._enricher.spawned == [entry.id]