feat: 信息流广告改"所见即所得"发奖——直接发客户端小球显示金币 (#86)

Reviewed-on: #86
Co-authored-by: wuqi <wuqi@wonderable.ai>
Co-committed-by: wuqi <wuqi@wonderable.ai>

app/api/v1/ad.py

@@ -411,6 +411,7 @@ def feed_reward(payload: FeedRewardIn, user: CurrentUser, db: DbSession) -> Feed
         app_env=payload.app_env,
         our_code_id=payload.our_code_id,
         aborted=payload.aborted,
+        display_coin=payload.display_coin,
     )
     logger.info(
         "feed ad reward user_id=%d event=%s status=%s units=%d coin=%d",

app/repositories/ad_ecpm.py

@@ -53,11 +53,16 @@ def create_ecpm_record(
         db.commit()
     except IntegrityError:
         db.rollback()
-        if ad_session_id:
-            existing = find_by_session(db, user_id=user_id, ad_session_id=ad_session_id)
-            if existing is not None:
-                return existing
-        raise
+        # 撞唯一约束 uq_ad_ecpm_record_session(全局按 ad_session_id、不含 user_id):并发同会话重复上报,
+        # 或同一 ad_session_id 已被先到的上报占用。本接口 fire-and-forget、best-effort —— 丢一条不影响业务
+        # (穿山甲后台才是结算权威),绝不向客户端抛 500。兜底查找须与唯一约束**同口径**(只按 ad_session_id、
+        # 不带 user_id):否则不同 user 上报了同一 ad_session_id 时,带 user_id 的查找会漏掉那条别人的记录 →
+        # 旧逻辑在此 raise 成 500(本应静默吞掉)。
+        existing = _find_by_session_global(db, ad_session_id)
+        if existing is not None:
+            return existing
+        # 极少:rollback 后既存记录又查不到(并发删除 / 竞态)。吞掉、返回未入库的内存对象(调用方不读返回值)。
+        return rec
     db.refresh(rec)
     return rec
 
@@ -76,6 +81,20 @@ def find_by_session(
     ).scalar_one_or_none()
 
 
+def _find_by_session_global(db: Session, ad_session_id: str | None) -> AdEcpmRecord | None:
+    """按 ad_session_id **全局**查找(与唯一约束 uq_ad_ecpm_record_session 同口径,不含 user_id)。
+
+    仅 create_ecpm_record 撞约束后兜底用:此时撞的是全局会话约束,既存记录可能属于**另一个 user**,
+    带 user_id 的 find_by_session 会漏掉它、导致误判「查无 → raise 500」。其它业务查「某 user 的某次
+    展示 eCPM」仍用 find_by_session(带 user_id,语义更准),不走这里。
+    """
+    if not ad_session_id:
+        return None
+    return db.execute(
+        select(AdEcpmRecord).where(AdEcpmRecord.ad_session_id == ad_session_id)
+    ).scalar_one_or_none()
+
+
 def count_today(db: Session, user_id: int) -> int:
     """该用户今日(北京时间)上报的 eCPM 条数,排查/对账辅助用。"""
     return db.execute(

app/repositories/ad_feed_reward.py

@@ -73,16 +73,19 @@ def grant_feed_reward(
     app_env: str | None = None,
     our_code_id: str | None = None,
     aborted: bool = False,
+    display_coin: int = 0,
 ) -> AdFeedRewardRecord:
     """**每条**信息流广告(客户端每条各上报一次)结算奖励。client_event_id 幂等,同号重试不重复发。
 
-    发奖规则:**一条广告 = 一个单次公式值**(rewards.calculate_ad_reward_coin),因子2(LT)按账号累计
-    **条**数递进;看满一份时长(unit_count>=1, 即 ≥10 秒)才发,**不逐份累加**。
+    发奖规则(所见即所得, 2026-06-27 用户拍板「显示多少给多少」):优先**直接发客户端小球显示的金币
+    display_coin**;防刷钳到本条「1 份满额」(eCPM 已钳 AD_ECPM_MAX_FEN, 因子2 按账号累计已发条数取档),
+    合法显示(实际因子2 × 进度 p ≤ 1 份)不被砍, 只挡伪造天价值。旧客户端不传 display_coin 时退回
+    「看满 10 秒发整份」(兼容不断币)。因子2(LT)由**客户端**按 granted 行 COUNT(拉自 /feed-reward/units)
+    算进 display_coin, 后端只记 granted 行让该计数自增, 不再服务端重算份值。
     - aborted=True(用户中途 ✕ 关闭这条):本条不发,记 status='closed_early'。
-    - 时长不足 10 秒(unit_count==0):记 status='too_short' 不发。
+    - display_coin 为 0 且时长不足一份:记 status='too_short' 不发(不计 LT / 当日上限)。
     - 命中当日条数上限:记 status='capped' 不发。
-    duration_seconds 是**这一条**的观看秒数。服务端两道硬闸防刷:时长钳到 FEED_MAX_DURATION_SECONDS、
-    eCPM 在 calculate_ad_reward_coin 内钳到 AD_ECPM_MAX_FEN;叠加每日 get_ad_daily_limit 条数上限。
+    duration_seconds 落库留痕(unit_count 字段), 旧端兼容路径据它判是否满 1 份。
     feed_scene:点位场景(comparison/coupon/welfare),仅归类落库,不参与计算。
     ad_type:广告形态(feed 信息流 / draw Draw 信息流),仅归类落库;**每日上限与因子2(LT)仍按本表
     全表 unit 累计(feed+draw 共享同一发奖池/上限),不按 ad_type 拆分**。
@@ -139,14 +142,26 @@ def grant_feed_reward(
         )
         return _commit_record(db, rec, client_event_id)
 
-    # 整场总时长不足 10 秒,凑不满一份 → 不发,记 too_short 留痕。
-    if unit_count == 0:
+    # 所见即所得(用户 2026-06-27「显示多少给多少」): 优先发**客户端小球显示**的金币 display_coin,
+    # 钳到本条「1 份满额」防刷(eCPM 已钳 AD_ECPM_MAX_FEN; 合法显示=因子2×p≤1份, 不会被砍)。
+    # 因子2(LT)按账号累计已发条数(granted 行 COUNT), 第 existing_ads+1 条。
+    existing_ads = granted_unit_total(db, user_id)
+    unit_cap = rewards.calculate_ad_reward_coin(ecpm, existing_ads + 1)
+    if display_coin > 0:
+        coin = min(display_coin, unit_cap)              # 新端: 所见即所得(直接发小球显示金币)
+    elif unit_count >= 1:
+        coin = unit_cap                                 # 旧端没传 display_coin: 退回「看满 1 份发整份」(兼容)
+    else:
+        coin = 0
+
+    # 显示金币为 0 且没满一份 → 不发, 记 too_short 留痕(不写 granted 行 → 不计 LT / 当日上限)。
+    if coin <= 0:
         rec = AdFeedRewardRecord(
             client_event_id=client_event_id,
             user_id=user_id,
             reward_date=today,
             duration_seconds=safe_duration,
-            unit_count=0,
+            unit_count=unit_count,
             ad_session_id=ad_session_id,
             ecpm_raw=ecpm,
             adn=adn,
@@ -161,15 +176,11 @@ def grant_feed_reward(
         )
         return _commit_record(db, rec, client_event_id)
 
-    # 一条广告 = 一个「单次公式值」(因子2 按账号累计**条**数, 即第 existing_ads+1 条);看满一份(unit_count>=1)即发,不逐份累加。
-    existing_ads = granted_unit_total(db, user_id)
-    coin = rewards.calculate_ad_reward_coin(ecpm, existing_ads + 1)
-    if coin > 0:
-        crud_wallet.grant_coins(
-            db, user_id, coin,
-            biz_type="feed_ad_reward", ref_id=client_event_id,
-            remark="信息流广告奖励",
-        )
+    crud_wallet.grant_coins(
+        db, user_id, coin,
+        biz_type="feed_ad_reward", ref_id=client_event_id,
+        remark="信息流广告奖励",
+    )
     rec = AdFeedRewardRecord(
         client_event_id=client_event_id,
         user_id=user_id,

app/schemas/ad.py

@@ -167,6 +167,11 @@ class FeedRewardIn(BaseModel):
     aborted: bool = Field(
         False, description="用户中途 ✕ 关闭广告(未走完比价):整场不发,记 closed_early"
     )
+    display_coin: int = Field(
+        0, ge=0,
+        description="客户端金币小球**本条显示**的金币(所见即所得):后端直接发这个数,钳到本条最大 1 份"
+        "满额防刷。缺省 0 = 旧客户端不传,退回服务端「看满 10 秒发整份」",
+    )
 
 
 class FeedRewardOut(BaseModel):