yml2213
6e729e7a80
时间线从此覆盖**全部**业务事实，`douyu_tasks` 也有了可控的增长边界。
## 一、删掉收集器（它必然漏数据）
`jobs/account_events_collect.py` 连同 `_task_to_event` / `COLLECTED_TASK_TYPES` /
水位线一起删除，迁移 `20261004_0043` 清掉它的作业行。删除理由是实测出来的：
- **靠 allowlist 认业务类型，新增类型就静默丢**：原有名单只有 5 个，
而 `exchange_esports_goods`（电竞兑换）、`exchange_xpd_goods`（和平小店兑换）、
`pay_locked_order`（锁单支付）、`donate_esports_*`（电竞礼物积分）**从来没进过时间线**；
- **「任务 success」≠「业务真的发生了」**（见下）；
- **字段路径靠抽样猜**：二维码充值的 `result` 里根本没有金额，收集器恒写
`amount=None`，`recharge_total` 系统性少计；
- 作业默认 `enabled=False`，生产上一直没跑。
历史**不回填**（既有结论不变）：需要历史走斗鱼侧权威接口
（`query_exchange_records` / `query_xpd_purchase_records` / `query_points` 都是现成任务类型）。
## 二、业务事实直发：命名构造函数
新增 `services/business_events.py`：`record_recharge` / `record_exchange` /
`record_points_credit` / `record_bind`。每类事实的字段、标题、金额符号只定义一次
（与 S4 的 `record_account_action`、`on_ck_verdict` 同一套路）。
**9 处成功点接线**：`exchange_goods` 真兑换分支、`exchange_esports_goods`、
`exchange_xpd_goods`、`pay_locked_order`、`create_gold_qr`（二维码 + 供应商轮询 +
**供应商回调**三条路径）、`donate_elite_gift`、`donate_esports_gift`、
`confirm_bind`、`confirm_esports_bind`。
**只记真实事实**（按你的决定）：`exchange_goods` 里「预约到货 / 已预兑 / 预兑确认」
三条 success 分支**不发事件** —— 它们不消耗积分也没换到东西，旧收集器只按 status
收集，于是每预约一次就生成一条金额为空的 `exchange` 把 `exchange_count` 灌大。
`confirm_bind` 里「检测到账号本来就已绑定」那条也跳过。
## 三、修掉两个真 bug（都是钱账）
1. **到账判据 `!=` → `>`**（`douyu_runner_gold.py`）：余额**下降**也算「鱼翅已到账」。
批内并发（最高 10）时同账号送礼会消耗鱼翅，余额一变就被充值任务读成到账，
**往台账记一笔根本没发生的充值**。
2. **供应商回调晚于 runner 超时**：原来守卫是「任务状态没被改过才更新」，
于是 runner 先判 failed、回调随后说成功的那笔真金白银**永远不进时间线**。
改成按**回调自己的终态**判断，并补发事件。
## 四、发射顺序：一个会静默丢事件的坑
`_mark_task` 内部 `db.commit()`，而 `emit` 用 `db.begin_nested()` **只 RELEASE
SAVEPOINT、不提交外层事务**；runner 的 `finally` 只 `close()`（未提交即回滚）。
所以**在 `_mark_task` 之后发事件会被静默丢弃，且 `emit` 仍返回 True**。
所有接线一律紧贴 `_mark_task(success)` **之前**，并加了源码扫描用例钉住
（已反向验证：故意挪到之后，用例立刻报出 file:line）。
## 五、`douyu_tasks` 保留策略（新增作业，默认关闭）
`jobs/douyu_tasks_retention.py`，两档：
| 档 | 保留 |
| --- | --- |
| 噪声查询（只读看一眼就走，且前端不拿它派生账号状态） | **7 天** |
| 其余一切（业务事实 + 状态派生依赖的查询 + **以后新增的类型**） | **90 天** |
长保留档用 `task_type NOT IN (噪声名单)` 表达 → **新增类型默认长保留**，不会因为
「忘了加名单」被误删（正是收集器翻车的反面）。在途任务（planned/pending/running）
**绝不删**；分批删避免长锁；两个窗口都能在作业 `config` 里改，不必发版。
⚠️ 「前端从任务列表反推账号当前状态」那几类查询（`query_xpd_bind_info` /
`query_esports_game_name` 等）**故意留在长保留档** —— 删早了工作台账号行会变
「未查」，那是把**当前显示**搞坏。根治要靠 §3.6 的状态列拆解（后续阶段）。
作业默认 `enabled=False`，由运维决定何时打开。
## 验证
- `TZ=Asia/Shanghai`：**450 passed / 6 skipped**（S4 后 426 → 净增 24）
- `TZ=UTC`（相关文件）101 passed；ruff check/format 全绿
- 前端 build 通过、lint 0 errors
- 真实开发库：迁移已应用（head `20261004_0043`，孤儿作业行已清）；
保留作业实跑确认「作业行默认关闭、SQL 在 MySQL 上正确、近期行不误删」
- **两处反向验证**：把 `record_bind` 挪到 `_mark_task` 之后 → 顺序用例报红；
修掉 `registered_jobs()` 那个**假绿用例**（它返回 dict，原来的集合推导式
遍历空字典恒成立），并在断言里加了反向确认
未做（后续）：S6 账号页重构 + `bind_status` 三列拆解（做完才能把状态派生类查询
也纳入短保留）、S7 账号检测改账号维度、S8 丢 `login_tasks` 旧凭据列。
485 lines
17 KiB
Python
485 lines
17 KiB
Python
"""账号事件流回归：幂等发射、旁路隔离、任务映射、收集器水位线。
|
||
|
||
三条必须守住：
|
||
|
||
1. **发射是旁路**：`emit` 任何失败都不能影响主流程（登录/兑换照常）；
|
||
2. **幂等**：同一个 `source_ref` 重复发射只落一条 —— 收集器重跑的可靠性全押在这上面；
|
||
3. **水位线只进不退**：解析失败要跳过而不是卡住，否则一行坏数据会让事件流永久停摆。
|
||
"""
|
||
|
||
import os
|
||
import time
|
||
from collections.abc import Iterator
|
||
from contextlib import contextmanager
|
||
from datetime import UTC, datetime
|
||
|
||
from sqlalchemy.orm import sessionmaker
|
||
|
||
import web.backend.database as database_module
|
||
from tests.support.db import get_test_engine, make_test_session
|
||
from web.backend.models import Account, AccountEvent, User
|
||
from web.backend.services import account_events_service as svc
|
||
|
||
|
||
@contextmanager
|
||
def _force_non_utc_timezone(name: str = "Asia/Shanghai") -> Iterator[None]:
|
||
"""临时把进程时区切到非 UTC，让「naive 被当成本地时间」这类 bug 必然暴露。
|
||
|
||
⚠️ 不加这一段，时区相关的回归在 **UTC 机器上（CI / 生产容器）会假绿**：
|
||
旧实现 `naive.astimezone(UTC)` 在 TZ=UTC 下恰好等于正确结果，
|
||
只有开发机（UTC+8）才现形 —— 而「只在某些机器上失败」的测试等于没有测试。
|
||
"""
|
||
previous = os.environ.get("TZ")
|
||
os.environ["TZ"] = name
|
||
time.tzset()
|
||
try:
|
||
yield
|
||
finally:
|
||
if previous is None:
|
||
os.environ.pop("TZ", None)
|
||
else:
|
||
os.environ["TZ"] = previous
|
||
time.tzset()
|
||
|
||
|
||
class _Base:
|
||
def setup_method(self):
|
||
self.engine = get_test_engine()
|
||
database_module.SessionLocal.configure(bind=self.engine)
|
||
self.session = make_test_session(self.engine)
|
||
self.session.query(AccountEvent).delete()
|
||
self.session.commit()
|
||
|
||
user = User(username="evt-admin", password_hash="h", role="super_admin")
|
||
self.session.add(user)
|
||
self.session.flush()
|
||
self.user_id = user.id
|
||
self.account = Account(
|
||
username="事件测试账号",
|
||
password="p",
|
||
email="evt@example.com",
|
||
email_password="ep",
|
||
)
|
||
self.session.add(self.account)
|
||
self.session.commit()
|
||
self.account_id = self.account.id
|
||
|
||
def teardown_method(self):
|
||
self.session.close()
|
||
|
||
def _fresh(self):
|
||
return sessionmaker(bind=self.engine)()
|
||
|
||
|
||
class TestEmit(_Base):
|
||
def test_naive_occurred_at_is_treated_as_utc(self):
|
||
"""⚠️ 回归：naive datetime 必须按「已经是 UTC」处理，不能当本地时间。
|
||
|
||
仓库的既定约定写在 `database.UTCDateTime` 里：
|
||
**写：aware → 转 UTC 去 tzinfo；naive → 视为已是 UTC，原样存**。
|
||
`scheduler._as_utc` / `yyb_service._as_utc` / `audit.py` 都遵守了这条。
|
||
|
||
但 `emit()` 曾经直接 `.astimezone(UTC)` —— 对 naive 值，Python 会把它
|
||
当作**服务器本地时间**再换算。本机是 UTC+8 时，事件时间会被整整扣掉
|
||
8 小时（实测库里 `account_events.occurred_at` 比源 `douyu_tasks.finished_at`
|
||
早 8 小时）。而 `occurred_at` 是「业务发生时间」，错了整条时间线都是错的。
|
||
|
||
naive 值的来源很具体：收集器用**裸 SQL** 读 `douyu_tasks`，绕过了
|
||
`UTCDateTime` 装饰器，拿到的就是 naive。
|
||
|
||
用例显式切到 UTC+8 执行：在 UTC 机器上旧实现与正确实现结果相同，
|
||
不切时区就会假绿。
|
||
"""
|
||
with _force_non_utc_timezone():
|
||
# 故意造 naive 值：本用例就是要验证 naive 的处理
|
||
naive_probe = datetime(2026, 10, 3, 11, 46, 7) # noqa: DTZ001
|
||
assert naive_probe.astimezone(UTC).replace(tzinfo=None) != naive_probe, (
|
||
"时区未生效，本用例失去意义"
|
||
)
|
||
db = self._fresh()
|
||
try:
|
||
# 模拟从 DATETIME 列读出来的裸值（语义是 UTC，但没有 tzinfo）
|
||
naive_utc = datetime(2026, 10, 3, 11, 46, 7) # noqa: DTZ001
|
||
assert naive_utc.tzinfo is None
|
||
ok = svc.emit(
|
||
db,
|
||
account_id=self.account_id,
|
||
event_type=svc.EVENT_RECHARGE,
|
||
title="充值 98",
|
||
amount=98,
|
||
occurred_at=naive_utc,
|
||
source="task",
|
||
source_ref="naive-utc-regression",
|
||
)
|
||
assert ok
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
|
||
db = self._fresh()
|
||
try:
|
||
event = (
|
||
db.query(AccountEvent)
|
||
.filter(AccountEvent.source_ref == "naive-utc-regression")
|
||
.one()
|
||
)
|
||
# 存进去的应当是同一个「墙上时间」，不能被时区换算挪动
|
||
assert event.occurred_at == datetime(2026, 10, 3, 11, 46, 7, tzinfo=UTC)
|
||
finally:
|
||
db.close()
|
||
|
||
def test_aware_occurred_at_is_converted_to_utc(self):
|
||
"""aware 值仍要正常换算（别为了修 naive 把这条路弄坏）。"""
|
||
from datetime import timedelta, timezone
|
||
|
||
db = self._fresh()
|
||
try:
|
||
ok = svc.emit(
|
||
db,
|
||
account_id=self.account_id,
|
||
event_type=svc.EVENT_RECHARGE,
|
||
title="充值 1",
|
||
occurred_at=datetime(
|
||
2026, 10, 3, 19, 46, 7, tzinfo=timezone(timedelta(hours=8))
|
||
),
|
||
source="task",
|
||
source_ref="aware-utc-regression",
|
||
)
|
||
assert ok
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
|
||
db = self._fresh()
|
||
try:
|
||
event = (
|
||
db.query(AccountEvent)
|
||
.filter(AccountEvent.source_ref == "aware-utc-regression")
|
||
.one()
|
||
)
|
||
assert event.occurred_at == datetime(2026, 10, 3, 11, 46, 7, tzinfo=UTC)
|
||
finally:
|
||
db.close()
|
||
|
||
def test_stores_category_subject_and_actor(self):
|
||
"""台账检索列（category / subject_* / actor_*）要真的能写进去。
|
||
|
||
S1 的迁移 0042 建了这 5 列与复合索引 `ix_account_events_ledger`，但 `emit()`
|
||
一直没有暴露它们 —— 列建好了却没有写入方（和 `status` 一样的「空壳」）。
|
||
这条用例锁住「能写」，S3/S5 才能靠索引做「该账号该商品最近一次兑换」这类查询。
|
||
"""
|
||
db = self._fresh()
|
||
try:
|
||
ok = svc.emit(
|
||
db,
|
||
account_id=self.account_id,
|
||
event_type=svc.EVENT_EXCHANGE,
|
||
title="兑换 兔兔白日梦",
|
||
category=svc.CATEGORY_BUSINESS,
|
||
subject_type="commodity",
|
||
subject_key="20260120QYOOB_bag10",
|
||
actor_type="system",
|
||
actor_id=7,
|
||
source_ref="ledger-columns-regression",
|
||
)
|
||
assert ok
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
|
||
db = self._fresh()
|
||
try:
|
||
event = (
|
||
db.query(AccountEvent)
|
||
.filter(AccountEvent.source_ref == "ledger-columns-regression")
|
||
.one()
|
||
)
|
||
assert event.category == "business"
|
||
assert event.subject_type == "commodity"
|
||
assert event.subject_key == "20260120QYOOB_bag10"
|
||
assert event.actor_type == "system"
|
||
assert event.actor_id == 7
|
||
finally:
|
||
db.close()
|
||
|
||
def test_writes_event(self):
|
||
db = self._fresh()
|
||
try:
|
||
ok = svc.emit(
|
||
db,
|
||
account_id=self.account_id,
|
||
event_type=svc.EVENT_EXCHANGE,
|
||
title="兑换 套装-兔兔白日梦",
|
||
amount=-1280,
|
||
balance_after=0,
|
||
detail={"commodity_name": "套装-兔兔白日梦"},
|
||
source="manual",
|
||
)
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
assert ok is True
|
||
|
||
db = self._fresh()
|
||
try:
|
||
row = db.query(AccountEvent).one()
|
||
assert row.account_id == self.account_id
|
||
assert row.event_type == svc.EVENT_EXCHANGE
|
||
assert row.amount == -1280
|
||
assert row.balance_after == 0
|
||
assert row.detail["commodity_name"] == "套装-兔兔白日梦"
|
||
finally:
|
||
db.close()
|
||
|
||
def test_same_source_ref_is_idempotent(self):
|
||
"""同一 source_ref 只能落一条 —— 收集器重跑靠它兜住。"""
|
||
db = self._fresh()
|
||
try:
|
||
for _ in range(3):
|
||
svc.emit(
|
||
db,
|
||
account_id=self.account_id,
|
||
event_type=svc.EVENT_RECHARGE,
|
||
title="充值 98",
|
||
amount=98,
|
||
source="task",
|
||
source_ref="douyu_task:1001",
|
||
)
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
|
||
db = self._fresh()
|
||
try:
|
||
assert db.query(AccountEvent).count() == 1
|
||
finally:
|
||
db.close()
|
||
|
||
def test_multiple_null_source_refs_coexist(self):
|
||
"""没有幂等键的事件（手工记录）不能互相冲突 —— 这正是 source_ref 要可空的原因。"""
|
||
db = self._fresh()
|
||
try:
|
||
for _ in range(3):
|
||
svc.emit(
|
||
db,
|
||
account_id=self.account_id,
|
||
event_type=svc.EVENT_ASSIGNED,
|
||
title="分配客服",
|
||
source="manual",
|
||
)
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
|
||
db = self._fresh()
|
||
try:
|
||
assert db.query(AccountEvent).count() == 3
|
||
finally:
|
||
db.close()
|
||
|
||
def test_failure_is_swallowed(self):
|
||
"""旁路原则：account_id 不存在（外键失败）也不能抛出去。"""
|
||
db = self._fresh()
|
||
try:
|
||
ok = svc.emit(
|
||
db,
|
||
account_id=99999999,
|
||
event_type=svc.EVENT_LOGIN_OK,
|
||
title="登录成功",
|
||
source="login",
|
||
)
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
assert ok is False
|
||
|
||
def test_event_does_not_break_caller_transaction(self):
|
||
"""emit 失败后，调用方在同一事务里已做的写入仍然能提交。
|
||
|
||
注意必须用**同一个 session**：换 session 就测不到「SAVEPOINT 隔离」这件事
|
||
（改动会落在另一个事务里，断言必然失败）。
|
||
"""
|
||
db = self._fresh()
|
||
try:
|
||
account = db.query(Account).filter(Account.id == self.account_id).first()
|
||
account.remark = "改动要保住"
|
||
db.flush()
|
||
# 这条会因外键失败
|
||
svc.emit(
|
||
db,
|
||
account_id=99999999,
|
||
event_type=svc.EVENT_LOGIN_OK,
|
||
title="x",
|
||
)
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
|
||
db = self._fresh()
|
||
try:
|
||
row = db.query(Account).filter(Account.id == self.account_id).first()
|
||
assert row.remark == "改动要保住"
|
||
finally:
|
||
db.close()
|
||
|
||
def test_empty_account_or_type_is_noop(self):
|
||
db = self._fresh()
|
||
try:
|
||
assert svc.emit(db, account_id=0, event_type="x", title="y") is False
|
||
assert (
|
||
svc.emit(db, account_id=self.account_id, event_type="", title="y")
|
||
is False
|
||
)
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
class TestAccountApi(_Base):
|
||
"""账号详情接口：概览统计 + 时间线分页 + 删除账号时清理事件。"""
|
||
|
||
def _admin(self):
|
||
return self.session.query(User).filter(User.username == "evt-admin").first()
|
||
|
||
def _seed(self):
|
||
"""造三条不同类型的事件，供统计与筛选断言。"""
|
||
db = self._fresh()
|
||
try:
|
||
svc.emit(
|
||
db,
|
||
account_id=self.account_id,
|
||
event_type=svc.EVENT_RECHARGE,
|
||
title="充值 98",
|
||
amount=98,
|
||
occurred_at=datetime(2026, 10, 1, 3, 0, tzinfo=UTC),
|
||
source="task",
|
||
source_ref="douyu_task:9001",
|
||
)
|
||
svc.emit(
|
||
db,
|
||
account_id=self.account_id,
|
||
event_type=svc.EVENT_EXCHANGE,
|
||
title="兑换 兔兔白日梦",
|
||
amount=-1280,
|
||
balance_after=0,
|
||
occurred_at=datetime(2026, 10, 2, 3, 0, tzinfo=UTC),
|
||
source="task",
|
||
source_ref="douyu_task:9002",
|
||
)
|
||
svc.emit(
|
||
db,
|
||
account_id=self.account_id,
|
||
event_type=svc.EVENT_ELITE_OPENED,
|
||
title="精英宝典开通",
|
||
occurred_at=datetime(2026, 10, 3, 3, 0, tzinfo=UTC),
|
||
source="manual",
|
||
source_ref="elite_opened:1:1",
|
||
)
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
# ⚠️ MySQL 默认 REPEATABLE READ：self.session 在 setup 阶段就开了事务，
|
||
# 快照早于上面那批写入，不结束事务就查不到刚写的事件（P1b 踩过同一个坑）。
|
||
self.session.commit()
|
||
|
||
def test_overview_aggregates_stats(self):
|
||
from web.backend.routers import accounts as api
|
||
|
||
self._seed()
|
||
data = api.account_overview(
|
||
account_id=self.account_id, db=self.session, current=self._admin()
|
||
)
|
||
assert data["username"] == "事件测试账号"
|
||
assert data["created_at"] is not None, "导入时间必须返回"
|
||
stats = data["stats"]
|
||
assert stats["recharge_total"] == 98
|
||
assert stats["exchange_count"] == 1
|
||
assert stats["points_spent"] == 1280, "消耗要按正数展示给用户"
|
||
assert stats["event_count"] == 3
|
||
# 开通时间由事件表提供（不回填历史，所以只有上线后开通的才有）
|
||
assert data["opened_at"]["elite"] is not None
|
||
|
||
def test_events_paginated_desc(self):
|
||
from web.backend.routers import accounts as api
|
||
|
||
self._seed()
|
||
page = api.account_events(
|
||
account_id=self.account_id,
|
||
page=1,
|
||
page_size=2,
|
||
db=self.session,
|
||
current=self._admin(),
|
||
)
|
||
assert page["total"] == 3
|
||
assert len(page["items"]) == 2
|
||
# 倒序：最新的精英宝典开通在最前
|
||
assert page["items"][0]["event_type"] == svc.EVENT_ELITE_OPENED
|
||
|
||
def test_events_filter_by_type(self):
|
||
from web.backend.routers import accounts as api
|
||
|
||
self._seed()
|
||
page = api.account_events(
|
||
account_id=self.account_id,
|
||
event_type=svc.EVENT_EXCHANGE,
|
||
db=self.session,
|
||
current=self._admin(),
|
||
)
|
||
assert page["total"] == 1
|
||
assert page["items"][0]["title"] == "兑换 兔兔白日梦"
|
||
|
||
def test_events_isolated_between_accounts(self):
|
||
"""事件必须按账号隔离 —— 串号会让客服看到别人的记录。"""
|
||
from web.backend.routers import accounts as api
|
||
|
||
self._seed()
|
||
other = Account(
|
||
username="另一个账号",
|
||
password="p",
|
||
email="o@example.com",
|
||
email_password="ep",
|
||
)
|
||
self.session.add(other)
|
||
self.session.commit()
|
||
db = self._fresh()
|
||
try:
|
||
svc.emit(
|
||
db,
|
||
account_id=other.id,
|
||
event_type=svc.EVENT_RECHARGE,
|
||
title="别人的充值",
|
||
amount=1,
|
||
source="task",
|
||
source_ref="douyu_task:9003",
|
||
)
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
|
||
page = api.account_events(
|
||
account_id=self.account_id, db=self.session, current=self._admin()
|
||
)
|
||
assert all("别人" not in item["title"] for item in page["items"])
|
||
|
||
def test_delete_account_cleans_events(self):
|
||
"""事件表有指向 accounts 的外键：删账号必须连带清事件，否则删不掉。"""
|
||
from web.backend.routers import accounts as api
|
||
|
||
self._seed()
|
||
api.delete_account(
|
||
account_id=self.account_id, db=self.session, current=self._admin()
|
||
)
|
||
|
||
db = self._fresh()
|
||
try:
|
||
left = (
|
||
db.query(AccountEvent)
|
||
.filter(AccountEvent.account_id == self.account_id)
|
||
.count()
|
||
)
|
||
assert left == 0
|
||
assert (
|
||
db.query(Account).filter(Account.id == self.account_id).first() is None
|
||
)
|
||
finally:
|
||
db.close()
|