feat(analytics): POST /events/batch with fail-close validation
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」的集成行为。正则本体与文档一致。
This commit is contained in:
@@ -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，不会返回部分值")
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user