修复：引导视频并发刷金币 —— (user_id,seq) 唯一键 + 发币条件更新幂等

start_play 无锁 check-then-insert，并发 /start 会算出同一个 seq、各拿一个
play_token 绕过次数上限刷金币；加 (user_id, seq) 唯一键，撞键即降级为不放视频。
grant_play 改成 status='playing'→'granted' 条件更新（compare-and-set），
播完与关闭抢跑/重试只会命中一条，不再二次铸币。含迁移与测试。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

alembic/versions/guide_video_play_user_seq_unique.py

@@ -0,0 +1,40 @@
+"""guide_video_play 加 (user_id, seq) 唯一约束:堵住并发 /start 绕过次数上限
+
+start_play 是无锁 check-then-insert(读 COUNT(*) 算 seq=used+1 再插一行),N 个并发
+/start 会都读到同一个 used、算出同一个 seq、各插一行拿到各自的 play_token,于是 3 次
+上限被绕过、每个 token 都能换 120 金币。加唯一键后并发同 seq 必撞,start_play 捕获
+IntegrityError 降级为 should_play=false(客户端照旧放广告)。
+
+用 unique index 而不是 batch_alter_table 加 UniqueConstraint:SQLite 加约束要整表重建,
+而 CREATE UNIQUE INDEX 两边都原生支持,回滚也干净。
+
+注:若库里已有并发产生的重复 (user_id, seq),建索引会失败 —— 本功能尚未上线,表通常是空的;
+真撞上了先按 seq 去重(留 id 最小的一行,多发的金币按 scripts/reset_guide_video.py 的口径退)。
+
+Revision ID: guide_video_user_seq_uq
+Revises: d9c03cc3ea07
+Create Date: 2026-07-24 10:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "guide_video_user_seq_uq"
+down_revision: Union[str, Sequence[str], None] = "d9c03cc3ea07"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_index(
+        "uq_guide_video_play_user_seq",
+        "guide_video_play",
+        ["user_id", "seq"],
+        unique=True,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("uq_guide_video_play_user_seq", table_name="guide_video_play")

app/models/guide_video.py

@@ -8,7 +8,10 @@
 - **开播即计数**:客户端每次要展示浮层时调 `/api/v1/guide-video/start`,服务端当场
   写一行(status='playing')并返回 play_token;`COUNT(*)` 即已用次数。用户中途 kill
   App 也算用掉一次(产品选定口径,防反复进出刷金币)。
-- **发币幂等**靠 play_token 唯一键:同一次播放重复上报只发一次。
+- **发币幂等**靠 play_token 定位 + `status='playing'` 条件更新:并发两次上报只有一次
+  改到行(另一次 rowcount=0),所以只发一次币。光有 play_token 唯一键挡不住 —— 发币走的是
+  UPDATE,不 INSERT,撞不到任何唯一键。
+- **次数上限**靠 (user_id, seq) 唯一键兜底,防并发 /start 绕过 COUNT 判定(见下)。
 
 与广告收益(ad_feed_reward_record)彻底分离:引导视频不是广告,不该进广告收益报表。
 """
@@ -16,7 +19,7 @@ from __future__ import annotations
 
 from datetime import datetime
 
-from sqlalchemy import DateTime, ForeignKey, Integer, String, UniqueConstraint, func
+from sqlalchemy import DateTime, ForeignKey, Index, Integer, String, UniqueConstraint, func
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.db.base import Base
@@ -29,6 +32,13 @@ class GuideVideoPlay(Base):
     __table_args__ = (
         # 客户端幂等键:同一次播放重复上报奖励只发一次。
         UniqueConstraint("play_token", name="uq_guide_video_play_token"),
+        # 次数上限的**硬约束**:start_play 是无锁 check-then-insert(读 COUNT 算 seq 再插),
+        # N 个并发 /start 会都读到同一个已用次数、算出同一个 seq,不拦就能各拿一个 token、
+        # 各发一次金币,3 次上限形同虚设(改包即可无限刷)。seq 唯一 → 并发同 seq 必撞,
+        # start_play 捕获 IntegrityError 降级成"这次不放视频"。
+        # 用 unique Index 而非 UniqueConstraint:与迁移里的 create_index 对齐(SQLite 加约束
+        # 要整表重建),autogenerate 才不会每次报一条假 diff。
+        Index("uq_guide_video_play_user_seq", "user_id", "seq", unique=True),
     )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
