from __future__ import annotations

import sqlite3
from collections.abc import Callable, Iterable, Mapping
from contextlib import AbstractContextManager
from pathlib import Path
from threading import Event, Thread
from typing import cast

import pytest
import uvicorn
from fastapi.testclient import TestClient
from starlette.types import ASGIApp

import game_data_capture.runtime as runtime_module
from game_data_capture.config import AppConfig, CaptureLimits, LifecycleConfig, ServerConfig
from game_data_capture.ingest.clipboard import IntakeReceipt
from game_data_capture.ingest.spool_helper import PNG_SIGNATURE
from game_data_capture.paths import RuntimePaths
from game_data_capture.runtime import (
    CaptureRuntime,
    InstanceLock,
    IntakePort,
    RuntimeFailure,
    _CaptureControl,
)
from game_data_capture.storage import CaptureCorpus, CaptureRepository, StorageDatabase
from game_data_capture.web.app import WebConfig

_PNG = PNG_SIGNATURE + b"unit-png"


def _paths(tmp_path: Path) -> RuntimePaths:
    root = tmp_path / "xdg"
    return RuntimePaths(
        config_dir=(root / "config").resolve(),
        data_dir=(root / "data").resolve(),
        cache_dir=(root / "cache").resolve(),
        runtime_dir=(root / "runtime").resolve(),
    )


def _runtime(tmp_path: Path, *, limits: CaptureLimits | None = None) -> CaptureRuntime:
    return CaptureRuntime(
        AppConfig(
            lifecycle=LifecycleConfig(lock_timeout_seconds=0), limits=limits or CaptureLimits()
        ),
        _paths(tmp_path),
        migrations_dir=(Path(__file__).parents[3] / "migrations").resolve(),
    )


def test_one_shot_persists_distinct_occurrences_and_one_blob(tmp_path: Path) -> None:
    runtime = _runtime(tmp_path)
    image = tmp_path / "image.png"
    image.write_bytes(_PNG)

    first = runtime.ingest_image(image)
    second = runtime.ingest_image(image)
    database = StorageDatabase(runtime.paths.data_dir, runtime.migrations_dir)

    with database.connect() as connection:
        assert first != second
        assert connection.execute("SELECT count(*) FROM captures").fetchone()[0] == 2
        assert connection.execute("SELECT count(*) FROM blobs").fetchone()[0] == 1


@pytest.mark.parametrize("kind", ["malformed", "oversize", "symlink", "broad-mode"])
def test_one_shot_rejects_unsafe_png_inputs(tmp_path: Path, kind: str) -> None:
    runtime = _runtime(tmp_path, limits=CaptureLimits(max_image_bytes=len(_PNG)))
    image = tmp_path / "image.png"
    image.write_bytes(_PNG)
    if kind == "malformed":
        image.write_bytes(b"not-a-png")
    elif kind == "oversize":
        image.write_bytes(_PNG + b"x")
    elif kind == "symlink":
        linked = tmp_path / "linked.png"
        linked.symlink_to(image)
        image = linked
    else:
        image.chmod(0o666)
    try:
        with pytest.raises(RuntimeFailure, match="safe PNG"):
            runtime.ingest_image(image)
    finally:
        if image.exists() and not image.is_symlink():
            image.chmod(0o600)


def test_stale_lock_file_is_reused_but_an_active_lock_is_not(tmp_path: Path) -> None:
    paths = _paths(tmp_path)
    paths.runtime_dir.mkdir(parents=True)
    paths.instance_lock_file.write_text("stale", encoding="utf-8")
    paths.instance_lock_file.chmod(0o600)
    first = InstanceLock.acquire(paths.instance_lock_file, 0)
    try:
        with pytest.raises(RuntimeFailure, match="another capture instance is active"):
            InstanceLock.acquire(paths.instance_lock_file, 0)
    finally:
        first.release()
    second = InstanceLock.acquire(paths.instance_lock_file, 0)
    second.release()
    assert paths.instance_lock_file.exists()


