"""Full HTTP contract tests for key-free demo mode."""

import asyncio

import httpx
from fastapi.testclient import TestClient
from pydantic import TypeAdapter
from starlette.types import ASGIApp, Message, Receive, Scope, Send

from app.adapters.demo.scenario import load_scenarios
from app.api.schemas import StreamEvent
from app.config import Settings
from app.main import create_app


class _MetadataBarrier:
    """Pause one marked ASGI response immediately after its metadata event."""

    def __init__(self, app: ASGIApp) -> None:
        self.app = app
        self.metadata_sent = asyncio.Event()
        self.resume_response = asyncio.Event()

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        is_refinement = dict(scope.get("headers", ())).get(b"x-demo-request") == b"refinement"

        async def send_with_barrier(message: Message) -> None:
            await send(message)
            if message["type"] == "http.response.body" and b'"type":"metadata"' in message.get(
                "body", b""
            ):
                self.metadata_sent.set()
                await self.resume_response.wait()

        await self.app(scope, receive, send_with_barrier if is_refinement else send)


def test_demo_request_streams_valid_events_without_a_session() -> None:
    app = create_app(Settings(demo_chunk_delay_seconds=0))
    with TestClient(app) as client:
        response = client.post(
            "/api/recommendations",
            json={"schema_version": 1, "query": "Focus while coding"},
        )

    adapter: TypeAdapter[StreamEvent] = TypeAdapter(StreamEvent)
    events = [adapter.validate_json(line) for line in response.text.splitlines()]
    assert response.status_code == 200
    assert events[0].type == "metadata"
    assert events[-1].type == "done"
    assert sum(event.type == "track" for event in events) == 15


def test_unknown_demo_query_discloses_the_replayed_scenario_before_tracks() -> None:
    app = create_app(Settings(demo_chunk_delay_seconds=0))
    with TestClient(app) as client:
        response = client.post(
            "/api/recommendations",
            json={"schema_version": 1, "query": "Focus while codign"},
        )

    adapter: TypeAdapter[StreamEvent] = TypeAdapter(StreamEvent)
    events = [adapter.validate_json(line) for line in response.text.splitlines()]
    warning_index = next(
        index
        for index, event in enumerate(events)
        if event.type == "warning" and event.code == "demo_replay"
    )
    first_track_index = next(index for index, event in enumerate(events) if event.type == "track")
    warning = events[warning_index]
    assert warning_index < first_track_index
    assert warning.type == "warning"
    assert "Focus while coding" in warning.message


def test_demo_refinement_reconstructs_the_pool_recorded_with_its_cassette() -> None:
    app = create_app(Settings(demo_chunk_delay_seconds=0))
    with TestClient(app) as client:
        initial_response = client.post(
            "/api/recommendations",
            json={"schema_version": 1, "query": "Focus while coding"},
        )
        initial_adapter: TypeAdapter[StreamEvent] = TypeAdapter(StreamEvent)
        initial_events: list[StreamEvent] = [
            initial_adapter.validate_json(line) for line in initial_response.text.splitlines()
        ]
        prior_recommendations = [
            {
                "rank": event.rank,
                "track_id": event.track.id,
                "title": event.track.title,
                "artists": event.track.artists,
            }
            for event in initial_events
            if event.type == "track"
        ]
        response = client.post(
            "/api/recommendations",
            json={
                "schema_version": 1,
                "query": "More electronic",
                "prior_recommendations": prior_recommendations,
            },
        )

    adapter: TypeAdapter[StreamEvent] = TypeAdapter(StreamEvent)
    events = [adapter.validate_json(line) for line in response.text.splitlines()]
    warning_codes = [event.code for event in events if event.type == "warning"]
    assert "rerank_fallback" not in warning_codes
    assert sum(event.type == "track" for event in events) == 15


def test_concurrent_demo_refinement_uses_its_request_local_recorded_pool() -> None:
    async def run() -> None:
        settings = Settings(demo_chunk_delay_seconds=0)
        previous = [
            {
                "rank": 1,
                "track_id": "previous",
                "title": "Previous track",
                "artists": ["Previous artist"],
            }
        ]
        payload = {
            "schema_version": 1,
            "query": "More electronic",
            "prior_recommendations": previous,
        }

        baseline_app = create_app(settings)
        baseline_transport = httpx.ASGITransport(app=baseline_app)
        async with (
            baseline_app.router.lifespan_context(baseline_app),
            httpx.AsyncClient(
                transport=baseline_transport,
                base_url="http://test",
            ) as client,
        ):
            baseline_response = await client.post("/api/recommendations", json=payload)
        baseline_events = _stream_events(baseline_response)
        expected_track_ids = [event.track.id for event in baseline_events if event.type == "track"]

        concurrent_app = create_app(settings)
        barrier = _MetadataBarrier(concurrent_app)
        transport = httpx.ASGITransport(app=barrier)
        async with (
            concurrent_app.router.lifespan_context(concurrent_app),
            httpx.AsyncClient(transport=transport, base_url="http://test") as client,
        ):
            refinement_task = asyncio.create_task(
                client.post(
                    "/api/recommendations",
                    json=payload,
                    headers={"x-demo-request": "refinement"},
                )
            )
            try:
                await asyncio.wait_for(barrier.metadata_sent.wait(), timeout=2)
                unrelated_response = await client.post(
                    "/api/recommendations",
                    json={"schema_version": 1, "query": "Energy for the gym"},
                )
            finally:
                barrier.resume_response.set()
            refinement_response = await refinement_task

        events = _stream_events(refinement_response)
        warning_codes = [event.code for event in events if event.type == "warning"]
        track_ids = [event.track.id for event in events if event.type == "track"]
        assert unrelated_response.status_code == 200
        assert refinement_response.status_code == 200
        assert "rerank_fallback" not in warning_codes
        assert track_ids == expected_track_ids

    asyncio.run(run())


def test_demo_auth_playlist_and_suggestions_are_explicitly_simulated() -> None:
    app = create_app(Settings(demo_chunk_delay_seconds=0))
    with TestClient(app, follow_redirects=False) as client:
        current_user = client.get("/api/auth/me")
        login = client.get("/api/auth/login")
        playlist = client.post(
            "/api/playlists",
            json={
                "schema_version": 1,
                "name": "Night drive",
                "track_uris": ["spotify:track:demo"],
            },
        )
        suggestions = client.get("/api/suggestions")

    expected_suggestions = [
        {"chip": scenario.chip, "query": scenario.query}
        for scenario in load_scenarios()
        if not scenario.is_refinement
    ]
    assert current_user.json() == {"display_name": "Demo Listener"}
    assert login.headers["location"] == "/?login=demo"
    assert playlist.json()["url"].startswith("https://open.spotify.com/playlist/demo-")
    assert suggestions.json() == expected_suggestions
    assert not hasattr(app.state, "http")
    assert not hasattr(app.state, "anthropic")


def _stream_events(response: httpx.Response) -> list[StreamEvent]:
    adapter: TypeAdapter[StreamEvent] = TypeAdapter(StreamEvent)
    return [adapter.validate_json(line) for line in response.text.splitlines()]
