"""
Files whose full path is longer than Windows' MAX_PATH (260 characters).

A deep library reaches it without trying: "Artist / Album (year, edition) /
01 - a long track title.flac" below a folder that is already deep. With the
system's default setting (`LongPathsEnabled` off) every call given a plain path
that long refuses it, whatever the program declares, and such a file was
invisible to the members: the walk could not list its directory, and said so
only in the log. The node now hands the system the extended form (`\\\\?\\`) at
each call that touches a group's disk, which works whatever the setting, and
keeps the plain form everywhere else: in the index, the hash cache, the log.

ffmpeg and ffprobe are given the plain form on purpose: they extend a long path
themselves, which the probe and thumbnail tests below hold.

These run only where the limit exists: with the setting on, a plain path works
too and they would prove nothing.
"""

import asyncio
import os
import shutil
import sys
from pathlib import Path

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.protocol import file_chunk_plaintext
from meshbay_node import ops
from meshbay_node.indexer.indexer import DirectoryIndexer
from meshbay_node.roots import RootSet, publish_upload, safe_subdir
from meshbay_node.transport.webrtc.disk import (
    _append_chunk,
    _free_bytes,
    _is_empty_dir,
    _locate,
    _mkdir_if_absent,
    _read_and_encrypt,
    _rmdir_if_empty,
)


def _limit_in_force() -> bool:
    if sys.platform != "win32":
        return False
    import winreg
    try:
        with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE,
                            r"SYSTEM\CurrentControlSet\Control\FileSystem") as key:
            return winreg.QueryValueEx(key, "LongPathsEnabled")[0] == 0
    except OSError:
        return True


pytestmark = pytest.mark.skipif(
    not _limit_in_force(), reason="MAX_PATH binds only on Windows with LongPathsEnabled off")

GID = "g" * 32
SEGMENT = "A folder name as long as the ones a real library has"


def _x(path: Path) -> str:
    """The extended form, for the test's own setup and checks."""
    return "\\\\?\\" + str(path)


def _write(path: Path, data: bytes) -> Path:
    with open(_x(path), "wb") as f:
        f.write(data)
    return path


def _read(path: Path) -> bytes:
    with open(_x(path), "rb") as f:
        return f.read()


@pytest.fixture
def library(tmp_path):
    """A root with one short file at the top and a directory past the limit.
    Removed here: the plain-path cleanup pytest does would stop at it."""
    root = tmp_path / "Library"
    deep = root
    while len(str(deep)) < 280:
        deep = deep / SEGMENT
    os.makedirs(_x(deep))
    _write(root / "short.txt", b"near the top")
    yield root, deep
    shutil.rmtree(_x(root), ignore_errors=True)


def _indexer(roots: RootSet, **kw) -> DirectoryIndexer:
    return DirectoryIndexer(roots=roots, group_id=GID,
                            sk_node=Ed25519PrivateKey.generate(), gek=None, **kw)


def _names(idx: DirectoryIndexer) -> set[str]:
    return {e.name for e in idx.index.entries}


def _entry(idx: DirectoryIndexer, name: str):
    (entry,) = [e for e in idx.index.entries if e.name == name]
    return entry


def _ffmpeg() -> str:
    exe = shutil.which("ffmpeg")
    if exe is None:
        pytest.skip("ffmpeg is not on PATH")
    return exe


async def _make_media(args: list[str], out: Path) -> None:
    proc = await asyncio.create_subprocess_exec(
        _ffmpeg(), "-v", "error", "-y", *args, str(out),
        stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.PIPE)
    _, err = await proc.communicate()
    assert proc.returncode == 0, err.decode(errors="replace")


def test_the_fixture_is_past_the_limit(library):
    """The premise: a plain path this long is refused on this machine."""
    _, deep = library
    track = _write(deep / "track.mkv", b"deep down")
    assert len(str(track)) > 260
    assert not os.path.exists(str(track))
    with pytest.raises(OSError):
        os.scandir(str(deep))


# ── Indexing ────────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_a_scan_indexes_a_file_past_the_limit(library):
    root, deep = library
    _write(deep / "track.mkv", b"deep down" * 100)
    idx = _indexer(RootSet.build([{"path": str(root)}]))

    await idx.initial_scan()

    assert _names(idx) == {"short.txt", "track.mkv"}
    assert _entry(idx, "track.mkv").size == 900
    assert idx.unread == {}


