from __future__ import annotations

from pathlib import Path
from typing import Any

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from k1link.simulation.gaussian_pipeline_gateway import (
    GaussianPipelineUnavailableError,
    GaussianSourceBundleUpload,
    GaussianSourceMemberUpload,
)
from k1link.simulation.projects import (
    SimulationProjectConflictError,
    SimulationProjectService,
    SimulationProjectStore,
    _world_manifest,
)
from k1link.web.simulation_projects_api import build_simulation_projects_router


def _folder_files() -> list[dict[str, object]]:
    return [
        {"logical_path": "export/scene.lcc", "byte_length": 23},
        {"logical_path": "export/index.bin", "byte_length": 5},
        {"logical_path": "export/data.bin", "byte_length": 4},
    ]


def _upload_all(store: SimulationProjectStore, project: dict[str, Any]) -> None:
    payloads = {
        "export/scene.lcc": b'{"fileType":"Portable"}',
        "export/index.bin": b"index",
        "export/data.bin": b"data",
    }
    for source_file in project["source"]["files"]:
        store.append_upload(
            project["project_id"],
            source_file["file_id"],
            offset=0,
            payload=payloads[source_file["logical_path"]],
        )


def test_store_persists_resumable_folder_upload_and_state_transitions(tmp_path: Path) -> None:
    store = SimulationProjectStore(tmp_path)
    project = store.create(
        name="  Smolensky   Boulevard  ",
        scene_type="outdoor",
        source_kind="folder",
        files=_folder_files(),
    )
    assert project["name"] == "Smolensky Boulevard"
    first = project["source"]["files"][0]
    store.append_upload(
        project["project_id"],
        first["file_id"],
        offset=0,
        payload=b'{"fileType":',
    )
    with pytest.raises(SimulationProjectConflictError, match="offset"):
        store.append_upload(
            project["project_id"],
            first["file_id"],
            offset=0,
            payload=b'"Portable"}',
        )
    store.append_upload(
        project["project_id"],
        first["file_id"],
        offset=len(b'{"fileType":'),
        payload=b'"Portable"}',
    )
    for source_file, payload in zip(
        project["source"]["files"][1:],
        (b"index", b"data"),
        strict=True,
    ):
        store.append_upload(
            project["project_id"],
            source_file["file_id"],
            offset=0,
            payload=payload,
        )
    queued = store.begin_build(project["project_id"])
    assert queued["status"] == "queued"
    with pytest.raises(SimulationProjectConflictError, match="active"):
        store.delete(project["project_id"])


class _ReadyProvider:
    def __init__(self) -> None:
        self.deleted: list[str] = []
        self.upload_calls = 0
        self.submit_calls = 0
        self.submitted_document: dict[str, object] | None = None

    def capabilities(self) -> dict[str, object]:
        return {"outputs": ["preview.sog", "streamed-sog"]}

    def upload_source_bundle(
        self,
        _root: Path,
        *,
        entrypoint: str,
        source_format: str,
    ) -> GaussianSourceBundleUpload:
        self.upload_calls += 1
        members = (
            GaussianSourceMemberUpload("upload-1", entrypoint, "a" * 64, 23),
            GaussianSourceMemberUpload("upload-2", "export/data.bin", "b" * 64, 4),
            GaussianSourceMemberUpload("upload-3", "export/index.bin", "c" * 64, 5),
        )
        return GaussianSourceBundleUpload(
            format=source_format,
            entrypoint=entrypoint,
            bundle_sha256="d" * 64,
            total_byte_length=32,
            members=members,
        )

    def submit_build(self, document: dict[str, object]) -> dict[str, object]:
        self.submit_calls += 1
        self.submitted_document = document
        return {
            "schema_version": "gaussian-pipeline.job/v1",
            "job_id": "gsp-20260826000000-deadbeef",
            "state": "queued",
        }

    def get_job(self, _job_id: str) -> dict[str, object]:
        return {
            "schema_version": "gaussian-pipeline.job/v1",
            "job_id": "gsp-20260826000000-deadbeef",
            "state": "ready",
            "progress": {"completed_steps": 4, "total_steps": 4},
        }

    def get_result(self, _job_id: str) -> dict[str, object]:
        return {
            "schema_version": "gaussian-pipeline.build-result/v1",
            "job_id": "gsp-20260826000000-deadbeef",
            "runtime": {
                "source_revision": "e" * 40,
                "image_digest": f"sha256:{'f' * 64}",
            },
            "artifacts": [
                {
                    "role": "preview",
                    "logical_path": "preview.sog",
                    "media_type": "application/octet-stream",
                    "sha256": "1" * 64,
                    "byte_length": 7,
                },
                {
                    "role": "stream-manifest",
                    "logical_path": "streamed/lod-meta.json",
                    "media_type": "application/json",
                    "sha256": "2" * 64,
                    "byte_length": 2,
                },
            ],
        }

    def download_artifact(
        self,
        _job_id: str,
        descriptor: dict[str, object],
        destination: Path,
    ) -> Path:
        destination.parent.mkdir(parents=True, exist_ok=True)
        destination.write_bytes(b"preview" if descriptor["role"] == "preview" else b"{}")
        return destination

    def delete_job(self, job_id: str) -> None:
        self.deleted.append(job_id)

    def close(self) -> None:
        return None


