from __future__ import annotations

import io
import subprocess
from collections import deque
from collections.abc import Iterable, Mapping
from contextlib import AbstractContextManager, nullcontext
from pathlib import Path
from threading import Event

import pytest
import uvicorn

import game_data_capture.cli as cli
import game_data_capture.runtime as runtime_module
from game_data_capture.config import AppConfig, LifecycleConfig, ServerConfig
from game_data_capture.ingest.clipboard import (
    ClipboardIntake,
    FocusedWindowMetadata,
    IntakeReceipt,
    WatchConfig,
)
from game_data_capture.ingest.spool_helper import PNG_SIGNATURE, receipt_name, spool_png
from game_data_capture.paths import RuntimePaths
from game_data_capture.runtime import CaptureRuntime, InstanceLock, RuntimeFailure, _CaptureControl
from game_data_capture.storage import CaptureRepository, ContentAddressedBlobStore, StorageDatabase

_PNG = PNG_SIGNATURE + b"watch-png"


class FakeIntake:
    def __init__(self, cancelled: Event, payloads: list[dict[str, object]]) -> None:
        self.cancelled = cancelled
        self.payloads = deque(payloads)
        self.receipts: deque[IntakeReceipt] = deque()
        self.armed = self.reconciled = self.shutdown_called = False
        self.opened = 0
        self.acknowledged: list[str] = []
        self.seen_tokens: set[str] = set()
        self._ticks = 0

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

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

    def tick(self, now: float | None = None) -> None:
        self._ticks += 1
        if self._ticks > 1:
            self.cancelled.set()

    def take_helper_receipt(self) -> dict[str, object] | None:
        return self.payloads.popleft() if self.payloads else None

    def accept_helper_receipt(
        self,
        payload: Mapping[str, object],
        *,
        occurrence_id: str,
        processing_version: str,
    ) -> bool:
        token = str(payload["receipt_token"])
        if token in self.seen_tokens:
            return False
        self.seen_tokens.add(token)
        self.receipts.append(
            IntakeReceipt(
                occurrence_id,
                token,
                "0" * 64,
                len(_PNG),
                "image/png",
                Path("/private/spool.png"),
                "capture-v1",
                processing_version,
                FocusedWindowMetadata(None, None, None),
            )
        )
        return True

    def take(self) -> IntakeReceipt | None:
        return self.receipts.popleft() if self.receipts else None

    def open_verified(self, receipt: IntakeReceipt) -> AbstractContextManager[Iterable[bytes]]:
        self.opened += 1
        return nullcontext([_PNG])

    def acknowledge(self, receipt: IntakeReceipt) -> bool:
        self.acknowledged.append(receipt.receipt_token)
        return True

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


def _runtime(tmp_path: Path, intake: FakeIntake) -> CaptureRuntime:
    root = tmp_path / "xdg"
    paths = RuntimePaths(
        config_dir=(root / "config").resolve(),
        data_dir=(root / "data").resolve(),
        cache_dir=(root / "cache").resolve(),
        runtime_dir=(root / "runtime").resolve(),
    )
    return CaptureRuntime(
        AppConfig(lifecycle=LifecycleConfig(lock_timeout_seconds=0)),
        paths,
        migrations_dir=(Path(__file__).parents[2] / "migrations").resolve(),
        intake_factory=lambda _config: intake,
    )


def _payload(token: str) -> dict[str, object]:
    return {
        "receipt_version": 1,
        "receipt_token": token,
        "sha256": "0" * 64,
        "byte_size": len(_PNG),
        "media_type": "image/png",
        "spool_name": "ignored.png",
        "policy_version": "capture-v1",
    }


class PassiveClipboardIntake(ClipboardIntake):
    """Use real private spool sidecars without launching a Wayland watcher."""

    def __init__(
        self, config: WatchConfig, cancelled: Event, *, acknowledge_result: bool = True
    ) -> None:
        super().__init__(config)
        self.cancelled = cancelled
        self.acknowledge_result = acknowledge_result
        self.acknowledged: list[str] = []
        self.reconcile_calls = 0
        self._ticks = 0

    def reconcile(self) -> list[str]:
        self.reconcile_calls += 1
        return super().reconcile()

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

    def tick(self, now: float | None = None) -> None:
        self._ticks += 1
        if self._ticks > 1:
            self.cancelled.set()

    def acknowledge(self, receipt: IntakeReceipt) -> bool:
        self.acknowledged.append(receipt.receipt_token)
        return self.acknowledge_result and super().acknowledge(receipt)


