质量模型设计 §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 返回受影响行数区分「真首玩 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 返回受影响行数保证同一玩家次日多次开局只计一次 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 常态下该 (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 best-effort + 独立旁路两表——不动既有 game_telemetry_game_stat 结构、quality_score 链路零风险；
+ * 纯 Mockito 单测不注入本 Provider，{@link #resolveRetentionService()} 空值守卫跳过（既有单测行为不变）。
+ */
+ @Resource
+ private ObjectProvider 只吃 {@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）';