"""The preview on the guest-facing routes (tasks 6.6, 6b.4, design D3, D9).

These need the database because what they assert is about the route, not about the model —
`test_link_preview.py` already covers the model in isolation. The two properties here are
the ones the tokenized route's safety rests on: that its preview is worth no more to a
crawler than the public page's, and that a crawler's fetch does not get recorded as a guest
opening their invitation.

The user-agent matching itself lives in the web app (`web/lib/crawlers.ts`) and is verified
against the running stack rather than here — `web/` has no test runner, and adding one is a
larger decision than this change should make on its own.
"""

from typing import Any

import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.main import app

from .conftest import requires_db

pytestmark = requires_db


async def _tokens(session: AsyncSession, event_id: Any) -> list[str]:
    rows = await session.execute(
        text("SELECT token FROM invitation WHERE event_id = :e ORDER BY token"),
        {"e": event_id},
    )
    return [r[0] for r in rows]


async def _open_state(session: AsyncSession, token: str) -> tuple[int, str]:
    row = (
        await session.execute(
            text("SELECT open_count, status FROM invitation WHERE token = :t"), {"t": token}
        )
    ).one()
    return int(row[0]), str(row[1])


async def _history_count(session: AsyncSession, token: str) -> int:
    count = await session.scalar(
        text(
            "SELECT count(*) FROM rsvp_history h JOIN invitation i ON i.id = h.invitation_id "
            "WHERE i.token = :t"
        ),
        {"t": token},
    )
    return int(count or 0)


@pytest.mark.usefixtures("app_engine_per_loop")
async def test_two_guests_of_one_event_get_an_identical_preview(
    session: AsyncSession, fixture_event: dict[str, Any]
) -> None:
    """The basis for previewing a tokenized URL at all: what a crawler can read from one
    guest's link has to be exactly what it can read from any other's."""
    tokens = await _tokens(session, fixture_event["event_id"])
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        first = await client.get(f"/api/invitations/by-token/{tokens[0]}?preview=1")
        second = await client.get(f"/api/invitations/by-token/{tokens[1]}?preview=1")

    assert first.status_code == 200, first.text
    assert first.json()["preview"] == second.json()["preview"]


@pytest.mark.usefixtures("app_engine_per_loop")
async def test_no_token_or_guest_detail_appears_in_the_preview(
    session: AsyncSession, fixture_event: dict[str, Any]
) -> None:
    tokens = await _tokens(session, fixture_event["event_id"])
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.get(f"/api/invitations/by-token/{tokens[0]}?preview=1")

    rendered = str(response.json()["preview"])
    assert tokens[0] not in rendered
    assert "Guest 0" not in rendered
    assert "@example.test" not in rendered


@pytest.mark.usefixtures("app_engine_per_loop")
async def test_the_open_route_and_the_token_route_preview_identically(
    session: AsyncSession, fixture_event: dict[str, Any]
) -> None:
    tokens = await _tokens(session, fixture_event["event_id"])
    slug = f"walima-{fixture_event['tag']}"
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        tokenised = await client.get(f"/api/invitations/by-token/{tokens[0]}?preview=1")
        public = await client.get(f"/api/events/{slug}")

    assert public.status_code == 200, public.text
    assert tokenised.json()["preview"] == public.json()["preview"]


@pytest.mark.usefixtures("app_engine_per_loop")
async def test_the_canonical_url_is_the_public_page_not_the_token(
    session: AsyncSession, fixture_event: dict[str, Any]
) -> None:
    tokens = await _tokens(session, fixture_event["event_id"])
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.get(f"/api/invitations/by-token/{tokens[0]}?preview=1")

    canonical = response.json()["preview"]["canonical_url"]
    assert canonical.endswith(f"/e/walima-{fixture_event['tag']}")
    assert "/i/" not in canonical


@pytest.mark.usefixtures("app_engine_per_loop")
async def test_an_event_with_no_card_still_previews(
    session: AsyncSession, fixture_event: dict[str, Any]
) -> None:
    """The fixture publishes no card design, so this is the degradation path."""
    tokens = await _tokens(session, fixture_event["event_id"])
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.get(f"/api/invitations/by-token/{tokens[0]}?preview=1")

    preview = response.json()["preview"]
    assert preview["image"] is None
    assert preview["title"] and preview["description"]


# ------------------------------------------------------- a crawler's fetch is not an open


@pytest.mark.usefixtures("app_engine_per_loop")
async def test_preview_mode_records_nothing(
    session: AsyncSession, fixture_event: dict[str, Any]
) -> None:
    """An admin pasting a link into WhatsApp must not mark that guest's invitation opened."""
    tokens = await _tokens(session, fixture_event["event_id"])
    token = tokens[0]

    before = await _open_state(session, token)
    history_before = await _history_count(session, token)

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.get(f"/api/invitations/by-token/{token}?preview=1")
    assert response.status_code == 200, response.text

    session.expire_all()
    assert await _open_state(session, token) == before
    assert await _history_count(session, token) == history_before


@pytest.mark.usefixtures("app_engine_per_loop")
async def test_a_normal_read_still_records_the_open(
    session: AsyncSession, fixture_event: dict[str, Any]
) -> None:
    """The counterweight: skipping the open for crawlers must not skip it for guests."""
    tokens = await _tokens(session, fixture_event["event_id"])
    token = tokens[0]

    count_before, _ = await _open_state(session, token)

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.get(f"/api/invitations/by-token/{token}")
    assert response.status_code == 200, response.text

    session.expire_all()
    count_after, _ = await _open_state(session, token)
    assert count_after == count_before + 1


@pytest.mark.usefixtures("app_engine_per_loop")
async def test_preview_mode_returns_the_same_payload(
    session: AsyncSession, fixture_event: dict[str, Any]
) -> None:
    """Only the recording is skipped. If the payload differed, this would be a second way to
    read an invitation, and the two would drift apart."""
    tokens = await _tokens(session, fixture_event["event_id"])
    token = tokens[0]

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        quiet = await client.get(f"/api/invitations/by-token/{token}?preview=1")
        normal = await client.get(f"/api/invitations/by-token/{token}")

    # `status` and the open itself are the only things a read is allowed to change, so they
    # are compared after removing the one field the normal read legitimately moved.
    a, b = quiet.json(), normal.json()
    a.pop("status"), b.pop("status")
    assert a == b


@pytest.mark.usefixtures("app_engine_per_loop")
async def test_preview_mode_does_not_reveal_an_unknown_token(
    session: AsyncSession, fixture_event: dict[str, Any]
) -> None:
    """It must not become a quieter way to probe for tokens."""
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        missing = await client.get("/api/invitations/by-token/not-a-real-token?preview=1")
    assert missing.status_code == 404
    assert missing.json()["detail"] == "Invitation not found"