class StopTimeoutReceiptIntake(PassiveClipboardIntake):
    def __init__(self, config: WatchConfig, cancelled: Event) -> None:
        super().__init__(config, cancelled)
        self.tick_entered = Event()
        self.release_tick = Event()

    def tick(self, now: float | None = None) -> None:
        self.tick_entered.set()
        self.release_tick.wait()

    def shutdown(self) -> None:
        self.release_tick.set()
        super().shutdown()


def _durable_runtime(
    tmp_path: Path, cancelled: Event, *, acknowledge_result: bool = True
) -> tuple[CaptureRuntime, list[PassiveClipboardIntake]]:
    created: list[PassiveClipboardIntake] = []
    root = tmp_path / "durable-xdg"
    paths = RuntimePaths(
        config_dir=(root / "config").resolve(),
        data_dir=(root / "data").resolve(),
        cache_dir=(root / "cache").resolve(),
        runtime_dir=(root / "runtime").resolve(),
    )

    def factory(config: WatchConfig) -> PassiveClipboardIntake:
        intake = PassiveClipboardIntake(config, cancelled, acknowledge_result=acknowledge_result)
        created.append(intake)
        return intake

    return (
        CaptureRuntime(
            AppConfig(lifecycle=LifecycleConfig(lock_timeout_seconds=0)),
            paths,
            migrations_dir=(Path(__file__).parents[2] / "migrations").resolve(),
            intake_factory=factory,
        ),
        created,
    )


def _durable_receipt(runtime: CaptureRuntime, token: str) -> None:
    spool_png(
        io.BytesIO(_PNG),
        spool=runtime.paths.cache_dir / "spool",
        max_bytes=100,
        receipt_token=token,
        policy_version="capture-v1",
    )


def test_watch_streams_verified_receipts_and_replays_exact_tokens_idempotently(
    tmp_path: Path,
) -> None:
    cancelled = Event()
    intake = FakeIntake(cancelled, [_payload("watch-receipt-1"), _payload("watch-receipt-1")])
    runtime = _runtime(tmp_path, intake)
    runtime.watch(cancelled=cancelled, poll_seconds=0.001)
    database = StorageDatabase(runtime.paths.data_dir, runtime.migrations_dir)
    with database.connect() as connection:
        assert connection.execute("SELECT count(*) FROM captures").fetchone()[0] == 1
        assert connection.execute("SELECT count(*) FROM blobs").fetchone()[0] == 1
    assert intake.armed and intake.reconciled and intake.shutdown_called and intake.opened == 1
    assert intake.acknowledged == ["watch-receipt-1"]


def test_watch_ignores_an_absent_helper_receipt_without_error_noise(tmp_path: Path) -> None:
    cancelled = Event()
    intake = FakeIntake(cancelled, [])
    runtime = _runtime(tmp_path, intake)
    runtime.watch(cancelled=cancelled, poll_seconds=0.001)
    database = StorageDatabase(runtime.paths.data_dir, runtime.migrations_dir)
    with database.connect() as connection:
        assert connection.execute("SELECT count(*) FROM captures").fetchone()[0] == 0
    assert intake.shutdown_called and intake.opened == 0


def test_startup_reconciles_partial_blob_storage_before_arming(tmp_path: Path) -> None:
    cancelled = Event()
    intake = FakeIntake(cancelled, [])
    runtime = _runtime(tmp_path, intake)
    blobs = ContentAddressedBlobStore(runtime.paths.data_dir)
    blobs.temporary.mkdir(parents=True)
    partial = blobs.temporary / ".leftover.partial"
    partial.write_bytes(b"partial")
    runtime.watch(cancelled=cancelled, poll_seconds=0.001)
    assert not partial.exists()
    assert intake.reconciled and intake.shutdown_called