class InterruptingIntake:
    def __init__(self) -> None:
        self.shutdown_called = False

    def reconcile(self) -> list[str]:
        return []

    def arm(self) -> None:
        return None

    def tick(self) -> None:
        raise KeyboardInterrupt

    def take_helper_receipt(self) -> dict[str, object] | None:
        return None

    def accept_helper_receipt(
        self, payload: Mapping[str, object], *, occurrence_id: str, processing_version: str
    ) -> bool:
        return False

    def take(self) -> IntakeReceipt | None:
        return None

    def open_verified(self, receipt: IntakeReceipt) -> AbstractContextManager[Iterable[bytes]]:
        raise AssertionError("no receipt should be opened")

    def acknowledge(self, receipt: IntakeReceipt) -> bool:
        raise AssertionError("no receipt should be acknowledged")

    def shutdown(self) -> None:
        self.shutdown_called = True


def test_keyboard_interrupt_reaps_intake_and_releases_runtime_lock(tmp_path: Path) -> None:
    intake = InterruptingIntake()
    runtime = CaptureRuntime(
        AppConfig(lifecycle=LifecycleConfig(lock_timeout_seconds=0)),
        _paths(tmp_path),
        migrations_dir=(Path(__file__).parents[3] / "migrations").resolve(),
        intake_factory=lambda _config: intake,
    )
    runtime.watch(poll_seconds=0.001)

    assert intake.shutdown_called
    reusable = InstanceLock.acquire(runtime.paths.instance_lock_file, 0)
    reusable.release()


def test_cancelled_watch_cleans_up_without_entering_the_tick_loop(tmp_path: Path) -> None:
    cancelled = Event()
    cancelled.set()
    intake = InterruptingIntake()
    runtime = CaptureRuntime(
        AppConfig(lifecycle=LifecycleConfig(lock_timeout_seconds=0)),
        _paths(tmp_path),
        migrations_dir=(Path(__file__).parents[3] / "migrations").resolve(),
        intake_factory=lambda _config: intake,
    )
    runtime.watch(cancelled=cancelled, poll_seconds=0.001)
    assert intake.shutdown_called