@pytest.mark.asyncio
async def test_reconcile_finds_a_new_one_and_leaves_a_known_one_alone(library):
    """Reconcile compares the disk with the index by path. Were the two sides
    to spell a long path differently, it would rewrite the entry at every
    pass and push the change to every member."""
    root, deep = library
    _write(deep / "track.mkv", b"deep down")
    idx = _indexer(RootSet.build([{"path": str(root)}]))
    await idx.initial_scan()

    _write(deep / "later.mkv", b"added while the node ran")
    await idx.reconcile()
    assert "later.mkv" in _names(idx)

    version = idx.index.version
    await idx.reconcile()
    assert idx.index.version == version, "a pass with nothing new changed the index"


@pytest.mark.asyncio
async def test_a_watcher_event_indexes_it(library):
    root, deep = library
    idx = _indexer(RootSet.build([{"path": str(root)}]), debounce_secs=0.01)
    await idx.initial_scan()
    idx._loop = asyncio.get_running_loop()

    _write(deep / "arrived.mkv", b"one more")
    idx._schedule_update(deep / "arrived.mkv")
    for _ in range(200):
        if "arrived.mkv" in _names(idx):
            break
        await asyncio.sleep(0.01)

    assert "arrived.mkv" in _names(idx)


@pytest.mark.asyncio
async def test_the_real_watcher_sees_it(library):
    root, deep = library
    idx = _indexer(RootSet.build([{"path": str(root)}]), debounce_secs=0.05)
    await idx.start()
    try:
        _write(deep / "watched.mkv", b"seen by the system's own notification")
        for _ in range(100):
            if "watched.mkv" in _names(idx):
                break
            await asyncio.sleep(0.05)
    finally:
        await idx.stop()

    assert "watched.mkv" in _names(idx)


@pytest.mark.asyncio
async def test_a_root_that_is_itself_past_the_limit(library):
    _, deep = library
    _write(deep / "track.mkv", b"deep down")
    roots = RootSet.build([{"path": str(deep)}])
    assert roots.roots[0].is_live()
    idx = _indexer(roots)

    await idx.initial_scan()

    assert _names(idx) == {"track.mkv"}


# ── Serving and deleting ────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_it_is_served_to_a_member(library):
    root, deep = library
    data = os.urandom(5000)
    _write(deep / "track.mkv", data)
    roots = RootSet.build([{"path": str(root)}])
    idx = _indexer(roots)
    await idx.initial_scan()
    entry = _entry(idx, "track.mkv")
    gek = os.urandom(32)

    path, refusal = _locate(roots, entry)
    assert refusal is None
    msg = _read_and_encrypt(gek, path, 0, bytes.fromhex(entry.id), entry.id)

    assert file_chunk_plaintext(gek, msg) == data


@pytest.mark.asyncio
async def test_it_is_served_over_quic_too(library):
    from meshbay_node.transport import quic_server

    root, deep = library
    _write(deep / "track.mkv", b"over the other transport")
    roots = RootSet.build([{"path": str(root)}])
    idx = _indexer(roots)
    await idx.initial_scan()
    entry = _entry(idx, "track.mkv")
    gek = os.urandom(32)

    path, refusal = _locate(roots, entry)
    msg = quic_server._read_and_encrypt(gek, path, 0, bytes.fromhex(entry.id), entry.id)

    assert file_chunk_plaintext(gek, msg) == b"over the other transport"


@pytest.mark.asyncio
async def test_the_operator_can_delete_it(library):
    root, deep = library
    track = _write(deep / "track.mkv", b"to be deleted")
    roots = RootSet.build([{"path": str(root)}])
    idx = _indexer(roots)
    await idx.initial_scan()
    state = {"groups_ctx": {GID: {"index": idx.index, "roots": roots}}}

    await ops.delete_file(state, GID, _entry(idx, "track.mkv").id)

    assert not os.path.exists(_x(track))
    assert "track.mkv" not in _names(idx)


# ── Uploads and folders ─────────────────────────────────────────────────────

def test_an_upload_lands_in_a_directory_past_the_limit(library):
    root, deep = library
    roots = RootSet.build([{"path": str(root), "writable": True}])
    rel = "/".join(["Library", *deep.relative_to(root).parts])
    target = safe_subdir(roots, rel)
    assert target == deep

    assert _free_bytes(target) > 0
    part = target / "up.bin.0123abcd.part"
    _append_chunk(part, b"first ", True)
    _append_chunk(part, b"second", False)
    assert publish_upload(part, target, "up.bin", "up.bin") == "up.bin"
    assert _read(target / "up.bin") == b"first second"

    again = target / "up.bin.4567cdef.part"
    _append_chunk(again, b"same name", True)
    assert publish_upload(again, target, "up.bin", "up.bin") == "up (2).bin", \
        "a long path made the existing file look absent, and it was replaced"
    assert _read(target / "up.bin") == b"first second"


