"""analytics 域 service —— ``ob_event`` 落库通道（FR-SYS-06 服务端部分）。

校验拆成模块级纯函数 :func:`validate_batch`（不碰 DB，单测直接覆盖），
``AnalyticsService.record`` 只负责取配置、组装与落库。整批 fail-close：
任一条不合法 → ``EVENTS_BATCH_REJECTED``（40020），绝不部分落库 ——
半批数据会污染看板口径（TD-24 三张看板都吃这张表），宁可让客户端整批重试。
"""

from __future__ import annotations

import json
from collections.abc import Sequence
from datetime import UTC, datetime, timedelta
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from packages.analytics.schemas import EVENT_NAME_PATTERN, EventIn
from packages.domain.ids import next_id
from packages.shared.config import get_settings
from packages.shared.errors import AppError, ErrorCode
from packages.shared.logging import get_logger
from packages.shared.models import ObEvent

logger = get_logger(__name__)

#: 客户端时钟最多比服务端快 5 分钟（超出视为异常时间戳，防未来事件污染当日看板）。
_CLIENT_TS_MAX_SKEW = timedelta(minutes=5)
#: 离线缓存上报最多回溯 7 天（再久的旧事件口径已失真，直接拒收）。
_CLIENT_TS_MAX_AGE = timedelta(days=7)

#: props 顶层出现即剔除的 PII 键（NFR-SEC-03：埋点与日志同族，都不落 PII）。
_PII_PROPS_KEYS = frozenset({"phone", "token", "authorization"})


def _drop_pii_keys(props: dict[str, Any], *, event_name: str) -> dict[str, Any]:
    """剔除 props 顶层的 PII 键（大小写不敏感，``Token`` 这类变体也拦）。

    日志只记被剔除的**键名**，不记值 —— 值本身就是 PII，不能二次泄漏。
    """
    if not any(key.lower() in _PII_PROPS_KEYS for key in props):
        return props
    dropped = [key for key in props if key.lower() in _PII_PROPS_KEYS]
    logger.warning("events_pii_dropped", event_name=event_name, dropped_keys=dropped)
    return {key: value for key, value in props.items() if key.lower() not in _PII_PROPS_KEYS}


def validate_batch(
    events: Sequence[EventIn],
    *,
    user_id: int,
    max_batch: int,
    max_bytes: int,
    now: datetime,
) -> list[ObEvent]:
    """整批 fail-close 校验并组装 ``ObEvent`` 行（纯函数，不触碰 DB）。

    返回的行尚未进 session，由 ``record`` 统一 ``add_all`` + ``flush``，
    保证「要么全落、要么全拒」在服务边界内成立。``now`` 由调用方注入，
    既让单测可以钉死时间，也保证 server_ts 与 client_ts 窗口用的是同一时刻。
    """
    if not 1 <= len(events) <= max_batch:
        raise AppError(ErrorCode.EVENTS_BATCH_REJECTED)

    rows: list[ObEvent] = []
    for event in events:
        # 事件字典枚举校验等 TD-23 定稿后再收紧，V1 只约束形状。
        if EVENT_NAME_PATTERN.fullmatch(event.event_name) is None:
            raise AppError(ErrorCode.EVENTS_BATCH_REJECTED)

        props = _drop_pii_keys(event.props, event_name=event.event_name)

        # 字节上限以 UTF-8 序列化后计（与客户端计算口径一致），不可序列化同样拒收。
        try:
            encoded = json.dumps(props, ensure_ascii=False).encode()
        except (TypeError, ValueError) as exc:
            raise AppError(ErrorCode.EVENTS_BATCH_REJECTED) from exc
        if len(encoded) > max_bytes:
            raise AppError(ErrorCode.EVENTS_BATCH_REJECTED)

        client_ts = event.client_ts
        if client_ts is not None:
            # 客户端漏发时区时按 UTC 解释（与服务端口径一致），仍落在窗口外则拒收。
            if client_ts.tzinfo is None:
                client_ts = client_ts.replace(tzinfo=UTC)
            if client_ts < now - _CLIENT_TS_MAX_AGE or client_ts > now + _CLIENT_TS_MAX_SKEW:
                raise AppError(ErrorCode.EVENTS_BATCH_REJECTED)

        rows.append(
            ObEvent(
                event_id=next_id(),
                user_id=user_id,
                event_name=event.event_name,
                props=props,
                client_ts=client_ts,
                server_ts=now,
            )
        )
    return rows


class AnalyticsService:
    def __init__(self, session: AsyncSession) -> None:
        self._session = session

    async def record(self, *, user_id: int, batch: list[EventIn]) -> int:
        """校验并落库一批埋点事件，返回落库条数。

        只 flush 不 commit —— 批量原子性交给 ``session_dep`` 的统一提交/回滚，
        校验失败时一条都不会写入。
        """
        settings = get_settings()
        rows = validate_batch(
            batch,
            user_id=user_id,
            max_batch=settings.events_max_batch,
            max_bytes=settings.events_props_max_bytes,
            now=datetime.now(UTC),
        )
        self._session.add_all(rows)
        await self._session.flush()
        return len(rows)