def test_durable_receipt_is_acknowledged_once_after_repository_confirmation(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    cancelled = Event()
    runtime, created = _durable_runtime(tmp_path, cancelled)
    _durable_receipt(runtime, "confirmed")
    events: list[str] = []
    original_receive = CaptureRepository.receive
    original_acknowledge = PassiveClipboardIntake.acknowledge

    def tracked_receive(
        self: CaptureRepository,
        capture_id: str,
        receipt_token: str,
        chunks: Iterable[bytes],
        *,
        media_type: str,
        max_bytes: int,
    ) -> str:
        events.append("receive")
        return original_receive(
            self,
            capture_id,
            receipt_token,
            chunks,
            media_type=media_type,
            max_bytes=max_bytes,
        )

    monkeypatch.setattr(CaptureRepository, "receive", tracked_receive)

    def tracked_acknowledge(self: PassiveClipboardIntake, receipt: IntakeReceipt) -> bool:
        events.append("acknowledge")
        return original_acknowledge(self, receipt)

    monkeypatch.setattr(PassiveClipboardIntake, "acknowledge", tracked_acknowledge)
    runtime.watch(cancelled=cancelled, poll_seconds=0.001)

    assert events == ["receive", "acknowledge"]
    assert created[0].acknowledged == ["confirmed"]
    assert created[0].reconcile_calls == 1
    assert not (runtime.paths.cache_dir / "spool" / receipt_name("confirmed")).exists()


def test_storage_failure_leaves_durable_receipt_unacknowledged(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    cancelled = Event()
    runtime, created = _durable_runtime(tmp_path, cancelled)
    _durable_receipt(runtime, "storage-failure")

    def failed_receive(
        self: CaptureRepository,
        capture_id: str,
        receipt_token: str,
        chunks: Iterable[bytes],
        *,
        media_type: str,
        max_bytes: int,
    ) -> str:
        raise OSError("storage unavailable")

    monkeypatch.setattr(CaptureRepository, "receive", failed_receive)
    with pytest.raises(RuntimeFailure, match="capture watch failed"):
        runtime.watch(cancelled=cancelled, poll_seconds=0.001)

    assert created[0].acknowledged == []
    assert (runtime.paths.cache_dir / "spool" / receipt_name("storage-failure")).exists()


def test_cancellation_after_storage_confirmation_does_not_acknowledge(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    cancelled = Event()
    runtime, created = _durable_runtime(tmp_path, cancelled)
    _durable_receipt(runtime, "cancelled-after-storage")
    original_receive = CaptureRepository.receive

    def cancel_after_receive(
        self: CaptureRepository,
        capture_id: str,
        receipt_token: str,
        chunks: Iterable[bytes],
        *,
        media_type: str,
        max_bytes: int,
    ) -> str:
        result = original_receive(
            self,
            capture_id,
            receipt_token,
            chunks,
            media_type=media_type,
            max_bytes=max_bytes,
        )
        cancelled.set()
        return result

    monkeypatch.setattr(CaptureRepository, "receive", cancel_after_receive)
    runtime.watch(cancelled=cancelled, poll_seconds=0.001)

    assert created[0].acknowledged == []
    assert (runtime.paths.cache_dir / "spool" / receipt_name("cancelled-after-storage")).exists()


def test_acknowledgement_failure_replays_idempotently_and_releases_lock(tmp_path: Path) -> None:
    cancelled = Event()
    runtime, failed = _durable_runtime(tmp_path, cancelled, acknowledge_result=False)
    _durable_receipt(runtime, "replay-token")
    with pytest.raises(RuntimeFailure, match="acknowledgement failed"):
        runtime.watch(cancelled=cancelled, poll_seconds=0.001)
    assert failed[0].acknowledged == ["replay-token"]
    sidecar = runtime.paths.cache_dir / "spool" / receipt_name("replay-token")
    assert sidecar.exists()
    reusable = InstanceLock.acquire(runtime.paths.instance_lock_file, 0)
    reusable.release()

    replay_cancelled = Event()
    replay_runtime, replayed = _durable_runtime(tmp_path, replay_cancelled)
    replay_runtime.watch(cancelled=replay_cancelled, poll_seconds=0.001)
    database = StorageDatabase(replay_runtime.paths.data_dir, replay_runtime.migrations_dir)
    with database.connect() as connection:
        assert connection.execute("SELECT count(*) FROM captures").fetchone()[0] == 1
    assert replayed[0].acknowledged == ["replay-token"]
    assert not sidecar.exists()


def test_service_stop_timeout_preserves_unacknowledged_durable_sidecar(tmp_path: Path) -> None:
    cancelled = Event()
    created: list[StopTimeoutReceiptIntake] = []
    root = tmp_path / "stop-timeout-xdg"
    paths = RuntimePaths(
        config_dir=(root / "config").resolve(),
        data_dir=(root / "data").resolve(),
        cache_dir=(root / "cache").resolve(),
        runtime_dir=(root / "runtime").resolve(),
    )

    def factory(config: WatchConfig) -> StopTimeoutReceiptIntake:
        intake = StopTimeoutReceiptIntake(config, cancelled)
        created.append(intake)
        return intake

    runtime = CaptureRuntime(
        AppConfig(lifecycle=LifecycleConfig(lock_timeout_seconds=0, shutdown_timeout_seconds=1)),
        paths,
        migrations_dir=(Path(__file__).parents[2] / "migrations").resolve(),
        intake_factory=factory,
    )
    _durable_receipt(runtime, "timeout-receipt")
    resources = runtime._start()
    control = _CaptureControl(runtime, resources.repository)
    try:
        assert control.start_session() == "started"
        assert created[0].tick_entered.wait(0.2)
        assert control.stop_session() == "stopped"
        assert created[0].acknowledged == []
        assert (runtime.paths.cache_dir / "spool" / receipt_name("timeout-receipt")).exists()
    finally:
        control.shutdown()
        resources.close()


def test_cli_bare_watch_is_a_safe_refusal_and_modes_never_start_live_processes(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
    root = tmp_path / "fresh-xdg"
    for key, name in (
        ("HOME", "home"),
        ("XDG_CONFIG_HOME", "config"),
        ("XDG_DATA_HOME", "data"),
        ("XDG_CACHE_HOME", "cache"),
        ("XDG_RUNTIME_DIR", "runtime"),
    ):
        monkeypatch.setenv(key, str(root / name))

    def forbidden(*_arguments: object, **_kwargs: object) -> None:
        raise AssertionError("test started a process")

    monkeypatch.setattr(subprocess, "Popen", forbidden)
    assert cli.main(["watch"]) == cli.EXIT_UNIMPLEMENTED
    assert "requires gdc-runtime-integration" in capsys.readouterr().err
    assert cli.main(["watch", "--dry-run"]) == 0
    assert capsys.readouterr().out == "watch dry-run: capture runtime plan is valid\n"
    assert not root.exists()


def test_cli_arm_reaches_only_a_mocked_runtime(monkeypatch: pytest.MonkeyPatch) -> None:
    calls: list[str] = []

    class FakeRuntime:
        def __init__(self, _config: object, _paths: object) -> None:
            calls.append("created")

        def watch(self) -> None:
            calls.append("watch")

    monkeypatch.setattr(cli, "CaptureRuntime", FakeRuntime)
    assert cli.main(["watch", "--arm"]) == 0
    assert calls == ["created", "watch"]


def test_cli_serve_validates_before_constructing_resources_and_passes_exact_authority(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    calls: list[tuple[str, object]] = []

    class FakeRuntime:
        def __init__(self, config: AppConfig, _paths: object) -> None:
            calls.append(("created", config.server))

        def serve(self) -> None:
            calls.append(("serve", None))

    monkeypatch.setattr(cli, "CaptureRuntime", FakeRuntime)
    assert cli.main(["serve", "--host", "127.0.0.1", "--port", "9877"]) == 0
    assert calls == [("created", ServerConfig("127.0.0.1", 9877)), ("serve", None)]

    calls.clear()
    assert cli.main(["serve", "--host", "0.0.0.0"]) == cli.EXIT_CONFIG_ERROR
    assert calls == []


def test_cli_serve_converts_uvicorn_startup_exit_without_leaking_its_token(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
    root = tmp_path / "serve-xdg"
    for key, name in (
        ("HOME", "home"),
        ("XDG_CONFIG_HOME", "config"),
        ("XDG_DATA_HOME", "data"),
        ("XDG_CACHE_HOME", "cache"),
        ("XDG_RUNTIME_DIR", "runtime"),
    ):
        monkeypatch.setenv(key, str(root / name))
    token = "Z" * 43
    monkeypatch.setattr(runtime_module, "new_bootstrap_token", lambda: token)

    def failed_server(*_: object, **__: object) -> None:
        raise SystemExit(2)

    monkeypatch.setattr(uvicorn, "run", failed_server)
    assert cli.main(["serve"]) == cli.EXIT_RUNTIME_ERROR
    output = capsys.readouterr()
    assert output.err == "capture runtime failed\n"
    assert token not in output.err
    assert "SystemExit" not in output.err
    reusable = InstanceLock.acquire(root / "runtime" / "shd-capture" / "instance.lock", 0)
    reusable.release()