def test_service_materializes_world_manifest_and_deletes_both_copies(tmp_path: Path) -> None:
    store = SimulationProjectStore(tmp_path)
    project = store.create(
        name="MOJJOKER",
        scene_type="interior",
        source_kind="folder",
        files=_folder_files(),
    )
    _upload_all(store, project)
    store.begin_build(project["project_id"])
    provider = _ReadyProvider()
    service = SimulationProjectService(store, provider_factory=lambda: provider)  # type: ignore[arg-type]
    service.process(project["project_id"])

    ready = store.get(project["project_id"])
    assert ready["status"] == "ready"
    assert ready["world_manifest"]["visual"]["preview_sog_url"].endswith("/preview.sog")
    assert ready["world_manifest"]["collision"]["available"] is False
    assert ready["world_manifest"]["transforms"]["world_from_visual"] == [
        1, 0, 0, 0, 0, -1, 0, 0, 0, 0, -1, 0, 0, 0, 0, 1,
    ]
    assert ready["world_manifest"]["transforms"]["world_from_collision"] == [
        -1, 0, 0, 0, 0, 1, 0, 0, 0, 0, -1, 0, 0, 0, 0, 1,
    ]
    assert provider.upload_calls == 1
    assert provider.submit_calls == 1
    assert provider.submitted_document is not None
    assert provider.submitted_document["outputs"] == {
        "preview_sog": True,
        "streamed_sog": True,
        "collision": True,
    }
    assert provider.submitted_document["collision_profile"] == {
        "scene_type": "interior",
        "seed_position": [0, 1, 0],
        "capsule_height": 1.6,
        "capsule_radius": 0.2,
        "voxel_size": 0.1,
        "mesh_shape": "smooth",
    }
    service.delete(project["project_id"])
    assert provider.deleted == ["gsp-20260826000000-deadbeef"]


def test_world_manifest_prefers_seed_connected_rover_terrain() -> None:
    manifest = _world_manifest(
        "sim-" + "a" * 32,
        [
            {
                "role": "collision-mesh",
                "logical_path": "collision/scene.collision.glb",
            },
            {
                "role": "rover-terrain-mesh",
                "logical_path": "rover/terrain.collision.glb",
            },
        ],
    )

    assert manifest["collision"]["available"] is True
    assert manifest["collision"]["mesh_url"].endswith(
        "/artifacts/rover/terrain.collision.glb"
    )


