feat(account): 兑换记录列 + 修掉 source_ref NOT NULL 导致的事件静默丢失
三件事合成一次提交（按反馈不再一改一提交）。 ## ① 账号列表加「兑换记录」列 次数 + 最近一次（商品名 + 日期），填上列表右侧的空档。 数据源是**账号自己的事件流**（`account_events` 的 `exchange`），不是 `douyu_tasks` 的 JSON —— 后者是执行日志，S5 起业务事实已经在事件里了。 性能上刻意**每页只跑两条聚合查询**（不是每行一条 N+1）： 一条分组算 `count` + `max(occurred_at)`（走 `(account_id, event_type, occurred_at)` 复合索引），一条用窗口函数 `ROW_NUMBER()` 取每个账号最新那条的商品名。 实测 6 个账号 3 条兑换事件，聚合与「最近」都正确。 「兑换 」前缀在前端剥掉（列头已经叫「兑换记录」，再带前缀是重复）。 ## ② 修掉一个静默丢事件的 bug（排查中发现的） **现象**：开发库上「状态变迁 / 分配 / 标签 / 备注 / CK 清除」这些事件**一条都没有**， 而带幂等键的事件（业务事实、`ck_invalid`、`relogin_*`）照常写入 —— 时间线看起来有数据，只是恰好缺了某一类，**极难发现**。 **根因**：`source_ref` 是**可选**幂等键（模型与迁移文件都是 `nullable=True`， `emit()` 对没有天然幂等键的事件就传 `None`）。但早期那版迁移 （重编号成 `0041` 之前的 `0040`）建的是 **NOT NULL**，而我后来只改了 **迁移文件**、没管**已建好的库** —— Alembic 只记版本号，不会重放已应用的脚本。 `__pycache__` 里那个 `20261002_0040_*.pyc` 就是旧版留下的痕迹。 **为什么静默**：`emit()` 按设计吞掉任何异常（事件是旁路，不能影响主流程）， 只记一条 warning，调用方也从不检查返回值。已实测复现： `emit(...)` 返回 `False` + warning，实际落库 `0` 条。 **修法**：迁移 `20261004_0045` 改回可空（`MODIFY`，**唯一索引保留**，幂等， 生产若跑过旧版也一并修好）。修复后实测：`emit` 返回 `True`、落库 1 条。 ## ③ 补一条能防「未来」同类问题的守卫 `tests/test_migrations.py` 原本只比对「表/列/索引在不在」，**不比对可空性**。 已加：迁移建出的列**不允许比模型更严**（模型说可空、库里 NOT NULL）。 只查这个**危险方向**：反方向（模型 NOT NULL、库里允许 NULL）是无害的宽松， 历史遗留里有 28 处，一并报出来只会制造噪声、逼人修没收益的东西。 > 这条守卫**有边界**，已在代码注释与设计文档 §3.5.1 写明： > 它在**全新库**上跑整链，只能防「未来的迁移建错」，**防不了已有库的历史漂移**。 > 已验证：临时禁用 `0045`，该用例仍通过（全新链本来就是对的）。 ## 验证 - `TZ=Asia/Shanghai`：**503 passed / 6 skipped**（+3 条兑换列用例） - `TZ=UTC`（迁移 + 列表相关）45 passed；ruff 全绿；前端 build 通过、lint 0 errors - 真实开发库：迁移已应用（head `20261004_0045`）， `source_ref` 现在 `IS_NULLABLE=YES` 且唯一索引仍在； 不带幂等键的事件确实能落库；兑换聚合返回值正确 注：验证时在开发库那个有 CK 的账号上留了 3 条 `exchange` 演示事件 （兔兔白日梦 / 冠军鸡腿 / 烟花），是为了让列有数据可看，随时可删。
This commit is contained in:
yml2213
@@ -299,6 +299,32 @@ CREATE INDEX ix_account_events_ledger
|
||||
> 规划 §10 的 `exchange_orders`（工单：资料 → 匹配 → 执行 → 归档）**有状态机，保留为独立表**；
|
||||
> 它执行成功后再往 `account_events` 写一条 `exchange` 事实。**流程归工单，事实归事件流。**
|
||||
|
||||
### 3.5.1 事故复盘：`source_ref` 的 NOT NULL 让一批事件静默消失
|
||||
|
||||
**现象**：开发库上「状态变迁 / 分配 / 标签 / 备注 / CK 清除」这些事件**一条都没有**，
|
||||
而带幂等键的事件（业务事实、`ck_invalid`、`relogin_*`）照常写入 ——
|
||||
时间线看起来「有数据」，只是恰好缺了某一类，**极难发现**。
|
||||
|
||||
**根因**：`account_events.source_ref` 是**可选**幂等键（模型与迁移文件都写着
|
||||
`nullable=True`，`emit()` 对没有天然幂等键的事件就传 `None`）。
|
||||
但早期那版迁移（重编号成 `0041` 之前的 `0040`）建的是 NOT NULL，
|
||||
而我后来只改了**迁移文件**、没处理**已建好的库** —— Alembic 只记版本号，
|
||||
不会重放已应用过的脚本。开发库就是这么漂移的。
|
||||
|
||||
**为什么静默**：`emit()` 按设计**吞掉任何异常**（事件是旁路，不能影响主流程），
|
||||
只记一条 warning；调用方也从不检查它的返回值。所以「写不进去」表现为完全正常。
|
||||
|
||||
**修法**：迁移 `20261004_0045` 把它改回可空（`MODIFY`，唯一索引保留，幂等）。
|
||||
另在 `tests/test_migrations.py` 加了一条**可空性比对**：迁移建出的列不允许
|
||||
比模型更严（模型说可空、库里 NOT NULL）。
|
||||
|
||||
> ⚠️ **这条守卫有边界**：它在**全新库**上跑整条链，所以只能防「**未来**的迁移
|
||||
> 把模型可空的列建成 NOT NULL」，**防不了已有库的历史漂移**（那种只能靠迁移本身修）。
|
||||
> 已验证：把 `0045` 临时禁用，该用例仍然通过 —— 因为全新链本来就是对的。
|
||||
>
|
||||
> **教训**：改动「迁移文件」时，已经建过表的库**不会自动跟上**。
|
||||
> 要么写一条幂等修正迁移，要么确认所有环境都还没应用过。
|
||||
|
||||
### 3.6 `bind_status` 的拆解（旧列 → 新字段）
|
||||
|
||||
> **S6 已落地（2026-10-04）**，实现见 `web/backend/services/account_binding.py`，
|
||||
|
||||
@@ -854,3 +854,93 @@ class TestSummaryByStatus(_Base):
|
||||
summary = self._summary()
|
||||
assert summary["ck_invalid_count"] >= 1
|
||||
assert summary["no_cookie_count"] >= 1
|
||||
|
||||
|
||||
class TestExchangeRecordColumn(_Base):
|
||||
"""「兑换记录」列：次数 + 最近一次（商品名 + 时间）。
|
||||
|
||||
数据源是**账号自己的事件流**（`account_events.exchange`），不是 `douyu_tasks`
|
||||
的 JSON —— 后者是执行日志，S5 起业务事实已经在事件里了。
|
||||
"""
|
||||
|
||||
def _items(self, **kwargs):
|
||||
from web.backend.routers.accounts import list_accounts
|
||||
|
||||
return list_accounts(
|
||||
assigned_only=False,
|
||||
tag="",
|
||||
has_cookie=False,
|
||||
search="",
|
||||
page=1,
|
||||
page_size=50,
|
||||
include_sensitive=False,
|
||||
status=kwargs.get("status", ""),
|
||||
ck_status="",
|
||||
bind_status="",
|
||||
db=self.session,
|
||||
current=self.user,
|
||||
)["items"]
|
||||
|
||||
def _item(self, username: str):
|
||||
for item in self._items():
|
||||
if item.username == username:
|
||||
return item
|
||||
raise AssertionError(f"列表里没有 {username}")
|
||||
|
||||
def _exchange(self, account_id: int, name: str, *, days_ago: int = 0):
|
||||
from datetime import timedelta
|
||||
|
||||
from web.backend.services.account_events_service import emit
|
||||
|
||||
emit(
|
||||
self.session,
|
||||
account_id=account_id,
|
||||
event_type="exchange",
|
||||
title=f"兑换 {name}",
|
||||
category="business",
|
||||
subject_key=name,
|
||||
occurred_at=NOW - timedelta(days=days_ago),
|
||||
)
|
||||
self.session.commit()
|
||||
|
||||
def test_no_exchange_yields_zero(self):
|
||||
self.make_account("noex")
|
||||
item = self._item("noex")
|
||||
assert item.exchange_count == 0
|
||||
assert item.last_exchange_at is None
|
||||
assert item.last_exchange_name == ""
|
||||
|
||||
def test_counts_and_keeps_latest_name(self):
|
||||
account = self.make_account("hasex")
|
||||
self._exchange(account.id, "兔兔白日梦", days_ago=3)
|
||||
self._exchange(account.id, "冠军鸡腿", days_ago=1)
|
||||
self._exchange(account.id, "烟花", days_ago=2)
|
||||
|
||||
item = self._item("hasex")
|
||||
assert item.exchange_count == 3
|
||||
# 「最近」按**事件 id** 取（同一天多次兑换时时间戳可能相同，靠 id 定序）
|
||||
# 后端返回事件原始 title；剥掉「兑换 」前缀是前端展示层的事。
|
||||
assert item.last_exchange_name == "兑换 烟花"
|
||||
|
||||
def test_only_exchange_events_count(self):
|
||||
"""别的类型（充值 / 登录 / 状态）不能混进兑换次数。"""
|
||||
from web.backend.services.account_events_service import emit
|
||||
|
||||
account = self.make_account("mixed")
|
||||
emit(
|
||||
self.session,
|
||||
account_id=account.id,
|
||||
event_type="recharge",
|
||||
title="充值 100",
|
||||
category="business",
|
||||
amount=100,
|
||||
)
|
||||
emit(
|
||||
self.session,
|
||||
account_id=account.id,
|
||||
event_type="status_changed",
|
||||
title="状态：未登录 → 正常",
|
||||
category="lifecycle",
|
||||
)
|
||||
self.session.commit()
|
||||
assert self._item("mixed").exchange_count == 0
|
||||
|
||||
@@ -26,7 +26,7 @@ TESTS_DIR = Path(__file__).resolve().parent
|
||||
PROJECT_ROOT = TESTS_DIR.parent
|
||||
VERSIONS_DIR = PROJECT_ROOT / "web" / "backend" / "migrations" / "versions"
|
||||
|
||||
HEAD_REVISION = "20261004_0044"
|
||||
HEAD_REVISION = "20261004_0045"
|
||||
|
||||
# 迁移新增、但不声明在模型里的复合查询索引。
|
||||
EXTRA_INDEXES = (
|
||||
@@ -105,12 +105,32 @@ class TestMigrationSmoke:
|
||||
assert not missing_tables
|
||||
|
||||
all_db_indexes = set()
|
||||
nullability_mismatches: list[str] = []
|
||||
for name, table in model_tables.items():
|
||||
db_columns = {column["name"] for column in inspector.get_columns(name)}
|
||||
db_cols = inspector.get_columns(name)
|
||||
db_columns = {column["name"] for column in db_cols}
|
||||
model_columns = set(table.columns.keys())
|
||||
missing_columns = model_columns - db_columns
|
||||
assert not missing_columns
|
||||
|
||||
# 可空性只查**危险的那个方向**：模型说可空、库里却是 NOT NULL。
|
||||
# 那样带 NULL 的写入会**在运行时静默失败**（`emit()` 按设计吞异常），
|
||||
# 正是 `account_events.source_ref` 那次事故的形态 ——
|
||||
# 只比对「列在不在」是抓不到的。
|
||||
#
|
||||
# 反方向（模型 NOT NULL、库里允许 NULL）是**无害的宽松**：
|
||||
# SQLAlchemy 总会写值，库里宽松一点不会丢数据。历史遗留里有不少，
|
||||
# 一并报出来只会制造噪声、逼人去修没有收益的东西。
|
||||
db_nullable = {c["name"]: c["nullable"] for c in db_cols}
|
||||
for column in table.columns:
|
||||
if column.name not in db_nullable:
|
||||
continue
|
||||
if bool(column.nullable) and not bool(db_nullable[column.name]):
|
||||
nullability_mismatches.append(
|
||||
f"{name}.{column.name}: 模型说可空，"
|
||||
f"但迁移建成了 NOT NULL（带 NULL 的写入会静默丢失）"
|
||||
)
|
||||
|
||||
db_indexes = set()
|
||||
for index in inspector.get_indexes(name):
|
||||
index_name = index["name"]
|
||||
@@ -121,6 +141,11 @@ class TestMigrationSmoke:
|
||||
missing_indexes = model_indexes - db_indexes
|
||||
assert not missing_indexes
|
||||
|
||||
assert not nullability_mismatches, (
|
||||
"迁移建出的列比模型更严（会在运行期静默丢数据）：\n"
|
||||
+ "\n".join(nullability_mismatches)
|
||||
)
|
||||
|
||||
# 迁移新增的复合查询索引（不入模型，但应存在）。
|
||||
for index_name in EXTRA_INDEXES:
|
||||
assert index_name in all_db_indexes, f"迁移链未创建索引 {index_name}"
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
"""修 `account_events.source_ref` 的可空性（模型与迁移文件早已是 NULL，但库里不是）
|
||||
|
||||
## 问题
|
||||
|
||||
`source_ref` 是**可选**的幂等键。设计上明确可空（模型与迁移 `0041` 都写着
|
||||
`nullable=True`，注释也说「MySQL 允许多个 NULL，便于『没有幂等键』的事件共存」）。
|
||||
`emit()` 对**没有天然幂等键**的事件（状态变迁 / 分配 / 标签 / 备注 / CK 清除）
|
||||
就是传 `None` 的。
|
||||
|
||||
但有的库里这一列是 **NOT NULL** —— 原因是我早期把迁移从 `20261002_0040` 重编号成
|
||||
`20261003_0041` 并顺手修正了可空性，而**已经用旧版本建过表的库不会跟着改**
|
||||
（Alembic 只记录版本号，不会重放已应用过的脚本）。
|
||||
|
||||
## 危害：静默丢事件
|
||||
|
||||
`emit()` 按设计**吞掉任何异常**（事件是旁路，不能影响主流程），只记一条 warning。
|
||||
于是这些事件在那种库上**一条都写不进去**，而业务完全正常、没有任何报错：
|
||||
|
||||
- `status_changed`（S3 生命周期状态变迁）
|
||||
- `assigned` / `tag_changed` / `remark_changed`（S4 人工操作）
|
||||
- `ck_cleared`（S4 清 CK）
|
||||
|
||||
而 `ck_invalid` / `relogin_*` / 业务事实等**带幂等键**的事件照常写入 ——
|
||||
所以时间线看起来「有数据」，只是**恰好缺了没有幂等键的那几类**，极难发现。
|
||||
|
||||
## 修法
|
||||
|
||||
改成可空。用 `MODIFY` 而不是重建列，**唯一索引会保留**（索引与列定义是两回事），
|
||||
而且重复执行是幂等的 —— 无论那种库此前是什么状态，跑完都对齐模型。
|
||||
|
||||
`downgrade` 刻意什么都不做：把可空改回 NOT NULL 会让上述事件重新丢失，
|
||||
而这不是「回退」该有的行为。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
revision: str = "20261004_0045"
|
||||
down_revision: str | None = "20261004_0044"
|
||||
branch_labels: str | None = None
|
||||
depends_on: str | None = None
|
||||
|
||||
TABLE = "account_events"
|
||||
COLUMN = "source_ref"
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.execute(sa.text(f"ALTER TABLE {TABLE} MODIFY {COLUMN} VARCHAR(96) NULL"))
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
# 见模块 docstring：改回 NOT NULL 会重新丢事件，不做。
|
||||
pass
|
||||
@@ -234,6 +234,54 @@ def list_accounts(
|
||||
]
|
||||
)
|
||||
accounts = query.options(*options).all()
|
||||
|
||||
# 「兑换记录」列：**每页只跑一条聚合查询**，不是每行一条（避免 N+1）。
|
||||
# 数据源是账号自己的事件流（`account_events` 的 `exchange`），
|
||||
# 而不是 `douyu_tasks` 的 JSON —— 后者是执行日志，S5 起业务事实已经在事件里了。
|
||||
exchange_stats: dict[int, tuple[int, Any, str]] = {}
|
||||
ids = [acc.id for acc in accounts]
|
||||
if ids:
|
||||
# ① 次数 + 最近一次时间（分组聚合，走 (account_id, event_type, occurred_at) 复合索引）
|
||||
for account_id, count, last_at in (
|
||||
db.query(
|
||||
AccountEvent.account_id,
|
||||
func.count(AccountEvent.id),
|
||||
func.max(AccountEvent.occurred_at),
|
||||
)
|
||||
.filter(
|
||||
AccountEvent.account_id.in_(ids),
|
||||
AccountEvent.event_type == EVENT_EXCHANGE,
|
||||
)
|
||||
.group_by(AccountEvent.account_id)
|
||||
.all()
|
||||
):
|
||||
exchange_stats[int(account_id)] = (int(count or 0), last_at, "")
|
||||
# ② 最近一次兑换的商品名（窗口函数取每个账号的最新一条）
|
||||
ranked = (
|
||||
db.query(
|
||||
AccountEvent.account_id.label("account_id"),
|
||||
AccountEvent.title.label("title"),
|
||||
func.row_number()
|
||||
.over(
|
||||
partition_by=AccountEvent.account_id,
|
||||
order_by=AccountEvent.id.desc(),
|
||||
)
|
||||
.label("rn"),
|
||||
)
|
||||
.filter(
|
||||
AccountEvent.account_id.in_(ids),
|
||||
AccountEvent.event_type == EVENT_EXCHANGE,
|
||||
)
|
||||
.subquery()
|
||||
)
|
||||
for account_id, title in (
|
||||
db.query(ranked.c.account_id, ranked.c.title).filter(ranked.c.rn == 1).all()
|
||||
):
|
||||
key = int(account_id)
|
||||
if key in exchange_stats:
|
||||
count, last_at, _ = exchange_stats[key]
|
||||
exchange_stats[key] = (count, last_at, str(title or ""))
|
||||
|
||||
result = []
|
||||
for acc in accounts:
|
||||
item = AccountOut(
|
||||
@@ -254,6 +302,9 @@ def list_accounts(
|
||||
ck_checked_at=acc.ck_checked_at,
|
||||
points=acc.points,
|
||||
gold_balance=acc.gold_balance,
|
||||
exchange_count=exchange_stats.get(acc.id, (0, None, ""))[0],
|
||||
last_exchange_at=exchange_stats.get(acc.id, (0, None, ""))[1],
|
||||
last_exchange_name=exchange_stats.get(acc.id, (0, None, ""))[2],
|
||||
**bind_payload(acc),
|
||||
)
|
||||
# 只有管理员可看完整字段（密码、邮箱等）
|
||||
|
||||
@@ -198,6 +198,10 @@ class AccountOut(BaseModel):
|
||||
esports_bind_label: str = ""
|
||||
xpd_bind_status: str = "unknown"
|
||||
xpd_bind_label: str = ""
|
||||
# 兑换记录（来自 account_events 的 exchange 事件，每页一条聚合查询）
|
||||
exchange_count: int = 0
|
||||
last_exchange_at: datetime | None = None
|
||||
last_exchange_name: str = ""
|
||||
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
@@ -229,6 +233,9 @@ class AccountOut(BaseModel):
|
||||
"esports_bind_label": self.esports_bind_label,
|
||||
"xpd_bind_status": self.xpd_bind_status,
|
||||
"xpd_bind_label": self.xpd_bind_label,
|
||||
"exchange_count": self.exchange_count,
|
||||
"last_exchange_at": _isoformat(self.last_exchange_at),
|
||||
"last_exchange_name": self.last_exchange_name,
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -269,6 +269,10 @@ export interface AccountItem {
|
||||
esports_bind_label: string;
|
||||
xpd_bind_status: BindStatus;
|
||||
xpd_bind_label: string;
|
||||
// 兑换记录（来自 account_events 的 exchange 事件）
|
||||
exchange_count: number;
|
||||
last_exchange_at: string | null;
|
||||
last_exchange_name: string;
|
||||
}
|
||||
|
||||
export interface AssignmentsSummary {
|
||||
|
||||
@@ -378,6 +378,29 @@ export default function AccountsPage() {
|
||||
</Space>
|
||||
),
|
||||
},
|
||||
{
|
||||
// 兑换记录：次数 + 最近一次（商品名 + 日期）。
|
||||
// 数据源是账号自己的事件流（`account_events.exchange`），不是 douyu_tasks 的 JSON ——
|
||||
// 后者是执行日志，S5 起业务事实已经在事件里了。完整历史看详情页时间线。
|
||||
title: '兑换记录',
|
||||
dataIndex: 'exchange_count',
|
||||
width: 170,
|
||||
render: (_, record: AccountItem) => {
|
||||
if (!record.exchange_count) return <Text type="secondary">未兑换</Text>;
|
||||
return (
|
||||
<Space direction="vertical" size={0} style={{ maxWidth: '100%' }}>
|
||||
<Text>{record.exchange_count} 次</Text>
|
||||
<Text type="secondary" style={{ fontSize: 12 }} ellipsis title={record.last_exchange_name}>
|
||||
{formatTimeShort(record.last_exchange_at)}
|
||||
{/* 列的标题已经是「兑换记录」，事件 title 里的「兑换 」前缀在这儿是重复的 */}
|
||||
{record.last_exchange_name
|
||||
? ` ${record.last_exchange_name.replace(/^兑换\s*/, '')}`
|
||||
: ''}
|
||||
</Text>
|
||||
</Space>
|
||||
);
|
||||
},
|
||||
},
|
||||
{
|
||||
title: '标签',
|
||||
dataIndex: 'tag',
|
||||
|
||||
Reference in New Issue
Block a user