def test_an_abandoned_upload_is_found(library):
    from meshbay_node import uploads

    root, deep = library
    roots = RootSet.build([{"path": str(root), "writable": True}])
    part = _write(deep / "film.mkv.89abcdef.part", b"half a film")

    found = [p for p, _ in uploads.find_parts(roots.roots)]

    assert found == [part]


def test_members_see_the_folders_past_the_limit(library):
    """The folder tree a member browses and uploads into."""
    from meshbay_node.transport.wire import list_dirs

    root, deep = library
    os.makedirs(_x(deep / "Deeper still"))
    roots = RootSet.build([{"path": str(root)}])

    dirs = list_dirs(roots)

    rel = "/".join(["Library", *deep.relative_to(root).parts])
    assert rel in dirs
    assert f"{rel}/Deeper still" in dirs


def test_a_folder_is_created_and_removed_past_the_limit(library):
    _, deep = library
    folder = deep / "New folder"
    assert _mkdir_if_absent(folder) is None
    assert _mkdir_if_absent(folder) == "Already exists"
    assert _is_empty_dir(folder)
    assert _rmdir_if_empty(folder)
    assert not os.path.exists(_x(folder))


# ── Enrichment ──────────────────────────────────────────────────────────────

@pytest.mark.asyncio
@pytest.mark.daemon_loop      # a subprocess needs the proactor loop
async def test_audio_tags_and_the_cover_beside_it_are_read(library):
    from meshbay_node.indexer.enrich_audio import _read_tags_and_cover
    from PIL import Image

    _, deep = library
    track = deep / "01 - A Long Track Title.flac"
    await _make_media(["-f", "lavfi", "-i", "sine=duration=1",
                       "-metadata", "title=Some Song", "-metadata", "artist=Some Artist",
                       "-metadata", "album=Some Album"], track)
    with open(_x(deep / "cover.jpg"), "wb") as f:
        Image.new("RGB", (8, 8), "red").save(f, "JPEG")

    tags, duration, cover = await asyncio.to_thread(_read_tags_and_cover, track)

    assert tags["title"] == "Some Song" and tags["album"] == "Some Album"
    assert duration == pytest.approx(1.0, abs=0.1)
    assert cover == _read(deep / "cover.jpg")


@pytest.mark.asyncio
async def test_a_photo_is_read(library):
    from meshbay_node.indexer.enrich_photo import _read_image
    from PIL import Image

    _, deep = library
    with open(_x(deep / "photo.jpg"), "wb") as f:
        Image.new("RGB", (64, 48), "blue").save(f, "JPEG")

    thumb, width, height, _, _ = await asyncio.to_thread(_read_image, deep / "photo.jpg")

    assert (width, height) == (64, 48) and thumb


@pytest.mark.asyncio
@pytest.mark.daemon_loop      # a subprocess needs the proactor loop
async def test_ffmpeg_probes_and_thumbnails_it_from_the_plain_path(library):
    from meshbay_node.indexer.enrich import _make_thumbnail
    from meshbay_node.media_probe import probe_video

    _, deep = library
    clip = deep / "clip.mp4"
    await _make_media(["-f", "lavfi", "-i", "testsrc=size=64x48:rate=5:duration=2",
                       "-pix_fmt", "yuv420p"], clip)

    probe = await probe_video(str(clip))
    thumb = await _make_thumbnail(clip, probe.duration)

    assert probe.duration == pytest.approx(2.0, abs=0.3)
    assert thumb and thumb[:2] == b"\xff\xd8"


@pytest.mark.asyncio
async def test_a_show_folder_past_the_limit_numbers_its_episodes(library):
    from meshbay_node.indexer.enrich import (
        _season_and_show_from_ancestors,
        _synthetic_episode_number,
        _title_from_siblings,
    )

    _, deep = library
    season = deep / "Some Show" / "Season 1"
    os.makedirs(_x(season))
    _write(season / "Some Show S01E01.mkv", b"")
    _write(season / "Pilot without a number.mkv", b"")
    _write(season / "Second without a number.mkv", b"")
    later = season / "Second without a number.mkv"

    found = _season_and_show_from_ancestors(later)
    assert found is not None
    number, show = found
    assert await asyncio.to_thread(_synthetic_episode_number, later, show, number) == 2
    assert await asyncio.to_thread(_title_from_siblings, later) == "Some Show"