def test_one_shot_never_connects_outside_the_utility_data_root(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    runtime = _runtime(tmp_path)
    image = tmp_path / "image.png"
    image.write_bytes(_PNG)
    real_connect = cast(Callable[..., sqlite3.Connection], sqlite3.connect)
    connected: list[Path] = []

    def checked_connect(database: str | Path, **kwargs: object) -> sqlite3.Connection:
        connected.append(Path(database))
        return real_connect(database, **kwargs)

    monkeypatch.setattr(sqlite3, "connect", checked_connect)
    runtime.ingest_image(image)
    assert connected
    assert all(path.parent == runtime.paths.data_dir for path in connected)


class ControlledIntake:
    def __init__(self, *, fail_tick: bool = False) -> None:
        self.armed = False
        self.reconciled = False
        self.shutdown_called = Event()
        self.queue_depth = 0
        self.helper_queue_depth = 0
        self.fail_tick = fail_tick

    def reconcile(self) -> list[str]:
        self.reconciled = True
        return []

    def arm(self) -> None:
        self.armed = True

    def tick(self) -> None:
        if self.fail_tick:
            raise OSError("helper fault")

    def take_helper_receipt(self) -> dict[str, object] | None:
        return None

    def accept_helper_receipt(
        self, payload: Mapping[str, object], *, occurrence_id: str, processing_version: str
    ) -> bool:
        return False

    def take(self) -> IntakeReceipt | None:
        return None

    def open_verified(self, receipt: IntakeReceipt) -> AbstractContextManager[Iterable[bytes]]:
        raise AssertionError("no receipt should be opened")

    def acknowledge(self, receipt: IntakeReceipt) -> bool:
        raise AssertionError("no receipt should be acknowledged")

    def shutdown(self) -> None:
        self.shutdown_called.set()


class FaultingCleanupIntake(ControlledIntake):
    def __init__(self) -> None:
        super().__init__(fail_tick=True)
        self.cleanup_started = Event()
        self.cleanup_release = Event()
        self._shutdown_calls = 0

    def shutdown(self) -> None:
        self._shutdown_calls += 1
        self.cleanup_started.set()
        if self._shutdown_calls == 1:
            self.cleanup_release.wait()
        self.shutdown_called.set()


class MalformedQueueIntake:
    @property
    def queue_depth(self) -> object:
        raise OSError("private intake failure")

    @property
    def helper_queue_depth(self) -> object:
        return True


class EmptyRepository:
    def receive(
        self,
        capture_id: str,
        receipt_token: str,
        chunks: Iterable[bytes],
        *,
        media_type: str,
        max_bytes: int,
    ) -> str:
        raise AssertionError("no receipt should be stored")


def test_control_serializes_repeated_start_stop_and_reports_bounded_queue(tmp_path: Path) -> None:
    intake = ControlledIntake()
    runtime = CaptureRuntime(
        AppConfig(lifecycle=LifecycleConfig(lock_timeout_seconds=0)),
        _paths(tmp_path),
        migrations_dir=(Path(__file__).parents[3] / "migrations").resolve(),
        intake_factory=lambda _config: intake,
    )
    control = _CaptureControl(runtime, cast(CaptureRepository, EmptyRepository()))
    outcomes: list[str] = []

    def start() -> None:
        outcomes.append(control.start_session())

    first, second = Thread(target=start), Thread(target=start)
    first.start()
    second.start()
    first.join()
    second.join()

    assert sorted(outcomes) == ["already_armed", "started"]
    assert intake.reconciled and intake.armed
    assert control.status().queue_depth == 0
    assert control.status().queue_capacity == runtime.config.limits.max_queue_items
    assert control.stop_session() in {"stopped", "already_stopped"}
    assert control.stop_session() == "already_stopped"
    assert intake.shutdown_called.wait(0.1)


def test_control_status_combines_and_bounds_both_intake_queues(tmp_path: Path) -> None:
    intake = ControlledIntake()
    runtime = CaptureRuntime(
        AppConfig(lifecycle=LifecycleConfig(lock_timeout_seconds=0)),
        _paths(tmp_path),
        migrations_dir=(Path(__file__).parents[3] / "migrations").resolve(),
        intake_factory=lambda _config: intake,
    )
    control = _CaptureControl(runtime, cast(CaptureRepository, EmptyRepository()))

    intake.queue_depth, intake.helper_queue_depth = 2, 3
    control._intake = cast(IntakePort, intake)
    assert control.status().queue_depth == 5
    intake.queue_depth, intake.helper_queue_depth = 1000, -1
    assert control.status().queue_depth == runtime.config.limits.max_queue_items
    control._intake = cast(IntakePort, MalformedQueueIntake())
    assert control.status().queue_depth == 0


def test_control_turns_background_intake_fault_into_bounded_disarmed_state(tmp_path: Path) -> None:
    intake = ControlledIntake(fail_tick=True)
    runtime = CaptureRuntime(
        AppConfig(lifecycle=LifecycleConfig(lock_timeout_seconds=0)),
        _paths(tmp_path),
        migrations_dir=(Path(__file__).parents[3] / "migrations").resolve(),
        intake_factory=lambda _config: intake,
    )
    control = _CaptureControl(runtime, cast(CaptureRepository, EmptyRepository()))

    assert control.start_session() == "started"
    assert intake.shutdown_called.wait(0.2)
    assert control.stop_session() in {"stopped", "already_stopped"}
    assert control.status().armed is False
    assert control.status().queue_depth == 0


def test_service_retains_its_lock_until_faulting_worker_cleanup_finishes(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    intake = FaultingCleanupIntake()
    runtime = CaptureRuntime(
        AppConfig(lifecycle=LifecycleConfig(lock_timeout_seconds=0, shutdown_timeout_seconds=1)),
        _paths(tmp_path),
        migrations_dir=(Path(__file__).parents[3] / "migrations").resolve(),
        intake_factory=lambda _config: intake,
    )
    captured: dict[str, _CaptureControl] = {}

    def factory(**kwargs: object) -> object:
        captured["control"] = cast(_CaptureControl, kwargs["capture_control"])
        return object()

    def runner(_: object, **__: object) -> None:
        assert captured["control"].start_session() == "started"
        assert intake.cleanup_started.wait(0.2)

    monkeypatch.setattr(runtime_module, "create_app", factory)
    monkeypatch.setattr(uvicorn, "run", runner)
    failure: list[RuntimeFailure] = []

    def serve() -> None:
        try:
            runtime.serve()
        except RuntimeFailure as error:
            failure.append(error)

    service = Thread(target=serve)
    service.start()
    try:
        service.join(timeout=2)
        assert not service.is_alive()
        assert failure
        with pytest.raises(RuntimeFailure, match="another capture instance"):
            InstanceLock.acquire(runtime.paths.instance_lock_file, 0)
        outcomes: list[str] = []
        start = Thread(target=lambda: outcomes.append(captured["control"].start_session()))
        stop = Thread(target=lambda: outcomes.append(captured["control"].stop_session()))
        start.start()
        stop.start()
        start.join()
        stop.join()
        assert sorted(outcomes) == ["conflict", "conflict"]
    finally:
        intake.cleanup_release.set()
    for _ in range(50):
        try:
            reusable = InstanceLock.acquire(runtime.paths.instance_lock_file, 0)
        except RuntimeFailure:
            Event().wait(0.01)
        else:
            reusable.release()
            break
    else:
        pytest.fail("service lock was not released after worker cleanup")


def test_serve_constructs_durable_corpus_without_auto_arming_and_uses_one_fragment_token(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
    created: list[ControlledIntake] = []
    token = "A" * 43

    def factory(_: object) -> ControlledIntake:
        intake = ControlledIntake()
        created.append(intake)
        return intake

    runtime = CaptureRuntime(
        AppConfig(
            server=ServerConfig(host="127.0.0.1", port=9876),
            lifecycle=LifecycleConfig(lock_timeout_seconds=0),
        ),
        _paths(tmp_path),
        migrations_dir=(Path(__file__).parents[3] / "migrations").resolve(),
        intake_factory=factory,
    )
    monkeypatch.setattr(runtime_module, "new_bootstrap_token", lambda: token)
    received: dict[str, object] = {}

    def runner(app: object, **kwargs: object) -> None:
        received.update(kwargs)
        with TestClient(cast(ASGIApp, app), base_url="http://127.0.0.1:9876") as client:
            assert client.get("/health").json() == {"status": "ok"}
            assert client.get("/").status_code == 200
            assert created == []
            bootstrapped = client.post(
                "/api/bootstrap",
                json={"token": token},
                headers={"Origin": "http://127.0.0.1:9876"},
            )
            csrf = bootstrapped.json()["csrf_token"]
            assert client.get("/api/session").json()["armed"] is False
            started = client.post(
                "/api/session/start",
                json={"confirmation": "START"},
                headers={"Origin": "http://127.0.0.1:9876", "X-CSRF-Token": csrf},
            )
            assert started.json()["result"] == "started"
            assert created[0].reconciled and created[0].armed

    monkeypatch.setattr(uvicorn, "run", runner)
    runtime.serve()

    output = capsys.readouterr().out
    assert output.count(token) == 1
    assert f"http://127.0.0.1:9876/#bootstrap={token}" in output
    assert received == {
        "host": "127.0.0.1",
        "port": 9876,
        "reload": False,
        "workers": 1,
        "proxy_headers": False,
    }
    assert created[0].shutdown_called.wait(0.1)
    reusable = InstanceLock.acquire(runtime.paths.instance_lock_file, 0)
    reusable.release()


def test_serve_loads_bounded_corpus_history_independently_of_intake_capacity(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    runtime = CaptureRuntime(
        AppConfig(
            limits=CaptureLimits(max_queue_items=64),
            lifecycle=LifecycleConfig(lock_timeout_seconds=0),
        ),
        _paths(tmp_path),
        migrations_dir=(Path(__file__).parents[3] / "migrations").resolve(),
    )
    captured: dict[str, object] = {}

    def factory(**kwargs: object) -> object:
        captured.update(kwargs)
        return object()

    def runner(_: object, **__: object) -> None:
        control = cast(_CaptureControl, captured["capture_control"])
        assert control.status().queue_capacity == runtime.config.limits.max_queue_items

    monkeypatch.setattr(runtime_module, "create_app", factory)
    monkeypatch.setattr(uvicorn, "run", runner)

    runtime.serve()

    web_config = cast(WebConfig, captured["config"])
    assert web_config.max_capture_items == 256
    assert web_config.max_capture_items > runtime.config.limits.max_queue_items


def test_service_startup_reconciliation_failure_releases_the_instance_lock(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    runtime = _runtime(tmp_path)

    def failed_reconciliation(self: CaptureCorpus, *, limit: int = 64) -> int:
        raise OSError("private deletion recovery fault")

    monkeypatch.setattr(CaptureCorpus, "reconcile_deletions", failed_reconciliation)
    with pytest.raises(RuntimeFailure, match="initialization failed"):
        runtime.serve()
    reusable = InstanceLock.acquire(runtime.paths.instance_lock_file, 0)
    reusable.release()