From 64175a1ea18e62614a3e01bad7542b00d56f364d Mon Sep 17 00:00:00 2001
From: xsl <youmiss@163.com>
Date: Fri, 14 Aug 2026 16:16:26 +0800
Subject: [PATCH] feat(analytics): POST /events/batch with fail-close
 validation
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

ob_event 落库通道：形状校验 + props 字节上限 + client_ts 合理性 +
PII 键剔除，整批原子（FR-SYS-06 服务端部分）。

实现：packages/analytics/schemas.py（EventIn/EventBatchRequest/
EventBatchResponse + 事件名形状正则常量）；service.py 拆出纯函数
validate_batch（批量/props 字节/事件名形状/client_ts 窗口/PII 剔除，
任一不过整批 40020），AnalyticsService.record 组装 ObEvent 逐条
add + 一次 flush；routers/events.py 提供 POST /events/batch
（/api/v1 与 /v1 别名，认证必需）。

测试：单测 tests/unit/test_events_validation.py 8 例（合法批、超批量、
props 超字节/不可序列化、事件名形状含尾随换行、client_ts 双边界、
naive 时区按 UTC、PII 键含大小写变体剔除）；集成
tests/integration/test_events.py 3 例（两前缀可用+落库断言、
非法批 40020 且库无半条、无 token 40100）。门禁全绿：ruff
check/format、mypy（T2 列表 + domain strict）、lint-imports 4 契约、
单测 177 通过、集成 6 通过。

偏差说明：event_name 正则未写成 Field(pattern=...)，而是作为
EVENT_NAME_PATTERN 常量放 schemas.py、由服务层 validate_batch 强制——
若挂在请求体校验上，非法事件名会先被 RequestValidationError 拦截、
返回通用 40001/422 信封，无法满足任务契约 §4「大写 event_name →
40020」的集成行为。正则本体与文档一致。
---
 apps/api/routers/events.py           |  19 +++-
 packages/analytics/schemas.py        |  43 +++++++++
 packages/analytics/service.py        | 114 +++++++++++++++++++++-
 tests/integration/test_events.py     | 137 +++++++++++++++++++++++++++
 tests/unit/test_events_validation.py | 136 ++++++++++++++++++++++++++
 5 files changed, 446 insertions(+), 3 deletions(-)
 create mode 100644 packages/analytics/schemas.py
 create mode 100644 tests/integration/test_events.py
 create mode 100644 tests/unit/test_events_validation.py

diff --git a/apps/api/routers/events.py b/apps/api/routers/events.py
index 4212253..a607a72 100644
--- a/apps/api/routers/events.py
+++ b/apps/api/routers/events.py
@@ -1,10 +1,25 @@
-"""``POST /events/batch`` — 客户端埋点批量上报（骨架，T3 填充实现）。
+"""``POST /events/batch`` — 客户端埋点批量上报（FR-SYS-06 服务端入口）。
 
