from __future__ import annotations

import hashlib
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from types import SimpleNamespace
from typing import Any

import pytest

from k1link.sessions.camera_frame import RecordedCameraFrame, RecordedCameraFrameService
from k1link.sessions.media import (
    RecordedMediaEpoch,
    RecordedMediaManifest,
    RecordedMediaSegment,
)
from k1link.sessions.models import RecordedMediaArtifact, SessionIntegrityError

_JPEG = b"\xff\xd8\xff\xc0\x00\x07\x08\x00\x01\x00\x01\xff\xd9"


class _Store:
    def __init__(self, *, block_prepare: bool = False) -> None:
        self.prepare_calls = 0
        self.list_calls = 0
        self.prepare_started = threading.Event()
        self.release_prepare = threading.Event()
        if not block_prepare:
            self.release_prepare.set()
        self.artifact = RecordedMediaArtifact(
            session_id="recorded-session",
            public_source_id="camera",
            artifact_id="camera-artifact",
            source_path=Path("/sealed/sensor.camera.right"),
            byte_length=123,
        )

    def prepare_replay(self, session_id: str, **_: Any) -> SimpleNamespace:
        self.prepare_calls += 1
        self.prepare_started.set()
        assert self.release_prepare.wait(timeout=2.0)
        return SimpleNamespace(session_id=session_id)

    def list_recorded_media(self, session_id: str) -> tuple[RecordedMediaArtifact, ...]:
        self.list_calls += 1
        return (self.artifact,)


class _Inspector:
    def __init__(self, manifest: RecordedMediaManifest) -> None:
        self.manifest = manifest
        self.inspect_calls = 0

    def inspect(self, artifact: RecordedMediaArtifact, replay: object) -> RecordedMediaManifest:
        assert artifact.artifact_id == "camera-artifact"
        assert replay is not None
        self.inspect_calls += 1
        return self.manifest


def _manifest(tmp_path: Path) -> RecordedMediaManifest:
    epoch_path = tmp_path / "epoch-1"
    segments = tuple(
        RecordedMediaSegment(
            sequence=sequence,
            path=epoch_path / "segments" / f"{sequence:08d}.m4s",
            byte_length=10,
            sha256=hashlib.sha256(str(sequence).encode()).hexdigest(),
            random_access=True,
            end_time_seconds=float(sequence),
        )
        for sequence in range(1, 4)
    )
    return RecordedMediaManifest(
        session_id="recorded-session",
        public_source_id="camera",
        artifact_id="camera-artifact",
        synchronization="recorded",
        generation_sha256="a" * 64,
        timeline_start_seconds=0.0,
        timeline_end_seconds=3.0,
        byte_length=123,
        epochs=(
            RecordedMediaEpoch(
                ordinal=1,
                path=epoch_path,
                init_path=epoch_path / "init.mp4",
                init_byte_length=10,
                init_sha256="b" * 64,
                media_type="video/mp4",
                timeline_start_seconds=0.0,
                timeline_end_seconds=3.0,
                segments=segments,
            ),
        ),
    )


def _service(
    tmp_path: Path,
    store: _Store,
    inspector: _Inspector,
    *,
    max_decode_lanes: int = 32,
    max_source_manifests: int = 32,
) -> RecordedCameraFrameService:
    return RecordedCameraFrameService(
        store,  # type: ignore[arg-type]
        inspector,  # type: ignore[arg-type]
        ffmpeg_path=Path(sys.executable),
        cache_root=tmp_path / "cache",
        max_decode_lanes=max_decode_lanes,
        max_source_manifests=max_source_manifests,
    )


def _wait_for_latest(service: RecordedCameraFrameService, ticket: int) -> None:
    deadline = time.monotonic() + 2.0
    source_key = ("recorded-session", "sensor.camera.right")
    while time.monotonic() < deadline:
        with service._coordination:  # noqa: SLF001
            lane = service._lanes.get(source_key)  # noqa: SLF001
            if lane is not None and lane.latest_ticket == ticket:
                return
        time.sleep(0.005)
    raise AssertionError(f"camera request ticket {ticket} was not registered")


def _frame(epoch: RecordedMediaEpoch, sequence: int) -> RecordedCameraFrame:
    return RecordedCameraFrame(
        payload=_JPEG,
        media_type="image/jpeg",
        width=1,
        height=1,
        sha256=hashlib.sha256(_JPEG).hexdigest(),
        source_fragment_sha256=epoch.segments[sequence - 1].sha256,
    )


