from __future__ import annotations

import asyncio
import json
from collections.abc import AsyncIterator, Callable
from typing import Any

import httpx
import pytest
from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import JSONResponse, StreamingResponse
from fastapi.routing import APIRoute

from k1link.web.map_api import (
    MAP_GATEWAY_URL_ENV,
    MAP_SANDBOX_HIDE_CREDITS_ENV,
    MapGatewayConfiguration,
    MapGatewayProxy,
    build_map_router,
)


class _AsyncContent(httpx.AsyncByteStream):
    def __init__(self, content: bytes) -> None:
        self._content = content

    async def __aiter__(self) -> AsyncIterator[bytes]:
        yield self._content


def _request(
    path: str,
    *,
    method: str = "GET",
    query: str = "",
    headers: list[tuple[bytes, bytes]] | None = None,
    client: tuple[str, int] = ("127.0.0.1", 8000),
) -> Request:
    async def receive() -> dict[str, Any]:
        return {"type": "http.request", "body": b"", "more_body": False}

    return Request(
        {
            "type": "http",
            "http_version": "1.1",
            "method": method,
            "scheme": "http",
            "path": path,
            "raw_path": path.encode("ascii"),
            "query_string": query.encode("ascii"),
            "headers": headers or [],
            "client": client,
            "server": ("127.0.0.1", 8000),
        },
        receive,
    )


def _endpoint(router: APIRouter, path: str, method: str) -> Callable[..., Any]:
    for route in router.routes:
        if (
            isinstance(route, APIRoute)
            and route.path == path
            and route.methods is not None
            and method in route.methods
        ):
            return route.endpoint
    raise AssertionError(f"{method} {path} route is missing")


def _response_body(response: JSONResponse) -> bytes:
    return bytes(response.body)


def _service(
    handler: Callable[[httpx.Request], httpx.Response],
) -> tuple[MapGatewayProxy, list[httpx.Request]]:
    requests: list[httpx.Request] = []

    def recording_handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)
        response = handler(request)
        if response.is_stream_consumed:
            return httpx.Response(
                response.status_code,
                headers=response.headers,
                stream=_AsyncContent(response.content),
            )
        return response

    client = httpx.AsyncClient(transport=httpx.MockTransport(recording_handler))
    return (
        MapGatewayProxy(
            MapGatewayConfiguration("http://map-gateway.internal:18103"),
            client=client,
        ),
        requests,
    )