-FR-SYS-06 服务端入口；实现契约见 docs/开发任务包/T3-埋点上报.md。
+埋点链路对客户端静默降级：非法批返回 40020（无 hint），客户端丢弃本地
+缓存即可，不打扰用户。校验规则见 packages/analytics/service.py。
 """
 
 from __future__ import annotations
 
 from fastapi import APIRouter
 
+from apps.api.deps import AnalyticsServiceDep, CurrentUserId
+from packages.analytics.schemas import EventBatchRequest, EventBatchResponse
+from packages.shared.response import ok
+
 router = APIRouter(prefix="/events", tags=["events"])
+
+
+@router.post("/batch", summary="批量上报埋点事件（整批校验，任一非法整批拒绝）")
+async def post_event_batch(
+    user_id: CurrentUserId,
+    analytics: AnalyticsServiceDep,
+    payload: EventBatchRequest,
+) -> dict[str, object]:
+    accepted = await analytics.record(user_id=user_id, batch=payload.events)
+    return ok(EventBatchResponse(accepted=accepted).model_dump())
diff --git a/packages/analytics/schemas.py b/packages/analytics/schemas.py
new file mode 100644
index 0000000..be0a1ab
--- /dev/null
+++ b/packages/analytics/schemas.py
@@ -0,0 +1,43 @@
+"""埋点上报的请求/响应模型（架构文档 §9.3 events，FR-SYS-06 服务端部分）。
+
+统一信封（§9.2）之外，这里只描述 ``data`` 载荷。事件字典（event dictionary）
+V1 不做服务端枚举校验 —— 客户端事件名集合由 TD-23 埋点字典维护、尚未定稿，
+先只用形状正则兜底，字典定稿后再收紧。
+"""
+
+from __future__ import annotations
+
+import re
+from datetime import datetime
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+#: 事件名形状：小写字母开头，仅 [a-z0-9_.]，总长 2~64（架构 §6.2 event_name）。
+#:
+#: 注意它刻意不写成 ``Field(pattern=...)`` —— 请求体校验失败会落进通用的
+#: 40001/422 信封，而埋点链路要求 fail-close 地返回域错误 40020（客户端据此
+#: 丢弃本地缓存、静默降级）。因此正则在 service 层 ``validate_batch`` 里强制。
+EVENT_NAME_PATTERN = re.compile(r"^[a-z][a-z0-9_.]{1,63}$")
+
+
+class EventIn(BaseModel):
+    """单条埋点事件（客户端时钟不可信，client_ts 仅作参考）。"""
+
+    event_name: str = Field(description="事件名；形状见 EVENT_NAME_PATTERN，服务层 fail-close 校验")
+    props: dict[str, Any] = Field(
+        default_factory=dict, description="事件属性；PII 键会被服务端剔除"
+    )
+    client_ts: datetime | None = Field(
+        default=None, description="客户端事件时间（ISO 8601）；缺省时仅用服务端时间"
+    )
+
+
+class EventBatchRequest(BaseModel):
+    """整批上报：任一条非法则整批拒绝，不部分落库。"""
+
+    events: list[EventIn] = Field(min_length=1, description="本批事件，上限见 EVENTS_MAX_BATCH")
+
+
+class EventBatchResponse(BaseModel):
+    accepted: int = Field(description="本批成功落库的事件条数；失败走 40020，不会返回部分值")
diff --git a/packages/analytics/service.py b/packages/analytics/service.py
index 0ab23e1..6d0abea 100644
--- a/packages/analytics/service.py
+++ b/packages/analytics/service.py
@@ -1,10 +1,122 @@
-"""analytics 域 service — 骨架（T3 任务填充实现）。"""
+"""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)
diff --git a/tests/integration/test_events.py b/tests/integration/test_events.py
new file mode 100644
index 0000000..5c693db
--- /dev/null
+++ b/tests/integration/test_events.py
@@ -0,0 +1,137 @@
+"""埋点上报集成用例（真实 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
diff --git a/tests/unit/test_events_validation.py b/tests/unit/test_events_validation.py
new file mode 100644
index 0000000..cd510ce
--- /dev/null
+++ b/tests/unit/test_events_validation.py
@@ -0,0 +1,136 @@
+"""埋点整批校验的纯逻辑用例（FR-SYS-06，不连库）。
+
+校验入口是 packages.analytics.service.validate_batch —— 模块级纯函数，
+时间与上限全部注入，便于钉死边界值逐条断言。
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+import pytest
+
+from packages.analytics.schemas import EventIn
+from packages.analytics.service import validate_batch
+from packages.shared.errors import AppError, ErrorCode
+
+#: 钉死的"服务端当前时间"，所有 client_ts 边界都相对它构造。
+NOW = datetime(2026, 8, 14, 12, 0, 0, tzinfo=UTC)
+USER_ID = 7_100_000_000_000_000_001
+
+MAX_BATCH = 50
+MAX_BYTES = 8192
+
+
+def _ev(
+    name: str = "page_view",
+    props: dict[str, Any] | None = None,
+    client_ts: datetime | None = None,
+) -> EventIn:
+    return EventIn(event_name=name, props=props if props is not None else {}, client_ts=client_ts)
+
+
+def _run(events: list[EventIn]) -> list[Any]:
+    return validate_batch(
+        events, user_id=USER_ID, max_batch=MAX_BATCH, max_bytes=MAX_BYTES, now=NOW
+    )
+
+
+def _assert_rejected(events: list[EventIn]) -> None:
+    """fail-close：任一规则不过 → 整批 40020。"""
+    with pytest.raises(AppError) as excinfo:
+        _run(events)
+    assert excinfo.value.code == ErrorCode.EVENTS_BATCH_REJECTED
+    assert int(excinfo.value.code) == 40020
+
+
+def test_valid_batch_passes_and_assembles_rows() -> None:
+    rows = _run(
+        [
+            _ev("outfit_card.tap", {"card_id": 42}, NOW - timedelta(minutes=1)),
+            _ev("page_view"),
+        ]
+    )
+    assert len(rows) == 2
+    assert all(row.user_id == USER_ID for row in rows)
+    # server_ts 用注入的 now，与 client_ts 窗口判定同一时刻
+    assert all(row.server_ts == NOW for row in rows)
+    assert rows[0].event_name == "outfit_card.tap"
+    assert rows[0].props == {"card_id": 42}
+    assert rows[0].client_ts == NOW - timedelta(minutes=1)
+    assert rows[1].client_ts is None  # 缺省 client_ts 允许
+    # 雪花 id 逐条独立分配
+    assert len({row.event_id for row in rows}) == 2
+
+
+def test_empty_and_oversized_batch_are_rejected() -> None:
+    _assert_rejected([])  # schema 的 min_length=1 之外，服务层同样兜底
+    _assert_rejected([_ev() for _ in range(MAX_BATCH + 1)])
+    # 恰好等于上限是合法的
+    assert len(_run([_ev() for _ in range(MAX_BATCH)])) == MAX_BATCH
+
+
+def test_oversize_props_are_rejected() -> None:
+    # 默认上限 8192 字节；8KB 的值序列化后必然超限
+    big = {"blob": "x" * 8192}
+    assert len(json.dumps(big, ensure_ascii=False).encode()) > MAX_BYTES
+    _assert_rejected([_ev(props=big)])
+    # 与合法事件混在同一批也要整批拒绝（fail-close，不部分落库）
+    _assert_rejected([_ev(), _ev(props=big)])
+
+
+def test_unserializable_props_are_rejected() -> None:
+    # set 不可 JSON 序列化 → fail-close 拒收，而不是落库时炸事务
+    _assert_rejected([_ev(props={"bad": {1, 2}})])  # type: ignore[dict-item]
+
+
+def test_event_name_shape_is_enforced() -> None:
+    for bad_name in (
+        "PageView",  # 大写
+        "1page",  # 数字开头
+        "p",  # 太短（正则要求 >= 2 位）
+        "page view",  # 空格
+        "page_view!",  # 非法字符
+        "a" * 65,  # 超长（正则上限 64）
+        "page_view\n",  # $ 可被尾随换行骗过，fullmatch 不行
+    ):
+        _assert_rejected([_ev(bad_name)])
+    # 64 位以内、含点与下划线、数字开头之外的位置都是合法的
+    assert len(_run([_ev("a" + "b" * 63)])) == 1
+    assert len(_run([_ev("wardrobe.item_2.sync_ok")])) == 1
+
+
+def test_client_ts_window_boundaries() -> None:
+    # 不晚于 now+5min：恰好 +5min 合法，再多 1 秒拒收
+    assert len(_run([_ev(client_ts=NOW + timedelta(minutes=5))])) == 1
+    _assert_rejected([_ev(client_ts=NOW + timedelta(minutes=5, seconds=1))])
+    # 不早于 7 天前：恰好 7 天前合法，再早 1 秒拒收
+    assert len(_run([_ev(client_ts=NOW - timedelta(days=7))])) == 1
+    _assert_rejected([_ev(client_ts=NOW - timedelta(days=7, seconds=1))])
+    # 同批一条合法 + 一条越界 → 整批拒
+    _assert_rejected([_ev(client_ts=NOW), _ev(client_ts=NOW + timedelta(hours=1))])
+
+
+def test_naive_client_ts_is_interpreted_as_utc() -> None:
+    # 客户端漏发时区按 UTC 解释，窗口内仍可落库
+    rows = _run([_ev(client_ts=datetime(2026, 8, 14, 11, 0, 0))])
+    assert rows[0].client_ts == datetime(2026, 8, 14, 11, 0, 0, tzinfo=UTC)
+
+
+def test_pii_keys_are_stripped_from_props() -> None:
+    rows = _run(
+        [
+            _ev(
+                "report.share",
+                {
+                    "phone": "13800001234",
+                    "Token": "Bearer abc",  # 大小写变体也拦
+                    "authorization": "Basic xyz",
+                    "scene": "home",
+                },
+            )
+        ]
+    )
+    assert rows[0].props == {"scene": "home"}