def test_service_resumes_a_persisted_provider_job_without_reupload(tmp_path: Path) -> None:
    store = SimulationProjectStore(tmp_path)
    project = store.create(
        name="Restart-safe scene",
        scene_type="outdoor",
        source_kind="folder",
        files=_folder_files(),
    )
    _upload_all(store, project)
    store.begin_build(project["project_id"])
    store.update_processing(
        project["project_id"],
        status="processing",
        provider_job_id="gsp-20260826000000-deadbeef",
        provider_state="building_streamed_sog",
        bundle_sha256="d" * 64,
    )
    provider = _ReadyProvider()
    service = SimulationProjectService(store, provider_factory=lambda: provider)  # type: ignore[arg-type]

    service.process(project["project_id"])

    assert store.get(project["project_id"])["status"] == "ready"
    assert provider.upload_calls == 0
    assert provider.submit_calls == 0


def test_failed_project_retries_from_retained_source_and_releases_old_job(tmp_path: Path) -> None:
    store = SimulationProjectStore(tmp_path)
    project = store.create(
        name="Retry scene",
        scene_type="outdoor",
        source_kind="folder",
        files=_folder_files(),
    )
    _upload_all(store, project)
    store.begin_build(project["project_id"])
    store.update_processing(
        project["project_id"],
        status="processing",
        provider_job_id="gsp-20260826000000-deadbeef",
        provider_state="failed",
        progress={"completed_steps": 1, "total_steps": 4},
        bundle_sha256="d" * 64,
    )
    store.fail(project["project_id"], "provider failure")
    provider = _ReadyProvider()
    service = SimulationProjectService(store, provider_factory=lambda: provider)  # type: ignore[arg-type]

    queued = service.begin_build(project["project_id"])

    assert provider.deleted == ["gsp-20260826000000-deadbeef"]
    assert queued["status"] == "queued"
    assert queued["error"] is None
    assert queued["provider"]["job_id"] is None
    assert queued["source"]["uploaded_byte_length"] == queued["source"]["total_byte_length"]


def test_failed_local_project_reattaches_to_live_provider_job_without_rebuild(
    tmp_path: Path,
) -> None:
    store = SimulationProjectStore(tmp_path)
    project = store.create(
        name="Reconnect scene",
        scene_type="outdoor",
        source_kind="folder",
        files=_folder_files(),
    )
    _upload_all(store, project)
    store.begin_build(project["project_id"])
    store.update_processing(
        project["project_id"],
        status="processing",
        provider_job_id="gsp-20260826000000-deadbeef",
        provider_state="building_streamed_sog",
        progress={"completed_steps": 3, "total_steps": 5},
        bundle_sha256="d" * 64,
    )
    store.fail(project["project_id"], "temporary provider transport failure")
    provider = _ReadyProvider()
    service = SimulationProjectService(
        store,
        provider_factory=lambda: provider,
    )  # type: ignore[arg-type]

    resumed = service.begin_build(project["project_id"])

    assert resumed["status"] == "processing"
    assert resumed["error"] is None
    assert resumed["provider"]["job_id"] == "gsp-20260826000000-deadbeef"
    assert resumed["provider"]["state"] == "ready"
    assert provider.deleted == []
    assert provider.upload_calls == 0
    assert provider.submit_calls == 0