def test_configuration_is_fail_closed_and_rejects_non_origin_values(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.delenv(MAP_GATEWAY_URL_ENV, raising=False)
    assert MapGatewayConfiguration.from_environment().internal_url is None

    monkeypatch.setenv(MAP_GATEWAY_URL_ENV, "http://user:pass@gateway/path?token=no")
    with pytest.raises(ValueError, match="HTTP\\(S\\) origin"):
        MapGatewayConfiguration.from_environment()


def test_runtime_config_exposes_only_same_origin_contract_and_hides_internal_origin(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.delenv(MAP_SANDBOX_HIDE_CREDITS_ENV, raising=False)
    service, requests = _service(
        lambda request: httpx.Response(
            200,
            headers={"content-type": "application/json"},
            json={"ok": True, "rendererVersion": "1.143.0"},
        )
    )
    router = build_map_router(service)
    endpoint = _endpoint(router, "/api/v1/map/runtime-config", "GET")

    response = asyncio.run(endpoint(request=_request("/api/v1/map/runtime-config")))

    assert isinstance(response, JSONResponse)
    assert response.status_code == 200
    document = json.loads(_response_body(response))
    assert document["schema_version"] == "missioncore.map-runtime/v1"
    assert document["map_page_version"] == "0.1.0"
    assert document["renderer"] == {"id": "cesium", "version": "1.143.0"}
    assert document["gateway"]["cache_proxy_prefix"].startswith("/api/v1/map/")
    assert document["assets"] == {
        "imagery": 2,
        "terrain": 1,
        "buildings": 96188,
    }
    assert document["sandbox"] == {"hide_credit_overlay": False}
    assert response.headers["cache-control"] == "no-store"
    assert not requests
    assert "map-gateway.internal" not in _response_body(response).decode()


def test_runtime_config_hides_credits_only_with_explicit_sandbox_flag(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    service, _ = _service(lambda request: httpx.Response(200))
    monkeypatch.setenv(MAP_SANDBOX_HIDE_CREDITS_ENV, "true")

    response = service.runtime_configuration()

    assert isinstance(response, JSONResponse)
    document = json.loads(_response_body(response))
    assert document["sandbox"] == {"hide_credit_overlay": True}


def test_runtime_config_is_fail_closed_when_gateway_is_not_configured() -> None:
    service = MapGatewayProxy(MapGatewayConfiguration(None))
    response = service.runtime_configuration()

    assert isinstance(response, JSONResponse)
    assert response.status_code == 503
    assert json.loads(_response_body(response))["code"] == "map_gateway_not_configured"


def test_public_json_contract_rejects_secret_material() -> None:
    service, _ = _service(
        lambda request: httpx.Response(
            200,
            headers={"content-type": "application/json"},
            json={"ok": True, "accessToken": "must-not-cross-bff"},
        )
    )

    response = asyncio.run(
        service.proxy_json(
            _request("/api/v1/map/runtime-config"),
            upstream_path="/api/map/runtime-config",
        )
    )

    assert isinstance(response, JSONResponse)
    assert response.status_code == 502
    assert json.loads(_response_body(response))["code"] == "map_gateway_invalid_response"
    assert b"must-not-cross-bff" not in _response_body(response)


def test_cache_stream_forwards_only_allowlisted_headers_and_preserves_range() -> None:
    service, requests = _service(
        lambda request: httpx.Response(
            206,
            headers={
                "content-type": "application/octet-stream",
                "content-range": "bytes 0-3/10",
                "etag": '"tile-v1"',
                "set-cookie": "forbidden=1",
                "x-internal-secret": "forbidden",
            },
            content=b"tile",
        )
    )
    response = asyncio.run(
        service.proxy_cache(
            _request(
                "/api/v1/map/gateway/cache",
                headers=[
                    (b"range", b"bytes=0-3"),
                    (b"authorization", b"Bearer browser-secret"),
                    (b"x-nodedc-user-id", b"browser-forgery"),
                ],
            ),
            target_url="https://assets.cesium.com/tile.bin?nodedc_cache_profile=live",
        )
    )

    assert isinstance(response, StreamingResponse)

    async def read_body() -> bytes:
        chunks: list[bytes] = []
        async for chunk in response.body_iterator:
            chunks.append(chunk.encode() if isinstance(chunk, str) else bytes(chunk))
        return b"".join(chunks)

    assert asyncio.run(read_body()) == b"tile"
    assert response.headers["content-range"] == "bytes 0-3/10"
    assert response.headers["etag"] == '"tile-v1"'
    assert "set-cookie" not in response.headers
    assert "x-internal-secret" not in response.headers
    assert requests[0].headers["range"] == "bytes=0-3"
    assert "authorization" not in requests[0].headers
    assert requests[0].headers["x-nodedc-user-id"] == "mission-core-loopback-operator"
    assert requests[0].url.params["url"].startswith("https://assets.cesium.com/")


def test_cache_rejects_credentials_non_https_and_non_loopback_clients() -> None:
    service, requests = _service(
        lambda request: httpx.Response(200, content=b"unused")
    )
    unsafe_targets = (
        "http://assets.cesium.com/tile.bin",
        "https://user:pass@assets.cesium.com/tile.bin",
        "https://assets.cesium.com/tile.bin#secret",
        "https://assets.cesium.com/tile.bin?access_token=forbidden",
    )
    for target in unsafe_targets:
        with pytest.raises(HTTPException) as exc_info:
            asyncio.run(
                service.proxy_cache(
                    _request("/api/v1/map/gateway/cache"),
                    target_url=target,
                )
            )
        assert exc_info.value.status_code == 422

    with pytest.raises(HTTPException) as exc_info:
        asyncio.run(
            service.proxy_cache(
                _request(
                    "/api/v1/map/gateway/cache",
                    client=("192.0.2.10", 1234),
                ),
                target_url="https://assets.cesium.com/tile.bin",
            )
        )
    assert exc_info.value.status_code == 403
    assert not requests


def test_cache_accepts_only_the_public_bing_gateway_key_marker() -> None:
    service, requests = _service(
        lambda request: httpx.Response(
            200,
            headers={"content-type": "application/json"},
            content=b"{}",
        )
    )

    accepted = asyncio.run(
        service.proxy_cache(
            _request("/api/v1/map/gateway/cache"),
            target_url=(
                "https://dev.virtualearth.net/REST/v1/Imagery/Metadata/Aerial"
                "?key=nodedc-gateway"
            ),
        )
    )

    assert isinstance(accepted, StreamingResponse)
    assert requests[0].url.params["url"].endswith("key=nodedc-gateway")

    with pytest.raises(HTTPException) as exc_info:
        asyncio.run(
            service.proxy_cache(
                _request("/api/v1/map/gateway/cache"),
                target_url=(
                    "https://dev.virtualearth.net/REST/v1/Imagery/Metadata/Aerial"
                    "?key=browser-secret"
                ),
            )
        )
    assert exc_info.value.status_code == 422


def test_asset_route_is_a_fixed_allowlist() -> None:
    service, requests = _service(
        lambda request: httpx.Response(
            200,
            headers={"content-type": "application/json"},
            json={"ok": True, "assetId": 1, "credentialMode": "gateway"},
        )
    )
    router = build_map_router(service)
    endpoint = _endpoint(
        router,
        "/api/v1/map/gateway/ion/assets/{asset_id}/endpoint",
        "GET",
    )

    rejected = asyncio.run(
        endpoint(
            request=_request("/api/v1/map/gateway/ion/assets/42/endpoint"),
            asset_id=42,
        )
    )
    accepted = asyncio.run(
        endpoint(
            request=_request("/api/v1/map/gateway/ion/assets/1/endpoint"),
            asset_id=1,
        )
    )

    assert isinstance(rejected, JSONResponse)
    assert rejected.status_code == 404
    assert json.loads(_response_body(rejected))["code"] == "cesium_asset_not_allowed"
    assert isinstance(accepted, JSONResponse)
    assert len(requests) == 1
    assert requests[0].url.path.endswith("/api/map/ion/assets/1/endpoint")


def test_upstream_errors_preserve_only_safe_code() -> None:
    service, _ = _service(
        lambda request: httpx.Response(
            504,
            headers={"content-type": "application/json"},
            json={
                "code": "map_upstream_timeout",
                "detail": "private-host.internal?token=forbidden",
            },
        )
    )

    response = asyncio.run(
        service.proxy_cache(
            _request("/api/v1/map/gateway/cache"),
            target_url="https://assets.cesium.com/tile.bin",
        )
    )

    assert isinstance(response, JSONResponse)
    assert response.status_code == 504
    assert json.loads(_response_body(response)) == {
        "schema_version": "missioncore.map-gateway/v1",
        "code": "map_upstream_timeout",
        "retryable": True,
    }
    assert b"private-host" not in _response_body(response)