feat(wallet): 金币记录按 trace_id 聚合比价/领券看广告金币 (#225)
## 背景 「金币变动记录」里看广告金币按**每条广告**统计，一次比价/领券的等候期会连续看多条信息流广告，于是列表刷出一长串「比价奖励 +x」「领券奖励 +x」，淹没其它记录、观感差。 ## 改动 把**一次比价 / 一次领券**连续看广告获得的多条金币，按 `trace_id` **聚合成一条**展示（金额取该次会话合计）。 **纯后端**：比价、领券的结算上报都已把本场 `trace_id` 传到 `/feed-reward`（领券自 2026-07-15、客户端 `a98cab8` 起），无需 Android 改动即可生效。 ### 具体 - `coin_transaction` 新增 `trace_id` 列 + 索引；`grant_coins` 增 `trace_id` 参数；`grant_feed_reward` 发奖时写入本场 trace_id。 - `GET /api/v1/wallet/coin-transactions` 改为**按 trace_id 分组的游标分页**（CTE 聚合），响应新增 `merged_count`（合并条数，未合并=1）。 - Alembic 迁移：加列 + 索引 + **回填历史**（从 `ad_feed_reward_record` 按 `ref_id == client_event_id` 补 trace_id）。 - 更正 `ad_feed_reward_record.trace_id` 过时注释（领券自 2026-07-15 也带）。 - **只改 App 用户接口**；admin 审计接口（`list_all_coin_transactions`）保持每条一行不动。 ## 关键设计 - **分组键**：`biz_type ∈ (feed_ad_reward_comparison, feed_ad_reward_coupon)` 且 `trace_id` 非空 → 按 `trace_id` 合并；其余（激励视频/签到/通用信息流/空 trace 等）每条一行。 - **代表行**取组内 `MAX(id)`（供余额/时间/标题）；`amount = SUM`；游标 `rep_id < cursor` 施于**外层**查询——CTE 在用户全量行上分组，**不**按 `id<cursor` 裁剪，避免会话行跨游标时产生「残组」重复（有专门回归测试锁死）。 - 返回 `CoinLedgerRow` dataclass（非 ORM，杜绝把聚合后的 amount 误写回底层流水）。 - SQL 全程 **SQLite/PG 双方言通用**（dev/测试 SQLite，prod PG）。 --------- Co-authored-by: guke <guke@autohome.com.cn> Reviewed-on: #225
This commit was merged in pull request #225.
This commit is contained in:
@@ -0,0 +1,60 @@
|
||||
"""coin_transaction.trace_id (金币记录按会话聚合比价/领券看广告金币)
|
||||
|
||||
Revision ID: coin_transaction_trace_id
|
||||
Revises: savings_record_trace_id
|
||||
Create Date: 2026-08-07
|
||||
|
||||
比价/领券信息流发奖时把本场 trace_id 一并写入 coin_transaction;金币变动记录接口按
|
||||
trace_id 把一次比价/领券的多条广告金币聚合成一条。历史行从 ad_feed_reward_record
|
||||
回填(ref_id == client_event_id)。
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = 'coin_transaction_trace_id'
|
||||
down_revision: Union[str, Sequence[str], None] = 'savings_record_trace_id'
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
# SQLite 下 ADD COLUMN(可空) 与 CREATE INDEX 均原生支持,无需 batch_alter_table。
|
||||
op.add_column(
|
||||
'coin_transaction',
|
||||
sa.Column('trace_id', sa.String(length=64), nullable=True),
|
||||
)
|
||||
op.create_index(
|
||||
op.f('ix_coin_transaction_trace_id'),
|
||||
'coin_transaction',
|
||||
['trace_id'],
|
||||
unique=False,
|
||||
)
|
||||
# 历史回填:从 ad_feed_reward_record 按 ref_id==client_event_id 补 trace_id。仅比价/领券两类、
|
||||
# 仅当前为空、且广告行确有 trace_id 时补(EXISTS 守护);`trace_id IS NULL` 保证重跑幂等。
|
||||
# 相关子查询 SQLite/PG 通用。两类 biz_type 在应用侧为 rewards.FEED_AD_SESSION_BIZ_TYPES,
|
||||
# 此处按「迁移不可变」原则硬编码历史快照(勿改为 import 应用常量)。
|
||||
# 大表(prod PG)如需可改按 id 区间分批;此处一次性 UPDATE。
|
||||
op.execute(
|
||||
"""
|
||||
UPDATE coin_transaction SET trace_id = (
|
||||
SELECT r.trace_id FROM ad_feed_reward_record r
|
||||
WHERE r.client_event_id = coin_transaction.ref_id)
|
||||
WHERE biz_type IN ('feed_ad_reward_comparison', 'feed_ad_reward_coupon')
|
||||
AND trace_id IS NULL
|
||||
AND EXISTS (SELECT 1 FROM ad_feed_reward_record r2
|
||||
WHERE r2.client_event_id = coin_transaction.ref_id
|
||||
AND r2.trace_id IS NOT NULL)
|
||||
"""
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_index(
|
||||
op.f('ix_coin_transaction_trace_id'),
|
||||
table_name='coin_transaction',
|
||||
)
|
||||
op.drop_column('coin_transaction', 'trace_id')
|
||||
@@ -26,6 +26,16 @@ SIGNIN_REWARDS: tuple[int, ...] = (
|
||||
SIGNIN_CYCLE_LEN: int = len(SIGNIN_REWARDS)
|
||||
|
||||
|
||||
# ===== 信息流广告「按会话聚合」的 biz_type =====
|
||||
# 比价 / 领券等候期看的信息流广告,每条各写一条 coin_transaction;这两类在「金币变动记录」
|
||||
# 里按 trace_id 聚合成一条展示(见 repositories/wallet.list_coin_transactions)。其余类型
|
||||
# (reward_video / guide_video / 通用 feed_ad_reward / signin / task_* 等)不聚合。
|
||||
FEED_AD_SESSION_BIZ_TYPES: tuple[str, str] = (
|
||||
"feed_ad_reward_comparison",
|
||||
"feed_ad_reward_coupon",
|
||||
)
|
||||
|
||||
|
||||
def signin_reward(cycle_day: int) -> int:
|
||||
"""cycle_day 取值 1..SIGNIN_CYCLE_LEN。"""
|
||||
return SIGNIN_REWARDS[cycle_day - 1]
|
||||
|
||||
@@ -36,8 +36,8 @@ class AdFeedRewardRecord(Base):
|
||||
# 点位场景:comparison(比价等待) / coupon(领券) / welfare(福利页)。比价与领券共用同一信息流
|
||||
# 代码位,slot_id/our_code_id 分不出,只能客户端各调用点显式打标;NULL=历史/未升级客户端=未分类。
|
||||
feed_scene: Mapped[str | None] = mapped_column(String(16), nullable=True)
|
||||
# 本次比价 trace_id(仅 comparison 场景由客户端带上):把这场广告金币归属到对应比价记录,
|
||||
# 比价记录页按 trace_id 聚合本场实发金币显示「比价赚 N 金币」。领券/福利/旧客户端 = NULL。
|
||||
# 本次会话 trace_id:比价一直带;领券自 2026-07-15(客户端 a98cab8)起也带;福利/旧客户端 = NULL。
|
||||
# 用途:比价记录页按 trace_id 聚合「比价赚 N 金币」;金币记录列表把一次比价/领券的多条广告金币聚合成一条。
|
||||
trace_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
|
||||
# 来源(广告收益报表用):我们的应用环境 prod/test + 我们配置的代码位 104xxx,由客户端 feed-reward 上报带上。旧数据为 NULL。
|
||||
app_env: Mapped[str | None] = mapped_column(String(16), nullable=True)
|
||||
|
||||
@@ -84,6 +84,9 @@ class CoinTransaction(Base):
|
||||
# 关联业务 id(签到日期、任务 key 等),可空
|
||||
ref_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
||||
remark: Mapped[str | None] = mapped_column(String(128), nullable=True)
|
||||
# 会话键:仅比价/领券信息流发奖(feed_ad_reward_comparison/coupon)时写入本场 trace_id,
|
||||
# 金币记录列表据此把一次比价/领券的多条广告金币聚合成一条。其余类型 = NULL。
|
||||
trace_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
|
||||
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
|
||||
|
||||
@@ -219,7 +219,7 @@ def grant_feed_reward(
|
||||
crud_wallet.grant_coins(
|
||||
db, user_id, coin,
|
||||
biz_type=reward_biz, ref_id=client_event_id,
|
||||
remark=reward_remark,
|
||||
remark=reward_remark, trace_id=trace_id,
|
||||
)
|
||||
rec = AdFeedRewardRecord(
|
||||
client_event_id=client_event_id,
|
||||
|
||||
+83
-10
@@ -10,9 +10,10 @@ import logging
|
||||
import re
|
||||
import unicodedata
|
||||
import uuid
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from sqlalchemy import func, select, update
|
||||
from sqlalchemy import String, and_, case, cast, func, literal, select, update
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
@@ -170,6 +171,7 @@ def grant_coins(
|
||||
biz_type: str,
|
||||
ref_id: str | None = None,
|
||||
remark: str | None = None,
|
||||
trace_id: str | None = None,
|
||||
) -> tuple[CoinAccount, CoinTransaction]:
|
||||
"""金币变动入口(正数入账 / 负数出账)。更新余额 + 写流水,不 commit。
|
||||
|
||||
@@ -189,6 +191,7 @@ def grant_coins(
|
||||
biz_type=biz_type,
|
||||
ref_id=ref_id,
|
||||
remark=remark,
|
||||
trace_id=trace_id,
|
||||
created_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None), # 存北京 wall-clock(客户端原样切片显示)
|
||||
)
|
||||
db.add(txn)
|
||||
@@ -257,24 +260,94 @@ def grant_invite_cash(
|
||||
return acc, txn
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CoinLedgerRow:
|
||||
"""`list_coin_transactions` 的返回行。广告类按 trace_id 聚合后的展示行,
|
||||
**非 ORM 对象**(避免把聚合后的 amount 误写回底层流水)。"""
|
||||
id: int
|
||||
amount: int
|
||||
balance_after: int
|
||||
biz_type: str
|
||||
ref_id: str | None
|
||||
remark: str | None
|
||||
created_at: datetime
|
||||
merged_count: int
|
||||
|
||||
|
||||
def list_coin_transactions(
|
||||
db: Session,
|
||||
user_id: int,
|
||||
*,
|
||||
limit: int = 20,
|
||||
cursor: int | None = None,
|
||||
) -> tuple[list[CoinTransaction], int | None]:
|
||||
"""金币流水分页(按 id 倒序,游标式)。
|
||||
) -> tuple[list[CoinLedgerRow], int | None]:
|
||||
"""金币流水分页(游标式,按 id 倒序)。
|
||||
|
||||
cursor 为上一页最后一条的 id;返回 (本页列表, next_cursor)。
|
||||
next_cursor 为 None 表示没有下一页。
|
||||
比价/领券两类信息流广告(rewards.FEED_AD_SESSION_BIZ_TYPES)且带 trace_id 的行,
|
||||
按 trace_id 聚合成一条(一次比价/领券 = 一行):金额合计、代表行取组内最新一条
|
||||
(MAX(id))的余额/时间、merged_count=组内条数。其余每条一行。
|
||||
|
||||
分组必须在该用户**全量**行上算真实 rep_id 后再按 rep_id 过滤——**不可**把 CTE 输入
|
||||
裁成 id<cursor,否则会话行交错跨游标时会算出与上页重复的「残组」。
|
||||
|
||||
cursor 为上一页最后一条的 id(即其组 rep_id);返回 (本页列表, next_cursor)。
|
||||
"""
|
||||
stmt = select(CoinTransaction).where(CoinTransaction.user_id == user_id)
|
||||
if cursor is not None:
|
||||
stmt = stmt.where(CoinTransaction.id < cursor)
|
||||
stmt = stmt.order_by(CoinTransaction.id.desc()).limit(limit)
|
||||
ct = CoinTransaction
|
||||
group_key = case(
|
||||
(
|
||||
and_(
|
||||
ct.biz_type.in_(rewards.FEED_AD_SESSION_BIZ_TYPES),
|
||||
ct.trace_id.is_not(None),
|
||||
),
|
||||
literal("T:") + ct.trace_id,
|
||||
),
|
||||
else_=literal("I:") + cast(ct.id, String),
|
||||
).label("group_key")
|
||||
|
||||
items = list(db.execute(stmt).scalars().all())
|
||||
grp = (
|
||||
select(
|
||||
group_key,
|
||||
func.max(ct.id).label("rep_id"),
|
||||
func.sum(ct.amount).label("total_amount"),
|
||||
func.count().label("merged_count"),
|
||||
)
|
||||
.where(ct.user_id == user_id)
|
||||
.group_by(group_key)
|
||||
.cte("grp")
|
||||
)
|
||||
|
||||
stmt = (
|
||||
select(
|
||||
ct.id,
|
||||
grp.c.total_amount.label("amount"),
|
||||
ct.balance_after,
|
||||
ct.biz_type,
|
||||
ct.ref_id,
|
||||
ct.remark,
|
||||
ct.created_at,
|
||||
grp.c.merged_count,
|
||||
)
|
||||
.select_from(grp)
|
||||
.join(ct, ct.id == grp.c.rep_id)
|
||||
)
|
||||
if cursor is not None:
|
||||
stmt = stmt.where(grp.c.rep_id < cursor)
|
||||
stmt = stmt.order_by(grp.c.rep_id.desc()).limit(limit)
|
||||
|
||||
rows = db.execute(stmt).all()
|
||||
items = [
|
||||
CoinLedgerRow(
|
||||
id=r.id,
|
||||
amount=int(r.amount),
|
||||
balance_after=r.balance_after,
|
||||
biz_type=r.biz_type,
|
||||
ref_id=r.ref_id,
|
||||
remark=r.remark,
|
||||
created_at=r.created_at,
|
||||
merged_count=int(r.merged_count),
|
||||
)
|
||||
for r in rows
|
||||
]
|
||||
next_cursor = items[-1].id if len(items) == limit else None
|
||||
return items, next_cursor
|
||||
|
||||
|
||||
@@ -30,6 +30,7 @@ class CoinTransactionOut(BaseModel):
|
||||
ref_id: str | None = None
|
||||
remark: str | None = None
|
||||
created_at: datetime
|
||||
merged_count: int = Field(1, description="本行合并的底层流水条数(比价/领券按会话聚合;未合并=1)")
|
||||
|
||||
|
||||
class CoinTransactionPage(BaseModel):
|
||||
|
||||
@@ -0,0 +1,761 @@
|
||||
# 金币记录按会话汇总比价/领券看广告金币 Implementation Plan
|
||||
|
||||
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||||
|
||||
**Goal:** App「金币变动记录」里，一次比价 / 一次领券连续看广告获得的多条金币，按 `trace_id` 聚合成一条展示。
|
||||
|
||||
**Architecture:** 纯后端。给 `coin_transaction` 加 `trace_id` 列并在发奖时写入；`GET /api/v1/wallet/coin-transactions` 改为按 `trace_id` 分组的游标分页（比价/领券两类 feed 广告按会话合并，其余每条一行）；历史行从 `ad_feed_reward_record` 回填。领券自 2026-07-15 起结算已带 trace_id，无需改 Android。仅动 App 用户接口，admin 审计接口不变。
|
||||
|
||||
**Tech Stack:** FastAPI · SQLAlchemy 2.0 · Alembic · Pydantic v2 · pytest（SQLite）。
|
||||
|
||||
**Spec:** `docs/superpowers/specs/2026-08-07-coin-ledger-aggregate-ad-rewards-design.md`
|
||||
|
||||
**约定：** dev / pytest 均 **SQLite**，prod PostgreSQL——所有 SQL 保持 SQLite/PG 通用（不用 PG 专有语法）。金额单位为金币（非分）。
|
||||
|
||||
---
|
||||
|
||||
## File Structure
|
||||
|
||||
| 文件 | 改动 | 职责 |
|
||||
|---|---|---|
|
||||
| `app/core/rewards.py` | 加常量 `FEED_AD_SESSION_BIZ_TYPES` | 「按会话聚合」的两类 biz_type 单一来源（查询 + 写入 + 测试共用） |
|
||||
| `app/models/wallet.py` | `CoinTransaction` 加 `trace_id` 列 | 金币流水会话键 |
|
||||
| `app/repositories/wallet.py` | `grant_coins` 加 `trace_id` 参；`list_coin_transactions` 改分组分页；加 `CoinLedgerRow` | 写入透传 + 聚合读取 |
|
||||
| `app/repositories/ad_feed_reward.py` | `grant_feed_reward` 调 `grant_coins` 时传 `trace_id` | 把本场 trace_id 写进金币流水 |
|
||||
| `app/schemas/welfare.py` | `CoinTransactionOut` 加 `merged_count` | 下发合并条数 |
|
||||
| `app/models/ad_feed_reward.py` | 更正 `trace_id` 注释 | 文档（领券自 2026-07-15 也带） |
|
||||
| `alembic/versions/coin_transaction_trace_id.py` | 新建迁移 | 加列 + 索引 + 回填历史 |
|
||||
| `tests/test_welfare.py` | 追加测试 | 覆盖写入 / 回填 / 发奖透传 / 聚合 / 分页防残组 |
|
||||
|
||||
---
|
||||
|
||||
## Task 1: 常量 + `CoinTransaction.trace_id` 列 + `grant_coins` 透传
|
||||
|
||||
**Files:**
|
||||
- Modify: `app/core/rewards.py`（加常量）
|
||||
- Modify: `app/models/wallet.py:86`（`remark` 后加 `trace_id` 列）
|
||||
- Modify: `app/repositories/wallet.py:165-196`（`grant_coins` 加参数 + 写入）
|
||||
- Test: `tests/test_welfare.py`（追加）
|
||||
|
||||
- [ ] **Step 1: 写失败测试**
|
||||
|
||||
追加到 `tests/test_welfare.py` 末尾：
|
||||
|
||||
```python
|
||||
def test_grant_coins_persists_trace_id(client) -> None:
|
||||
"""grant_coins 传 trace_id 落库;不传则为 None。"""
|
||||
phone = "13800002001"
|
||||
_login(client, phone)
|
||||
with SessionLocal() as db:
|
||||
user = get_user_by_phone(db, phone)
|
||||
assert user is not None
|
||||
_, txn1 = crud_wallet.grant_coins(
|
||||
db, user.id, 5, biz_type="feed_ad_reward_comparison",
|
||||
ref_id="evt1", remark="比价奖励", trace_id="trace-A",
|
||||
)
|
||||
_, txn2 = crud_wallet.grant_coins(
|
||||
db, user.id, 30, biz_type="signin", remark="每日签到奖励",
|
||||
)
|
||||
db.commit()
|
||||
assert txn1.trace_id == "trace-A"
|
||||
assert txn2.trace_id is None
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 跑测试确认失败**
|
||||
|
||||
Run: `pytest tests/test_welfare.py::test_grant_coins_persists_trace_id -q`
|
||||
Expected: FAIL —— `TypeError: grant_coins() got an unexpected keyword argument 'trace_id'`
|
||||
|
||||
- [ ] **Step 3: 加常量**
|
||||
|
||||
在 `app/core/rewards.py` 的签到常量段之后（约第 32 行 `SIGNIN_CYCLE_LEN` 之后）插入：
|
||||
|
||||
```python
|
||||
# ===== 信息流广告「按会话聚合」的 biz_type =====
|
||||
# 比价 / 领券等候期看的信息流广告,每条各写一条 coin_transaction;这两类在「金币变动记录」
|
||||
# 里按 trace_id 聚合成一条展示(见 repositories/wallet.list_coin_transactions)。其余类型
|
||||
# (reward_video / guide_video / 通用 feed_ad_reward / signin / task_* 等)不聚合。
|
||||
FEED_AD_SESSION_BIZ_TYPES: tuple[str, str] = (
|
||||
"feed_ad_reward_comparison",
|
||||
"feed_ad_reward_coupon",
|
||||
)
|
||||
```
|
||||
|
||||
- [ ] **Step 4: 加模型列**
|
||||
|
||||
`app/models/wallet.py`，把 `remark` 行（第 86 行）后面补一列：
|
||||
|
||||
```python
|
||||
remark: Mapped[str | None] = mapped_column(String(128), nullable=True)
|
||||
# 会话键:仅比价/领券信息流发奖(feed_ad_reward_comparison/coupon)时写入本场 trace_id,
|
||||
# 金币记录列表据此把一次比价/领券的多条广告金币聚合成一条。其余类型 = NULL。
|
||||
trace_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
|
||||
```
|
||||
|
||||
（`String` 在该文件已导入。`index=True` 会在测试 `create_all` 时自动建 `ix_coin_transaction_trace_id`。）
|
||||
|
||||
- [ ] **Step 5: `grant_coins` 加参数 + 写入**
|
||||
|
||||
`app/repositories/wallet.py` 的 `grant_coins`：签名加 `trace_id`，构造 `CoinTransaction` 时带上。
|
||||
|
||||
签名（第 165-173 行）改为：
|
||||
|
||||
```python
|
||||
def grant_coins(
|
||||
db: Session,
|
||||
user_id: int,
|
||||
amount: int,
|
||||
*,
|
||||
biz_type: str,
|
||||
ref_id: str | None = None,
|
||||
remark: str | None = None,
|
||||
trace_id: str | None = None,
|
||||
) -> tuple[CoinAccount, CoinTransaction]:
|
||||
```
|
||||
|
||||
`CoinTransaction(...)` 构造（第 185-193 行）改为：
|
||||
|
||||
```python
|
||||
txn = CoinTransaction(
|
||||
user_id=user_id,
|
||||
amount=amount,
|
||||
balance_after=acc.coin_balance,
|
||||
biz_type=biz_type,
|
||||
ref_id=ref_id,
|
||||
remark=remark,
|
||||
trace_id=trace_id,
|
||||
created_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None), # 存北京 wall-clock(客户端原样切片显示)
|
||||
)
|
||||
```
|
||||
|
||||
- [ ] **Step 6: 跑测试确认通过**
|
||||
|
||||
Run: `pytest tests/test_welfare.py::test_grant_coins_persists_trace_id -q`
|
||||
Expected: PASS
|
||||
|
||||
- [ ] **Step 7: 回归 + lint**
|
||||
|
||||
Run: `pytest tests/test_welfare.py -q && ruff check app/core/rewards.py app/models/wallet.py app/repositories/wallet.py tests/test_welfare.py`
|
||||
Expected: 原有用例仍 PASS，无新增 lint。
|
||||
|
||||
- [ ] **Step 8: 提交**
|
||||
|
||||
```bash
|
||||
git add app/core/rewards.py app/models/wallet.py app/repositories/wallet.py tests/test_welfare.py
|
||||
git commit -m "feat(wallet): coin_transaction 增 trace_id 列 + grant_coins 透传
|
||||
|
||||
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Task 2: 迁移（加列 + 索引 + 回填历史）
|
||||
|
||||
**Files:**
|
||||
- Create: `alembic/versions/coin_transaction_trace_id.py`
|
||||
- Test: `tests/test_welfare.py`（追加回填语义测试）
|
||||
|
||||
- [ ] **Step 1: 写失败测试（回填语义）**
|
||||
|
||||
在 `tests/test_welfare.py` 顶部 import 区补两行（若尚无）：
|
||||
|
||||
```python
|
||||
from sqlalchemy import text
|
||||
from app.models.ad_feed_reward import AdFeedRewardRecord
|
||||
```
|
||||
|
||||
追加测试：
|
||||
|
||||
```python
|
||||
def test_backfill_coin_trace_id_from_ad_record(client) -> None:
|
||||
"""回填:coin_transaction(trace_id 空)按 ref_id==client_event_id 从 ad_feed_reward_record 补 trace_id;
|
||||
只补比价/领券两类,无关类型与无匹配的不动。SQL 与迁移 coin_transaction_trace_id 保持同步。"""
|
||||
phone = "13800002002"
|
||||
_login(client, phone)
|
||||
with SessionLocal() as db:
|
||||
user = get_user_by_phone(db, phone)
|
||||
assert user is not None
|
||||
# 广告表:两条带 trace_id 的记录(模拟客户端已上报)
|
||||
db.add(AdFeedRewardRecord(
|
||||
client_event_id="evt-cmp", user_id=user.id, reward_date="2026-08-07",
|
||||
duration_seconds=10, unit_count=1, ecpm_raw="1000",
|
||||
feed_scene="comparison", trace_id="trace-CMP", coin=5, status="granted",
|
||||
))
|
||||
db.add(AdFeedRewardRecord(
|
||||
client_event_id="evt-cpn", user_id=user.id, reward_date="2026-08-07",
|
||||
duration_seconds=10, unit_count=1, ecpm_raw="1000",
|
||||
feed_scene="coupon", trace_id="trace-CPN", coin=7, status="granted",
|
||||
))
|
||||
db.commit()
|
||||
# 老金币流水:trace_id 全空(模拟改动前),ref_id 指向上面的广告事件
|
||||
_, c1 = crud_wallet.grant_coins(db, user.id, 5, biz_type="feed_ad_reward_comparison", ref_id="evt-cmp", remark="比价奖励")
|
||||
_, c2 = crud_wallet.grant_coins(db, user.id, 7, biz_type="feed_ad_reward_coupon", ref_id="evt-cpn", remark="领券奖励")
|
||||
_, c3 = crud_wallet.grant_coins(db, user.id, 30, biz_type="signin", remark="每日签到奖励")
|
||||
db.commit()
|
||||
assert c1.trace_id is None and c2.trace_id is None
|
||||
|
||||
# 执行与迁移 upgrade() 等价的回填 SQL(务必与迁移保持一致)
|
||||
db.execute(text(
|
||||
"""
|
||||
UPDATE coin_transaction SET trace_id = (
|
||||
SELECT r.trace_id FROM ad_feed_reward_record r
|
||||
WHERE r.client_event_id = coin_transaction.ref_id)
|
||||
WHERE biz_type IN ('feed_ad_reward_comparison', 'feed_ad_reward_coupon')
|
||||
AND trace_id IS NULL
|
||||
AND EXISTS (SELECT 1 FROM ad_feed_reward_record r2
|
||||
WHERE r2.client_event_id = coin_transaction.ref_id
|
||||
AND r2.trace_id IS NOT NULL)
|
||||
"""
|
||||
))
|
||||
db.commit()
|
||||
db.refresh(c1); db.refresh(c2); db.refresh(c3)
|
||||
assert c1.trace_id == "trace-CMP" # 比价补上
|
||||
assert c2.trace_id == "trace-CPN" # 领券补上
|
||||
assert c3.trace_id is None # 签到不动
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 跑测试确认失败**
|
||||
|
||||
Run: `pytest tests/test_welfare.py::test_backfill_coin_trace_id_from_ad_record -q`
|
||||
Expected: FAIL —— `sqlite3.OperationalError: no such column: trace_id`（Task 1 已加列则此步应改为直接 PASS；若已 PASS 说明 create_all 已含列，跳到 Step 4 建迁移）。
|
||||
|
||||
> 说明：本测试验证的是回填 SQL 的 JOIN 语义，不经 Alembic。Task 1 完成后列已在测试库存在，测试可能直接 PASS——这是预期的（回填 SQL 本身是正确逻辑）。真正要新建的产物是迁移文件（Step 4），供 dev/prod 使用。
|
||||
|
||||
- [ ] **Step 3: 跑测试确认通过**
|
||||
|
||||
Run: `pytest tests/test_welfare.py::test_backfill_coin_trace_id_from_ad_record -q`
|
||||
Expected: PASS
|
||||
|
||||
- [ ] **Step 4: 建迁移文件**
|
||||
|
||||
Create `alembic/versions/coin_transaction_trace_id.py`：
|
||||
|
||||
```python
|
||||
"""coin_transaction.trace_id (金币记录按会话聚合比价/领券看广告金币)
|
||||
|
||||
Revision ID: coin_transaction_trace_id
|
||||
Revises: comparison_updated_at
|
||||
Create Date: 2026-08-07
|
||||
|
||||
比价/领券信息流发奖时把本场 trace_id 一并写入 coin_transaction;金币变动记录接口按
|
||||
trace_id 把一次比价/领券的多条广告金币聚合成一条。历史行从 ad_feed_reward_record
|
||||
回填(ref_id == client_event_id)。
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = 'coin_transaction_trace_id'
|
||||
down_revision: Union[str, Sequence[str], None] = 'comparison_updated_at'
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
# SQLite 下 ADD COLUMN(可空) 与 CREATE INDEX 均原生支持,无需 batch_alter_table。
|
||||
op.add_column(
|
||||
'coin_transaction',
|
||||
sa.Column('trace_id', sa.String(length=64), nullable=True),
|
||||
)
|
||||
op.create_index(
|
||||
op.f('ix_coin_transaction_trace_id'),
|
||||
'coin_transaction',
|
||||
['trace_id'],
|
||||
unique=False,
|
||||
)
|
||||
# 历史回填:从 ad_feed_reward_record 按 ref_id==client_event_id 补 trace_id。仅比价/领券两类、
|
||||
# 仅当前为空、且广告行确有 trace_id 时补(EXISTS 守护);`trace_id IS NULL` 保证重跑幂等。
|
||||
# 相关子查询 SQLite/PG 通用。两类 biz_type 在应用侧为 rewards.FEED_AD_SESSION_BIZ_TYPES,
|
||||
# 此处按「迁移不可变」原则硬编码历史快照(勿改为 import 应用常量)。
|
||||
# 大表(prod PG)如需可改按 id 区间分批;此处一次性 UPDATE。
|
||||
op.execute(
|
||||
"""
|
||||
UPDATE coin_transaction SET trace_id = (
|
||||
SELECT r.trace_id FROM ad_feed_reward_record r
|
||||
WHERE r.client_event_id = coin_transaction.ref_id)
|
||||
WHERE biz_type IN ('feed_ad_reward_comparison', 'feed_ad_reward_coupon')
|
||||
AND trace_id IS NULL
|
||||
AND EXISTS (SELECT 1 FROM ad_feed_reward_record r2
|
||||
WHERE r2.client_event_id = coin_transaction.ref_id
|
||||
AND r2.trace_id IS NOT NULL)
|
||||
"""
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_index(
|
||||
op.f('ix_coin_transaction_trace_id'),
|
||||
table_name='coin_transaction',
|
||||
)
|
||||
op.drop_column('coin_transaction', 'trace_id')
|
||||
```
|
||||
|
||||
- [ ] **Step 5: 验证迁移可正反向应用（用一次性 scratch SQLite，不碰 dev 库）**
|
||||
|
||||
Run:
|
||||
```bash
|
||||
DATABASE_URL="sqlite:///./data/_migcheck.db" python -m alembic upgrade head && \
|
||||
DATABASE_URL="sqlite:///./data/_migcheck.db" python -m alembic downgrade -1 && \
|
||||
DATABASE_URL="sqlite:///./data/_migcheck.db" python -m alembic upgrade head && \
|
||||
rm -f ./data/_migcheck.db
|
||||
```
|
||||
Expected: 三条 alembic 命令均无报错，最终 head 落在 `coin_transaction_trace_id`；`rm` 清掉临时库。
|
||||
|
||||
- [ ] **Step 6: lint**
|
||||
|
||||
Run: `ruff check alembic/versions/coin_transaction_trace_id.py tests/test_welfare.py`
|
||||
Expected: 无新增 lint。
|
||||
|
||||
- [ ] **Step 7: 提交**
|
||||
|
||||
```bash
|
||||
git add alembic/versions/coin_transaction_trace_id.py tests/test_welfare.py
|
||||
git commit -m "feat(wallet): 迁移加 coin_transaction.trace_id + 索引 + 回填历史
|
||||
|
||||
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Task 3: `grant_feed_reward` 把 trace_id 写进金币流水
|
||||
|
||||
**Files:**
|
||||
- Modify: `app/repositories/ad_feed_reward.py:219-223`（`grant_coins` 调用加 `trace_id`）
|
||||
- Test: `tests/test_welfare.py`（追加）
|
||||
|
||||
- [ ] **Step 1: 写失败测试**
|
||||
|
||||
在 `tests/test_welfare.py` 顶部 import 区补（若尚无）：
|
||||
|
||||
```python
|
||||
from sqlalchemy import select
|
||||
from app.models.wallet import CoinTransaction
|
||||
from app.repositories import ad_feed_reward as crud_feed
|
||||
```
|
||||
|
||||
追加测试：
|
||||
|
||||
```python
|
||||
def test_grant_feed_reward_sets_coin_trace_id(client) -> None:
|
||||
"""grant_feed_reward(comparison) 把 trace_id 透传给 grant_coins,coin_transaction 带上本场 trace_id。"""
|
||||
phone = "13800002003"
|
||||
_login(client, phone)
|
||||
with SessionLocal() as db:
|
||||
user = get_user_by_phone(db, phone)
|
||||
assert user is not None
|
||||
rec = crud_feed.grant_feed_reward(
|
||||
db, user.id,
|
||||
client_event_id="evt-fr-1", ecpm="1000", duration_seconds=10,
|
||||
feed_scene="comparison", trace_id="trace-FR", display_coin=5,
|
||||
)
|
||||
assert rec.status == "granted", rec.status
|
||||
txn = db.execute(
|
||||
select(CoinTransaction).where(
|
||||
CoinTransaction.user_id == user.id,
|
||||
CoinTransaction.ref_id == "evt-fr-1",
|
||||
)
|
||||
).scalar_one()
|
||||
assert txn.biz_type == "feed_ad_reward_comparison"
|
||||
assert txn.trace_id == "trace-FR"
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 跑测试确认失败**
|
||||
|
||||
Run: `pytest tests/test_welfare.py::test_grant_feed_reward_sets_coin_trace_id -q`
|
||||
Expected: FAIL —— `AssertionError: assert None == 'trace-FR'`（`grant_coins` 尚未收到 trace_id）。
|
||||
|
||||
- [ ] **Step 3: 传 trace_id**
|
||||
|
||||
`app/repositories/ad_feed_reward.py` 第 219-223 行 `grant_coins` 调用改为：
|
||||
|
||||
```python
|
||||
crud_wallet.grant_coins(
|
||||
db, user_id, coin,
|
||||
biz_type=reward_biz, ref_id=client_event_id,
|
||||
remark=reward_remark, trace_id=trace_id,
|
||||
)
|
||||
```
|
||||
|
||||
- [ ] **Step 4: 跑测试确认通过**
|
||||
|
||||
Run: `pytest tests/test_welfare.py::test_grant_feed_reward_sets_coin_trace_id -q`
|
||||
Expected: PASS
|
||||
|
||||
- [ ] **Step 5: lint**
|
||||
|
||||
Run: `ruff check app/repositories/ad_feed_reward.py tests/test_welfare.py`
|
||||
Expected: 无新增 lint。
|
||||
|
||||
- [ ] **Step 6: 提交**
|
||||
|
||||
```bash
|
||||
git add app/repositories/ad_feed_reward.py tests/test_welfare.py
|
||||
git commit -m "feat(ad-feed): 发奖把本场 trace_id 写入金币流水
|
||||
|
||||
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Task 4: 聚合查询 —— `list_coin_transactions` 分组 + `CoinLedgerRow` + `merged_count`
|
||||
|
||||
**Files:**
|
||||
- Modify: `app/repositories/wallet.py`（imports + 加 `CoinLedgerRow` + 重写 `list_coin_transactions`）
|
||||
- Modify: `app/schemas/welfare.py:23-32`（`CoinTransactionOut` 加 `merged_count`）
|
||||
- Test: `tests/test_welfare.py`（追加 3 个聚合测试 + 一个 `_seed_coin` 帮手）
|
||||
|
||||
- [ ] **Step 1: 写失败测试**
|
||||
|
||||
在 `tests/test_welfare.py` 追加帮手 + 测试：
|
||||
|
||||
```python
|
||||
def _seed_coin(db, user_id, amount, biz_type, *, trace_id=None, ref_id=None, remark=None):
|
||||
crud_wallet.grant_coins(
|
||||
db, user_id, amount, biz_type=biz_type, ref_id=ref_id, remark=remark, trace_id=trace_id
|
||||
)
|
||||
|
||||
|
||||
def test_coin_transactions_aggregate_by_trace(client) -> None:
|
||||
"""一次比价的多条广告金币聚合成一条:金额合计、merged_count=条数、balance_after 取最后一条。"""
|
||||
phone = "13800002004"
|
||||
token = _login(client, phone)
|
||||
with SessionLocal() as db:
|
||||
user = get_user_by_phone(db, phone)
|
||||
assert user is not None
|
||||
_seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="t1", ref_id="e1", remark="比价奖励")
|
||||
_seed_coin(db, user.id, 3, "feed_ad_reward_comparison", trace_id="t1", ref_id="e2", remark="比价奖励")
|
||||
_seed_coin(db, user.id, 4, "feed_ad_reward_comparison", trace_id="t1", ref_id="e3", remark="比价奖励")
|
||||
db.commit()
|
||||
r = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token))
|
||||
assert r.status_code == 200, r.text
|
||||
items = r.json()["items"]
|
||||
assert len(items) == 1
|
||||
row = items[0]
|
||||
assert row["biz_type"] == "feed_ad_reward_comparison"
|
||||
assert row["amount"] == 12 # 5+3+4
|
||||
assert row["merged_count"] == 3
|
||||
assert row["balance_after"] == 12 # 末条到账后余额(本用户从 0 起)
|
||||
|
||||
|
||||
def test_coin_transactions_distinct_traces_stay_separate(client) -> None:
|
||||
"""不同 trace(两次比价 / 一次领券)各成一条;不同会话不合并。"""
|
||||
phone = "13800002005"
|
||||
token = _login(client, phone)
|
||||
with SessionLocal() as db:
|
||||
user = get_user_by_phone(db, phone)
|
||||
assert user is not None
|
||||
_seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="cmpA", ref_id="a1", remark="比价奖励")
|
||||
_seed_coin(db, user.id, 6, "feed_ad_reward_comparison", trace_id="cmpB", ref_id="b1", remark="比价奖励")
|
||||
_seed_coin(db, user.id, 6, "feed_ad_reward_comparison", trace_id="cmpB", ref_id="b2", remark="比价奖励")
|
||||
_seed_coin(db, user.id, 7, "feed_ad_reward_coupon", trace_id="cpnC", ref_id="c1", remark="领券奖励")
|
||||
db.commit()
|
||||
items = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token)).json()["items"]
|
||||
assert len(items) == 3
|
||||
assert sorted(i["amount"] for i in items) == [5, 7, 12]
|
||||
cpn = next(i for i in items if i["biz_type"] == "feed_ad_reward_coupon")
|
||||
assert cpn["amount"] == 7 and cpn["merged_count"] == 1
|
||||
|
||||
|
||||
def test_coin_transactions_non_session_rows_stay_per_row(client) -> None:
|
||||
"""签到 / 无 trace 的通用信息流各自一行,不被聚合;夹在比价广告中间的签到不影响比价聚合。"""
|
||||
phone = "13800002006"
|
||||
token = _login(client, phone)
|
||||
with SessionLocal() as db:
|
||||
user = get_user_by_phone(db, phone)
|
||||
assert user is not None
|
||||
_seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="t1", ref_id="e1", remark="比价奖励")
|
||||
_seed_coin(db, user.id, 30, "signin", remark="每日签到奖励") # 夹在两条比价广告中间
|
||||
_seed_coin(db, user.id, 4, "feed_ad_reward_comparison", trace_id="t1", ref_id="e2", remark="比价奖励")
|
||||
_seed_coin(db, user.id, 8, "feed_ad_reward", trace_id=None, ref_id="w1", remark="信息流广告奖励")
|
||||
db.commit()
|
||||
items = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token)).json()["items"]
|
||||
assert len(items) == 3
|
||||
cmp_row = next(i for i in items if i["biz_type"] == "feed_ad_reward_comparison")
|
||||
assert cmp_row["amount"] == 9 and cmp_row["merged_count"] == 2
|
||||
signin_row = next(i for i in items if i["biz_type"] == "signin")
|
||||
assert signin_row["amount"] == 30 and signin_row["merged_count"] == 1
|
||||
feed_row = next(i for i in items if i["biz_type"] == "feed_ad_reward")
|
||||
assert feed_row["amount"] == 8 and feed_row["merged_count"] == 1
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 跑测试确认失败**
|
||||
|
||||
Run: `pytest tests/test_welfare.py::test_coin_transactions_aggregate_by_trace -q`
|
||||
Expected: FAIL —— 未聚合，返回 3 条 / `merged_count` 字段缺失（`KeyError` 或 `len(items)==3`）。
|
||||
|
||||
- [ ] **Step 3: schema 加 `merged_count`**
|
||||
|
||||
`app/schemas/welfare.py` 的 `CoinTransactionOut`（第 23-32 行）在 `created_at` 后加一行：
|
||||
|
||||
```python
|
||||
class CoinTransactionOut(BaseModel):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
id: int
|
||||
amount: int = Field(..., description="正=入账,负=出账")
|
||||
balance_after: int
|
||||
biz_type: str
|
||||
ref_id: str | None = None
|
||||
remark: str | None = None
|
||||
created_at: datetime
|
||||
merged_count: int = Field(1, description="本行合并的底层流水条数(比价/领券按会话聚合;未合并=1)")
|
||||
```
|
||||
|
||||
- [ ] **Step 4: 补 imports + 加 `CoinLedgerRow`**
|
||||
|
||||
`app/repositories/wallet.py` 顶部：把 `from sqlalchemy import func, select, update`（第 15 行）改为：
|
||||
|
||||
```python
|
||||
from sqlalchemy import String, and_, case, cast, func, literal, select, update
|
||||
```
|
||||
|
||||
在 stdlib import 区（约第 7-13 行）加一行：
|
||||
|
||||
```python
|
||||
from dataclasses import dataclass
|
||||
```
|
||||
|
||||
在 `list_coin_transactions` 之前加返回行类型：
|
||||
|
||||
```python
|
||||
@dataclass(frozen=True)
|
||||
class CoinLedgerRow:
|
||||
"""`list_coin_transactions` 的返回行。广告类按 trace_id 聚合后的展示行,
|
||||
**非 ORM 对象**(避免把聚合后的 amount 误写回底层流水)。"""
|
||||
id: int
|
||||
amount: int
|
||||
balance_after: int
|
||||
biz_type: str
|
||||
ref_id: str | None
|
||||
remark: str | None
|
||||
created_at: datetime
|
||||
merged_count: int
|
||||
```
|
||||
|
||||
- [ ] **Step 5: 重写 `list_coin_transactions`**
|
||||
|
||||
`app/repositories/wallet.py` 第 260-279 行整体替换为：
|
||||
|
||||
```python
|
||||
def list_coin_transactions(
|
||||
db: Session,
|
||||
user_id: int,
|
||||
*,
|
||||
limit: int = 20,
|
||||
cursor: int | None = None,
|
||||
) -> tuple[list[CoinLedgerRow], int | None]:
|
||||
"""金币流水分页(游标式,按 id 倒序)。
|
||||
|
||||
比价/领券两类信息流广告(rewards.FEED_AD_SESSION_BIZ_TYPES)且带 trace_id 的行,
|
||||
按 trace_id 聚合成一条(一次比价/领券 = 一行):金额合计、代表行取组内最新一条
|
||||
(MAX(id))的余额/时间、merged_count=组内条数。其余每条一行。
|
||||
|
||||
分组必须在该用户**全量**行上算真实 rep_id 后再按 rep_id 过滤——**不可**把 CTE 输入
|
||||
裁成 id<cursor,否则会话行交错跨游标时会算出与上页重复的「残组」。
|
||||
|
||||
cursor 为上一页最后一条的 id(即其组 rep_id);返回 (本页列表, next_cursor)。
|
||||
"""
|
||||
ct = CoinTransaction
|
||||
group_key = case(
|
||||
(
|
||||
and_(
|
||||
ct.biz_type.in_(rewards.FEED_AD_SESSION_BIZ_TYPES),
|
||||
ct.trace_id.is_not(None),
|
||||
),
|
||||
literal("T:") + ct.trace_id,
|
||||
),
|
||||
else_=literal("I:") + cast(ct.id, String),
|
||||
).label("group_key")
|
||||
|
||||
grp = (
|
||||
select(
|
||||
group_key,
|
||||
func.max(ct.id).label("rep_id"),
|
||||
func.sum(ct.amount).label("total_amount"),
|
||||
func.count().label("merged_count"),
|
||||
)
|
||||
.where(ct.user_id == user_id)
|
||||
.group_by(group_key)
|
||||
.cte("grp")
|
||||
)
|
||||
|
||||
stmt = (
|
||||
select(
|
||||
ct.id,
|
||||
grp.c.total_amount.label("amount"),
|
||||
ct.balance_after,
|
||||
ct.biz_type,
|
||||
ct.ref_id,
|
||||
ct.remark,
|
||||
ct.created_at,
|
||||
grp.c.merged_count,
|
||||
)
|
||||
.select_from(grp)
|
||||
.join(ct, ct.id == grp.c.rep_id)
|
||||
)
|
||||
if cursor is not None:
|
||||
stmt = stmt.where(grp.c.rep_id < cursor)
|
||||
stmt = stmt.order_by(grp.c.rep_id.desc()).limit(limit)
|
||||
|
||||
rows = db.execute(stmt).all()
|
||||
items = [
|
||||
CoinLedgerRow(
|
||||
id=r.id,
|
||||
amount=int(r.amount),
|
||||
balance_after=r.balance_after,
|
||||
biz_type=r.biz_type,
|
||||
ref_id=r.ref_id,
|
||||
remark=r.remark,
|
||||
created_at=r.created_at,
|
||||
merged_count=int(r.merged_count),
|
||||
)
|
||||
for r in rows
|
||||
]
|
||||
next_cursor = items[-1].id if len(items) == limit else None
|
||||
return items, next_cursor
|
||||
```
|
||||
|
||||
（端点 `app/api/v1/wallet.py` 无需改：`CoinTransactionOut.model_validate(it)` 对 `CoinLedgerRow` dataclass 按 `from_attributes` 读取即可。）
|
||||
|
||||
- [ ] **Step 6: 跑三个聚合测试确认通过**
|
||||
|
||||
Run: `pytest tests/test_welfare.py -k "coin_transactions_aggregate_by_trace or distinct_traces_stay_separate or non_session_rows_stay_per_row" -q`
|
||||
Expected: 3 PASS
|
||||
|
||||
- [ ] **Step 7: 全量回归 + lint**
|
||||
|
||||
Run: `pytest tests/test_welfare.py -q && ruff check app/repositories/wallet.py app/schemas/welfare.py tests/test_welfare.py`
|
||||
Expected: 原有用例(含 `test_signin_flow` / `test_exchange_flow` 等对 coin-transactions 的断言)仍 PASS；无新增 lint。
|
||||
|
||||
- [ ] **Step 8: 提交**
|
||||
|
||||
```bash
|
||||
git add app/repositories/wallet.py app/schemas/welfare.py tests/test_welfare.py
|
||||
git commit -m "feat(wallet): 金币记录按 trace_id 聚合比价/领券看广告金币
|
||||
|
||||
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Task 5: 分页防残组回归测试
|
||||
|
||||
**Files:**
|
||||
- Test: `tests/test_welfare.py`（追加）
|
||||
|
||||
- [ ] **Step 1: 写回归测试**
|
||||
|
||||
追加：
|
||||
|
||||
```python
|
||||
def test_coin_transactions_pagination_no_phantom_regroup(client) -> None:
|
||||
"""交错跨游标不产生残组:会话广告成员被其它记录隔开、rep_id 在游标上、成员在游标下时,
|
||||
翻到下一页该会话不得以「残组」重复出现(锁死 spec §11 反面优化警示)。"""
|
||||
phone = "13800002007"
|
||||
token = _login(client, phone)
|
||||
with SessionLocal() as db:
|
||||
user = get_user_by_phone(db, phone)
|
||||
assert user is not None
|
||||
_seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="t1", ref_id="e1", remark="比价奖励") # id=n+1
|
||||
_seed_coin(db, user.id, 30, "signin", remark="每日签到奖励") # id=n+2
|
||||
_seed_coin(db, user.id, 40, "signin", remark="每日签到奖励") # id=n+3
|
||||
_seed_coin(db, user.id, 4, "feed_ad_reward_comparison", trace_id="t1", ref_id="e2", remark="比价奖励") # id=n+4 = t1 的 rep
|
||||
db.commit()
|
||||
# 第 1 页(limit=2):按 rep 降序 = [t1(rep=n+4, 合计 9), signin(n+3, 40)]
|
||||
p1 = client.get("/api/v1/wallet/coin-transactions?limit=2", headers=_auth(token)).json()
|
||||
assert len(p1["items"]) == 2
|
||||
assert p1["items"][0]["biz_type"] == "feed_ad_reward_comparison"
|
||||
assert p1["items"][0]["amount"] == 9 and p1["items"][0]["merged_count"] == 2
|
||||
assert p1["items"][1]["biz_type"] == "signin" and p1["items"][1]["amount"] == 40
|
||||
assert p1["next_cursor"] is not None
|
||||
# 第 2 页:只剩另一条 signin(30);t1 的 rep 在游标上,成员虽在游标下也不得成残组重复
|
||||
p2 = client.get(
|
||||
f"/api/v1/wallet/coin-transactions?limit=2&cursor={p1['next_cursor']}",
|
||||
headers=_auth(token),
|
||||
).json()
|
||||
assert len(p2["items"]) == 1
|
||||
assert p2["items"][0]["biz_type"] == "signin" and p2["items"][0]["amount"] == 30
|
||||
assert all(i["biz_type"] != "feed_ad_reward_comparison" for i in p2["items"])
|
||||
assert p2["next_cursor"] is None
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 跑测试确认通过（锁定不变量）**
|
||||
|
||||
Run: `pytest tests/test_welfare.py::test_coin_transactions_pagination_no_phantom_regroup -q`
|
||||
Expected: PASS（Task 4 的全量分组实现已正确；若 FAIL 说明 CTE 被错误地按 cursor 裁剪，回到 Task 4 Step 5 修正）。
|
||||
|
||||
- [ ] **Step 3: lint**
|
||||
|
||||
Run: `ruff check tests/test_welfare.py`
|
||||
Expected: 无新增 lint。
|
||||
|
||||
- [ ] **Step 4: 提交**
|
||||
|
||||
```bash
|
||||
git add tests/test_welfare.py
|
||||
git commit -m "test(wallet): 锁死金币记录分组分页不产生残组
|
||||
|
||||
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Task 6: 更正 `ad_feed_reward_record.trace_id` 过时注释
|
||||
|
||||
**Files:**
|
||||
- Modify: `app/models/ad_feed_reward.py:39-40`
|
||||
|
||||
- [ ] **Step 1: 改注释**
|
||||
|
||||
`app/models/ad_feed_reward.py` 第 39-40 行两行注释替换为：
|
||||
|
||||
```python
|
||||
# 本次会话 trace_id:比价一直带;领券自 2026-07-15(客户端 a98cab8)起也带;福利/旧客户端 = NULL。
|
||||
# 用途:比价记录页按 trace_id 聚合「比价赚 N 金币」;金币记录列表把一次比价/领券的多条广告金币聚合成一条。
|
||||
trace_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 冒烟 + lint**
|
||||
|
||||
Run: `python -c "import app.models.ad_feed_reward" && ruff check app/models/ad_feed_reward.py`
|
||||
Expected: 无报错、无 lint。
|
||||
|
||||
- [ ] **Step 3: 提交**
|
||||
|
||||
```bash
|
||||
git add app/models/ad_feed_reward.py
|
||||
git commit -m "docs(ad-feed): 更正 trace_id 注释(领券自 2026-07-15 也带)
|
||||
|
||||
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 收尾验证
|
||||
|
||||
- [ ] **全量测试**：`pytest -q`（对齐 memory「preexisting-test-lint-debt」：先前已知失败数不变、无本改动引入的新失败）。
|
||||
- [ ] **lint**：`ruff check .`（无本改动引入的新问题）。
|
||||
- [ ] **迁移落 dev**：`python -m alembic upgrade head`（把 dev SQLite 迁到最新，确认 `./run.sh` 正常）。
|
||||
|
||||
---
|
||||
|
||||
## Self-Review（写完计划后自查）
|
||||
|
||||
**Spec 覆盖：**
|
||||
- §4 加列 → Task 1 Step 4 + Task 2。✓
|
||||
- §5 写入透传 → Task 1 Step 5 + Task 3。✓
|
||||
- §6 聚合查询（CASE 分组键、MAX(id) 代表/游标、biz_type 门控、常量）→ Task 4 Step 4-5 + Task 1 常量。✓
|
||||
- §7 `merged_count` 下发 → Task 4 Step 3。✓
|
||||
- §8 回填（幂等 + EXISTS 守护 + 通用 SQL）→ Task 2 Step 4。✓
|
||||
- §9 注释更正 → Task 6。✓
|
||||
- §10 App 端无需改 → 端点未改（Task 4 Step 5 注）。✓
|
||||
- §11 反面优化警示（残组）→ Task 4 docstring + Task 5 回归测试。✓
|
||||
- §3 只改用户接口、admin 不动 → 全程只碰 `crud_wallet.list_coin_transactions`，未触 `app/admin`。✓
|
||||
- §12 测试（分组/隔离/空 trace/分页/回填）→ Task 2/4/5。✓
|
||||
|
||||
**占位扫描：** 无 TBD / TODO；每个代码步骤均给出完整代码与确切命令。✓
|
||||
|
||||
**类型一致：** `grant_coins(..., trace_id=...)`（Task 1）↔ `grant_feed_reward` 调用（Task 3）↔ 查询 `rewards.FEED_AD_SESSION_BIZ_TYPES`（Task 4）↔ 迁移硬编码同两值（Task 2，有意快照）一致；`CoinLedgerRow` 字段（Task 4 Step 4）↔ `CoinTransactionOut` 字段（Task 4 Step 3）逐一对应（id/amount/balance_after/biz_type/ref_id/remark/created_at/merged_count）。✓
|
||||
@@ -0,0 +1,190 @@
|
||||
# 金币记录：比价/领券看广告金币按会话汇总成一条 设计
|
||||
|
||||
- 日期：2026-08-07
|
||||
- 分支：feat/coin-ledger-aggregate-ad-rewards
|
||||
- 相关：`app/repositories/wallet.py`（发币/流水）、`app/repositories/ad_feed_reward.py`（信息流发奖）、`app/api/v1/wallet.py`（金币流水接口）、Android `CoinHistoryViewModel`（列表渲染）
|
||||
|
||||
## 1. 背景与问题
|
||||
|
||||
App「金币变动记录」列表按**每看一次广告一条**展示。一次比价的等候期会连续看多条信息流广告，一次领券流程同理，于是列表里一次比价/领券会刷出一长串「比价奖励 +x」「领券奖励 +x」，淹没其它记录、观感差。
|
||||
|
||||
需求：**一次比价的所有看广告金币合并成一条、一次领券的合并成一条**，金额取该次会话合计。
|
||||
|
||||
## 2. 现状（关键事实）
|
||||
|
||||
- 发币唯一入口 `crud_wallet.grant_coins`（`app/repositories/wallet.py:165`）：每条广告写一条 `CoinTransaction`。金币账本 `coin_transaction` 与现金账本物理分离。
|
||||
- 信息流发奖 `grant_feed_reward`（`app/repositories/ad_feed_reward.py:71`）按点位场景拆 `biz_type`：
|
||||
- `feed_ad_reward_comparison`（比价，remark「比价奖励」）
|
||||
- `feed_ad_reward_coupon`（领券，remark「领券奖励」）
|
||||
- `feed_ad_reward`（通用/福利/旧端）
|
||||
- **trace_id 已全链贯通到结算**：
|
||||
- 比价：客户端结算上报早已带 trace_id。
|
||||
- 领券：`CouponForegroundService.reportFeedReward` 自 **2026-07-15**（commit `a98cab8`）起带 `feedScene="coupon" + traceId=sessionTraceId`。
|
||||
- 两者的 trace_id 都落到了 `ad_feed_reward_record.trace_id`（`grant_feed_reward` 写入）。
|
||||
- **唯一缺口**：`grant_feed_reward` 调 `grant_coins` 时**没把 trace_id 透传下去**，`coin_transaction` 表本身**没有 trace_id 列** → 金币流水层无法按会话分组。
|
||||
- 列表接口 `GET /api/v1/coin-transactions`（`app/api/v1/wallet.py:61` → `list_coin_transactions` `app/repositories/wallet.py:260`）：按 `id` 倒序游标分页。
|
||||
- 展示文案权在客户端：Android `CoinHistoryViewModel.coinTitle` 按 `bizType` 直显固定标题（比价奖励/领券奖励），后端 `remark` 只作兜底。
|
||||
|
||||
结论：这是**纯后端**改动，领券不需要额外 Android 改动即可生效。
|
||||
|
||||
## 3. 目标与范围
|
||||
|
||||
- 目标：金币记录列表里，同一次比价的多条「比价奖励」合并成一条、同一次领券的多条「领券奖励」合并成一条；金额为该次会话合计。
|
||||
- 会话键：`trace_id`（一次比价/一次领券 = 一个 trace_id）。与「比价记录页」现有的 trace_id 聚合口径一致。
|
||||
- **范围**：只圈两类 `biz_type` —— `feed_ad_reward_comparison`、`feed_ad_reward_coupon`。激励视频 `reward_video`、引导视频 `guide_video`、通用信息流 `feed_ad_reward`、签到、任务、兑换等**一律不动**，仍每条一行。
|
||||
- 聚合位置：**后端**（`/coin-transactions` 直接返回合并后的行）。客户端聚合被否决（见 §11）。
|
||||
- **只改 App 用户接口**：聚合仅作用于 `crud_wallet.list_coin_transactions`（`GET /api/v1/wallet/coin-transactions`）。**admin 接口不动**——`app/admin/routers/wallet.py` 走独立的 `queries.list_all_coin_transactions`（跨用户、可按 `biz_type` 筛），审计/客服必须能看到**每一条**广告发币，保持每条一行。两者物理分离，聚合天然不波及 admin，此处显式声明防误改。
|
||||
- 历史：**加列 + 一次性回填**，历史记录也合并。
|
||||
|
||||
## 4. 数据模型改动
|
||||
|
||||
`coin_transaction` 新增列：
|
||||
|
||||
| 列 | 类型 | 约束 | 说明 |
|
||||
|---|---|---|---|
|
||||
| `trace_id` | `String(64)` | nullable, index | 会话键；仅比价/领券信息流发奖时写入，其余为 NULL |
|
||||
|
||||
- 新增 Alembic 迁移（`render_as_batch` 兼容 SQLite）：加列 + 索引 `ix_coin_transaction_trace_id`。
|
||||
- 可选复合索引 `(user_id, trace_id)` 辅助分组扫描（视线上量级决定，MVP 可先只加单列索引）。
|
||||
|
||||
## 5. 写入路径改动
|
||||
|
||||
- `grant_coins(...)` 增加可选参数 `trace_id: str | None = None`，写入 `CoinTransaction.trace_id`。默认 None → 其余调用方（签到/任务/兑换/激励视频）零改动、保持 NULL。
|
||||
- `grant_feed_reward` 调 `grant_coins` 时透传 `trace_id=trace_id`（比价/领券自然有值；welfare/旧端为 None）。
|
||||
- 幂等不变：`client_event_id` 仍是幂等键；`grant_coins` 仍不 commit，由 `_commit_record` 同事务提交。
|
||||
|
||||
## 6. 读取/聚合查询（核心）
|
||||
|
||||
`list_coin_transactions` 改为「分组游标分页」。用 CTE 先分组、再 JOIN 回代表行取展示字段。
|
||||
|
||||
分组规则：
|
||||
|
||||
```
|
||||
group_key =
|
||||
若 biz_type ∈ (feed_ad_reward_comparison, feed_ad_reward_coupon) 且 trace_id 非空
|
||||
→ 'T:' || trace_id # 同一会话所有广告归一组
|
||||
否则
|
||||
→ 'I:' || id # 每条自成一组(行为等同现状)
|
||||
```
|
||||
|
||||
每组取：`rep_id = MAX(id)`、`total = SUM(amount)`、`merged_count = COUNT(*)`；再 JOIN `coin_transaction` 取 `rep_id` 那条的 `balance_after / biz_type / ref_id / remark / created_at / trace_id`。
|
||||
|
||||
等价 SQL（SQLite / PostgreSQL 通用）：
|
||||
|
||||
```sql
|
||||
WITH grp AS (
|
||||
SELECT
|
||||
CASE WHEN biz_type IN ('feed_ad_reward_comparison','feed_ad_reward_coupon')
|
||||
AND trace_id IS NOT NULL
|
||||
THEN 'T:' || trace_id
|
||||
ELSE 'I:' || CAST(id AS TEXT) END AS group_key,
|
||||
MAX(id) AS rep_id,
|
||||
SUM(amount) AS total_amount,
|
||||
COUNT(*) AS merged_count
|
||||
FROM coin_transaction
|
||||
WHERE user_id = :uid
|
||||
GROUP BY group_key
|
||||
)
|
||||
SELECT ct.id, grp.total_amount AS amount, ct.balance_after,
|
||||
ct.biz_type, ct.ref_id, ct.remark, ct.trace_id, ct.created_at,
|
||||
grp.merged_count
|
||||
FROM grp
|
||||
JOIN coin_transaction ct ON ct.id = grp.rep_id
|
||||
WHERE (:cursor IS NULL OR grp.rep_id < :cursor)
|
||||
ORDER BY grp.rep_id DESC
|
||||
LIMIT :limit;
|
||||
```
|
||||
|
||||
返回行字段口径：
|
||||
|
||||
- `id` = 组内 `MAX(id)`：做列表 key + 下一页游标。组间不重复（每个 id 只属一组），`rep_id DESC` 是全序，游标 `rep_id < cursor` 干净。
|
||||
- `amount` = 组内合计（该次会话总金币；这些广告行金额恒正）。
|
||||
- `balance_after` / `created_at` / `remark` / `ref_id` = 代表行（最后一条）的值。`balance_after` 即该次会话最后一条广告到账后的余额，正确。
|
||||
- `merged_count` = 合并条数（未合并 = 1）。
|
||||
- `next_cursor` = 本页最后一行的 `id`（= 其组 rep_id），够 limit 才给，否则 None。
|
||||
|
||||
非分组类型、`trace_id` 为空的旧广告行 → 各自成组（`'I:'||id`），行为与现状完全一致。
|
||||
|
||||
> 说明（分组安全性）：分组键按 `biz_type` **门控**——通用 `feed_ad_reward`（福利/旧端）即便偶带 trace_id 也走 `'I:'||id` 保持每条一行，只有比价/领券两类才按 trace_id 合并。trace_id 由后端按会话签发、全局唯一（一个 trace = 一次比价**或**一次领券），故仅按 trace_id 分组不会把两类混并，代表行的 `biz_type` 唯一确定展示标题。
|
||||
>
|
||||
> 实现约定：这两类 biz_type 抽成模块常量 `FEED_AD_SESSION_BIZ_TYPES`，**查询 / 回填 / 测试共用一处**，避免字符串散落三地漂移。
|
||||
|
||||
## 7. 接口契约
|
||||
|
||||
`GET /api/v1/coin-transactions` 响应新增 `merged_count`（`CoinTransactionOut` 加字段，Pydantic 默认 1）：
|
||||
|
||||
```jsonc
|
||||
{
|
||||
"items": [
|
||||
{ "id": 1520, "amount": 12, "balance_after": 3380,
|
||||
"biz_type": "feed_ad_reward_comparison", "ref_id": "evt_...",
|
||||
"remark": "比价奖励", "created_at": "2026-08-07T12:03:11",
|
||||
"merged_count": 5 }, // 本次比价看了 5 条广告，合计 +12
|
||||
{ "id": 1512, "amount": 30, "balance_after": 3368,
|
||||
"biz_type": "signin", "remark": "每日签到奖励",
|
||||
"merged_count": 1 }
|
||||
],
|
||||
"next_cursor": 1490
|
||||
}
|
||||
```
|
||||
|
||||
`trace_id` 是否在响应里下发：**可选**，MVP 不下发（客户端用不到，标题按 biz_type、金额已合计）。若后续要做「点开看明细」再补。
|
||||
|
||||
## 8. 历史回填
|
||||
|
||||
迁移里一次性把历史 `coin_transaction.trace_id` 从 `ad_feed_reward_record` 补齐（相关子查询 UPDATE，SQLite/PG 通用）：
|
||||
|
||||
```sql
|
||||
UPDATE coin_transaction SET trace_id = (
|
||||
SELECT r.trace_id FROM ad_feed_reward_record r
|
||||
WHERE r.client_event_id = coin_transaction.ref_id)
|
||||
WHERE biz_type IN ('feed_ad_reward_comparison','feed_ad_reward_coupon')
|
||||
AND trace_id IS NULL
|
||||
AND EXISTS (SELECT 1 FROM ad_feed_reward_record r2
|
||||
WHERE r2.client_event_id = coin_transaction.ref_id
|
||||
AND r2.trace_id IS NOT NULL);
|
||||
```
|
||||
|
||||
- 关联键：`coin_transaction.ref_id == ad_feed_reward_record.client_event_id`（`grant_feed_reward` 里两者同源，见 `app/repositories/ad_feed_reward.py:221`）。`client_event_id` 有唯一约束 → 相关子查询至多一行，无「子查询多行」风险。
|
||||
- 只补两类 biz_type；只在广告行确有 trace_id 时补（`EXISTS` 守护）。
|
||||
- **幂等**：`AND trace_id IS NULL` 保证重跑不重复改。
|
||||
- **大表守护**：`coin_transaction` 线上若很大，单条全表 UPDATE 在 PG 上是一次性长事务/行锁；必要时按 `id` 区间**分批**（每批数万行）跑。SQLite（dev）无所谓。
|
||||
- 2026-07-15 前的领券老行 `ad_feed_reward_record.trace_id` 本就为空 → 回填后仍为空 → 保持每条一行（可接受，会随时间自然淘汰）。
|
||||
|
||||
## 9. 顺手更正
|
||||
|
||||
更新 `app/models/ad_feed_reward.py:39` 的 `trace_id` 注释：现「仅 comparison 场景由客户端带上……领券/福利/旧客户端 = NULL」已过时，改为「比价一直带；领券自 2026-07-15 起也带；福利/旧客户端仍 NULL」。
|
||||
|
||||
## 10. App 端
|
||||
|
||||
- **核心无需改**：接口返回合并后的行，`CoinHistoryViewModel.coinTitle` 仍按 `bizType` 显示「比价奖励/领券奖励」，金额显示合计值。
|
||||
- **可选增强（后续、非本次必须）**：客户端读到 `merged_count > 1` 时把标题渲染成「比价奖励 ×5」或副行「看了 5 条广告」。做不做都不影响本功能生效。
|
||||
|
||||
## 11. 边界与取舍
|
||||
|
||||
- **时间交错重排**：一次比价的广告中间夹了签到等其它记录时，合并行锚定在「该会话最后一条广告」的 id 上，夹在中间的签到会排到合并行之后。这是折叠的固有效果，可接受。
|
||||
- **余额不连续**：相邻两展示行的 `balance_after` 之间存在被折叠的中间变动，属聚合固有现象；每行 `balance_after` 仍是该点真实运行余额。
|
||||
- **迟到广告**：会话已过、同 trace 的迟到广告到账后并入该组、金额变大、合并行上移（因 rep_id 变大）；用户未刷新则短暂 stale。会话结束后广告即时结算，迟到罕见，可接受。
|
||||
- **单条会话**：一次会话只看了 1 条广告 → 组内 1 行、`merged_count=1`、展示与现状一致，无需特判。
|
||||
- **查询成本**：分组 CTE 每页对该用户**全部** `coin_transaction` 行分组后再分页，O(N)/页，比原「`id < cursor LIMIT n`」的索引 seek 重。金币记录非热路径（偶尔打开、翻几页）；粗估重度用户万级行/年、单页分组数毫秒级（估算，非实测），可接受。
|
||||
- **索引口径**：分组按 CASE 表达式（`'T:'||trace_id` / `'I:'||id`）在内存做，`(user_id, trace_id)` 复合索引**并不加速**该 group-by；真正需要的只是「按 user_id 取该用户全部行」，现有 `ix_coin_transaction_user_id` 已够。故本设计**只加 `trace_id` 单列索引**（回填 / 潜在按 trace 查用），不加 `(user_id, trace_id)`；若要 index-only 可选覆盖索引 `(user_id, id, biz_type, trace_id, amount)`，MVP 不加。
|
||||
- ⚠️ **反面优化警示**：**不要**把 CTE 输入裁成 `WHERE id < :cursor` 求快。会话行若被其它记录交错、成员跨越游标（部分成员 `id < cursor`、但组的 `rep_id ≥ cursor`），裁剪后会算出一个 rep_id 更小的「残组」→ 与上一页已展示的整组**重复出现**。分组必须在该用户全量行上算出真实 `rep_id` 后，再 `rep_id < cursor` 过滤。
|
||||
- **未来读优化（YAGNI，暂不做）**：若量级压力显现，加只读会话聚合投影表 `coin_ad_session_agg(user_id, trace_id, total_coin, cnt, last_txn_id, last_created_at)`，写广告流水时 upsert，列表用「非广告行 UNION 该投影」走索引分页。这是**读优化**（明细账本仍每条一行、审计不动），与 §13 否决的「写时账本聚合」不同。
|
||||
|
||||
## 12. 测试
|
||||
|
||||
- **分组**：同 trace 多条比价广告 → 一行（`amount` 合计 / `merged_count=N` / `balance_after` 取最后 / `id=MAX`）；两次不同 trace 的比价 → 两行；领券同 trace → 一行。
|
||||
- **隔离**：比价广告序列中间夹一条签到 → 比价折叠成一行、签到独立成行；`reward_video` / `signin` / `exchange_out` → 仍 1:1，`merged_count=1`。
|
||||
- **空 trace_id**：`feed_ad_reward`（通用）或 trace_id 为 NULL 的广告行 → 每条一行。
|
||||
- **分页**：多组跨页时按 `rep_id` 游标翻页不重不漏；一个大组（跨越 limit 边界的多条底层行）不被拆成两页；`next_cursor` 到底返回 None。
|
||||
- **交错跨游标不重复（防残组回归）**：造一个会话，其广告成员 id 被一条非广告行隔开（部分成员在游标下、`rep_id` 在游标上），翻到下一页时该组**不得**再以「残组」重复出现——锁死 §11 的反面优化警示。
|
||||
- **回填迁移**：造 `coin_transaction`（比价/领券行，ref_id 指向带 trace_id 的 `ad_feed_reward_record`）+ 无关行，跑迁移后仅两类被正确补 trace_id、无关行不动。
|
||||
- **既有测试**：`/coin-transactions` 现有用例（非广告类型 1:1、分页）保持通过。
|
||||
|
||||
## 13. 不做（YAGNI）
|
||||
|
||||
- 不做写时聚合（会破坏实时到账 / 幂等 / 余额连续性）。
|
||||
- 不做客户端聚合（SWR 缓存 + 游标分页下，一个会话可能跨页，跨页分组脆弱）。
|
||||
- 不动激励视频 / 引导视频 / 通用信息流 / 签到等其它类型。
|
||||
- 不动「比价记录页」——它的「比价赚 N 金币」是独立查询（直接聚合 `ad_feed_reward_record`），不受本改动影响。
|
||||
- MVP 不下发 trace_id、不做「点开看明细」。
|
||||
+209
-1
@@ -4,7 +4,7 @@
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from sqlalchemy import select, text
|
||||
|
||||
from app.core.config import settings
|
||||
from app.core.rewards import (
|
||||
@@ -17,6 +17,9 @@ from app.core.rewards import (
|
||||
)
|
||||
from app.db.session import SessionLocal
|
||||
from app.integrations import pangle
|
||||
from app.models.ad_feed_reward import AdFeedRewardRecord
|
||||
from app.models.wallet import CoinTransaction
|
||||
from app.repositories import ad_feed_reward as crud_feed
|
||||
from app.repositories import wallet as crud_wallet
|
||||
from app.repositories.user import get_user_by_phone
|
||||
|
||||
@@ -384,3 +387,208 @@ def test_endpoints_require_auth(client) -> None:
|
||||
assert client.get("/api/v1/savings/summary").status_code == 401
|
||||
assert client.get("/api/v1/savings/battle").status_code == 401
|
||||
assert client.get("/api/v1/savings/records").status_code == 401
|
||||
|
||||
|
||||
def test_grant_coins_persists_trace_id(client) -> None:
|
||||
"""grant_coins 传 trace_id 落库;不传则为 None。"""
|
||||
phone = "13800002001"
|
||||
_login(client, phone)
|
||||
with SessionLocal() as db:
|
||||
user = get_user_by_phone(db, phone)
|
||||
assert user is not None
|
||||
_, txn1 = crud_wallet.grant_coins(
|
||||
db, user.id, 5, biz_type="feed_ad_reward_comparison",
|
||||
ref_id="evt1", remark="比价奖励", trace_id="trace-A",
|
||||
)
|
||||
_, txn2 = crud_wallet.grant_coins(
|
||||
db, user.id, 30, biz_type="signin", remark="每日签到奖励",
|
||||
)
|
||||
db.commit()
|
||||
assert txn1.trace_id == "trace-A"
|
||||
assert txn2.trace_id is None
|
||||
|
||||
|
||||
def test_backfill_coin_trace_id_from_ad_record(client) -> None:
|
||||
"""回填:coin_transaction(trace_id 空)按 ref_id==client_event_id 从 ad_feed_reward_record 补 trace_id;
|
||||
只补比价/领券两类,无关类型与无匹配的不动。SQL 与迁移 coin_transaction_trace_id 保持同步。"""
|
||||
phone = "13800002002"
|
||||
_login(client, phone)
|
||||
with SessionLocal() as db:
|
||||
user = get_user_by_phone(db, phone)
|
||||
assert user is not None
|
||||
db.add(AdFeedRewardRecord(
|
||||
client_event_id="evt-cmp", user_id=user.id, reward_date="2026-08-07",
|
||||
duration_seconds=10, unit_count=1, ecpm_raw="1000",
|
||||
feed_scene="comparison", trace_id="trace-CMP", coin=5, status="granted",
|
||||
))
|
||||
db.add(AdFeedRewardRecord(
|
||||
client_event_id="evt-cpn", user_id=user.id, reward_date="2026-08-07",
|
||||
duration_seconds=10, unit_count=1, ecpm_raw="1000",
|
||||
feed_scene="coupon", trace_id="trace-CPN", coin=7, status="granted",
|
||||
))
|
||||
# 广告行存在但 trace_id 为空(2026-07-15 前的老比价广告):回填必须跳过、保持 NULL(EXISTS 守护)。
|
||||
db.add(AdFeedRewardRecord(
|
||||
client_event_id="evt-null", user_id=user.id, reward_date="2026-08-07",
|
||||
duration_seconds=10, unit_count=1, ecpm_raw="1000",
|
||||
feed_scene="comparison", trace_id=None, coin=5, status="granted",
|
||||
))
|
||||
db.commit()
|
||||
_, c1 = crud_wallet.grant_coins(db, user.id, 5, biz_type="feed_ad_reward_comparison", ref_id="evt-cmp", remark="比价奖励")
|
||||
_, c2 = crud_wallet.grant_coins(db, user.id, 7, biz_type="feed_ad_reward_coupon", ref_id="evt-cpn", remark="领券奖励")
|
||||
_, c3 = crud_wallet.grant_coins(db, user.id, 30, biz_type="signin", remark="每日签到奖励")
|
||||
_, c4 = crud_wallet.grant_coins(db, user.id, 5, biz_type="feed_ad_reward_comparison", ref_id="evt-null", remark="比价奖励")
|
||||
db.commit()
|
||||
assert c1.trace_id is None and c2.trace_id is None
|
||||
|
||||
db.execute(text(
|
||||
"""
|
||||
UPDATE coin_transaction SET trace_id = (
|
||||
SELECT r.trace_id FROM ad_feed_reward_record r
|
||||
WHERE r.client_event_id = coin_transaction.ref_id)
|
||||
WHERE biz_type IN ('feed_ad_reward_comparison', 'feed_ad_reward_coupon')
|
||||
AND trace_id IS NULL
|
||||
AND EXISTS (SELECT 1 FROM ad_feed_reward_record r2
|
||||
WHERE r2.client_event_id = coin_transaction.ref_id
|
||||
AND r2.trace_id IS NOT NULL)
|
||||
"""
|
||||
))
|
||||
db.commit()
|
||||
db.refresh(c1)
|
||||
db.refresh(c2)
|
||||
db.refresh(c3)
|
||||
db.refresh(c4)
|
||||
assert c1.trace_id == "trace-CMP"
|
||||
assert c2.trace_id == "trace-CPN"
|
||||
assert c3.trace_id is None # 无关类型不动
|
||||
assert c4.trace_id is None # 广告行 trace_id 为空 → EXISTS 守护跳过、保持 NULL
|
||||
|
||||
|
||||
def test_grant_feed_reward_sets_coin_trace_id(client) -> None:
|
||||
"""grant_feed_reward(comparison) 把 trace_id 透传给 grant_coins,coin_transaction 带上本场 trace_id。"""
|
||||
phone = "13800002003"
|
||||
_login(client, phone)
|
||||
with SessionLocal() as db:
|
||||
user = get_user_by_phone(db, phone)
|
||||
assert user is not None
|
||||
rec = crud_feed.grant_feed_reward(
|
||||
db, user.id,
|
||||
client_event_id="evt-fr-1", ecpm="1000", duration_seconds=10,
|
||||
feed_scene="comparison", trace_id="trace-FR", display_coin=5,
|
||||
)
|
||||
assert rec.status == "granted", rec.status
|
||||
txn = db.execute(
|
||||
select(CoinTransaction).where(
|
||||
CoinTransaction.user_id == user.id,
|
||||
CoinTransaction.ref_id == "evt-fr-1",
|
||||
)
|
||||
).scalar_one()
|
||||
assert txn.biz_type == "feed_ad_reward_comparison"
|
||||
assert txn.trace_id == "trace-FR"
|
||||
|
||||
|
||||
def _seed_coin(db, user_id, amount, biz_type, *, trace_id=None, ref_id=None, remark=None):
|
||||
crud_wallet.grant_coins(
|
||||
db, user_id, amount, biz_type=biz_type, ref_id=ref_id, remark=remark, trace_id=trace_id
|
||||
)
|
||||
|
||||
|
||||
def test_coin_transactions_aggregate_by_trace(client) -> None:
|
||||
"""一次比价的多条广告金币聚合成一条:金额合计、merged_count=条数、balance_after 取最后一条。"""
|
||||
phone = "13800002004"
|
||||
token = _login(client, phone)
|
||||
with SessionLocal() as db:
|
||||
user = get_user_by_phone(db, phone)
|
||||
assert user is not None
|
||||
_seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="t1", ref_id="e1", remark="比价奖励")
|
||||
_seed_coin(db, user.id, 3, "feed_ad_reward_comparison", trace_id="t1", ref_id="e2", remark="比价奖励")
|
||||
_seed_coin(db, user.id, 4, "feed_ad_reward_comparison", trace_id="t1", ref_id="e3", remark="比价奖励")
|
||||
db.commit()
|
||||
r = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token))
|
||||
assert r.status_code == 200, r.text
|
||||
items = r.json()["items"]
|
||||
assert len(items) == 1
|
||||
row = items[0]
|
||||
assert row["biz_type"] == "feed_ad_reward_comparison"
|
||||
assert row["amount"] == 12
|
||||
assert row["merged_count"] == 3
|
||||
assert row["balance_after"] == 12
|
||||
|
||||
|
||||
def test_coin_transactions_distinct_traces_stay_separate(client) -> None:
|
||||
"""不同 trace(两次比价 / 一次领券)各成一条;不同会话不合并。"""
|
||||
phone = "13800002005"
|
||||
token = _login(client, phone)
|
||||
with SessionLocal() as db:
|
||||
user = get_user_by_phone(db, phone)
|
||||
assert user is not None
|
||||
_seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="cmpA", ref_id="a1", remark="比价奖励")
|
||||
_seed_coin(db, user.id, 6, "feed_ad_reward_comparison", trace_id="cmpB", ref_id="b1", remark="比价奖励")
|
||||
_seed_coin(db, user.id, 6, "feed_ad_reward_comparison", trace_id="cmpB", ref_id="b2", remark="比价奖励")
|
||||
_seed_coin(db, user.id, 7, "feed_ad_reward_coupon", trace_id="cpnC", ref_id="c1", remark="领券奖励")
|
||||
db.commit()
|
||||
items = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token)).json()["items"]
|
||||
assert len(items) == 3
|
||||
assert sorted(i["amount"] for i in items) == [5, 7, 12]
|
||||
cpn = next(i for i in items if i["biz_type"] == "feed_ad_reward_coupon")
|
||||
assert cpn["amount"] == 7 and cpn["merged_count"] == 1
|
||||
|
||||
|
||||
def test_coin_transactions_non_session_rows_stay_per_row(client) -> None:
|
||||
"""签到 / 无 trace 的通用信息流各自一行,不被聚合;夹在比价广告中间的签到不影响比价聚合。"""
|
||||
phone = "13800002006"
|
||||
token = _login(client, phone)
|
||||
with SessionLocal() as db:
|
||||
user = get_user_by_phone(db, phone)
|
||||
assert user is not None
|
||||
_seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="t1", ref_id="e1", remark="比价奖励")
|
||||
_seed_coin(db, user.id, 30, "signin", remark="每日签到奖励")
|
||||
_seed_coin(db, user.id, 4, "feed_ad_reward_comparison", trace_id="t1", ref_id="e2", remark="比价奖励")
|
||||
_seed_coin(db, user.id, 8, "feed_ad_reward", trace_id=None, ref_id="w1", remark="信息流广告奖励")
|
||||
db.commit()
|
||||
items = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token)).json()["items"]
|
||||
assert len(items) == 3
|
||||
cmp_row = next(i for i in items if i["biz_type"] == "feed_ad_reward_comparison")
|
||||
assert cmp_row["amount"] == 9 and cmp_row["merged_count"] == 2
|
||||
signin_row = next(i for i in items if i["biz_type"] == "signin")
|
||||
assert signin_row["amount"] == 30 and signin_row["merged_count"] == 1
|
||||
feed_row = next(i for i in items if i["biz_type"] == "feed_ad_reward")
|
||||
assert feed_row["amount"] == 8 and feed_row["merged_count"] == 1
|
||||
|
||||
|
||||
def test_coin_transactions_pagination_no_phantom_regroup(client) -> None:
|
||||
"""交错跨游标不产生残组:会话广告成员被其它记录隔开、rep_id 在游标上、成员在游标下时,
|
||||
翻到下一页该会话不得以「残组」重复出现(锁死 spec §11 反面优化警示)。"""
|
||||
phone = "13800002007"
|
||||
token = _login(client, phone)
|
||||
with SessionLocal() as db:
|
||||
user = get_user_by_phone(db, phone)
|
||||
assert user is not None
|
||||
_seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="t1", ref_id="e1", remark="比价奖励") # id=n+1
|
||||
_seed_coin(db, user.id, 30, "signin", remark="每日签到奖励") # id=n+2
|
||||
_seed_coin(db, user.id, 40, "signin", remark="每日签到奖励") # id=n+3
|
||||
_seed_coin(db, user.id, 4, "feed_ad_reward_comparison", trace_id="t1", ref_id="e2", remark="比价奖励") # id=n+4 = t1 的 rep
|
||||
db.commit()
|
||||
t1_ids = db.execute(
|
||||
select(CoinTransaction.id)
|
||||
.where(CoinTransaction.user_id == user.id, CoinTransaction.trace_id == "t1")
|
||||
.order_by(CoinTransaction.id)
|
||||
).scalars().all()
|
||||
# 第 1 页(limit=2):按 rep 降序 = [t1(rep=n+4, 合计 9), signin(n+3, 40)]
|
||||
p1 = client.get("/api/v1/wallet/coin-transactions?limit=2", headers=_auth(token)).json()
|
||||
assert len(p1["items"]) == 2
|
||||
assert p1["items"][0]["biz_type"] == "feed_ad_reward_comparison"
|
||||
assert p1["items"][0]["amount"] == 9 and p1["items"][0]["merged_count"] == 2
|
||||
assert p1["items"][1]["biz_type"] == "signin" and p1["items"][1]["amount"] == 40
|
||||
assert p1["next_cursor"] is not None
|
||||
# 前置条件:游标(第1页末条 rep)必须严格落在 t1 两成员 id 之间,才是真正的「跨游标」场景;
|
||||
# 否则用例会退化成非跨游标、悄悄测错形状仍通过,失去回归守护意义。
|
||||
assert t1_ids[0] < p1["next_cursor"] < t1_ids[-1]
|
||||
# 第 2 页:只剩另一条 signin(30);t1 的 rep 在游标上,成员虽在游标下也不得成残组重复
|
||||
p2 = client.get(
|
||||
f"/api/v1/wallet/coin-transactions?limit=2&cursor={p1['next_cursor']}",
|
||||
headers=_auth(token),
|
||||
).json()
|
||||
assert len(p2["items"]) == 1
|
||||
assert p2["items"][0]["biz_type"] == "signin" and p2["items"][0]["amount"] == 30
|
||||
assert all(i["biz_type"] != "feed_ad_reward_comparison" for i in p2["items"])
|
||||
assert p2["next_cursor"] is None
|
||||
|
||||
Reference in New Issue
Block a user