"""埋点上报集成用例（真实 PostgreSQL，FR-SYS-06）。

覆盖：/v1 与 /api/v1 两个前缀、落库行内容（含 PII 剔除）、
非法批 40020 且整批原子（库里没有半条）、无 token 40100。
"""

from __future__ import annotations

import os
import uuid
from datetime import UTC, datetime
from typing import Any

import pytest


def _infra_available() -> bool:
    return bool(os.getenv("MEIDA_RUN_INTEGRATION"))


pytestmark = [
    pytest.mark.integration,
    pytest.mark.skipif(
        not _infra_available(),
        reason="set MEIDA_RUN_INTEGRATION=1 with PostgreSQL and Redis running",
    ),
]


@pytest.fixture
async def client() -> Any:
    import httpx

    from apps.api.main import create_app
    from packages.shared.config import get_settings

    # 埋点链路不碰 LLM，无需替换 provider；lifespan 照常起网关与连接池
    app = create_app(get_settings())
    transport = httpx.ASGITransport(app=app)
    async with (
        httpx.AsyncClient(transport=transport, base_url="http://test") as http,
        app.router.lifespan_context(app),
    ):
        yield http


@pytest.fixture
async def auth(client: Any) -> dict[str, Any]:
    """dev-token 换一个专属测试用户，避免与其它用例的 ob_event 行串号。"""
    resp = await client.post(
        "/api/v1/auth/dev-token", json={"device_key": f"t3-{uuid.uuid4().hex[:8]}"}
    )
    assert resp.json()["code"] == 0
    data = resp.json()["data"]
    headers = {"Authorization": f"Bearer {data['access_token']}"}
    return {"headers": headers, "user_id": int(data["user_id"])}


async def _count_events(user_id: int) -> int:
    from sqlalchemy import func, select

    from packages.shared.db import get_session_factory
    from packages.shared.models import ObEvent

    async with get_session_factory()() as session:
        stmt = select(func.count()).select_from(ObEvent).where(ObEvent.user_id == user_id)
        return int(await session.scalar(stmt))


async def _fetch_events(user_id: int) -> list[Any]:
    from sqlalchemy import select

    from packages.shared.db import get_session_factory
    from packages.shared.models import ObEvent

    async with get_session_factory()() as session:
        result = await session.execute(
            select(ObEvent).where(ObEvent.user_id == user_id).order_by(ObEvent.event_id)
        )
        return list(result.scalars().all())


async def test_batch_lands_two_rows_with_pii_stripped(client: Any, auth: dict[str, Any]) -> None:
    resp = await client.post(
        "/v1/events/batch",
        json={
            "events": [
                {
                    "event_name": "page_view",
                    "props": {"page": "home", "phone": "13800001234"},
                    "client_ts": datetime.now(UTC).isoformat(),
                },
                {"event_name": "outfit_card.tap", "props": {"card_id": 42}},
            ]
        },
        headers=auth["headers"],
    )
    assert resp.status_code == 200
    body = resp.json()
    assert body["code"] == 0
    assert body["data"] == {"accepted": 2}

    rows = await _fetch_events(auth["user_id"])
    assert len(rows) == 2
    assert rows[0].event_name == "page_view"
    assert rows[0].props == {"page": "home"}  # phone 已在服务端剔除
    assert rows[0].client_ts is not None
    assert rows[0].server_ts is not None
    assert rows[1].event_name == "outfit_card.tap"
    assert rows[1].props == {"card_id": 42}
    assert rows[1].client_ts is None


async def test_invalid_batch_is_rejected_atomically(client: Any, auth: dict[str, Any]) -> None:
    before = await _count_events(auth["user_id"])

    resp = await client.post(
        "/api/v1/events/batch",
        json={
            "events": [
                {"event_name": "page_view"},
                {"event_name": "PageView"},  # 大写 → 形状非法
            ]
        },
        headers=auth["headers"],
    )
    assert resp.status_code == 400
    assert resp.json()["code"] == 40020

    # 整批原子：合法的那条也不能半落库
    assert await _count_events(auth["user_id"]) == before


async def test_events_require_authentication(client: Any) -> None:
    resp = await client.post("/api/v1/events/batch", json={"events": [{"event_name": "page_view"}]})
    assert resp.status_code == 401
    assert resp.json()["code"] == 40100