def test_live_provider_reattach_retries_temporary_unavailability(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    store = SimulationProjectStore(tmp_path)
    project = store.create(
        name="Flaky tunnel scene",
        scene_type="outdoor",
        source_kind="folder",
        files=_folder_files(),
    )
    _upload_all(store, project)
    store.begin_build(project["project_id"])
    store.update_processing(
        project["project_id"],
        status="processing",
        provider_job_id="gsp-20260826000000-deadbeef",
        provider_state="building_collision",
        bundle_sha256="d" * 64,
    )
    store.fail(project["project_id"], "temporary provider transport failure")

    class _FlakyProvider(_ReadyProvider):
        def __init__(self) -> None:
            super().__init__()
            self.get_job_calls = 0

        def get_job(self, job_id: str) -> dict[str, object]:
            self.get_job_calls += 1
            if self.get_job_calls < 3:
                raise GaussianPipelineUnavailableError("temporary tunnel failure")
            return super().get_job(job_id)

    provider = _FlakyProvider()
    monkeypatch.setattr("k1link.simulation.projects.time.sleep", lambda _delay: None)
    service = SimulationProjectService(
        store,
        provider_factory=lambda: provider,
    )  # type: ignore[arg-type]

    resumed = service.begin_build(project["project_id"])

    assert resumed["status"] == "processing"
    assert provider.get_job_calls == 3
    assert provider.deleted == []


def test_ready_project_rebuilds_from_retained_source_and_preserves_viewer_settings(
    tmp_path: Path,
) -> None:
    store = SimulationProjectStore(tmp_path)
    project = store.create(
        name="Rebuild scene",
        scene_type="outdoor",
        source_kind="folder",
        files=_folder_files(),
    )
    _upload_all(store, project)
    store.begin_build(project["project_id"])
    provider = _ReadyProvider()
    service = SimulationProjectService(
        store,
        provider_factory=lambda: provider,
    )  # type: ignore[arg-type]
    service.process(project["project_id"])
    settings = {
        "schema_version": "missioncore.simulation-viewer-settings/v1",
        "quality": "medium",
        "visual": {"inverted": True, "axis": "z"},
        "collision": {"inverted": True, "axis": "y"},
        "camera": {"invert_horizontal": False, "invert_vertical": True},
    }
    store.update_viewer_settings(project["project_id"], settings)

    queued = service.begin_build(project["project_id"])

    assert provider.deleted == ["gsp-20260826000000-deadbeef"]
    assert queued["status"] == "queued"
    assert queued["viewer_settings"] == settings
    assert queued["source"]["uploaded_byte_length"] == queued["source"]["total_byte_length"]


def test_api_exposes_same_origin_resumable_upload_contract(tmp_path: Path) -> None:
    store = SimulationProjectStore(tmp_path)
    service = SimulationProjectService(store, provider_factory=lambda: None)
    app = FastAPI()
    app.include_router(build_simulation_projects_router(store=store, service=service))
    client = TestClient(app)
    created = client.post("/api/v1/simulation-worlds/projects", json={
        "schema_version": "missioncore.simulation-project-create/v1",
        "name": "Archive scene",
        "scene_type": "outdoor",
        "source_kind": "archive",
        "files": [{"logical_path": "scan.rar", "byte_length": 6}],
    })
    assert created.status_code == 201
    project = created.json()
    assert project["viewer_settings"] == {
        "schema_version": "missioncore.simulation-viewer-settings/v1",
        "quality": "high",
        "visual": {"inverted": True, "axis": "x"},
        "collision": {"inverted": True, "axis": "y"},
        "camera": {"invert_horizontal": True, "invert_vertical": False},
    }
    source_file = project["source"]["files"][0]
    upload_url = (
        f"/api/v1/simulation-worlds/projects/{project['project_id']}"
        f"/source/{source_file['file_id']}"
    )
    assert client.head(upload_url).headers["upload-offset"] == "0"
    first = client.patch(upload_url, headers={"Upload-Offset": "0"}, content=b"abc")
    assert first.status_code == 204
    assert first.headers["upload-offset"] == "3"
    mismatch = client.patch(upload_url, headers={"Upload-Offset": "0"}, content=b"def")
    assert mismatch.status_code == 409
    second = client.patch(upload_url, headers={"Upload-Offset": "3"}, content=b"def")
    assert second.status_code == 204
    catalog = client.get("/api/v1/simulation-worlds/projects").json()
    assert catalog["projects"][0]["project_id"] == project["project_id"]

    settings = {
        "schema_version": "missioncore.simulation-viewer-settings/v1",
        "quality": "medium",
        "visual": {"inverted": True, "axis": "z"},
        "collision": {"inverted": False, "axis": "x"},
        "camera": {"invert_horizontal": False, "invert_vertical": True},
    }
    saved = client.put(
        f"/api/v1/simulation-worlds/projects/{project['project_id']}/viewer-settings",
        json=settings,
    )
    assert saved.status_code == 200
    assert saved.json()["viewer_settings"] == settings
    persisted = client.get(
        f"/api/v1/simulation-worlds/projects/{project['project_id']}"
    ).json()
    assert persisted["viewer_settings"] == settings