from __future__ import annotations

import asyncio
import json
from copy import deepcopy
from typing import Any

import pytest
from fastapi.testclient import TestClient
from jsonschema import Draft202012Validator, FormatChecker
from sqlalchemy.ext.asyncio import create_async_engine

from hub_core.contracts import workload_projection_contract_root
from hub_core.runtime.app import create_app
from hub_core.runtime.config import RuntimeSettings
from hub_core.runtime.postgres_store import PostgresPortStore
from hub_core.runtime.store import InMemoryPortStore
from hub_core.runtime.tables import (
    runtime_metadata,
    runtime_workload_projection_records,
    runtime_workload_projection_state,
)
from hub_core.runtime.workload_projection import (
    WorkloadCursorMismatch,
    WorkloadProjectionRejected,
    WorkloadProjectionService,
)


def fixture() -> dict[str, Any]:
    resource = workload_projection_contract_root().joinpath(
        "fixtures", "repo-manager-nine-workloads.json"
    )
    return json.loads(resource.read_text(encoding="utf-8"))


class PageClient:
    def __init__(self, pages: list[dict[str, Any]]) -> None:
        self.pages = pages
        self.calls: list[str | None] = []

    async def fetch_workload_page(self, cursor: str | None) -> dict[str, Any]:
        self.calls.append(cursor)
        for page in self.pages:
            if page["snapshot"]["page_cursor"] == cursor:
                return deepcopy(page)
        raise RuntimeError(f"unexpected cursor {cursor!r}")


class FailingClient:
    async def fetch_workload_page(self, cursor: str | None) -> dict[str, Any]:
        raise RuntimeError("repo-manager workload index unavailable")


def next_snapshot(
    page: dict[str, Any], *, marker: str, generated_at: str
) -> dict[str, Any]:
    changed = deepcopy(page)
    changed["snapshot"].update(
        {"snapshot_id": marker * 64, "generated_at": generated_at}
    )
    return changed


def test_ingests_nine_records_and_resolves_only_exact_reference() -> None:
    async def run() -> None:
        store = InMemoryPortStore()
        service = WorkloadProjectionService(client=PageClient([fixture()]), store=store)

        result = await service.refresh()
        resolved = await service.resolve(
            rapp_id="rapp-qonto", name="qonto", deployable="rapp-qonto"
        )
        wrong_name = await service.resolve(
            rapp_id="rapp-qonto", name="qonto-assistant"
        )
        wrong_deployable = await service.resolve(
            rapp_id="rapp-openbao", name="openbao", deployable="rapp-openbao"
        )

        assert result.workload_count == 9
        assert resolved is not None and resolved["status"] == "resolved"
        assert wrong_name is not None and wrong_name["reason"] == "not_found"
        assert wrong_deployable is not None
        assert wrong_deployable["reason"] == "deployable_not_declared"
        assert await service.readiness_checks() == {"workload_projection": "ok"}

    asyncio.run(run())


def test_deletion_and_full_rebuild_restore_equivalent_content() -> None:
    async def run() -> None:
        source = fixture()
        store = InMemoryPortStore()
        initial = WorkloadProjectionService(client=PageClient([source]), store=store)
        original = await initial.refresh()

        reduced = next_snapshot(
            source, marker="a", generated_at="2026-08-22T12:05:00Z"
        )
        reduced["workloads"] = reduced["workloads"][:-1]
        reduced["snapshot"]["total_workload_count"] = 8
        await WorkloadProjectionService(
            client=PageClient([reduced]), store=store
        ).refresh()
        after_delete = await store.get_workload_projection()

        restored = next_snapshot(
            source, marker="b", generated_at="2026-08-22T12:10:00Z"
        )
        rebuilt = await WorkloadProjectionService(
            client=PageClient([restored]), store=store
        ).refresh()

        assert after_delete is not None and len(after_delete.workloads) == 8
        assert rebuilt.workload_count == 9
        assert rebuilt.content_hash == original.content_hash

    asyncio.run(run())


def test_invalid_non_utc_input_preserves_last_generation_as_stale() -> None:
    async def run() -> None:
        store = InMemoryPortStore()
        await WorkloadProjectionService(
            client=PageClient([fixture()]), store=store
        ).refresh()
        before = await store.get_workload_projection()
        invalid = next_snapshot(
            fixture(), marker="c", generated_at="2026-08-22T14:00:00+02:00"
        )

        with pytest.raises(WorkloadProjectionRejected, match="invalid workload page"):
            await WorkloadProjectionService(
                client=PageClient([invalid]), store=store
            ).refresh()
        after = await store.get_workload_projection()

        assert before is not None and after is not None
        assert after.content_hash == before.content_hash
        assert after.workloads == before.workloads
        assert after.projection_status == "stale"

    asyncio.run(run())


