import hashlib
import io
import json
from concurrent.futures import ThreadPoolExecutor

import pytest
from PIL import Image

from pigway_frame_flow.errors import FlowError
from pigway_frame_flow.providers.photo_sources import PhotoSourceInput
from pigway_frame_flow.providers.photos_cache import PhotosCache
from pigway_frame_flow.providers.synology_photos import SynologyPhotos


def jpeg(color="red"):
    out = io.BytesIO()
    Image.new("RGB", (24, 16), color).save(out, "JPEG")
    return out.getvalue()


def session(settings, value="fixture-session"):
    settings.config_dir.mkdir(parents=True, exist_ok=True)
    (settings.config_dir / "synology-photos-session.json").write_text(
        json.dumps({"sid": value})
    )


def row(id, folder="/family", name="same.jpg"):
    return {
        "id": id,
        "filename": name,
        "type": "photo",
        "filesize": 9000,
        "indexed_time": 42,
        "additional": {
            "folder": folder,
            "resolution": {"width": 24, "height": 16},
            "thumbnail": {"unit_id": id, "cache_key": "fixture-key"},
        },
    }


@pytest.mark.parametrize(
    "space,prefix", [("shared", "SYNO.FotoTeam"), ("personal", "SYNO.Foto")]
)
@pytest.mark.parametrize("mode,expected", [("any", ["1", "2", "3"]), ("all", ["2"])])
def test_api_identity_selection(settings, monkeypatch, space, prefix, mode, expected):
    session(settings)
    p = SynologyPhotos(settings)
    monkeypatch.setattr(p, "people", lambda space: [{"id": 10}, {"id": 20}])

    def pages(api, version, **params):
        assert api == prefix + ".Browse.Item"
        return [row(i) for i in ([1, 2] if params["person_id"] == 10 else [2, 3])]

    monkeypatch.setattr(p, "pages", pages)
    assert [str(x["id"]) for x in p.items([10, 20], mode, space=space)] == expected
    with pytest.raises(FlowError):
        p.items([99], space=space)


def test_people_pagination_does_not_stop_at_500(settings, monkeypatch):
    session(settings)
    p = SynologyPhotos(settings)
    calls = []

    def request(api, method, version, **params):
        calls.append(params["offset"])
        return {
            "list": [
                {"id": i}
                for i in range(params["offset"], min(1334, params["offset"] + 500))
            ]
        }

    monkeypatch.setattr(p, "request", request)
    assert len(p.people()) == 1334
    assert calls == [0, 500, 1000]


def test_caption_people_use_actual_membership_and_cache(settings, monkeypatch):
    session(settings)
    provider = SynologyPhotos(settings)
    monkeypatch.setattr(
        provider,
        "people",
        lambda space: [
            {"id": 1, "name": "Alice", "named": True},
            {"id": 2, "name": "Bob", "named": True},
            {"id": 3, "name": "3", "named": False},
        ],
    )
    calls = []

    def pages(api, version, **params):
        calls.append(api)
        return [{"id": params["person_id"]}, {"id": 10}]

    monkeypatch.setattr(provider, "pages", pages)
    assert provider.item_people(1) == ["Alice"]
    assert provider.item_people(10) == ["Alice", "Bob"]
    assert provider.item_people(99) == []
    assert calls == ["SYNO.FotoTeam.Browse.Item"] * 2
    assert provider.item_people(10, space="personal") == ["Alice", "Bob"]
    assert calls[-2:] == ["SYNO.Foto.Browse.Item"] * 2


@pytest.mark.parametrize("labels_available", [True, False])
def test_api_caption_keeps_filename_and_folder_with_optional_people(
    service, settings, monkeypatch, labels_available
):
    session(settings)
    provider = service.photo_sources
    monkeypatch.setattr(
        provider.synology,
        "items",
        lambda *a, **k: [row(1, "/20261017 Family@Park", "one.jpg")],
    )
    monkeypatch.setattr(provider.synology, "download", lambda *a, **k: jpeg())

    def names(*a, **k):
        if not labels_available:
            raise FlowError("CLOUD_UNAVAILABLE", "unavailable", 502)
        return ["Alice", "Bob"]

    monkeypatch.setattr(provider.synology, "item_people", names)
    provider.save(
        "captions",
        PhotoSourceInput(
            name="Family", kind="synology-photos", location=".", folder_caption=True
        ),
    )
    _, _, caption = provider.read("captions", "1")
    assert caption["date"] == "2026-10-17"
    assert caption["title"] == "Family"
    assert caption["filename"] == "one.jpg"
    assert caption.get("people", "") == ("Alice · Bob" if labels_available else "")


