feat(aigc): WU2② 注册 claim 消费者——消费 WU1 注册成功事件从池 CAS 绑玩家
WU1 afterCommit 发「注册成功」事件（topic passport-register），本单作其消费者，不进注册 事务：渠道过滤（只 password/invite claim，sms/sms-mock 跳过防白占池）→ 从池单条 CAS 抢一个 FREE 绑 game_player（biz_no=claim_<id>，幂等靠 status='FREE'+两 uk）。 - 池空不抛异常（补池是离线动作、非 MQ 重试窗口内可恢复）：记 WARN + 触发水位告警， 该玩家暂不绑，首次生成由懒 claim 兜。 - 跨模块不依赖 system-server：按同名字段建消费侧镜像体 PassportRegisterEvent（topic 是唯一 wire 契约）。 - 主开关 aigc.newapi-quota.enabled 默认 false，关时消费者不装配、服务全旁路（WU2 整体 opt-in）。 单测 9 例：claim 幂等/渠道过滤/池空不抛/并发 dup 收敛/身份分流/懒 claim fail-fast。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
lili
parent
fe6079ba62
commit
4cf0e14ec3
@ -0,0 +1,65 @@
|
||||
package com.wanxiang.huijing.game.module.aigc.mq;
|
||||
|
||||
import com.wanxiang.huijing.game.module.aigc.mq.message.PassportRegisterEvent;
|
||||
import com.wanxiang.huijing.game.module.aigc.service.quota.NewapiQuotaService;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
|
||||
import org.apache.rocketmq.spring.core.RocketMQListener;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* new-api 额度注册 claim 消费者（WU2 §3.4）——消费 WU1「注册成功」事件，从预置池 claim 一个 FREE 绑玩家。
|
||||
*
|
||||
* <p><b>接缝</b>：订阅 WU1 outbox 投递器的 topic {@code passport-register}（生产侧常量
|
||||
* {@code PassportRegisterOutboxDeliverer.TOPIC}；跨模块不依赖 system-server，故此处以字面量对齐 wire 契约）。
|
||||
* 消息载荷是 JSON，反序列化进本模块镜像体 {@link PassportRegisterEvent}（字段与生产侧 {@code PassportRegisterMessage} 同名）。
|
||||
*
|
||||
* <p><b>装配开关</b>：随 {@code aigc.newapi-quota.enabled=true} 装配——关闭（默认）时不起消费者、不连 NameServer，
|
||||
* WU2 整体未启用时零副作用（与 {@code NewapiQuotaServiceImpl} 主开关同源）。
|
||||
*
|
||||
* <p><b>幂等 + 不外抛</b>：claim 幂等由 service 层 uk + 单 CAS 保证（重投/并发收敛成一条）；池空/异常已在 service 内吞
|
||||
* （池空是离线补池才能恢复、非 MQ 重试窗口内可恢复，抛异常只灌 DLQ 空转）。此处防御性兜底 catch，正常返回=ack，不无限重投。
|
||||
* 消费模型 = 默认 CLUSTERING（每条消息集群内单实例消费）。
|
||||
*
|
||||
* @author 绘境AI
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
@ConditionalOnProperty(prefix = "aigc.newapi-quota", name = "enabled", havingValue = "true")
|
||||
@RocketMQMessageListener(
|
||||
topic = "passport-register", // = PassportRegisterOutboxDeliverer.TOPIC（wire 契约，跨模块不引 -server）
|
||||
consumerGroup = "newapi_quota_claim_consumer"
|
||||
)
|
||||
public class NewapiQuotaClaimConsumer implements RocketMQListener<PassportRegisterEvent> {
|
||||
|
||||
/** 额度池服务（claim 落地：渠道过滤 + CAS 抢占 + 池空处置）。 */
|
||||
private final NewapiQuotaService quotaService;
|
||||
|
||||
public NewapiQuotaClaimConsumer(NewapiQuotaService quotaService) {
|
||||
this.quotaService = quotaService;
|
||||
}
|
||||
|
||||
/**
|
||||
* 消费一条「注册成功」事件 → 交 service 做渠道过滤 + 幂等 CAS claim。
|
||||
*
|
||||
* @param event 注册成功事件镜像体（userId=game_player.id / registerChannel 归因）
|
||||
*/
|
||||
@Override
|
||||
public void onMessage(PassportRegisterEvent event) {
|
||||
if (event == null || event.getUserId() == null) {
|
||||
// 毒消息（体空/无 userId）：重投无益，丢弃不重投，只告警留痕。
|
||||
log.error("[newapi-quota-mq] 注册事件体空或缺 userId，丢弃不重投 event={}", event);
|
||||
return;
|
||||
}
|
||||
try {
|
||||
quotaService.claimForRegister(event.getUserId(), event.getRegisterChannel());
|
||||
log.info("[newapi-quota-mq] 消费注册成功事件已处置 userId={}, channel={}", event.getUserId(), event.getRegisterChannel());
|
||||
} catch (Exception e) {
|
||||
// 防御性兜底：service 内已吞池空/常规失败，此处只兜意外异常——正常返回=ack，不让 MQ 无限重投风暴。
|
||||
log.error("[newapi-quota-mq] 消费注册事件异常（吞掉不重投，待补池后懒 claim 兜）userId={}, channel={}",
|
||||
event.getUserId(), event.getRegisterChannel(), e);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,29 @@
|
||||
package com.wanxiang.huijing.game.module.aigc.mq.message;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
import java.io.Serializable;
|
||||
|
||||
/**
|
||||
* 「注册成功」事件消费侧镜像体（WU2 §3.4）
|
||||
*
|
||||
* <p>WU1 的生产侧消息类 {@code PassportRegisterMessage} 落在 huijing-module-system-server（-server，非 -api），
|
||||
* 跨模块不依赖对方 -server（守门④）。故本模块按同名字段建一个消费侧镜像 DTO——RocketMQ 以 JSON 传输载荷，
|
||||
* 消费端 {@code RocketMQListener<PassportRegisterEvent>} 按字段名反序列化即得，与生产侧解耦（topic 是唯一 wire 契约）。
|
||||
* 字段须与生产侧 {@code PassportRegisterMessage} 逐一同名（userId/registerChannel/anonId/idempotentKey）。
|
||||
*
|
||||
* @author 绘境AI
|
||||
*/
|
||||
@Data
|
||||
public class PassportRegisterEvent implements Serializable {
|
||||
|
||||
/** 新注册玩家编号（game_player.id；claim 幂等键） */
|
||||
private Long userId;
|
||||
/** 注册通道：sms / invite / password（渠道过滤据此） */
|
||||
private String registerChannel;
|
||||
/** 注册时携带的客户端 anonId（本消费不使用，保真镜像） */
|
||||
private String anonId;
|
||||
/** 幂等键（=register:{userId}） */
|
||||
private String idempotentKey;
|
||||
|
||||
}
|
||||
@ -0,0 +1,49 @@
|
||||
package com.wanxiang.huijing.game.module.aigc.service.quota;
|
||||
|
||||
/**
|
||||
* new-api per-user 额度池服务（WU2 §3.4 注册 claim + §3.6 派发期 per-user 门）
|
||||
*
|
||||
* <p>账本在 new-api 网关，本服务只做两件纯 game-cloud 内部 DB 操作、零 new-api 外呼：
|
||||
* <ol>
|
||||
* <li><b>注册 claim</b>（{@link #claimForRegister}）：消费 WU1「注册成功」事件，从预置池 CAS 抢一个 FREE 绑玩家；</li>
|
||||
* <li><b>派发取 token</b>（{@link #resolveUserTokenForDispatch}）：真实 member 在线创作时取其 claim 到的 token 随 job 下发。</li>
|
||||
* </ol>
|
||||
*
|
||||
* <p><b>主开关</b>：{@code aigc.newapi-quota.enabled}（默认 false）——关时两个入口全旁路（现行行为，worker 回落全局 key），
|
||||
* 开时注册 claim 消费者装配 + per-user 门生效。这样 WU2 可整体 opt-in，池未预置前不误伤既有生成/批跑验证路。
|
||||
*
|
||||
* @author 绘境AI
|
||||
*/
|
||||
public interface NewapiQuotaService {
|
||||
|
||||
/**
|
||||
* 注册 claim（§3.4）：消费「注册成功」事件，对种子渠道从池 CAS 抢一个 FREE 绑该玩家（幂等）。
|
||||
*
|
||||
* <p>渠道过滤：只对 {@code password/invite} claim；{@code sms} 自动注册与 {@code sms-mock} 后门跳过
|
||||
* （避免一次性手机号每次首登占掉一个 ¥100 池条目，这类若真去生成由懒 claim 兜）。
|
||||
* <p>幂等：uk_claimed_by + uk_biz_no + 单条 CAS 三重收敛，重投/并发不给同一玩家占第二条。
|
||||
* <p>池空：不抛异常（补池是离线人工动作、非 MQ 重试窗口内可恢复，抛异常只灌 DLQ 空转），记 WARN + 触发水位告警，
|
||||
* 该玩家暂不绑，首次生成由懒 claim 兜（§3.6）。
|
||||
*
|
||||
* @param gamePlayerId 新注册玩家编号（game_player.id）
|
||||
* @param registerChannel 注册通道（sms/invite/password）
|
||||
*/
|
||||
void claimForRegister(Long gamePlayerId, String registerChannel);
|
||||
|
||||
/**
|
||||
* 派发期解析 per-user token（§3.6）——per-user 门只对真实 member 在线创作生效。
|
||||
*
|
||||
* <ul>
|
||||
* <li>主开关关 / creatorUserId 空 → 返 null（旁路，worker 回落全局 key）；</li>
|
||||
* <li><b>无 game_player 行</b>（系统/编排/bake-off/admin 触发）→ 返 null（旁路，不查池不懒 claim）；</li>
|
||||
* <li><b>命中 game_player 行</b>（真实 member）→ 取其 CLAIMED 条目的 token_key；无 CLAIMED 则单 CAS 懒 claim 抢一个 FREE：
|
||||
* 抢到返 token_key，池空抢不到 → 抛 {@link QuotaPoolExhaustedException}（当次干净失败给可读拒因 + 水位告警）。</li>
|
||||
* </ul>
|
||||
*
|
||||
* @param creatorUserId 任务创作者（{@code AigcTaskDO.creatorUserId}）
|
||||
* @return per-user token_key（member 命中/懒 claim 成功）；null（旁路：系统触发或主开关关）
|
||||
* @throws QuotaPoolExhaustedException member 需要 token 但池空、懒 claim 抢不到
|
||||
*/
|
||||
String resolveUserTokenForDispatch(Long creatorUserId);
|
||||
|
||||
}
|
||||
@ -0,0 +1,181 @@
|
||||
package com.wanxiang.huijing.game.module.aigc.service.quota;
|
||||
|
||||
import com.wanxiang.huijing.game.module.aigc.dal.dataobject.quota.NewapiQuotaPoolDO;
|
||||
import com.wanxiang.huijing.game.module.aigc.dal.mysql.quota.NewapiQuotaPoolMapper;
|
||||
import com.wanxiang.huijing.framework.tenant.core.util.TenantUtils;
|
||||
import com.wanxiang.huijing.module.system.api.passport.PlayerApi;
|
||||
import com.wanxiang.huijing.module.system.api.passport.dto.PlayerRespDTO;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.dao.DuplicateKeyException;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import jakarta.annotation.Resource;
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
/**
|
||||
* new-api per-user 额度池服务实现（WU2 §3.4/§3.6）
|
||||
*
|
||||
* <p>claim 与派发均是纯 game-cloud 内部 DB 操作、零 new-api 外呼；池是系统级资源，故每次 Mapper 调用都经
|
||||
* {@link TenantUtils#executeIgnore} 跨租户（注册消费/派发线均无租户上下文）。主开关 {@code aigc.newapi-quota.enabled}
|
||||
* 默认 false，关时两入口全旁路 = 现行行为（不误伤既有生成/批跑验证路）。
|
||||
*
|
||||
* @author 绘境AI
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
public class NewapiQuotaServiceImpl implements NewapiQuotaService {
|
||||
|
||||
/** 种子渠道：只对这两类 claim（一次性手机号/后门跳过，避免白占 ¥100 池条目，§3.4）。 */
|
||||
private static final String CHANNEL_PASSWORD = "password";
|
||||
private static final String CHANNEL_INVITE = "invite";
|
||||
|
||||
/**
|
||||
* 主开关（默认 false）：关时 claim/派发全旁路。同 {@code NewapiQuotaClaimConsumer} 的 @ConditionalOnProperty，
|
||||
* 开则消费者装配 + per-user 门生效。
|
||||
*/
|
||||
@Value("${aigc.newapi-quota.enabled:false}")
|
||||
private boolean enabled;
|
||||
|
||||
/** 池水位告警阈值：FREE 计数低于此值即 WARN 提示 ops 补池（§3.4，默认 20）。 */
|
||||
@Value("${aigc.newapi-quota.water-level-threshold:20}")
|
||||
private long waterLevelThreshold;
|
||||
|
||||
@Resource
|
||||
private NewapiQuotaPoolMapper poolMapper;
|
||||
|
||||
/**
|
||||
* 玩家身份 seam（跨模块只依赖 system 的 -api）：判 creatorUserId 是否命中一行 game_player。
|
||||
* 单体内由 system 的 PlayerApiImpl(@Primary) 就地解析；getPlayer 对无此行返 data=null（不抛）。
|
||||
*/
|
||||
@Resource
|
||||
private PlayerApi playerApi;
|
||||
|
||||
// ================== §3.4 注册 claim ==================
|
||||
|
||||
@Override
|
||||
public void claimForRegister(Long gamePlayerId, String registerChannel) {
|
||||
if (!enabled) {
|
||||
// 主开关关：WU2 未启用，注册 claim 全旁路（现行行为，玩家生成走全局 key）。
|
||||
return;
|
||||
}
|
||||
if (gamePlayerId == null) {
|
||||
log.warn("[newapi-quota] 注册 claim 收到空 gamePlayerId，跳过 channel={}", registerChannel);
|
||||
return;
|
||||
}
|
||||
// 渠道过滤：只对种子渠道 claim；sms 自动注册 / sms-mock 后门跳过（避免白占池条目，真去生成由懒 claim 兜）。
|
||||
if (!CHANNEL_PASSWORD.equals(registerChannel) && !CHANNEL_INVITE.equals(registerChannel)) {
|
||||
log.info("[newapi-quota] 注册渠道非种子渠道，跳过 claim gamePlayerId={}, channel={}", gamePlayerId, registerChannel);
|
||||
return;
|
||||
}
|
||||
// 池是系统级资源、消费线无租户上下文 → executeIgnore 跨租户存取（block 体=void，绑 Runnable 重载，丢弃返回值）。
|
||||
TenantUtils.executeIgnore(() -> {
|
||||
tryClaimCore(gamePlayerId, "register");
|
||||
});
|
||||
}
|
||||
|
||||
// ================== §3.6 派发期 per-user 门 ==================
|
||||
|
||||
@Override
|
||||
public String resolveUserTokenForDispatch(Long creatorUserId) {
|
||||
if (!enabled || creatorUserId == null) {
|
||||
// 主开关关 / 无创作者 → 旁路，worker 回落全局 key（现行行为）。
|
||||
return null;
|
||||
}
|
||||
// 身份门：只对真实 member（命中一行 game_player）生效。getPlayer 对无此行返 null（系统/编排/bake-off/admin 触发）。
|
||||
PlayerRespDTO player = playerApi.getPlayer(creatorUserId).getCheckedData();
|
||||
if (player == null) {
|
||||
// 无 game_player 行 → 旁路：不查池、不懒 claim，userToken 留空，worker 回落全局 key（§3.6 越界不误伤）。
|
||||
log.debug("[newapi-quota] 无 game_player 行（系统/编排触发），per-user 门旁路 creatorUserId={}", creatorUserId);
|
||||
return null;
|
||||
}
|
||||
// 真实 member：取其 CLAIMED token；无则懒 claim。全程 executeIgnore 跨租户（派发线无租户上下文）。
|
||||
// ⚠ executeIgnore(Callable) 会把内部抛的任何异常包成 RuntimeException（丢失类型），故池空信号用「返回 null」表达、
|
||||
// 在 executeIgnore 外再抛 QuotaPoolExhaustedException，保证 dispatchGeneric 能按类型精确捕获。
|
||||
String tokenKey = TenantUtils.executeIgnore(() -> {
|
||||
NewapiQuotaPoolDO claimed = poolMapper.selectClaimedByPlayer(creatorUserId);
|
||||
if (claimed != null && StringUtils.hasText(claimed.getNewapiTokenKey())) {
|
||||
return claimed.getNewapiTokenKey();
|
||||
}
|
||||
// 无 CLAIMED 条目（存量 null / 注册时池空未绑）→ 派发线单 CAS 懒 claim（纯 DB CAS，非慢外呼，派发热路安全）。
|
||||
boolean claimedNow = tryClaimCore(creatorUserId, "lazy-dispatch");
|
||||
if (claimedNow) {
|
||||
NewapiQuotaPoolDO after = poolMapper.selectClaimedByPlayer(creatorUserId);
|
||||
if (after != null && StringUtils.hasText(after.getNewapiTokenKey())) {
|
||||
return after.getNewapiTokenKey();
|
||||
}
|
||||
}
|
||||
// member 路返回 null 专表「池空抢不到」（外部据此抛耗尽）；水位告警已在 tryClaimCore 内触发。
|
||||
return null;
|
||||
});
|
||||
if (tokenKey == null) {
|
||||
// 池空抢不到 → 当次干净失败给可读拒因（quota_exhausted）。绝不派无 token 的 member create job 下去静默烧共享额度。
|
||||
throw new QuotaPoolExhaustedException(creatorUserId);
|
||||
}
|
||||
return tokenKey;
|
||||
}
|
||||
|
||||
// ================== 内部：幂等 CAS claim（调用方须已 executeIgnore 包裹） ==================
|
||||
|
||||
/**
|
||||
* 幂等 CAS claim 核心（假定调用方已 {@code executeIgnore} 包裹）：
|
||||
* 已绑复用 / 抢一个 FREE / 并发 dup 收敛 / 池空不抛。
|
||||
*
|
||||
* @param gamePlayerId game_player.id
|
||||
* @param scene 场景（register / lazy-dispatch，日志用）
|
||||
* @return true=该玩家已有 CLAIMED 条目（既有复用 / 本次抢到 / 并发已被抢占同玩家）；false=池空未绑
|
||||
*/
|
||||
private boolean tryClaimCore(Long gamePlayerId, String scene) {
|
||||
// 幂等前置：已绑则 no-op 复用（重投事件 / 并发天然收敛）。
|
||||
NewapiQuotaPoolDO existing = poolMapper.selectClaimedByPlayer(gamePlayerId);
|
||||
if (existing != null) {
|
||||
log.info("[newapi-quota] 玩家已绑池条目，claim no-op 复用 scene={}, gamePlayerId={}, poolId={}",
|
||||
scene, gamePlayerId, existing.getId());
|
||||
return true;
|
||||
}
|
||||
String bizNo = "claim_" + gamePlayerId;
|
||||
try {
|
||||
int affected = poolMapper.claimOneFree(gamePlayerId, bizNo, LocalDateTime.now());
|
||||
if (affected == 1) {
|
||||
log.info("[newapi-quota] claim 成功 FREE→CLAIMED scene={}, gamePlayerId={}, bizNo={}", scene, gamePlayerId, bizNo);
|
||||
alertIfLowWater(scene);
|
||||
return true;
|
||||
}
|
||||
// affected=0：子查询无 FREE 命中 = 池空（非并发；并发同玩家走下方 DuplicateKeyException 分支）。
|
||||
log.warn("[newapi-quota] 池空（无 FREE 条目），本次未绑 scene={}, gamePlayerId={}", scene, gamePlayerId);
|
||||
triggerWaterLevelAlert(0L, scene);
|
||||
return false;
|
||||
} catch (DuplicateKeyException e) {
|
||||
// 并发同玩家 claim：两路各抢一条不同 FREE，第二路写 claimed_by/biz_no 撞 uk_claimed_by/uk_biz_no →
|
||||
// 幂等收敛为一条（该玩家已被并发路绑好），视为成功复用，不抛。
|
||||
log.info("[newapi-quota] 并发同玩家 claim 收敛（uk 拦截），幂等复用 scene={}, gamePlayerId={}", scene, gamePlayerId);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* claim 成功后查水位，低于阈值即告警（§3.4 池水位是核心运维信号，缺了池会静默见底）。
|
||||
*/
|
||||
private void alertIfLowWater(String scene) {
|
||||
try {
|
||||
Long free = poolMapper.countFree();
|
||||
long freeVal = free == null ? 0L : free;
|
||||
if (freeVal < waterLevelThreshold) {
|
||||
triggerWaterLevelAlert(freeVal, scene);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
// 水位查询失败不阻断 claim（best-effort 观测信号）；错误路径留痕。
|
||||
log.warn("[newapi-quota] 池水位查询失败（不阻断 claim）scene={}", scene, e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 触发池水位告警——内测阶段以带唯一标记的 WARN 落日志供 ops/观测线捕获；对接真实告警通道属观测线 follow-up。
|
||||
*/
|
||||
private void triggerWaterLevelAlert(long freeCount, String scene) {
|
||||
log.warn("[newapi-quota-alert] 池水位低，请 ops 重跑预置脚本补池 free={}, threshold={}, scene={}",
|
||||
freeCount, waterLevelThreshold, scene);
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,26 @@
|
||||
package com.wanxiang.huijing.game.module.aigc.service.quota;
|
||||
|
||||
/**
|
||||
* 额度池空信号异常（WU2 §3.6）——真实 member 在派发热路懒 claim 时池已见底、抢不到 FREE 条目。
|
||||
*
|
||||
* 由 {@link NewapiQuotaService#resolveUserTokenForDispatch} 抛出、{@code AigcGenerateExecutor.dispatchGeneric} 捕获，
|
||||
* 翻成 {@code FailureReasonEnum.QUOTA_EXHAUSTED} 当次干净失败（文案「额度账户准备中，请稍后重试」），
|
||||
* 而不是派一个没 token 的 create 路 job 下去静默烧共享额度。运行时异常：不强制调用方 try（dispatch 显式捕获处置）。
|
||||
*
|
||||
* @author 绘境AI
|
||||
*/
|
||||
public class QuotaPoolExhaustedException extends RuntimeException {
|
||||
|
||||
/** 触发耗尽的创作者（game_player.id），便于日志/告警定位 */
|
||||
private final Long creatorUserId;
|
||||
|
||||
public QuotaPoolExhaustedException(Long creatorUserId) {
|
||||
super("newapi 额度池空，玩家懒 claim 抢不到 FREE 条目 creatorUserId=" + creatorUserId);
|
||||
this.creatorUserId = creatorUserId;
|
||||
}
|
||||
|
||||
public Long getCreatorUserId() {
|
||||
return creatorUserId;
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,69 @@
|
||||
package com.wanxiang.huijing.game.module.aigc.mq;
|
||||
|
||||
import com.wanxiang.huijing.game.module.aigc.mq.message.PassportRegisterEvent;
|
||||
import com.wanxiang.huijing.game.module.aigc.service.quota.NewapiQuotaService;
|
||||
import com.wanxiang.huijing.framework.test.core.ut.BaseMockitoUnitTest;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockito.InjectMocks;
|
||||
import org.mockito.Mock;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.anyString;
|
||||
import static org.mockito.ArgumentMatchers.eq;
|
||||
import static org.mockito.Mockito.doThrow;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verify;
|
||||
|
||||
/**
|
||||
* {@link NewapiQuotaClaimConsumer} 单元测试（纯 Mockito）——把守「委托 service 做渠道过滤+claim」与「不外抛」契约。
|
||||
*
|
||||
* <p>覆盖：正常事件透传 userId+channel 给 service；毒消息（体空/无 userId）丢弃不委托不抛；service 异常被兜底吞（不无限重投）。
|
||||
*
|
||||
* @author 绘境AI
|
||||
*/
|
||||
class NewapiQuotaClaimConsumerTest extends BaseMockitoUnitTest {
|
||||
|
||||
@Mock
|
||||
private NewapiQuotaService quotaService;
|
||||
|
||||
@InjectMocks
|
||||
private NewapiQuotaClaimConsumer consumer;
|
||||
|
||||
private PassportRegisterEvent event(Long userId, String channel) {
|
||||
PassportRegisterEvent e = new PassportRegisterEvent();
|
||||
e.setUserId(userId);
|
||||
e.setRegisterChannel(channel);
|
||||
e.setIdempotentKey(userId == null ? null : "register:" + userId);
|
||||
return e;
|
||||
}
|
||||
|
||||
/** 正常事件：透传 userId + channel 给 service（渠道过滤在 service 内做）。 */
|
||||
@Test
|
||||
void testOnMessage_delegatesToService() {
|
||||
consumer.onMessage(event(5L, "password"));
|
||||
verify(quotaService).claimForRegister(eq(5L), eq("password"));
|
||||
}
|
||||
|
||||
/** 毒消息：event 为 null → 不委托、不抛。 */
|
||||
@Test
|
||||
void testOnMessage_nullEvent_discardNoThrow() {
|
||||
assertDoesNotThrow(() -> consumer.onMessage(null));
|
||||
verify(quotaService, never()).claimForRegister(any(), anyString());
|
||||
}
|
||||
|
||||
/** 毒消息：userId 为空 → 不委托、不抛。 */
|
||||
@Test
|
||||
void testOnMessage_nullUserId_discardNoThrow() {
|
||||
assertDoesNotThrow(() -> consumer.onMessage(event(null, "password")));
|
||||
verify(quotaService, never()).claimForRegister(any(), anyString());
|
||||
}
|
||||
|
||||
/** service 异常：兜底吞掉不外抛（正常返回=ack，不让 MQ 无限重投）。 */
|
||||
@Test
|
||||
void testOnMessage_serviceThrows_swallowedNoThrow() {
|
||||
doThrow(new RuntimeException("db down")).when(quotaService).claimForRegister(eq(5L), eq("password"));
|
||||
assertDoesNotThrow(() -> consumer.onMessage(event(5L, "password")));
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,194 @@
|
||||
package com.wanxiang.huijing.game.module.aigc.service.quota;
|
||||
|
||||
import com.wanxiang.huijing.game.module.aigc.dal.dataobject.quota.NewapiQuotaPoolDO;
|
||||
import com.wanxiang.huijing.game.module.aigc.dal.mysql.quota.NewapiQuotaPoolMapper;
|
||||
import com.wanxiang.huijing.framework.common.pojo.CommonResult;
|
||||
import com.wanxiang.huijing.framework.test.core.ut.BaseMockitoUnitTest;
|
||||
import com.wanxiang.huijing.module.system.api.passport.PlayerApi;
|
||||
import com.wanxiang.huijing.module.system.api.passport.dto.PlayerRespDTO;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockito.InjectMocks;
|
||||
import org.mockito.Mock;
|
||||
import org.springframework.dao.DuplicateKeyException;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.anyLong;
|
||||
import static org.mockito.ArgumentMatchers.anyString;
|
||||
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 NewapiQuotaServiceImpl} 单元测试（纯 Mockito，不触真 DB）——把守 WU2 §3.4 注册 claim 与 §3.6 派发 per-user 门契约。
|
||||
*
|
||||
* <p>覆盖：claim CAS 幂等（既有复用不二次占池 / 并发 dup 收敛）、渠道过滤（sms 跳过）、池空不抛、主开关旁路、
|
||||
* per-user 门身份分流（member 取 token / 系统旁路 userToken 空）、懒 claim fail-fast（池空抛 QuotaPoolExhaustedException）。
|
||||
*
|
||||
* @author 绘境AI
|
||||
*/
|
||||
class NewapiQuotaServiceImplTest extends BaseMockitoUnitTest {
|
||||
|
||||
@InjectMocks
|
||||
private NewapiQuotaServiceImpl service;
|
||||
|
||||
@Mock
|
||||
private NewapiQuotaPoolMapper poolMapper;
|
||||
@Mock
|
||||
private PlayerApi playerApi;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
// @Value 字段 Mockito 不注入，反射置默认：主开关开、水位阈值 20。
|
||||
ReflectionTestUtils.setField(service, "enabled", true);
|
||||
ReflectionTestUtils.setField(service, "waterLevelThreshold", 20L);
|
||||
}
|
||||
|
||||
private PlayerRespDTO member(long userId) {
|
||||
PlayerRespDTO dto = new PlayerRespDTO();
|
||||
dto.setUserId(userId);
|
||||
dto.setStatus(0);
|
||||
dto.setCreatorFlag(1);
|
||||
return dto;
|
||||
}
|
||||
|
||||
private NewapiQuotaPoolDO claimedEntry(long playerId, String tokenKey) {
|
||||
NewapiQuotaPoolDO entry = new NewapiQuotaPoolDO();
|
||||
entry.setId(9001L);
|
||||
entry.setClaimedByGamePlayerId(playerId);
|
||||
entry.setNewapiTokenKey(tokenKey);
|
||||
entry.setStatus("CLAIMED");
|
||||
return entry;
|
||||
}
|
||||
|
||||
// ===================== §3.4 注册 claim =====================
|
||||
|
||||
/** password 渠道 + 无既有条目 → 执行一次 CAS 抢占，bizNo=claim_<id>。 */
|
||||
@Test
|
||||
void testClaimForRegister_password_casOnce() {
|
||||
when(poolMapper.selectClaimedByPlayer(5L)).thenReturn(null);
|
||||
when(poolMapper.claimOneFree(eq(5L), eq("claim_5"), any(LocalDateTime.class))).thenReturn(1);
|
||||
|
||||
service.claimForRegister(5L, "password");
|
||||
|
||||
verify(poolMapper).claimOneFree(eq(5L), eq("claim_5"), any(LocalDateTime.class));
|
||||
}
|
||||
|
||||
/** 幂等：该玩家已有 CLAIMED 条目 → no-op 复用，不再 CAS（重投不二次占池）。 */
|
||||
@Test
|
||||
void testClaimForRegister_idempotent_existingClaimed_noCas() {
|
||||
when(poolMapper.selectClaimedByPlayer(5L)).thenReturn(claimedEntry(5L, "sk-EXIST"));
|
||||
|
||||
service.claimForRegister(5L, "invite");
|
||||
|
||||
verify(poolMapper, never()).claimOneFree(anyLong(), anyString(), any(LocalDateTime.class));
|
||||
}
|
||||
|
||||
/** 渠道过滤：sms 自动注册跳过 claim，池 mapper 全程不碰。 */
|
||||
@Test
|
||||
void testClaimForRegister_smsChannel_skip() {
|
||||
service.claimForRegister(5L, "sms");
|
||||
verifyNoInteractions(poolMapper);
|
||||
}
|
||||
|
||||
/** 渠道过滤：sms-mock 后门跳过。 */
|
||||
@Test
|
||||
void testClaimForRegister_smsMockChannel_skip() {
|
||||
service.claimForRegister(5L, "sms-mock");
|
||||
verifyNoInteractions(poolMapper);
|
||||
}
|
||||
|
||||
/** 池空：CAS 返 0 → 不抛异常、不绑（记 WARN + 水位告警，注册照常成功）。 */
|
||||
@Test
|
||||
void testClaimForRegister_poolEmpty_noThrow() {
|
||||
when(poolMapper.selectClaimedByPlayer(5L)).thenReturn(null);
|
||||
when(poolMapper.claimOneFree(eq(5L), eq("claim_5"), any(LocalDateTime.class))).thenReturn(0);
|
||||
|
||||
assertDoesNotThrow(() -> service.claimForRegister(5L, "password"));
|
||||
}
|
||||
|
||||
/** 并发同玩家 claim：CAS 撞 uk 抛 DuplicateKeyException → 幂等收敛，不外抛。 */
|
||||
@Test
|
||||
void testClaimForRegister_concurrentDuplicate_idempotentNoThrow() {
|
||||
when(poolMapper.selectClaimedByPlayer(5L)).thenReturn(null);
|
||||
when(poolMapper.claimOneFree(eq(5L), eq("claim_5"), any(LocalDateTime.class)))
|
||||
.thenThrow(new DuplicateKeyException("uk_claimed_by 冲突"));
|
||||
|
||||
assertDoesNotThrow(() -> service.claimForRegister(5L, "password"));
|
||||
}
|
||||
|
||||
/** 主开关关：注册 claim 全旁路，playerApi/poolMapper 均不碰（现行行为）。 */
|
||||
@Test
|
||||
void testClaimForRegister_disabled_bypass() {
|
||||
ReflectionTestUtils.setField(service, "enabled", false);
|
||||
service.claimForRegister(5L, "password");
|
||||
verifyNoInteractions(poolMapper, playerApi);
|
||||
}
|
||||
|
||||
// ===================== §3.6 派发 per-user 门 =====================
|
||||
|
||||
/** 主开关关：派发解析旁路返 null，不查身份、不查池。 */
|
||||
@Test
|
||||
void testResolve_disabled_null() {
|
||||
ReflectionTestUtils.setField(service, "enabled", false);
|
||||
assertNull(service.resolveUserTokenForDispatch(5L));
|
||||
verifyNoInteractions(playerApi, poolMapper);
|
||||
}
|
||||
|
||||
/** 系统/编排触发（无 game_player 行）：getPlayer 返 null → 旁路返 null，不查池不懒 claim。 */
|
||||
@Test
|
||||
void testResolve_nonMember_bypassNull() {
|
||||
when(playerApi.getPlayer(999L)).thenReturn(CommonResult.success((PlayerRespDTO) null));
|
||||
|
||||
assertNull(service.resolveUserTokenForDispatch(999L));
|
||||
|
||||
verifyNoInteractions(poolMapper);
|
||||
}
|
||||
|
||||
/** 真实 member 且已有 CLAIMED 条目 → 取其 token_key，不懒 claim。 */
|
||||
@Test
|
||||
void testResolve_memberWithClaimed_returnsToken() {
|
||||
when(playerApi.getPlayer(5L)).thenReturn(CommonResult.success(member(5L)));
|
||||
when(poolMapper.selectClaimedByPlayer(5L)).thenReturn(claimedEntry(5L, "sk-TOKEN-5"));
|
||||
|
||||
assertEquals("sk-TOKEN-5", service.resolveUserTokenForDispatch(5L));
|
||||
|
||||
verify(poolMapper, never()).claimOneFree(anyLong(), anyString(), any(LocalDateTime.class));
|
||||
}
|
||||
|
||||
/** 真实 member 无 CLAIMED → 懒 claim 抢到 → 取 token_key。 */
|
||||
@Test
|
||||
void testResolve_memberNoClaimed_lazyClaimSuccess_returnsToken() {
|
||||
when(playerApi.getPlayer(5L)).thenReturn(CommonResult.success(member(5L)));
|
||||
// 首查无 CLAIMED（含 tryClaimCore 内的幂等前置也走此桩），CAS 抢到后再查得已绑条目。
|
||||
when(poolMapper.selectClaimedByPlayer(5L))
|
||||
.thenReturn(null)
|
||||
.thenReturn(null)
|
||||
.thenReturn(claimedEntry(5L, "sk-LAZY-5"));
|
||||
when(poolMapper.claimOneFree(eq(5L), eq("claim_5"), any(LocalDateTime.class))).thenReturn(1);
|
||||
|
||||
assertEquals("sk-LAZY-5", service.resolveUserTokenForDispatch(5L));
|
||||
|
||||
verify(poolMapper).claimOneFree(eq(5L), eq("claim_5"), any(LocalDateTime.class));
|
||||
}
|
||||
|
||||
/** 真实 member 无 CLAIMED 且池空 → 懒 claim fail-fast 抛 QuotaPoolExhaustedException（当次干净失败）。 */
|
||||
@Test
|
||||
void testResolve_memberNoClaimed_poolEmpty_throwsExhausted() {
|
||||
when(playerApi.getPlayer(5L)).thenReturn(CommonResult.success(member(5L)));
|
||||
when(poolMapper.selectClaimedByPlayer(5L)).thenReturn(null);
|
||||
when(poolMapper.claimOneFree(eq(5L), eq("claim_5"), any(LocalDateTime.class))).thenReturn(0);
|
||||
|
||||
assertThrows(QuotaPoolExhaustedException.class, () -> service.resolveUserTokenForDispatch(5L));
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user