def test_query_order_cursor_and_generation_binding() -> None:
    async def run() -> None:
        store = InMemoryPortStore()
        service = WorkloadProjectionService(client=PageClient([fixture()]), store=store)
        await service.refresh()
        first = await service.query(limit=4)

        assert first is not None and first["next_cursor"]
        assert [item["rapp_id"] for item in first["workloads"]] == sorted(
            item["rapp_id"] for item in first["workloads"]
        )
        second = await service.query(cursor=first["next_cursor"], limit=4)
        assert second is not None and len(second["workloads"]) == 4

        changed = next_snapshot(
            fixture(), marker="d", generated_at="2026-08-22T12:15:00Z"
        )
        await WorkloadProjectionService(
            client=PageClient([changed]), store=store
        ).refresh()
        with pytest.raises(WorkloadCursorMismatch, match="cursor_snapshot_mismatch"):
            await service.query(cursor=first["next_cursor"], limit=4)

    asyncio.run(run())


def test_outage_without_generation_fails_readiness_closed() -> None:
    async def run() -> None:
        service = WorkloadProjectionService(
            client=FailingClient(), store=InMemoryPortStore()
        )
        with pytest.raises(WorkloadProjectionRejected, match="unavailable"):
            await service.refresh()
        assert await service.readiness_checks() == {"workload_projection": "unavailable"}

    asyncio.run(run())


def test_http_and_mcp_backing_routes_are_read_only_and_schema_valid() -> None:
    settings = RuntimeSettings(environment="test", backend="memory", allow_ephemeral=True)
    app = create_app(
        settings=settings,
        port_store=InMemoryPortStore(),
        workload_projection_client=PageClient([fixture()]),
    )
    with TestClient(app) as runtime:
        response = runtime.get(
            "/ports/projections/workloads", params={"deployable": "openbao"}
        )
        resolved = runtime.get(
            "/ports/projections/workloads/resolve",
            params={"rapp_id": "rapp-openbao", "name": "openbao"},
        )
        unknown = runtime.get(
            "/ports/projections/workloads/resolve",
            params={"rapp_id": "rapp-openbao", "name": "railiance-platform"},
        )
        ready = runtime.get("/readyz")
        openapi = runtime.get("/openapi.json").json()

    schema_root = workload_projection_contract_root().joinpath("schemas")
    projection_schema = json.loads(
        schema_root.joinpath("workload-projection.schema.json").read_text(encoding="utf-8")
    )
    resolution_schema = json.loads(
        schema_root.joinpath("workload-resolution.schema.json").read_text(encoding="utf-8")
    )
    Draft202012Validator(projection_schema, format_checker=FormatChecker()).validate(
        response.json()
    )
    Draft202012Validator(resolution_schema, format_checker=FormatChecker()).validate(
        resolved.json()
    )
    Draft202012Validator(resolution_schema, format_checker=FormatChecker()).validate(
        unknown.json()
    )
    assert response.status_code == resolved.status_code == unknown.status_code == 200
    assert [record["rapp_id"] for record in response.json()["workloads"]] == [
        "rapp-openbao"
    ]
    assert resolved.json()["status"] == "resolved"
    assert unknown.json()["status"] == "unknown"
    assert ready.json()["checks"]["workload_projection"] == "ok"
    paths = {
        path: item
        for path, item in openapi["paths"].items()
        if path.startswith("/ports/projections/workloads")
    }
    assert paths and all(set(item) == {"get"} for item in paths.values())


def test_postgres_workload_projection_survives_reopen(tmp_path) -> None:
    database_url = f"sqlite+aiosqlite:///{tmp_path / 'workloads.db'}"

    async def run() -> None:
        engine = create_async_engine(database_url)
        async with engine.begin() as connection:
            await connection.run_sync(runtime_metadata.create_all)
        await engine.dispose()
        first_store = PostgresPortStore.from_url(database_url)
        result = await WorkloadProjectionService(
            client=PageClient([fixture()]), store=first_store
        ).refresh()
        await first_store.aclose()
        second_store = PostgresPortStore.from_url(database_url)
        projection = await second_store.get_workload_projection()
        await second_store.aclose()

        assert projection is not None
        assert projection.content_hash == result.content_hash
        assert len(projection.workloads) == 9
        assert all(item["observed_at"].endswith("Z") for item in projection.workloads)

    asyncio.run(run())


def test_projection_tables_and_module_have_no_private_repo_manager_coupling() -> None:
    source = (
        __import__("inspect")
        .getsource(__import__("hub_core.runtime.workload_projection", fromlist=["*"]))
    )

    assert "repo_manager." not in source
    assert not runtime_workload_projection_state.foreign_keys
    assert not runtime_workload_projection_records.foreign_keys