from __future__ import annotations

import asyncio
from datetime import UTC, datetime
from typing import Any

import pytest
from sqlalchemy import insert, select
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker

from script_build_host.domain.canonical_json import canonical_sha256
from script_build_host.domain.errors import HttpIdempotencyConflict, ProtocolViolation
from script_build_host.domain.records import Principal
from script_build_host.infrastructure.http_commands import HttpCommandJournal
from script_build_host.infrastructure.tables import http_command_table


@pytest.mark.asyncio
async def test_http_command_replays_completed_response_and_rejects_fingerprint_conflict(
    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
) -> None:
    _, sessions = database
    journal = HttpCommandJournal(sessions)
    principal = Principal("subject", "tenant")
    calls = 0

    async def command() -> tuple[int, dict[str, Any]]:
        nonlocal calls
        calls += 1
        return 201, {"success": True, "resource_id": 9}

    first = await journal.execute(
        principal=principal,
        route_family="script-build:test",
        idempotency_key="same",
        request_payload={"value": 1},
        command=command,
        resource_id=9,
    )
    second = await journal.execute(
        principal=principal,
        route_family="script-build:test",
        idempotency_key="same",
        request_payload={"value": 1},
        command=command,
        resource_id=9,
    )
    assert first == second == (201, {"success": True, "resource_id": 9})
    assert calls == 1

    with pytest.raises(HttpIdempotencyConflict):
        await journal.execute(
            principal=principal,
            route_family="script-build:test",
            idempotency_key="same",
            request_payload={"value": 2},
            command=command,
        )


@pytest.mark.asyncio
async def test_concurrent_same_http_command_executes_once_and_waits_for_completion(
    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
) -> None:
    _, sessions = database
    journal = HttpCommandJournal(sessions)
    entered = asyncio.Event()
    release = asyncio.Event()
    calls = 0

    async def command() -> tuple[int, dict[str, Any]]:
        nonlocal calls
        calls += 1
        entered.set()
        await release.wait()
        return 200, {"success": True}

    first = asyncio.create_task(
        journal.execute(
            principal=Principal("subject"),
            route_family="script-build:concurrent",
            idempotency_key="one",
            request_payload={"build": 1},
            command=command,
        )
    )
    await entered.wait()
    second = asyncio.create_task(
        journal.execute(
            principal=Principal("subject"),
            route_family="script-build:concurrent",
            idempotency_key="one",
            request_payload={"build": 1},
            command=command,
        )
    )
    release.set()
    assert await first == await second == (200, {"success": True})
    assert calls == 1


@pytest.mark.asyncio
async def test_failed_http_command_is_not_blindly_replayed(
    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
) -> None:
    _, sessions = database
    journal = HttpCommandJournal(sessions)

    async def command() -> tuple[int, dict[str, Any]]:
        raise RuntimeError("lost acknowledgement")

    values = dict(
        principal=Principal("subject"),
        route_family="script-build:failed",
        idempotency_key="one",
        request_payload={"build": 1},
        command=command,
    )
    with pytest.raises(RuntimeError, match="lost acknowledgement"):
        await journal.execute(**values)
    with pytest.raises(ProtocolViolation, match="reconciliation"):
        await journal.execute(**values)


@pytest.mark.asyncio
async def test_reserved_http_command_resumes_preallocated_identity(
    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
) -> None:
    _, sessions = database
    principal = Principal("subject", "tenant")
    route = "script-build:start"
    key = "crash-window"
    payload = {"topic_id": 3}
    scope = canonical_sha256(
        {"subject": principal.subject, "tenant_id": principal.tenant_id}
    ).hex_value
    fingerprint = canonical_sha256({"route_family": route, "request": payload}).hex_value
    now = datetime.now(UTC)
    async with sessions() as session, session.begin():
        await session.execute(
            insert(http_command_table).values(
                principal_scope_sha256=scope,
                route_family=route,
                idempotency_key=key,
                request_fingerprint=fingerprint,
                preallocated_root_trace_id="root-preallocated",
                resource_id=41,
                state="reserved",
                created_at=now,
                updated_at=now,
            )
        )

    async def command() -> tuple[int, dict[str, Any]]:
        raise AssertionError("reserved command must not be blindly replayed")

    async def resume(root: str | None, resource: int | None) -> tuple[int, dict[str, Any]]:
        assert (root, resource) == ("root-preallocated", 41)
        return 200, {"script_build_id": 41, "root_trace_id": root}

    result = await HttpCommandJournal(sessions).execute(
        principal=principal,
        route_family=route,
        idempotency_key=key,
        request_payload=payload,
        command=command,
        resume_reserved=resume,
    )
    assert result == (
        200,
        {"script_build_id": 41, "root_trace_id": "root-preallocated"},
    )
    async with sessions() as session:
        assert (
            await session.scalar(
                select(http_command_table.c.state).where(
                    http_command_table.c.idempotency_key == key
                )
            )
            == "completed"
        )


@pytest.mark.asyncio
async def test_oversized_idempotent_response_is_failed_not_left_reserved(
    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
) -> None:
    _, sessions = database

    async def command() -> tuple[int, dict[str, Any]]:
        return 200, {"value": "x" * (65 * 1024)}

    with pytest.raises(ProtocolViolation, match="exceeds"):
        await HttpCommandJournal(sessions).execute(
            principal=Principal("subject"),
            route_family="script-build:oversized",
            idempotency_key="oversized",
            request_payload={},
            command=command,
        )
    async with sessions() as session:
        assert (
            await session.scalar(
                select(http_command_table.c.state).where(
                    http_command_table.c.idempotency_key == "oversized"
                )
            )
            == "failed"
        )


@pytest.mark.asyncio
async def test_idempotency_key_is_scoped_by_principal_and_route(
    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
) -> None:
    _, sessions = database
    journal = HttpCommandJournal(sessions)
    calls: list[str] = []

    async def execute(label: str) -> tuple[int, dict[str, Any]]:
        calls.append(label)
        return 200, {"label": label}

    common = {
        "idempotency_key": "same-key",
        "request_payload": {"script_build_id": 7},
    }
    first = await journal.execute(
        principal=Principal("subject", "tenant-a"),
        route_family="script-build:phase-three-advance",
        command=lambda: execute("advance-a"),
        **common,
    )
    second = await journal.execute(
        principal=Principal("subject", "tenant-b"),
        route_family="script-build:phase-three-advance",
        command=lambda: execute("advance-b"),
        **common,
    )
    third = await journal.execute(
        principal=Principal("subject", "tenant-a"),
        route_family="script-build:finalize",
        command=lambda: execute("finalize-a"),
        **common,
    )

    assert first[1]["label"] == "advance-a"
    assert second[1]["label"] == "advance-b"
    assert third[1]["label"] == "finalize-a"
    assert calls == ["advance-a", "advance-b", "finalize-a"]