def test_source_without_mount_pagination_scope_and_session(
    client, service, settings, monkeypatch
):
    session(settings)
    settings.photo_root = settings.data_dir / "must-not-exist"
    p = service.photo_sources
    monkeypatch.setattr(
        p.synology,
        "items",
        lambda *a, **k: [row(1), row(2, "/family/child"), row(3, "/other")],
    )
    monkeypatch.setattr(p.synology, "download", lambda *a, **k: jpeg())
    source = PhotoSourceInput(
        name="API",
        kind="synology-photos",
        location="family",
        people_filter=True,
        person_ids=[10],
        orientation="landscape",
    )
    p.save("api", source)
    response = client.get("/api/v1/photo-sources/api/images?offset=1&limit=1")
    assert response.status_code == 200, response.text
    assert response.json()["total"] == 2
    assert response.json()["items"][0]["id"] == "2"
    assert "cache_key" not in response.text and "_photos" not in response.text
    data, name, _ = p.read("api", "1")
    Image.open(io.BytesIO(data)).load()
    assert name == "same.jpg" and not settings.photo_root.exists()
    with pytest.raises(FlowError):
        p.read("api", "3")
    p.synology.disconnect()
    with pytest.raises(FlowError) as error:
        p.catalog("api")
    assert error.value.status == 409
    session(settings, "different-account")
    monkeypatch.setattr(p.synology, "items", lambda *a, **k: [])
    assert p.catalog("api") == []


def test_existing_local_sources_are_not_silently_converted(service, settings):
    settings.photo_root = settings.data_dir / "missing"
    with pytest.raises(FlowError):
        service.photo_sources.save(
            "local", PhotoSourceInput(name="Legacy", kind="local", location=".")
        )
    with pytest.raises(FlowError):
        service.photo_sources.save(
            "api",
            PhotoSourceInput(name="API", kind="synology-photos", location="../secrets"),
        )


def test_cache_coalesces_concurrency_and_recovers_partial_files(settings):
    settings.photos_cache_bytes = 2000
    settings.max_bytes = 1000
    cache = PhotosCache(settings)

    def download(handle, cap, deadline):
        handle.write(jpeg())

    with ThreadPoolExecutor(max_workers=8) as pool:
        results = list(pool.map(lambda _: cache.read("one", download), range(8)))
    assert all(x == jpeg() for x in results)
    assert cache.downloads == 1 and cache.hits == 7
    (cache.root / "interrupted.part").write_bytes(b"partial")
    cache.read("two", download)
    cache.read("three", download)
    assert sum(p.stat().st_size for p in cache.root.glob("*.image")) <= 2000
    assert not list(cache.root.glob("*.part"))
    assert cache.read("three", download) == jpeg()


@pytest.mark.parametrize("failure", ["interrupt", "invalid", "oversize"])
def test_cache_failures_never_commit(settings, failure):
    settings.max_bytes = 1000
    cache = PhotosCache(settings)

    def download(handle, cap, deadline):
        handle.write(
            jpeg()
            if failure == "interrupt"
            else b"wrong"
            if failure == "invalid"
            else b"x" * 1001
        )
        if failure == "interrupt":
            raise TimeoutError()

    with pytest.raises(FlowError):
        cache.read("bad", download)
    assert not list(cache.root.glob("*.part"))
    assert not list(cache.root.glob("*.image"))


