import asyncio
import json
import os
from collections import Counter
from pathlib import Path
from uuid import uuid4

import httpx
from sqlalchemy import func, select
from test_after_sales import scenario  # noqa: F401

from zhifu.modules.after_sales.models import Ticket


async def test_replica_reads_and_idempotent_writes(scenario):  # noqa: F811
    phase = os.environ.get("REPLICA_PHASE", "both")
    assert phase in {"both", "single", "recovered"}
    expected = {"api1"} if phase == "single" else {"api1", "api2"}
    async with httpx.AsyncClient(
        base_url="http://127.0.0.1:18080",
        headers=scenario["headers"]["customer"],
        timeout=15,
        trust_env=False,
    ) as client:
        responses = await asyncio.gather(
            *(client.get(f"/openapi/v1/orders?store_id={scenario['store']}") for _ in range(64))
        )
        assert all(response.status_code == 200 for response in responses), [
            {"status": response.status_code, "body": response.text}
            for response in responses
            if response.status_code != 200
        ]
        counts = Counter(response.headers["X-Zhifu-Replica"] for response in responses)
        assert set(counts) == expected, counts
        key = uuid4().hex
        writes = await asyncio.gather(
            *(
                client.post(
                    "/openapi/v1/after-sales/proposals",
                    json=scenario["proposal"],
                    headers={"Idempotency-Key": key},
                )
                for _ in range(16)
            )
        )
        assert all(response.status_code == 201 for response in writes), [r.text for r in writes]
        assert {response.headers["X-Zhifu-Replica"] for response in writes} == expected
        assert len({response.json()["ticket_id"] for response in writes}) == 1
        async with scenario["app"].state.sessions() as session:
            tickets = await session.scalar(
                select(func.count())
                .select_from(Ticket)
                .where(Ticket.tenant_id == scenario["tenant"].bytes)
            )
        assert tickets == 1
        report = {
            "phase": phase,
            "read_statuses": dict(Counter(r.status_code for r in responses)),
            "read_replicas": dict(counts),
            "write_statuses": dict(Counter(r.status_code for r in writes)),
            "write_replicas": dict(Counter(r.headers["X-Zhifu-Replica"] for r in writes)),
            "tickets": tickets,
        }
        await asyncio.to_thread(
            Path(f"docs/verification/replica-{phase}.json").write_text,
            json.dumps(report, indent=2),
            encoding="utf-8",
        )