feat(telemetry): D1 次留旁路聚合(玩家×游戏×日去重·平台时区·quality_score 链路零风险) (W-REAL R6)
按质量模型设计 §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) <noreply@anthropic.com>
This commit is contained in:
lili
parent
f2d85a072a
commit
617abbb589
@ -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，游戏 × 玩家 粒度）
|
||||
*
|
||||
* <p>质量模型设计 §3.4 L3 留存结构落地件：每玩家对每游戏一行，记首玩自然日 + 是否在首玩日+1 自然日回访开局。
|
||||
* 继承 {@link TenantBaseDO} 自动携带 creator/create_time/updater/update_time/deleted + tenant_id。
|
||||
* 唯一约束 (game_id, player_key) 保证每玩家每游戏一行——既是首玩落点，也是 D1 去重幂等键。
|
||||
*
|
||||
* <p>玩家标识 {@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;
|
||||
|
||||
}
|
||||
@ -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日 粒度）
|
||||
*
|
||||
* <p>质量模型设计 §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;
|
||||
|
||||
}
|
||||
@ -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 次留旁路，游戏 × 玩家 粒度）
|
||||
*
|
||||
* <p>并发安全靠两条原子语句：{@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<PlayerFirstPlayDO> {
|
||||
|
||||
/**
|
||||
* 首玩幂等落点（INSERT IGNORE）：新玩家插入一行（d1_returned=0），命中 uk_game_player(game_id, player_key) 即忽略。
|
||||
*
|
||||
* <p>返回受影响行数区分「真首玩 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<PlayerFirstPlayDO>()
|
||||
.eq(PlayerFirstPlayDO::getGameId, gameId)
|
||||
.eq(PlayerFirstPlayDO::getPlayerKey, playerKey));
|
||||
}
|
||||
|
||||
/**
|
||||
* CAS 置位 D1 回访标记（d1_returned 0→1，恰一次去重）：仅当当前为 0 才置 1。
|
||||
*
|
||||
* <p>返回受影响行数保证同一玩家次日多次开局只计一次 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);
|
||||
|
||||
}
|
||||
@ -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日 粒度）
|
||||
*
|
||||
* <p>两条行级原子累加语句（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<RetentionStatDO> {
|
||||
|
||||
/**
|
||||
* 累加当日首玩新玩家数（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)。
|
||||
*
|
||||
* <p>常态下该 (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<RetentionStatDO>()
|
||||
.eq(RetentionStatDO::getGameId, gameId)
|
||||
.eq(RetentionStatDO::getStatDate, statDate));
|
||||
}
|
||||
|
||||
}
|
||||
@ -121,6 +121,14 @@ public class EventIngestServiceImpl implements EventIngestService {
|
||||
@Resource
|
||||
private EventIngestService self;
|
||||
|
||||
/**
|
||||
* D1 次留旁路聚合软注入（W-REAL R6）：消费/同步写单条摄取后附带记一次开局到次留旁路（仅 game_play_start）。
|
||||
* <p>best-effort + 独立旁路两表——不动既有 game_telemetry_game_stat 结构、quality_score 链路零风险；
|
||||
* 纯 Mockito 单测不注入本 Provider，{@link #resolveRetentionService()} 空值守卫跳过（既有单测行为不变）。
|
||||
*/
|
||||
@Resource
|
||||
private ObjectProvider<com.wanxiang.huijing.game.module.telemetry.service.retention.RetentionAggregateService> 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（部分成功语义、非整批失败）。
|
||||
|
||||
@ -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 留存结构）
|
||||
*
|
||||
* <p>只吃 {@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）。
|
||||
*
|
||||
* <p><b>零风险纪律</b>：内部 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);
|
||||
|
||||
}
|
||||
@ -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）
|
||||
*
|
||||
* <p>口径：同一 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。
|
||||
*
|
||||
* <p>与 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;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -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 库里跑<b>真 V27 迁移文件的 DDL</b> + 一个两日 D1 场景，验 new_player_count/d1_retained_count 计数正确。
|
||||
*
|
||||
* <p><b>覆盖边界</b>：真 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 冲突。
|
||||
*
|
||||
* <p><b>默认跳过</b>：{@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;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -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));
|
||||
}
|
||||
}
|
||||
@ -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）';
|
||||
Loading…
x
Reference in New Issue
Block a user