@@ -39,7 +49,8 @@ class GuideVideoPlay(Base):
     play_token: Mapped[str] = mapped_column(String(64), nullable=False)
     # 触发场景:目前只有 coupon(领券等候浮层);留字段以便日后比价等场景复用。
     scene: Mapped[str] = mapped_column(String(16), nullable=False, default="coupon")
-    # 本账号第几次(1-based),= 建行时已有行数 + 1。仅留痕/排查用,判定仍以 COUNT 为准。
+    # 本账号第几次(1-based),= 建行时已有行数 + 1。日常判定仍以 COUNT 为准,但 (user_id, seq)
+    # 唯一键让并发 /start 只能成一个 —— 见 __table_args__。
     seq: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
     # 当次下发的视频地址(运营换片后能回溯用户当时看的是哪支)。
     video_url: Mapped[str | None] = mapped_column(String(512), nullable=True)

app/repositories/guide_video.py

@@ -7,9 +7,13 @@
 **计次**按账号(user_id)、**开播即计数**:客户端每次要展示领券等候浮层时调
 `/api/v1/guide-video/start`,命中则当场写一行 guide_video_play(status='playing')。
 已用次数 = 该账号的行数,达到 max_plays 后不再下发,客户端改放广告(原逻辑)。
+COUNT 判定本身无锁,真正卡住次数上限的是 (user_id, seq) 唯一键:并发 /start 只能成一个。
 
-**发币**幂等键是 play_token:同一次播放重复上报只入账一次(网络重试 / 关闭与播完同时触发
-都靠它挡住)。中途关闭也照发 —— 产品拍板「中途关闭也算看完」。
+**发币**幂等键是 play_token,落地方式是 `status='playing' → 'granted'` 的**条件更新**:
+同一次播放重复上报只入账一次(网络重试 / 关闭与播完同时触发都靠它挡住)。
+中途关闭也照发 —— 产品拍板「中途关闭也算看完」。
+
+两处都是直接铸币的路径,改动前先看 `start_play` / `grant_play` 上的并发注释。
 """
 from __future__ import annotations
 
@@ -17,7 +21,7 @@ import uuid
 from datetime import datetime
 from typing import Any
 
-from sqlalchemy import func, select
+from sqlalchemy import func, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -169,16 +173,19 @@ def start_play(
     max_plays = int(cfg.get("max_plays") or 0)
     reward_coin = int(cfg.get("reward_coin") or 0)
     used = used_plays(db, user_id)
-    miss = {
-        "should_play": False,
-        "video_url": None,
-        "play_token": "",
-        "reward_coin": reward_coin,
-        "seq": used,
-        "remaining": max(0, max_plays - used),
-    }
+
+    def _miss(used_now: int) -> dict[str, Any]:
+        return {
+            "should_play": False,
+            "video_url": None,
+            "play_token": "",
+            "reward_coin": reward_coin,
+            "seq": used_now,
+            "remaining": max(0, max_plays - used_now),
+        }
+
     if not cfg.get("enabled") or not video_url or max_plays <= 0 or used >= max_plays:
-        return miss
+        return _miss(used)
 
     seq = used + 1
     play = GuideVideoPlay(
@@ -193,10 +200,17 @@ def start_play(
         started_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
     )
     db.add(play)
-    if commit:
-        db.commit()
-    else:
-        db.flush()
+    # 上面的 COUNT 判定是无锁 check-then-insert:并发 /start 会都算出同一个 seq。
+    # (user_id, seq) 唯一键让只有一个能落库,其余撞键 → 回滚后按"这次不放视频"降级,
+    # 客户端照旧走广告链路。没有它,并发就能绕过 max_plays 无限刷金币。
+    try:
+        if commit:
+            db.commit()
+        else:
+            db.flush()
+    except IntegrityError:
+        db.rollback()
+        return _miss(used_plays(db, user_id))
     return {
         "should_play": True,
         "video_url": video_url,
@@ -207,6 +221,16 @@ def start_play(
     }
 
 
+def _find_play(db: Session, user_id: int, token: str) -> GuideVideoPlay | None:
+    """按 (play_token, user_id) 取播放行 —— 带 user_id 是防拿别人的 token 来兑。"""
+    return db.execute(
+        select(GuideVideoPlay).where(
+            GuideVideoPlay.play_token == token,
+            GuideVideoPlay.user_id == user_id,
+        )
+    ).scalar_one_or_none()
+
+
 def grant_play(
     db: Session, user_id: int, *, play_token: str, completed: bool
 ) -> dict[str, Any]:
@@ -216,23 +240,43 @@ def grant_play(
     重复上报返回 granted=False + 已发金币(客户端据此不重复累加 toast 金额)。
     """
     token = (play_token or "").strip()
