From 617abbb58998a9399636163a0f3935baeba3fae2 Mon Sep 17 00:00:00 2001 From: lili Date: Thu, 2 Jul 2026 20:23:00 -0700 Subject: [PATCH] =?UTF-8?q?feat(telemetry):=20D1=20=E6=AC=A1=E7=95=99?= =?UTF-8?q?=E6=97=81=E8=B7=AF=E8=81=9A=E5=90=88(=E7=8E=A9=E5=AE=B6=C3=97?= =?UTF-8?q?=E6=B8=B8=E6=88=8F=C3=97=E6=97=A5=E5=8E=BB=E9=87=8D=C2=B7?= =?UTF-8?q?=E5=B9=B3=E5=8F=B0=E6=97=B6=E5=8C=BA=C2=B7quality=5Fscore=20?= =?UTF-8?q?=E9=93=BE=E8=B7=AF=E9=9B=B6=E9=A3=8E=E9=99=A9)=20(W-REAL=20R6)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 按质量模型设计 §3.4 口径立 L3 留存字段:同一 game_id、同一稳定玩家标识(登录取 userId、未登录取 anonId, 对齐 #5 契约 anonId 口径)在首玩自然日(平台时区,与 game_telemetry_game_stat.stat_date 同日界)后第 1 个 自然日内 ≥1 次 game_play_start,按玩家×游戏×日去重。 - V27 旁路两表:game_telemetry_player_first_play(游戏×玩家,首玩追踪+d1_returned CAS 去重) + game_telemetry_retention_stat(游戏×cohort日,new_player_count/d1_retained_count;D1率读时算)。 不动既有 game_telemetry_game_stat 结构,quality_score 链路零风险。 - RetentionAggregateService:仅吃 game_play_start,首玩 INSERT IGNORE 落点+累加 cohort;首玩日+1 回访 CAS 置位+累加分子;同日/非D1/已置位去重。挂进 R1 consumeFromMq(ingestOne 之后),best-effort 内部吞异常 不外抛——次留失败绝不连累 quality_score 主链;软注入(单测不注入即跳过,既有行为不变)。 - 玩家标识命名空间隔离(u:{userId}/a:{anonId})防碰撞;匿名→登录跨身份桥接留后续(user_login)。 验证:模块 68 单测绿(新增 9 个口径单测覆盖首玩/D1窗口/去重/匿名标识/best-effort;既有 EventIngestServiceImplTest 13 逐字不变=回灌行为不变);连 mini-infra 真 MySQL 8.4.8 聚合真跑 IT 绿(throwaway 库跑真 V27 DDL + D1 两日场景, new_player=2/d1_retained=1 断言通过)。Flyway 版本核对迁移目录 V26→取 V27。 Co-Authored-By: Claude Opus 4.8 (1M context) --- .../dataobject/stat/PlayerFirstPlayDO.java | 40 ++++ .../dal/dataobject/stat/RetentionStatDO.java | 37 ++++ .../dal/mysql/stat/PlayerFirstPlayMapper.java | 68 +++++++ .../dal/mysql/stat/RetentionStatMapper.java | 67 ++++++ .../service/event/EventIngestServiceImpl.java | 19 ++ .../retention/RetentionAggregateService.java | 28 +++ .../RetentionAggregateServiceImpl.java | 121 +++++++++++ .../retention/RetentionAggregateRealDbIT.java | 191 ++++++++++++++++++ .../RetentionAggregateServiceImplTest.java | 184 +++++++++++++++++ ...7.0.0__create_game_telemetry_retention.sql | 59 ++++++ 10 files changed, 814 insertions(+) create mode 100644 game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/dal/dataobject/stat/PlayerFirstPlayDO.java create mode 100644 game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/dal/dataobject/stat/RetentionStatDO.java create mode 100644 game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/dal/mysql/stat/PlayerFirstPlayMapper.java create mode 100644 game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/dal/mysql/stat/RetentionStatMapper.java create mode 100644 game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/service/retention/RetentionAggregateService.java create mode 100644 game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/service/retention/RetentionAggregateServiceImpl.java create mode 100644 game-cloud/game-module-telemetry/game-module-telemetry-server/src/test/java/com/wanxiang/huijing/game/module/telemetry/service/retention/RetentionAggregateRealDbIT.java create mode 100644 game-cloud/game-module-telemetry/game-module-telemetry-server/src/test/java/com/wanxiang/huijing/game/module/telemetry/service/retention/RetentionAggregateServiceImplTest.java create mode 100644 game-cloud/huijing-server/src/main/resources/db/migration/V27.0.0__create_game_telemetry_retention.sql diff --git a/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/dal/dataobject/stat/PlayerFirstPlayDO.java b/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/dal/dataobject/stat/PlayerFirstPlayDO.java new file mode 100644 index 00000000..ad9e498f --- /dev/null +++ b/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/dal/dataobject/stat/PlayerFirstPlayDO.java @@ -0,0 +1,40 @@ +package com.wanxiang.huijing.game.module.telemetry.dal.dataobject.stat; + +import com.wanxiang.huijing.framework.tenant.core.db.TenantBaseDO; +import com.baomidou.mybatisplus.annotation.KeySequence; +import com.baomidou.mybatisplus.annotation.TableName; +import lombok.Data; +import lombok.EqualsAndHashCode; + +import java.time.LocalDate; + +/** + * 玩家首玩追踪 DO（对应表 game_telemetry_player_first_play，游戏 × 玩家 粒度） + * + *

质量模型设计 §3.4 L3 留存结构落地件：每玩家对每游戏一行，记首玩自然日 + 是否在首玩日+1 自然日回访开局。 + * 继承 {@link TenantBaseDO} 自动携带 creator/create_time/updater/update_time/deleted + tenant_id。 + * 唯一约束 (game_id, player_key) 保证每玩家每游戏一行——既是首玩落点，也是 D1 去重幂等键。 + * + *

玩家标识 {@link #playerKey} 命名空间隔离：登录 = {@code u:{userId}}、未登录 = {@code a:{anonId}} + * （对齐 #5 契约 anonId 口径），防 userId 与 anonId 字面碰撞。 + * + * @author 绘境AI + */ +@TableName("game_telemetry_player_first_play") +@KeySequence("game_telemetry_player_first_play_seq") +@Data +@EqualsAndHashCode(callSuper = true) +public class PlayerFirstPlayDO extends TenantBaseDO { + + /** 记录 ID */ + private Long id; + /** 游戏 ID（= game_project.id） */ + private Long gameId; + /** 稳定玩家标识：登录 = u:{userId}、未登录 = a:{anonId}（命名空间隔离防碰撞） */ + private String playerKey; + /** 该玩家对该游戏的首玩自然日（平台时区，与 game_telemetry_game_stat.stat_date 同日界） */ + private LocalDate firstPlayDate; + /** 次留标记：0 未回 / 1 已在首玩日+1 自然日回访开局（CAS 置位一次、去重） */ + private Integer d1Returned; + +} diff --git a/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/dal/dataobject/stat/RetentionStatDO.java b/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/dal/dataobject/stat/RetentionStatDO.java new file mode 100644 index 00000000..a1376493 --- /dev/null +++ b/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/dal/dataobject/stat/RetentionStatDO.java @@ -0,0 +1,37 @@ +package com.wanxiang.huijing.game.module.telemetry.dal.dataobject.stat; + +import com.wanxiang.huijing.framework.tenant.core.db.TenantBaseDO; +import com.baomidou.mybatisplus.annotation.KeySequence; +import com.baomidou.mybatisplus.annotation.TableName; +import lombok.Data; +import lombok.EqualsAndHashCode; + +import java.time.LocalDate; + +/** + * 游戏维度 D1 次留聚合 DO（对应表 game_telemetry_retention_stat，游戏 × cohort日 粒度） + * + *

质量模型设计 §3.4 L3 留存结构聚合件：按 (game_id, stat_date=cohort日) 聚合当日首玩新玩家数 new_player_count + * 与其中次日回访开局的 d1_retained_count。D1 留存率 = d1_retained_count / new_player_count（读时计算、不落列）。 + * 独立于 game_telemetry_game_stat：不动既有聚合表结构、quality_score 链路零风险。 + * + * @author 绘境AI + */ +@TableName("game_telemetry_retention_stat") +@KeySequence("game_telemetry_retention_stat_seq") +@Data +@EqualsAndHashCode(callSuper = true) +public class RetentionStatDO extends TenantBaseDO { + + /** 聚合记录 ID */ + private Long id; + /** 游戏 ID（= game_project.id） */ + private Long gameId; + /** cohort 日 = 首玩自然日（平台时区） */ + private LocalDate statDate; + /** 当日首玩新玩家数（cohort 规模，玩家×游戏×日去重） */ + private Long newPlayerCount; + /** 当日 cohort 中在次日回访开局的玩家数（D1 留存分子） */ + private Long d1RetainedCount; + +} diff --git a/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/dal/mysql/stat/PlayerFirstPlayMapper.java b/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/dal/mysql/stat/PlayerFirstPlayMapper.java new file mode 100644 index 00000000..af802008 --- /dev/null +++ b/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/dal/mysql/stat/PlayerFirstPlayMapper.java @@ -0,0 +1,68 @@ +package com.wanxiang.huijing.game.module.telemetry.dal.mysql.stat; + +import com.wanxiang.huijing.game.module.telemetry.dal.dataobject.stat.PlayerFirstPlayDO; +import com.wanxiang.huijing.framework.mybatis.core.mapper.BaseMapperX; +import com.wanxiang.huijing.framework.mybatis.core.query.LambdaQueryWrapperX; +import org.apache.ibatis.annotations.Insert; +import org.apache.ibatis.annotations.Mapper; +import org.apache.ibatis.annotations.Param; +import org.apache.ibatis.annotations.Update; + +import java.time.LocalDate; + +/** + * 玩家首玩追踪 Mapper（D1 次留旁路，游戏 × 玩家 粒度） + * + *

并发安全靠两条原子语句：{@link #insertIgnoreFirstPlay}（INSERT IGNORE，首玩幂等落点，命中 uk_game_player 即忽略） + * + {@link #markD1ReturnedIfUnset}（CAS 置位，d1_returned 0→1 恰一次）。审计/租户缺省列依赖 DDL 默认值（MVP 单租户=0）， + * 与 {@code GameStatMapper.insertOrAccumulate} 同范式。 + * + * @author 绘境AI + */ +@Mapper +public interface PlayerFirstPlayMapper extends BaseMapperX { + + /** + * 首玩幂等落点（INSERT IGNORE）：新玩家插入一行（d1_returned=0），命中 uk_game_player(game_id, player_key) 即忽略。 + * + *

返回受影响行数区分「真首玩 vs 已存在」：1=真插入（该玩家对该游戏首次开局）→ 调用方据此累加 new_player_count； + * 0=已存在（同玩家此前已首玩）→ 调用方转 D1 回访判定。并发双首玩由 MySQL 在唯一键上原子裁决（一方插入一方忽略）。 + * + * @param gameId 游戏 ID + * @param playerKey 稳定玩家标识（u:{userId} / a:{anonId}） + * @param firstPlayDate 首玩自然日（平台时区） + * @return 受影响行数：1=真首玩、0=已存在（含并发被忽略） + */ + @Insert("INSERT IGNORE INTO game_telemetry_player_first_play " + + "(game_id, player_key, first_play_date, d1_returned) " + + "VALUES (#{gameId}, #{playerKey}, #{firstPlayDate}, 0)") + int insertIgnoreFirstPlay(@Param("gameId") Long gameId, @Param("playerKey") String playerKey, + @Param("firstPlayDate") LocalDate firstPlayDate); + + /** + * 取玩家对某游戏的首玩行（判 D1 回访窗口用：读 first_play_date 与 d1_returned）。 + * + * @param gameId 游戏 ID + * @param playerKey 稳定玩家标识 + * @return 首玩行；不存在返回 null + */ + default PlayerFirstPlayDO selectByGameAndPlayer(Long gameId, String playerKey) { + return selectOne(new LambdaQueryWrapperX() + .eq(PlayerFirstPlayDO::getGameId, gameId) + .eq(PlayerFirstPlayDO::getPlayerKey, playerKey)); + } + + /** + * CAS 置位 D1 回访标记（d1_returned 0→1，恰一次去重）：仅当当前为 0 才置 1。 + * + *

返回受影响行数保证同一玩家次日多次开局只计一次 D1：1=本次首次置位（调用方据此累加 d1_retained_count）； + * 0=已被置位（重复回访，去重跳过）。 + * + * @param id 首玩行主键 + * @return 受影响行数：1=首次置位、0=已置位（去重） + */ + @Update("UPDATE game_telemetry_player_first_play SET d1_returned = 1 " + + "WHERE id = #{id} AND d1_returned = 0") + int markD1ReturnedIfUnset(@Param("id") Long id); + +} diff --git a/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/dal/mysql/stat/RetentionStatMapper.java b/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/dal/mysql/stat/RetentionStatMapper.java new file mode 100644 index 00000000..3dafc833 --- /dev/null +++ b/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/dal/mysql/stat/RetentionStatMapper.java @@ -0,0 +1,67 @@ +package com.wanxiang.huijing.game.module.telemetry.dal.mysql.stat; + +import com.wanxiang.huijing.game.module.telemetry.dal.dataobject.stat.RetentionStatDO; +import com.wanxiang.huijing.framework.mybatis.core.mapper.BaseMapperX; +import com.wanxiang.huijing.framework.mybatis.core.query.LambdaQueryWrapperX; +import org.apache.ibatis.annotations.Insert; +import org.apache.ibatis.annotations.Mapper; +import org.apache.ibatis.annotations.Param; + +import java.time.LocalDate; + +/** + * 游戏维度 D1 次留聚合 Mapper（游戏 × cohort日 粒度） + * + *

两条行级原子累加语句（INSERT…ON DUPLICATE KEY UPDATE，命中 uk_game_date 累加、消除读-改-写并发丢增量）， + * 与 {@code GameStatMapper.insertOrAccumulate} 同范式；审计/租户缺省列依赖 DDL 默认值（MVP 单租户=0）。 + * 幂等边界：cohort 计数的「恰一次」由 first_play 表的 INSERT IGNORE / CAS 置位保证（见 RetentionAggregateService）， + * 本表只做纯加法。 + * + * @author 绘境AI + */ +@Mapper +public interface RetentionStatMapper extends BaseMapperX { + + /** + * 累加当日首玩新玩家数（cohort 规模 +1）：命中 (game_id, stat_date) 则 new_player_count+1，否则建行初值 (1,0)。 + * + * @param gameId 游戏 ID + * @param statDate cohort 日（= 首玩自然日） + * @return 受影响行数（插入=1 / 累加=2），仅日志排障用 + */ + @Insert("INSERT INTO game_telemetry_retention_stat " + + "(game_id, stat_date, new_player_count, d1_retained_count) " + + "VALUES (#{gameId}, #{statDate}, 1, 0) " + + "ON DUPLICATE KEY UPDATE new_player_count = new_player_count + 1") + int insertOrAccumulateNewPlayer(@Param("gameId") Long gameId, @Param("statDate") LocalDate statDate); + + /** + * 累加当日 cohort 的 D1 回访数（分子 +1）：命中 (game_id, stat_date) 则 d1_retained_count+1，否则建行初值 (0,1)。 + * + *

常态下该 (game_id, cohort日) 行在首玩日已由 {@link #insertOrAccumulateNewPlayer} 建出，故走累加分支； + * VALUES(0,1) 插入分支为防御性兜底（cohort 行意外缺失时不丢分子）。 + * + * @param gameId 游戏 ID + * @param statDate cohort 日（= 该玩家的首玩自然日） + * @return 受影响行数（插入=1 / 累加=2），仅日志排障用 + */ + @Insert("INSERT INTO game_telemetry_retention_stat " + + "(game_id, stat_date, new_player_count, d1_retained_count) " + + "VALUES (#{gameId}, #{statDate}, 0, 1) " + + "ON DUPLICATE KEY UPDATE d1_retained_count = d1_retained_count + 1") + int insertOrAccumulateD1Retained(@Param("gameId") Long gameId, @Param("statDate") LocalDate statDate); + + /** + * 按 游戏×cohort日 取次留聚合行（对账/看板/回流读口）。 + * + * @param gameId 游戏 ID + * @param statDate cohort 日 + * @return 聚合行；不存在返回 null + */ + default RetentionStatDO selectByGameAndDate(Long gameId, LocalDate statDate) { + return selectOne(new LambdaQueryWrapperX() + .eq(RetentionStatDO::getGameId, gameId) + .eq(RetentionStatDO::getStatDate, statDate)); + } + +} diff --git a/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/service/event/EventIngestServiceImpl.java b/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/service/event/EventIngestServiceImpl.java index bdf903ab..1e7dbd1e 100644 --- a/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/service/event/EventIngestServiceImpl.java +++ b/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/service/event/EventIngestServiceImpl.java @@ -121,6 +121,14 @@ public class EventIngestServiceImpl implements EventIngestService { @Resource private EventIngestService self; + /** + * D1 次留旁路聚合软注入（W-REAL R6）：消费/同步写单条摄取后附带记一次开局到次留旁路（仅 game_play_start）。 + *

best-effort + 独立旁路两表——不动既有 game_telemetry_game_stat 结构、quality_score 链路零风险； + * 纯 Mockito 单测不注入本 Provider，{@link #resolveRetentionService()} 空值守卫跳过（既有单测行为不变）。 + */ + @Resource + private ObjectProvider retentionAggregateServiceProvider; + // ============================== 上报入口（MQ 异步路由 / 同步回落）============================== /** @@ -180,6 +188,12 @@ public class EventIngestServiceImpl implements EventIngestService { boolean injected = injectSystemIdentityIfAnonymous(userId); try { ingestOne(reqVO, userId); + // R6：D1 次留旁路聚合（仅 game_play_start；best-effort 内部吞异常，与 quality_score 主链同事务但失败不连累它） + com.wanxiang.huijing.game.module.telemetry.service.retention.RetentionAggregateService retention = + resolveRetentionService(); + if (retention != null) { + retention.recordIfPlayStart(reqVO, userId); + } } finally { if (injected) { SecurityContextHolder.clearContext(); @@ -194,6 +208,11 @@ public class EventIngestServiceImpl implements EventIngestService { return telemetryEventProducerProvider == null ? null : telemetryEventProducerProvider.getIfAvailable(); } + /** 解析次留聚合服务（软注入 + 空值守卫）：Provider 未注入（单测）→ null → 跳过次留旁路（既有行为不变）。 */ + private com.wanxiang.huijing.game.module.telemetry.service.retention.RetentionAggregateService resolveRetentionService() { + return retentionAggregateServiceProvider == null ? null : retentionAggregateServiceProvider.getIfAvailable(); + } + /** * MQ 异步路径批量投递：逐条轻量校验，通过则投递 event tag（成功计 accepted、投递失败计 rejected 交前端重传）， * 校验失败计 rejected。返回 accepted/rejected + 回执 traceId（部分成功语义、非整批失败）。 diff --git a/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/service/retention/RetentionAggregateService.java b/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/service/retention/RetentionAggregateService.java new file mode 100644 index 00000000..9711ccc2 --- /dev/null +++ b/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/service/retention/RetentionAggregateService.java @@ -0,0 +1,28 @@ +package com.wanxiang.huijing.game.module.telemetry.service.retention; + +import com.wanxiang.huijing.game.module.telemetry.controller.app.event.vo.EnvelopeReqVO; + +/** + * D1 次留旁路聚合 Service（质量模型设计 §3.4 L3 留存结构） + * + *

只吃 {@code game_play_start}（开局）事件：追踪每玩家对每游戏的首玩自然日，并在首玩日+1 自然日回访开局时 + * 置位 D1 回访，产出 (game_id, cohort日) 粒度的 new_player_count / d1_retained_count。独立旁路两表， + * 不动既有 game_telemetry_game_stat 结构与 quality_score 链路。 + * + * @author 绘境AI + */ +public interface RetentionAggregateService { + + /** + * 若为 game_play_start 事件则记一次开局到次留旁路（否则 no-op）。 + * + *

零风险纪律：内部 best-effort——任何解析/落库异常都在内部吞掉并告警，绝不外抛， + * 故与它同事务的 quality_score 主链（event 落库 + game_stat 聚合 + feed 回灌）不受次留失败连累。 + * 幂等：首玩 INSERT IGNORE、D1 CAS 置位——同玩家同日多次开局不重复计数。 + * + * @param envelope 单条信封（已通过轻量校验） + * @param userId 上报时点登录用户 ID（匿名为 null；与信封 user.userId 一起决定稳定玩家标识） + */ + void recordIfPlayStart(EnvelopeReqVO envelope, Long userId); + +} diff --git a/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/service/retention/RetentionAggregateServiceImpl.java b/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/service/retention/RetentionAggregateServiceImpl.java new file mode 100644 index 00000000..a14ef525 --- /dev/null +++ b/game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/service/retention/RetentionAggregateServiceImpl.java @@ -0,0 +1,121 @@ +package com.wanxiang.huijing.game.module.telemetry.service.retention; + +import com.wanxiang.huijing.game.module.telemetry.controller.app.event.vo.EnvelopeReqVO; +import com.wanxiang.huijing.game.module.telemetry.dal.dataobject.stat.PlayerFirstPlayDO; +import com.wanxiang.huijing.game.module.telemetry.dal.mysql.stat.PlayerFirstPlayMapper; +import com.wanxiang.huijing.game.module.telemetry.dal.mysql.stat.RetentionStatMapper; +import com.wanxiang.huijing.game.module.telemetry.enums.TelemetryEventEnum; +import jakarta.annotation.Resource; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import org.springframework.util.StringUtils; + +import java.time.Instant; +import java.time.LocalDate; +import java.time.ZoneId; + +/** + * D1 次留旁路聚合实现（质量模型设计 §3.4 L3） + * + *

口径：同一 game_id、同一稳定玩家标识（登录取 userId、未登录取 anonId）在首玩自然日（平台时区，与 + * game_telemetry_game_stat.stat_date 同日界）之后第 1 个自然日内 ≥1 次 game_play_start，按「玩家×游戏×日」去重。 + * 增量落法：首玩 INSERT IGNORE 落 first_play 行 + new_player_count+1；次日回访 CAS 置位 d1_returned + d1_retained_count+1。 + * + *

与 quality_score 主链同事务但 best-effort：所有异常内部吞掉告警、不外抛，次留失败不连累 event 落库/聚合/回灌。 + * + * @author 绘境AI + */ +@Slf4j +@Service +public class RetentionAggregateServiceImpl implements RetentionAggregateService { + + @Resource + private PlayerFirstPlayMapper playerFirstPlayMapper; + + @Resource + private RetentionStatMapper retentionStatMapper; + + @Override + public void recordIfPlayStart(EnvelopeReqVO envelope, Long userId) { + // 只吃开局事件（game_play_start）；其余事件 no-op + if (envelope == null || !TelemetryEventEnum.GAME_PLAY_START.getEvent().equals(envelope.getEvent())) { + return; + } + // 解析落点三要素：gameId（聚合落点）、playerKey（稳定玩家标识）、playDate（平台时区自然日） + Long gameId = parseLong(envelope.getContext() == null ? null : envelope.getContext().getGameId()); + String playerKey = resolvePlayerKey(envelope, userId); + if (gameId == null || playerKey == null || envelope.getTs() == null) { + // 无游戏归属 / 无稳定玩家标识 / 无时间戳 → 无法归 cohort，跳过（不报错） + return; + } + LocalDate playDate = Instant.ofEpochMilli(envelope.getTs()).atZone(ZoneId.systemDefault()).toLocalDate(); + + try { + // ① 首玩幂等落点：INSERT IGNORE —— 真插入(=1)即该玩家对该游戏首次开局 → 累加 cohort 新玩家数 + int inserted = playerFirstPlayMapper.insertIgnoreFirstPlay(gameId, playerKey, playDate); + if (inserted > 0) { + retentionStatMapper.insertOrAccumulateNewPlayer(gameId, playDate); + log.info("[retention] 记首玩 gameId={} player={} firstPlayDate={}", gameId, playerKey, playDate); + return; + } + // ② 已有首玩记录：判是否首玩日+1 自然日回访（D1 窗口），且尚未置位 → CAS 置位 + 累加分子 + PlayerFirstPlayDO row = playerFirstPlayMapper.selectByGameAndPlayer(gameId, playerKey); + if (row == null || row.getFirstPlayDate() == null) { + return; // 极端并发下取不到（罕见）——跳过，下次回访再判 + } + boolean unset = row.getD1Returned() == null || row.getD1Returned() == 0; + boolean isD1Window = playDate.equals(row.getFirstPlayDate().plusDays(1)); + if (unset && isD1Window) { + int marked = playerFirstPlayMapper.markD1ReturnedIfUnset(row.getId()); + if (marked > 0) { + // cohort 日 = 该玩家的首玩日（分子累加落在首玩日的 cohort 行上） + retentionStatMapper.insertOrAccumulateD1Retained(gameId, row.getFirstPlayDate()); + log.info("[retention] 记 D1 回访 gameId={} player={} cohortDate={} returnDate={}", + gameId, playerKey, row.getFirstPlayDate(), playDate); + } + } + // 同日重复开局 / 更晚回访(非 D1 窗口) / 已置位 → 无操作（去重） + } catch (Exception e) { + // 零风险纪律：次留旁路任何失败都不连累 quality_score 主链（同事务内已完成的 event/聚合/回灌照常提交）。 + // 错误路径必须留痕（创始人铁律）。 + log.warn("[retention] 次留旁路聚合失败（best-effort，不影响 quality_score 主链）gameId={} player={} eventId={}", + gameId, playerKey, envelope.getEventId(), e); + } + } + + /** + * 稳定玩家标识：登录 = {@code u:{userId}}、未登录 = {@code a:{anonId}}（命名空间隔离防碰撞）。 + * userId 优先取信封 user.userId，回退上报时点登录 userId；均无则取信封 user.anonId。 + * + * @return 玩家标识；无 userId 也无 anonId 时返回 null（无法归属，跳过） + */ + private String resolvePlayerKey(EnvelopeReqVO envelope, Long userId) { + Long uid = userId; + if (envelope.getUser() != null && StringUtils.hasText(envelope.getUser().getUserId())) { + Long parsed = parseLong(envelope.getUser().getUserId()); + if (parsed != null) { + uid = parsed; + } + } + if (uid != null) { + return "u:" + uid; + } + if (envelope.getUser() != null && StringUtils.hasText(envelope.getUser().getAnonId())) { + return "a:" + envelope.getUser().getAnonId(); + } + return null; + } + + /** 字符串安全转 Long（空/非法返回 null）。 */ + private static Long parseLong(String s) { + if (!StringUtils.hasText(s)) { + return null; + } + try { + return Long.parseLong(s.trim()); + } catch (NumberFormatException ex) { + return null; + } + } + +} diff --git a/game-cloud/game-module-telemetry/game-module-telemetry-server/src/test/java/com/wanxiang/huijing/game/module/telemetry/service/retention/RetentionAggregateRealDbIT.java b/game-cloud/game-module-telemetry/game-module-telemetry-server/src/test/java/com/wanxiang/huijing/game/module/telemetry/service/retention/RetentionAggregateRealDbIT.java new file mode 100644 index 00000000..0adf8fbb --- /dev/null +++ b/game-cloud/game-module-telemetry/game-module-telemetry-server/src/test/java/com/wanxiang/huijing/game/module/telemetry/service/retention/RetentionAggregateRealDbIT.java @@ -0,0 +1,191 @@ +package com.wanxiang.huijing.game.module.telemetry.service.retention; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledIfSystemProperty; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.Statement; +import java.time.LocalDate; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * R6「D1 次留聚合真跑」集成证据——连 mini-infra 真 MySQL（{@code 100.64.0.8:3306} · Tailscale 直连、JDBC 不走 HTTP 代理）， + * 在 throwaway 库里跑真 V27 迁移文件的 DDL + 一个两日 D1 场景，验 new_player_count/d1_retained_count 计数正确。 + * + *

覆盖边界：真 MySQL 上 ① V27 DDL 建表可跑（双验 R2 单体 Flyway 前的 DDL 合法性）；② mapper 三条语句 + * （INSERT IGNORE 首玩 / ON DUP KEY 累加 / CAS 置位 D1）行为正确；③ D1 cohort 计数（首玩日+1 去重）正确。 + * 服务分支逻辑（首玩/D1窗口/去重）由 {@link RetentionAggregateServiceImplTest} 单测坐实，本 IT 用真 SQL 复演其决策序列。 + * 全程 throwaway 库 {@code telemetry_r6_e2e} 建/删，隔离、不碰真库真表、不与 Flyway 冲突。 + * + *

默认跳过：{@code @EnabledIfSystemProperty(telemetry.retention.e2e=1)}——常规 {@code mvn test} 不连 MySQL； + * 出证据时 {@code mvn test -Dtelemetry.retention.e2e=1 -Dtest=RetentionAggregateRealDbIT}。 + * + * @author 绘境AI + */ +@EnabledIfSystemProperty(named = "telemetry.retention.e2e", matches = "1") +class RetentionAggregateRealDbIT { + + private static final String HOST = System.getProperty("telemetry.retention.dbhost", "100.64.0.8"); + private static final String USER = System.getProperty("telemetry.retention.dbuser", "root"); + // 内网 MVP 凭据（铁律授权入仓，见 docs/内网凭据与端点.md）；可经 -D 覆盖 + private static final String PASS = System.getProperty("telemetry.retention.dbpass", "ZRH3jwYLOrntBcTAw29MW9BP"); + private static final String DB = "telemetry_r6_e2e"; + private static final long G = 999_999_001L; // 测试游戏 ID（throwaway 库，无关真数据） + private static final LocalDate D = LocalDate.of(2026, 7, 2); + + private String baseUrl() { + // JDBC/TCP 直连 Tailscale，不经 HTTP 代理；allowPublicKeyRetrieval + useSSL=false 内网 MVP + return "jdbc:mysql://" + HOST + ":3306/?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=UTC"; + } + + @Test + void v27Ddl_and_d1CohortCounting_onRealMysql() throws Exception { + // 内网直连绕系统 SOCKS 代理（macOS SOCKSEnable，127.0.0.1:7897）——MySQL Connector/J 默认走 JVM socks， + // 直连 Tailscale MySQL 必须显式禁（栽过的确切坑：不禁则握手期 EOF「read 0 bytes」）。 + System.setProperty("socksProxyHost", ""); + System.setProperty("java.net.useSystemProxies", "false"); + try (Connection conn = DriverManager.getConnection(baseUrl(), USER, PASS)) { + try (Statement st = conn.createStatement()) { + st.execute("DROP DATABASE IF EXISTS " + DB); + st.execute("CREATE DATABASE " + DB + " DEFAULT CHARACTER SET utf8mb4"); + st.execute("USE " + DB); + } + // ① 跑真 V27 迁移文件的 DDL（证 DDL 在真 MySQL 合法可建） + runV27Ddl(conn); + + // ② 两日 D1 场景（复演 RetentionAggregateService 的决策序列，用 mapper 同款 SQL）： + // Day D：玩家 u:1、u:2 首玩 → new_player_count=2 + // Day D+1：u:1 回访(D1 命中)、u:2 不回 → d1_retained_count=1 + recordPlayStart(conn, "u:1", D); // 首玩 + recordPlayStart(conn, "u:2", D); // 首玩 + recordPlayStart(conn, "u:1", D.plusDays(1)); // D1 回访 + // u:2 不回访 —— 不产生 D+1 事件 + + // ③ 断言 cohort 计数 + try (PreparedStatement ps = conn.prepareStatement( + "SELECT new_player_count, d1_retained_count FROM game_telemetry_retention_stat " + + "WHERE game_id=? AND stat_date=?")) { + ps.setLong(1, G); + ps.setObject(2, D); + try (ResultSet rs = ps.executeQuery()) { + assertTrue(rs.next(), "cohort 行应存在"); + assertEquals(2L, rs.getLong(1), "首玩新玩家数应为 2（u:1 + u:2）"); + assertEquals(1L, rs.getLong(2), "D1 回访数应为 1（仅 u:1 次日回访）"); + } + } + // 复核首玩表：u:1 已置位 d1_returned=1、u:2 未置位=0 + assertEquals(1, queryD1Returned(conn, "u:1"), "u:1 应被置位 D1 回访"); + assertEquals(0, queryD1Returned(conn, "u:2"), "u:2 未回访、d1_returned 应为 0"); + } finally { + // 清理 throwaway 库（隔离、不留痕） + try (Connection conn = DriverManager.getConnection(baseUrl(), USER, PASS); + Statement st = conn.createStatement()) { + st.execute("DROP DATABASE IF EXISTS " + DB); + } + } + } + + /** 读并执行真 V27 迁移文件的 DDL（去 -- 注释行，按 ; 切分执行）。 */ + private void runV27Ddl(Connection conn) throws Exception { + // 相对模块 basedir（surefire 工作目录）定位 huijing-server 下的迁移文件 + Path migration = Paths.get(System.getProperty("user.dir"), "..", "..", + "huijing-server", "src", "main", "resources", "db", "migration", + "V27.0.0__create_game_telemetry_retention.sql"); + assertTrue(Files.exists(migration), "应找到 V27 迁移文件：" + migration.toAbsolutePath().normalize()); + String sql = Files.readString(migration); + // 去掉整行 -- 注释后按 ; 切分 + StringBuilder cleaned = new StringBuilder(); + for (String line : sql.split("\n")) { + String trimmed = line.trim(); + if (trimmed.startsWith("--")) { + continue; + } + cleaned.append(line).append('\n'); + } + try (Statement st = conn.createStatement()) { + for (String stmt : cleaned.toString().split(";")) { + if (stmt.trim().isEmpty()) { + continue; + } + st.execute(stmt); + } + } + } + + /** 复演 service.recordIfPlayStart 决策序列（真 SQL）：首玩累加 new_player；D1 窗口 CAS 置位 + 累加分子。 */ + private void recordPlayStart(Connection conn, String playerKey, LocalDate playDate) throws Exception { + // ① INSERT IGNORE 首玩落点 + int inserted; + try (PreparedStatement ps = conn.prepareStatement( + "INSERT IGNORE INTO game_telemetry_player_first_play " + + "(game_id, player_key, first_play_date, d1_returned) VALUES (?,?,?,0)")) { + ps.setLong(1, G); + ps.setString(2, playerKey); + ps.setObject(3, playDate); + inserted = ps.executeUpdate(); + } + if (inserted > 0) { + accumulate(conn, "INSERT INTO game_telemetry_retention_stat " + + "(game_id, stat_date, new_player_count, d1_retained_count) VALUES (?,?,1,0) " + + "ON DUPLICATE KEY UPDATE new_player_count = new_player_count + 1", playDate); + return; + } + // ② 已存在：判 D1 窗口 + CAS 置位 + Long rowId = null; + LocalDate firstPlayDate = null; + int d1 = 0; + try (PreparedStatement ps = conn.prepareStatement( + "SELECT id, first_play_date, d1_returned FROM game_telemetry_player_first_play " + + "WHERE game_id=? AND player_key=?")) { + ps.setLong(1, G); + ps.setString(2, playerKey); + try (ResultSet rs = ps.executeQuery()) { + if (rs.next()) { + rowId = rs.getLong(1); + firstPlayDate = rs.getObject(2, LocalDate.class); + d1 = rs.getInt(3); + } + } + } + if (rowId != null && d1 == 0 && firstPlayDate != null && playDate.equals(firstPlayDate.plusDays(1))) { + int marked; + try (PreparedStatement ps = conn.prepareStatement( + "UPDATE game_telemetry_player_first_play SET d1_returned=1 WHERE id=? AND d1_returned=0")) { + ps.setLong(1, rowId); + marked = ps.executeUpdate(); + } + if (marked > 0) { + accumulate(conn, "INSERT INTO game_telemetry_retention_stat " + + "(game_id, stat_date, new_player_count, d1_retained_count) VALUES (?,?,0,1) " + + "ON DUPLICATE KEY UPDATE d1_retained_count = d1_retained_count + 1", firstPlayDate); + } + } + } + + private void accumulate(Connection conn, String sql, LocalDate statDate) throws Exception { + try (PreparedStatement ps = conn.prepareStatement(sql)) { + ps.setLong(1, G); + ps.setObject(2, statDate); + ps.executeUpdate(); + } + } + + private int queryD1Returned(Connection conn, String playerKey) throws Exception { + try (PreparedStatement ps = conn.prepareStatement( + "SELECT d1_returned FROM game_telemetry_player_first_play WHERE game_id=? AND player_key=?")) { + ps.setLong(1, G); + ps.setString(2, playerKey); + try (ResultSet rs = ps.executeQuery()) { + return rs.next() ? rs.getInt(1) : -1; + } + } + } +} diff --git a/game-cloud/game-module-telemetry/game-module-telemetry-server/src/test/java/com/wanxiang/huijing/game/module/telemetry/service/retention/RetentionAggregateServiceImplTest.java b/game-cloud/game-module-telemetry/game-module-telemetry-server/src/test/java/com/wanxiang/huijing/game/module/telemetry/service/retention/RetentionAggregateServiceImplTest.java new file mode 100644 index 00000000..09352399 --- /dev/null +++ b/game-cloud/game-module-telemetry/game-module-telemetry-server/src/test/java/com/wanxiang/huijing/game/module/telemetry/service/retention/RetentionAggregateServiceImplTest.java @@ -0,0 +1,184 @@ +package com.wanxiang.huijing.game.module.telemetry.service.retention; + +import com.wanxiang.huijing.game.module.telemetry.controller.app.event.vo.EnvelopeReqVO; +import com.wanxiang.huijing.game.module.telemetry.dal.dataobject.stat.PlayerFirstPlayDO; +import com.wanxiang.huijing.game.module.telemetry.dal.mysql.stat.PlayerFirstPlayMapper; +import com.wanxiang.huijing.game.module.telemetry.dal.mysql.stat.RetentionStatMapper; +import com.wanxiang.huijing.framework.test.core.ut.BaseMockitoUnitTest; +import org.junit.jupiter.api.Test; +import org.mockito.InjectMocks; +import org.mockito.Mock; + +import java.time.LocalDate; +import java.time.ZoneId; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +/** + * {@link RetentionAggregateServiceImpl} D1 次留口径单测（纯 Mockito）——把守质量模型 §3.4 口径： + * 首玩落点、D1 窗口（首玩日+1）、玩家×游戏×日去重、匿名/登录稳定标识、best-effort 零风险。 + * + * @author 绘境AI + */ +class RetentionAggregateServiceImplTest extends BaseMockitoUnitTest { + + @InjectMocks + private RetentionAggregateServiceImpl service; + + @Mock + private PlayerFirstPlayMapper playerFirstPlayMapper; + @Mock + private RetentionStatMapper retentionStatMapper; + + private static final LocalDate D = LocalDate.of(2026, 7, 2); + + /** 把 LocalDate 转成落在该自然日内的 ts（平台时区 = systemDefault，与实现同口径），取正午防边界。 */ + private static long tsForDate(LocalDate date) { + return date.atTime(12, 0).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(); + } + + private static EnvelopeReqVO playStart(String userId, String anonId, String gameId, LocalDate date) { + EnvelopeReqVO env = new EnvelopeReqVO(); + env.setEventId("evt-" + userId + "-" + date); + env.setEvent("game_play_start"); + env.setSchemaVersion("v1"); + env.setTs(tsForDate(date)); + env.setTraceId("t-" + date); + EnvelopeReqVO.UserVO user = new EnvelopeReqVO.UserVO(); + user.setUserId(userId); + user.setAnonId(anonId); + env.setUser(user); + EnvelopeReqVO.ContextVO ctx = new EnvelopeReqVO.ContextVO(); + ctx.setGameId(gameId); + env.setContext(ctx); + return env; + } + + /** 首玩：INSERT IGNORE 真插入(=1) → 累加 cohort 新玩家数、不置 D1。 */ + @Test + void testFirstPlay_countsNewPlayer() { + when(playerFirstPlayMapper.insertIgnoreFirstPlay(eq(1024L), eq("u:10086"), eq(D))).thenReturn(1); + + service.recordIfPlayStart(playStart("10086", null, "1024", D), 10086L); + + verify(retentionStatMapper).insertOrAccumulateNewPlayer(eq(1024L), eq(D)); + verify(retentionStatMapper, never()).insertOrAccumulateD1Retained(anyLong(), any()); + verify(playerFirstPlayMapper, never()).markD1ReturnedIfUnset(anyLong()); + } + + /** D1 回访：首玩日+1 开局，未置位 → CAS 置位成功 → 累加分子（cohort 日 = 首玩日 D）。 */ + @Test + void testD1Return_countsRetained() { + when(playerFirstPlayMapper.insertIgnoreFirstPlay(eq(1024L), eq("u:10086"), eq(D.plusDays(1)))).thenReturn(0); + PlayerFirstPlayDO row = new PlayerFirstPlayDO(); + row.setId(7L); + row.setGameId(1024L); + row.setPlayerKey("u:10086"); + row.setFirstPlayDate(D); + row.setD1Returned(0); + when(playerFirstPlayMapper.selectByGameAndPlayer(1024L, "u:10086")).thenReturn(row); + when(playerFirstPlayMapper.markD1ReturnedIfUnset(7L)).thenReturn(1); + + service.recordIfPlayStart(playStart("10086", null, "1024", D.plusDays(1)), 10086L); + + // 分子累加落在 cohort 日（首玩日 D），非回访日 D+1 + verify(retentionStatMapper).insertOrAccumulateD1Retained(eq(1024L), eq(D)); + verify(retentionStatMapper, never()).insertOrAccumulateNewPlayer(anyLong(), any()); + } + + /** 同日重复开局：非 D1 窗口 → 不置位、不累加分子（去重）。 */ + @Test + void testSameDayRepeat_noRetained() { + when(playerFirstPlayMapper.insertIgnoreFirstPlay(eq(1024L), eq("u:10086"), eq(D))).thenReturn(0); + PlayerFirstPlayDO row = new PlayerFirstPlayDO(); + row.setId(7L); + row.setFirstPlayDate(D); + row.setD1Returned(0); + when(playerFirstPlayMapper.selectByGameAndPlayer(1024L, "u:10086")).thenReturn(row); + + service.recordIfPlayStart(playStart("10086", null, "1024", D), 10086L); + + verify(playerFirstPlayMapper, never()).markD1ReturnedIfUnset(anyLong()); + verify(retentionStatMapper, never()).insertOrAccumulateD1Retained(anyLong(), any()); + } + + /** D+2 回访：超出 D1 窗口（首玩日+1）→ 不计 D1。 */ + @Test + void testD2Return_notCountedAsD1() { + when(playerFirstPlayMapper.insertIgnoreFirstPlay(eq(1024L), eq("u:10086"), eq(D.plusDays(2)))).thenReturn(0); + PlayerFirstPlayDO row = new PlayerFirstPlayDO(); + row.setId(7L); + row.setFirstPlayDate(D); + row.setD1Returned(0); + when(playerFirstPlayMapper.selectByGameAndPlayer(1024L, "u:10086")).thenReturn(row); + + service.recordIfPlayStart(playStart("10086", null, "1024", D.plusDays(2)), 10086L); + + verify(playerFirstPlayMapper, never()).markD1ReturnedIfUnset(anyLong()); + verify(retentionStatMapper, never()).insertOrAccumulateD1Retained(anyLong(), any()); + } + + /** 已置位（d1_returned=1）：D1 窗口内重复回访不再累加（CAS 去重的语义前置守卫）。 */ + @Test + void testAlreadyReturned_noDoubleCount() { + when(playerFirstPlayMapper.insertIgnoreFirstPlay(eq(1024L), eq("u:10086"), eq(D.plusDays(1)))).thenReturn(0); + PlayerFirstPlayDO row = new PlayerFirstPlayDO(); + row.setId(7L); + row.setFirstPlayDate(D); + row.setD1Returned(1); // 已置位 + when(playerFirstPlayMapper.selectByGameAndPlayer(1024L, "u:10086")).thenReturn(row); + + service.recordIfPlayStart(playStart("10086", null, "1024", D.plusDays(1)), 10086L); + + verify(playerFirstPlayMapper, never()).markD1ReturnedIfUnset(anyLong()); + verify(retentionStatMapper, never()).insertOrAccumulateD1Retained(anyLong(), any()); + } + + /** 未登录：无 userId、有 anonId → 玩家标识 a:{anonId}。 */ + @Test + void testAnonymousPlayer_usesAnonIdKey() { + when(playerFirstPlayMapper.insertIgnoreFirstPlay(eq(1024L), eq("a:anon-xyz"), eq(D))).thenReturn(1); + + service.recordIfPlayStart(playStart(null, "anon-xyz", "1024", D), null); + + verify(playerFirstPlayMapper).insertIgnoreFirstPlay(eq(1024L), eq("a:anon-xyz"), eq(D)); + verify(retentionStatMapper).insertOrAccumulateNewPlayer(eq(1024L), eq(D)); + } + + /** 非 game_play_start 事件 → no-op（不碰任何 mapper）。 */ + @Test + void testNonPlayStart_noOp() { + EnvelopeReqVO like = playStart("10086", null, "1024", D); + like.setEvent("like"); + + service.recordIfPlayStart(like, 10086L); + + verifyNoInteractions(playerFirstPlayMapper); + verifyNoInteractions(retentionStatMapper); + } + + /** 无 gameId / 无玩家标识 → no-op（无法归 cohort）。 */ + @Test + void testNoGameIdOrNoPlayer_noOp() { + service.recordIfPlayStart(playStart("10086", null, null, D), 10086L); // 无 gameId + service.recordIfPlayStart(playStart(null, null, "1024", D), null); // 无 userId 也无 anonId + verifyNoInteractions(playerFirstPlayMapper); + verifyNoInteractions(retentionStatMapper); + } + + /** best-effort：mapper 抛异常 → recordIfPlayStart 不外抛（保 quality_score 主链零风险）。 */ + @Test + void testBestEffort_swallowsMapperException() { + when(playerFirstPlayMapper.insertIgnoreFirstPlay(any(), any(), any())) + .thenThrow(new RuntimeException("DB down")); + + assertDoesNotThrow(() -> service.recordIfPlayStart(playStart("10086", null, "1024", D), 10086L)); + } +} diff --git a/game-cloud/huijing-server/src/main/resources/db/migration/V27.0.0__create_game_telemetry_retention.sql b/game-cloud/huijing-server/src/main/resources/db/migration/V27.0.0__create_game_telemetry_retention.sql new file mode 100644 index 00000000..3633d2c3 --- /dev/null +++ b/game-cloud/huijing-server/src/main/resources/db/migration/V27.0.0__create_game_telemetry_retention.sql @@ -0,0 +1,59 @@ +-- ============================================================================= +-- 契约 #2 DB 迁移 | 模块：telemetry（D1 次留旁路聚合）| owner：WS5 +-- 文件：V27.0.0__create_game_telemetry_retention.sql（Flyway，只新增；已合入禁止修改，回滚写新补偿迁移） +-- 内容：D1 次留旁路两表 —— 玩家首玩追踪 game_telemetry_player_first_play + 游戏维度次留聚合 game_telemetry_retention_stat +-- 背景：质量模型设计 §3.4 L3 留存结构立到「可建字段」。既有 game_telemetry_game_stat 结构与 quality_score 链路 +-- 零改动、零风险——次留是跨日玩家级去重 + cohort 追踪，不是 game_stat 那种当日行级累加，故独立旁路表承载。 +-- D1 口径（§3.4）：同一 game_id、同一稳定玩家标识（登录取 userId、未登录取匿名设备标识 anonId，对齐 #5 契约 anonId 口径） +-- 在首玩自然日（平台时区，与 game_telemetry_game_stat.stat_date 同日界）之后的第 1 个自然日内 ≥1 次开局事件（game_play_start）， +-- 按「玩家 × 游戏 × 日」去重。D1 留存率 = d1_retained_count / new_player_count（读时计算、不落列，避免除法陈旧）。 +-- 增量落法：消费侧对 game_play_start 事件旁路累加（best-effort，失败不连累 quality_score 主链），见 RetentionAggregateService。 +-- 约定：InnoDB + utf8mb4；显式列；含 Yudao 审计列 + 租户列。 +-- 错误码段：telemetry = 1-104-***-*** +-- ============================================================================= + +-- ----------------------------------------------------------------------------- +-- 表：game_telemetry_player_first_play —— 玩家首玩追踪（游戏 × 玩家 粒度） +-- 每玩家对每游戏一行：记首玩自然日 + 是否在首玩日+1 自然日回访开局（d1_returned CAS 置位一次去重）。 +-- 玩家标识命名空间隔离：登录=u:{userId}、未登录=a:{anonId}，防 userId 与 anonId 字面碰撞。 +-- 已知 MVP 边界：匿名→登录跨身份（先 anon 玩、次日登录玩）当前记为两个玩家，user_login 事件桥接留后续精化。 +-- ----------------------------------------------------------------------------- +CREATE TABLE `game_telemetry_player_first_play` ( + `id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '记录 ID', + `game_id` BIGINT NOT NULL COMMENT '游戏 ID（= game_project.id）', + `player_key` VARCHAR(80) NOT NULL COMMENT '稳定玩家标识：登录=u:{userId}，未登录=a:{anonId}（对齐 #5 契约 anonId 口径；两类命名空间隔离防碰撞）', + `first_play_date` DATE NOT NULL COMMENT '该玩家对该游戏的首玩自然日（平台时区，与 game_telemetry_game_stat.stat_date 同日界）', + `d1_returned` TINYINT NOT NULL DEFAULT 0 COMMENT '次留标记：0未回 1已在首玩日+1 自然日回访开局（CAS 置位一次、去重）', + `creator` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '创建者（Yudao 审计列；上报通道为 system/上报方）', + `create_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', + `updater` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '更新者', + `update_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', + `deleted` BIT(1) NOT NULL DEFAULT b'0' COMMENT '逻辑删除：0未删 1已删', + `tenant_id` BIGINT NOT NULL DEFAULT 0 COMMENT '租户 ID（Yudao 多租户兼容，MVP 单租户=0）', + PRIMARY KEY (`id`), + UNIQUE KEY `uk_game_player` (`game_id`, `player_key`) COMMENT '每玩家每游戏一行（首玩落点 + D1 去重的幂等键）', + KEY `idx_game_firstdate` (`game_id`, `first_play_date`) COMMENT '按游戏+首玩日取 cohort（对账/回算）' +) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4 COMMENT = '遥测玩家首玩追踪表（D1 次留旁路，游戏×玩家粒度）'; + +-- ----------------------------------------------------------------------------- +-- 表：game_telemetry_retention_stat —— 游戏维度 D1 次留聚合（游戏 × cohort日 粒度） +-- 由消费侧对 game_play_start 增量累加产出；唯一约束 (game_id, stat_date) 保证聚合幂等（行级原子累加）。 +-- new_player_count：cohort 规模（当日首玩新玩家数）；d1_retained_count：该 cohort 次日回访开局玩家数。 +-- 独立于 game_telemetry_game_stat：不动既有聚合表结构、quality_score 链路零风险；本表暂不回灌 feed（观测/回流用）。 +-- ----------------------------------------------------------------------------- +CREATE TABLE `game_telemetry_retention_stat` ( + `id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '聚合记录 ID', + `game_id` BIGINT NOT NULL COMMENT '游戏 ID（= game_project.id）', + `stat_date` DATE NOT NULL COMMENT 'cohort 日 = 首玩自然日（平台时区）', + `new_player_count` BIGINT NOT NULL DEFAULT 0 COMMENT '当日首玩新玩家数（cohort 规模，玩家×游戏×日去重）', + `d1_retained_count` BIGINT NOT NULL DEFAULT 0 COMMENT '当日 cohort 中在次日回访开局的玩家数（D1 留存分子；D1 率 = 本列/new_player_count 读时算）', + `creator` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '创建者（Yudao 审计列；聚合任务为 system）', + `create_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', + `updater` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '更新者', + `update_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', + `deleted` BIT(1) NOT NULL DEFAULT b'0' COMMENT '逻辑删除：0未删 1已删', + `tenant_id` BIGINT NOT NULL DEFAULT 0 COMMENT '租户 ID（Yudao 多租户兼容，MVP 单租户=0）', + PRIMARY KEY (`id`), + UNIQUE KEY `uk_game_date` (`game_id`, `stat_date`) COMMENT '聚合幂等：游戏×cohort日一行，消费侧行级原子累加', + KEY `idx_date` (`stat_date`) COMMENT '按日取留存（看板/数据飞轮回流）' +) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4 COMMENT = '游戏维度 D1 次留聚合表（旁路，不动既有 game_telemetry_game_stat）';