def test_camera_frame_burst_scans_manifest_once_and_decodes_only_latest(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    store = _Store(block_prepare=True)
    inspector = _Inspector(_manifest(tmp_path))
    service = _service(tmp_path, store, inspector)
    decoded: list[int] = []

    def decode(
        manifest: RecordedMediaManifest,
        epoch: RecordedMediaEpoch,
        sequence: int,
    ) -> RecordedCameraFrame:
        assert manifest is inspector.manifest
        decoded.append(sequence)
        return _frame(epoch, sequence)

    monkeypatch.setattr(service, "_decode", decode)
    with ThreadPoolExecutor(max_workers=3) as pool:
        first = pool.submit(service.extract, "recorded-session", 0)
        assert store.prepare_started.wait(timeout=2.0)
        second = pool.submit(service.extract, "recorded-session", 1)
        _wait_for_latest(service, 2)
        third = pool.submit(service.extract, "recorded-session", 2)
        _wait_for_latest(service, 3)
        store.release_prepare.set()

        with pytest.raises(SessionIntegrityError, match="superseded"):
            first.result(timeout=2.0)
        with pytest.raises(SessionIntegrityError, match="superseded"):
            second.result(timeout=2.0)
        assert third.result(timeout=2.0).source_fragment_sha256 == (
            inspector.manifest.epochs[0].segments[2].sha256
        )

    assert store.prepare_calls == 1
    assert store.list_calls == 1
    assert inspector.inspect_calls == 1
    assert decoded == [3]


def test_camera_frame_lane_keeps_only_latest_waiter_behind_active_decode(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    store = _Store()
    inspector = _Inspector(_manifest(tmp_path))
    service = _service(tmp_path, store, inspector)
    decode_started = threading.Event()
    release_decode = threading.Event()
    decoded: list[int] = []

    def decode(
        manifest: RecordedMediaManifest,
        epoch: RecordedMediaEpoch,
        sequence: int,
    ) -> RecordedCameraFrame:
        assert manifest is inspector.manifest
        decoded.append(sequence)
        if sequence == 1:
            decode_started.set()
            assert release_decode.wait(timeout=2.0)
        return _frame(epoch, sequence)

    monkeypatch.setattr(service, "_decode", decode)
    with ThreadPoolExecutor(max_workers=3) as pool:
        first = pool.submit(service.extract, "recorded-session", 0)
        assert decode_started.wait(timeout=2.0)
        second = pool.submit(service.extract, "recorded-session", 1)
        _wait_for_latest(service, 2)
        third = pool.submit(service.extract, "recorded-session", 2)
        _wait_for_latest(service, 3)
        release_decode.set()

        with pytest.raises(SessionIntegrityError, match="superseded"):
            first.result(timeout=2.0)
        with pytest.raises(SessionIntegrityError, match="superseded"):
            second.result(timeout=2.0)
        assert third.result(timeout=2.0).width == 1

    assert decoded == [1, 3]
    assert store.prepare_calls == 1
    assert inspector.inspect_calls == 1


def test_camera_frame_memory_caches_are_lru_bounded(tmp_path: Path) -> None:
    store = _Store()
    inspector = _Inspector(_manifest(tmp_path))
    service = _service(
        tmp_path,
        store,
        inspector,
        max_decode_lanes=2,
        max_source_manifests=2,
    )

    for ordinal in range(4):
        source_key = (f"session-{ordinal}", "sensor.camera.right")
        lane, _ = service._acquire_lane(source_key)  # noqa: SLF001
        service._release_lane(source_key, lane)  # noqa: SLF001
    assert tuple(service._lanes) == (  # noqa: SLF001
        ("session-2", "sensor.camera.right"),
        ("session-3", "sensor.camera.right"),
    )

    for ordinal in range(3):
        source_key = (f"session-{ordinal}", "sensor.camera.right")
        service._source_manifest(  # noqa: SLF001
            source_key,
            session_id=source_key[0],
            expected_source_name=source_key[1],
        )
    assert tuple(service._source_manifests) == (  # noqa: SLF001
        ("session-1", "sensor.camera.right"),
        ("session-2", "sensor.camera.right"),
    )

    service._source_manifest(  # noqa: SLF001
        ("session-1", "sensor.camera.right"),
        session_id="session-1",
        expected_source_name="sensor.camera.right",
    )
    service._source_manifest(  # noqa: SLF001
        ("session-3", "sensor.camera.right"),
        session_id="session-3",
        expected_source_name="sensor.camera.right",
    )
    assert tuple(service._source_manifests) == (  # noqa: SLF001
        ("session-1", "sensor.camera.right"),
        ("session-3", "sensor.camera.right"),
    )
    assert store.prepare_calls == 4
    assert inspector.inspect_calls == 4


def test_camera_playback_source_reuses_manifest_and_exposes_exact_segment_clock(
    tmp_path: Path,
) -> None:
    store = _Store()
    manifest = _manifest(tmp_path)
    manifest = RecordedMediaManifest(
        session_id=manifest.session_id,
        public_source_id=manifest.public_source_id,
        artifact_id=manifest.artifact_id,
        synchronization="host-arrival-best-effort",
        generation_sha256=manifest.generation_sha256,
        timeline_start_seconds=manifest.timeline_start_seconds,
        timeline_end_seconds=manifest.timeline_end_seconds,
        byte_length=manifest.byte_length,
        epochs=(
            RecordedMediaEpoch(
                ordinal=1,
                path=manifest.epochs[0].path,
                init_path=manifest.epochs[0].init_path,
                init_byte_length=manifest.epochs[0].init_byte_length,
                init_sha256=manifest.epochs[0].init_sha256,
                media_type='video/mp4; codecs="avc1.640028"',
                timeline_start_seconds=0.0,
                timeline_end_seconds=3.0,
                segments=manifest.epochs[0].segments,
            ),
        ),
    )
    inspector = _Inspector(manifest)
    service = _service(tmp_path, store, inspector)

    first = service.playback_source("recorded-session")
    second = service.playback_source("recorded-session")

    assert first == second
    assert first.segment_count == 3
    assert first.segment_sha256s == tuple(
        segment.sha256 for segment in manifest.epochs[0].segments
    )
    assert first.segment_start_times_ns == (0, 1_000_000_000, 2_000_000_000)
    assert store.prepare_calls == 1
    assert inspector.inspect_calls == 1


def test_camera_frame_rejects_unbounded_memory_cache_configuration(tmp_path: Path) -> None:
    store = _Store()
    inspector = _Inspector(_manifest(tmp_path))
    with pytest.raises(SessionIntegrityError, match="cache bounds"):
        _service(tmp_path, store, inspector, max_decode_lanes=0)