def transport(
    settings, monkeypatch, *, bad_pin=False, expired=False, interrupted=False
):
    monkeypatch.setenv("SYNOLOGY_PHOTOS_URL", "https://nas.test:9115")
    monkeypatch.setenv(
        "SYNOLOGY_PHOTOS_CERT_SHA256",
        "a" * 64 if bad_pin else hashlib.sha256(b"certificate").hexdigest(),
    )
    sent = []

    class Socket:
        def getpeercert(self, **kw):
            return b"certificate"

    class Response:
        status = 200

        def __init__(self):
            self.buffer = io.BytesIO(
                b'{"success":false,"error":{"code":106}}' if expired else jpeg()
            )

        def getheader(self, name):
            return (
                ("application/json" if expired else "image/jpeg")
                if name == "Content-Type"
                else str(len(self.buffer.getvalue()))
            )

        def read(self, size):
            if interrupted:
                raise TimeoutError()
            return self.buffer.read(size)

    class Connection:
        sock = Socket()

        def __init__(self, *a, **kw):
            pass

        def connect(self):
            pass

        def request(self, *args):
            sent.append(args)

        def getresponse(self):
            return Response()

        def close(self):
            pass

    monkeypatch.setattr("http.client.HTTPSConnection", Connection)
    session(settings)
    return SynologyPhotos(settings), sent


def test_binary_transport_pin_before_secret_and_read_allowlist(settings, monkeypatch):
    p, sent = transport(settings, monkeypatch, bad_pin=True)
    with pytest.raises(FlowError):
        p.download(row(1))
    assert not sent
    with pytest.raises(FlowError):
        p._download(
            "SYNO.FotoTeam.Browse.Item",
            "delete",
            {},
            "fixture",
            io.BytesIO(),
            1000,
            100,
        )
    assert not sent


def test_download_decodes_cache_reuse_and_rendition_rejection(settings, monkeypatch):
    p, sent = transport(settings, monkeypatch)
    assert p.download(row(1)) == jpeg()
    assert p.download(row(1)) == jpeg()
    assert len(sent) == 1
    assert "SYNO.FotoTeam.Thumbnail" in sent[0][2]
    assert p.download(row(1), space="personal") == jpeg()
    assert len(sent) == 2
    session(settings, "different-session")
    assert p.download(row(1)) == jpeg()
    assert len(sent) == 3
    with pytest.raises(FlowError):
        p.download(row(1), rendition="original")


def test_download_session_error_and_interruption(settings, monkeypatch):
    p, _ = transport(settings, monkeypatch, expired=True)
    with pytest.raises(FlowError) as error:
        p.download(row(1))
    assert error.value.status == 409
    assert not p.path.exists()
    assert not list(p.image_cache.root.glob("*.part"))
    p, _ = transport(settings, monkeypatch, interrupted=True)
    with pytest.raises(FlowError):
        p.download(row(1))
    assert not list(p.image_cache.root.glob("*.part"))


def cache_process(settings, marker):
    cache = PhotosCache(settings)

    def download(handle, cap, deadline):
        with open(marker, "ab") as counter:
            counter.write(b"download\n")
        handle.write(jpeg())

    assert cache.read("shared-process-key", download) == jpeg()


def test_cache_coalesces_across_processes(settings, tmp_path):
    import multiprocessing

    context = multiprocessing.get_context("spawn")
    marker = tmp_path / "downloads"
    processes = [
        context.Process(target=cache_process, args=(settings, marker)) for _ in range(3)
    ]
    for process in processes:
        process.start()
    for process in processes:
        process.join(15)
        assert process.exitcode == 0
    assert marker.read_bytes() == b"download\n"


def test_explicit_offline_conversion_preserves_task_reference(service, settings):
    import importlib.util
    from pathlib import Path

    spec = importlib.util.spec_from_file_location(
        "photos_migration", Path(__file__).parents[1] / "scripts/migrate-photos-api.py"
    )
    migration = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(migration)
    # The production service uses this filename; inspect the fixture connection
    # rather than modifying an unrelated database.
    path = Path(service.db.conn.execute("PRAGMA database_list").fetchone()[2])
    source = PhotoSourceInput(
        name="Existing",
        kind="local",
        location="family",
        people_filter=True,
        person_ids=[5],
        synology_previews=True,
        people_space="personal",
        person_match="all",
    )
    service.db.execute(
        "INSERT INTO app_config VALUES(?,?)",
        ("photo_source:existing", source.model_dump_json()),
    )
    before = service.photo_sources.get("existing")
    assert not migration.convert(path, ["existing"])["applied"]
    assert service.photo_sources.get("existing") == before
    result = migration.convert(path, ["existing"], apply=True)
    assert Path(result["backup"]).is_file()
    after = service.photo_sources.get("existing")
    assert after.kind == "synology-photos" and not after.synology_previews
    assert (
        after.person_ids == [5]
        and after.people_space == "personal"
        and after.person_match == "all"
        and after.location == "family"
    )