-    play = db.execute(
-        select(GuideVideoPlay).where(
-            GuideVideoPlay.play_token == token,
-            GuideVideoPlay.user_id == user_id,
-        )
-    ).scalar_one_or_none()
-    if play is None:
-        return {"granted": False, "coin": 0, "status": "not_found"}
-    if play.status == "granted":
-        return {"granted": False, "coin": play.coin, "status": "already_granted"}
-
     # 金币额度以**服务端配置**为准,不信客户端(客户端只上报"播完/关闭")。
     coin = int(get_config(db).get("reward_coin") or 0)
-    play.completed = 1 if completed else 0
-    play.status = "granted"
-    play.coin = coin
-    play.granted_at = datetime.now(rewards.CN_TZ).replace(tzinfo=None)
+
+    # 幂等核心:把 status 放进 WHERE 做条件更新(compare-and-set),而不是"先读再判再写"。
+    # 「播完」与「✕ 关闭」抢跑、或客户端超时重试时,两个请求会都读到 status='playing',
+    # 无锁的话就都往下发币、都 commit,金币入账两次(不用恶意,重试就会中招)。改成条件更新后
+    # 并发里只有一条 rowcount=1,另一条拿 0 → 按已发返回,不二次铸币。
+    # (PG READ COMMITTED 下后到的 UPDATE 阻塞到对手提交,再按新版本重判 status;SQLite 写串行。)
+    #
+    # 别指望 IntegrityError 兜底:这里只 UPDATE 不 INSERT,撞不到 uq_guide_video_play_token;
+    # 而 biz_type='guide_video' 的金币流水也不在 ux_coin_transaction_task_ref 的谓词
+    # (biz_type LIKE 'task%')覆盖范围内 —— 两个唯一键在这条路径上都是不生效的。
+    won = db.execute(
+        update(GuideVideoPlay)
+        .where(
+            GuideVideoPlay.play_token == token,
+            GuideVideoPlay.user_id == user_id,
+            GuideVideoPlay.status == "playing",
+        )
+        .values(
+            status="granted",
+            coin=coin,
+            completed=1 if completed else 0,
+            granted_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
+        )
+        .execution_options(synchronize_session=False)
+    ).rowcount
+
+    if not won:
+        # 没抢到:token 不存在 / 不是本人的 / 已被另一次上报发过。回滚拿干净快照再区分两者
+        # (对手此时必然已提交 —— 我们就是被它挡下的,所以读得到它写的 coin)。
+        db.rollback()
+        play = _find_play(db, user_id, token)
+        if play is None:
+            return {"granted": False, "coin": 0, "status": "not_found"}
+        return {"granted": False, "coin": play.coin, "status": "already_granted"}
+
     if coin > 0:
         crud_wallet.grant_coins(
             db,
@@ -242,16 +286,5 @@ def grant_play(
             ref_id=token,
             remark="新手引导视频奖励",
         )
-    try:
-        db.commit()
-    except IntegrityError:
-        # 并发双发(播完 + ✕ 同时到)时其中一条会撞唯一键/行锁,回滚后按已发返回。
-        db.rollback()
-        again = db.execute(
-            select(GuideVideoPlay).where(
-                GuideVideoPlay.play_token == token,
-                GuideVideoPlay.user_id == user_id,
-            )
-        ).scalar_one_or_none()
-        return {"granted": False, "coin": again.coin if again else 0, "status": "already_granted"}
+    db.commit()
     return {"granted": True, "coin": coin, "status": "granted"}

tests/test_guide_video.py

@@ -0,0 +1,263 @@
+"""新手引导视频:次数上限 + 发币幂等。
+
+这条链路直接铸币,两处并发缺陷曾经都是真漏洞,所以本文件的重点不是走通 happy path,
+而是**并发失败分支**:
+  - `/start` 并发算出同一个 seq → (user_id, seq) 唯一键挡下,降级为"这次不放视频";
+  - `/reward` 并发抢跑 → status 条件更新只让一个 rowcount=1,输的那个不得入账。
+
+真并发在 SQLite 测试库里复现不了(读事务会直接把写方锁死,而不是让它读到旧快照),
+所以用"把对手已经落库的状态先摆好、再让被测方按旧读数往下走"来模拟,断言的是同一件事:
+输的一方必须空手而归,且账不能乱。
+"""
+from __future__ import annotations
+
+import pytest
+from sqlalchemy import func, select
+
+from app.db.session import SessionLocal
+from app.models.guide_video import GuideVideoPlay
+from app.models.wallet import CoinAccount, CoinTransaction
+from app.repositories import guide_video as crud_guide
+from app.repositories.user import get_user_by_phone
+
+VIDEO_URL = "/media/guide_video/pytest_guide.mp4"
+
+
+def _login(client, phone: str) -> str:
+    client.post("/api/v1/auth/sms/send", json={"phone": phone})
+    r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
+    assert r.status_code == 200, r.text
+    return r.json()["access_token"]
+
+
+def _auth(token: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {token}"}
+
+
+def _user_id(phone: str) -> int:
+    with SessionLocal() as db:
+        user = get_user_by_phone(db, phone)
+        assert user is not None
+        return user.id
+
+
+def _coin_balance(user_id: int) -> int:
+    with SessionLocal() as db:
+        acc = db.get(CoinAccount, user_id)
+        return acc.coin_balance if acc else 0
+
+
+def _guide_txns(user_id: int) -> list[CoinTransaction]:
+    """该用户的引导视频金币流水(按 id 升序)。"""
+    with SessionLocal() as db:
+        return list(db.execute(
+            select(CoinTransaction)
+            .where(
+                CoinTransaction.user_id == user_id,
+                CoinTransaction.biz_type == crud_guide.BIZ_TYPE,
+            )
+            .order_by(CoinTransaction.id)
+        ).scalars().all())
+
+
+def _assert_ledger_balanced(user_id: int) -> None:
+    """coin_balance 必须恒等于流水总和 —— 双发/丢更新都会先在这里露馅。"""
+    with SessionLocal() as db:
+        total = db.execute(
+            select(func.coalesce(func.sum(CoinTransaction.amount), 0)).where(
+                CoinTransaction.user_id == user_id
+            )
+        ).scalar_one()
+        acc = db.get(CoinAccount, user_id)
+        assert acc is not None
+        assert acc.coin_balance == total, f"余额 {acc.coin_balance} != 流水总和 {total}"
+
+
+@pytest.fixture()
+def guide_configured():
+    """给全局配置塞一支片子(默认 max_plays=3 / reward_coin=120),用完还原成未配片。"""
+    with SessionLocal() as db:
+        crud_guide.set_video(db, VIDEO_URL, admin_id=1)
+        cfg = crud_guide.get_config(db)
+    yield cfg
+    with SessionLocal() as db:
+        crud_guide.set_video(db, None, admin_id=1)
+
+
+# ===== 基本闭环 =====
+
+
+def test_start_miss_when_no_video_configured(client) -> None:
+    """没配片 → should_play=False,客户端照旧放广告(本功能不配视频就等于没上线)。"""
+    token = _login(client, "13920000001")
+    r = client.post("/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token))
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["should_play"] is False
+    assert body["play_token"] == ""
+
+
+def test_start_then_reward_grants_once(client, guide_configured) -> None:
+    """开播 → 上报 → 到账 120;重复上报不再加钱,只回已发金额。"""
+    cfg = guide_configured
+    phone = "13920000002"
+    token = _login(client, phone)
+    uid = _user_id(phone)
+
+    r = client.post("/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token))
+    body = r.json()
+    assert body["should_play"] is True
+    assert body["video_url"] == VIDEO_URL
+    assert body["seq"] == 1
+    assert body["reward_coin"] == cfg["reward_coin"]
+    play_token = body["play_token"]
+    assert play_token
+
+    r = client.post(
+        "/api/v1/guide-video/reward",
+        json={"play_token": play_token, "completed": True},
+        headers=_auth(token),
+    )
+    assert r.status_code == 200, r.text
+    assert r.json() == {"granted": True, "coin": cfg["reward_coin"], "status": "granted"}
+    assert _coin_balance(uid) == cfg["reward_coin"]
+
+    # 重复上报(客户端超时重试 / 播完与 ✕ 都报了一次)
+    r = client.post(
+        "/api/v1/guide-video/reward",
+        json={"play_token": play_token, "completed": False},
+        headers=_auth(token),
+    )
+    assert r.json() == {"granted": False, "coin": cfg["reward_coin"], "status": "already_granted"}
+    assert _coin_balance(uid) == cfg["reward_coin"], "重复上报不得二次入账"
+    assert len(_guide_txns(uid)) == 1, "一个 play_token 只能有一条金币流水"
+    _assert_ledger_balanced(uid)
+
+
+def test_reward_rejects_unknown_and_others_token(client, guide_configured) -> None:
+    """乱填 token / 拿别人的 token 都发不出币(grant 按 user_id + token 双条件定位)。"""
+    victim = _login(client, "13920000003")
+    r = client.post("/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(victim))
+    stolen = r.json()["play_token"]
+
+    attacker_phone = "13920000004"
+    attacker = _login(client, attacker_phone)
+    attacker_uid = _user_id(attacker_phone)
+
+    r = client.post(
+        "/api/v1/guide-video/reward",
+        json={"play_token": stolen, "completed": True},
+        headers=_auth(attacker),
+    )
+    assert r.json() == {"granted": False, "coin": 0, "status": "not_found"}
+
+    r = client.post(
+        "/api/v1/guide-video/reward",
+        json={"play_token": "deadbeef" * 4, "completed": True},
+        headers=_auth(attacker),
+    )
+    assert r.json() == {"granted": False, "coin": 0, "status": "not_found"}
+    assert _coin_balance(attacker_uid) == 0
+
+
+def test_start_stops_at_max_plays(client, guide_configured) -> None:
+    """开播即计次:用满 max_plays 后不再下发,客户端回到广告链路。"""
+    max_plays = int(guide_configured["max_plays"])
+    token = _login(client, "13920000005")
+
+    for i in range(1, max_plays + 1):
+        body = client.post(
+            "/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token)
+        ).json()
+        assert body["should_play"] is True, f"第 {i} 次应该还能放"
+        assert body["seq"] == i
+        assert body["remaining"] == max_plays - i
+
+    body = client.post(
+        "/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token)
+    ).json()
+    assert body["should_play"] is False
+    assert body["remaining"] == 0
+
+
+# ===== 并发失败分支(回归) =====
+
+
+def test_start_loses_seq_race_degrades_to_ad(client, guide_configured, monkeypatch) -> None:
+    """并发 /start 抢到同一个 seq 时,输的一方降级成"不放视频",而不是多拿一个 token。
+
+    模拟:对手已经提交了 seq=1,而本次请求读到的还是旧计数(used=0)—— 这正是无锁
+    check-then-insert 的race window。没有 (user_id, seq) 唯一键的话,这里会插成第二行、
+    换回第二个 play_token,3 次上限就能被并发无限绕过(改包即可 N 倍刷币)。
+    """
+    phone = "13920000006"
+    token = _login(client, phone)
+    uid = _user_id(phone)
+
+    # 对手先落一行 seq=1
+    first = client.post(
+        "/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token)
+    ).json()
+    assert first["should_play"] is True and first["seq"] == 1
+
+    # 本次请求读到的是过期计数 → 仍会算出 seq=1
+    monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id: 0)
+    with SessionLocal() as db:
+        result = crud_guide.start_play(db, uid)
+
+    assert result["should_play"] is False, "撞 seq 唯一键后必须降级,不能再发一个 token"
+    assert result["play_token"] == ""
+
+    monkeypatch.undo()
+    with SessionLocal() as db:
+        rows = db.execute(
+            select(func.count()).select_from(GuideVideoPlay).where(
+                GuideVideoPlay.user_id == uid
+            )
+        ).scalar_one()
+    assert rows == 1, "抢输的那次不该留下播放行"
+
+
+def test_reward_loses_race_does_not_double_mint(client, guide_configured) -> None:
+    """并发 /reward 抢输的一方不得入账 —— 条件更新(status 进 WHERE)的失败分支。
+
+    模拟的是真并发的**必要条件**:输的一方在对手提交之前就已经读到 status='playing',
+    之后带着这个旧认知继续往下走(「播完」与「✕ 关闭」抢跑、或超时重试都会造出这一幕)。
+    这里靠 Session 的 identity map 固化那次旧读数 —— 后续同一 Session 再查同一行,拿回的
+    还是这份旧快照,等价于 PG READ COMMITTED 下两个事务都读到 playing。
+
+    旧实现在这里会照发第二笔:它先读再判再写,而 `except IntegrityError` 兜底根本不可能
+    触发 —— 发币走 UPDATE 撞不到 uq_guide_video_play_token,biz_type='guide_video' 的流水
+    也不在 ux_coin_transaction_task_ref 的谓词(biz_type LIKE 'task%')覆盖内。
+    """
+    coin = int(guide_configured["reward_coin"])
+    phone = "13920000007"
+    token = _login(client, phone)
+    uid = _user_id(phone)
+
+    play_token = client.post(
+        "/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token)
+    ).json()["play_token"]
+
+    db_loser = SessionLocal()
+    try:
+        # 输的一方先读到 playing —— 并发下两个请求都会读到它
+        stale = db_loser.execute(
+            select(GuideVideoPlay).where(GuideVideoPlay.play_token == play_token)
+        ).scalar_one()
+        assert stale.status == "playing"
+
+        # 对手抢先发币并提交
+        with SessionLocal() as db_winner:
+            won = crud_guide.grant_play(db_winner, uid, play_token=play_token, completed=True)
+        assert won == {"granted": True, "coin": coin, "status": "granted"}
+
+        # 输的一方带着旧认知继续:必须空手而归
+        lost = crud_guide.grant_play(db_loser, uid, play_token=play_token, completed=False)
+        assert lost == {"granted": False, "coin": coin, "status": "already_granted"}
+    finally:
+        db_loser.close()
+
+    assert _coin_balance(uid) == coin, "一次播放只能发一次币"
+    assert len(_guide_txns(uid)) == 1, "一个 play_token 只能有一条金币流水"
+    _assert_ledger_balanced(uid)