feat(p1r): 接入 Account 事件传播真实链路
This commit is contained in:
zizi
parent
0f498026bb
commit
0e10caf324
@ -40,6 +40,11 @@
|
||||
<artifactId>muse-module-infra-api</artifactId>
|
||||
<version>${revision}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>cn.iocoder.cloud</groupId>
|
||||
<artifactId>muse-module-events-api</artifactId>
|
||||
<version>${revision}</version>
|
||||
</dependency>
|
||||
|
||||
<!-- 业务组件 -->
|
||||
<dependency>
|
||||
|
||||
@ -0,0 +1,33 @@
|
||||
package cn.iocoder.muse.module.member.application.account;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
|
||||
/**
|
||||
* Account Events payload 安全白名单。
|
||||
*/
|
||||
public final class AccountEventPayloads {
|
||||
|
||||
private static final Set<String> PAYLOAD_KEYS = Set.of("type", "message", "resourceRef", "timestamp");
|
||||
private static final Set<String> RESOURCE_REF_KEYS = Set.of("resourceType", "resourceId");
|
||||
|
||||
private AccountEventPayloads() {
|
||||
}
|
||||
|
||||
public static boolean isValidQuotaAlertPayload(Map<String, Object> payload) {
|
||||
if (payload == null || !PAYLOAD_KEYS.equals(payload.keySet())) {
|
||||
return false;
|
||||
}
|
||||
if (!"quota_alert".equals(payload.get("type")) || !"额度已调整".equals(payload.get("message"))
|
||||
|| !(payload.get("timestamp") instanceof String)) {
|
||||
return false;
|
||||
}
|
||||
if (!(payload.get("resourceRef") instanceof Map<?, ?> resourceRef)
|
||||
|| !RESOURCE_REF_KEYS.equals(resourceRef.keySet())) {
|
||||
return false;
|
||||
}
|
||||
return "account_quota".equals(resourceRef.get("resourceType"))
|
||||
&& resourceRef.get("resourceId") instanceof Number;
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,18 @@
|
||||
package cn.iocoder.muse.module.member.application.account;
|
||||
|
||||
import cn.iocoder.muse.module.member.dal.dataobject.account.AccountEventPublishOutboxDO;
|
||||
import cn.iocoder.muse.module.member.dal.dataobject.account.MemberEntitlementAuditLogDO;
|
||||
|
||||
/**
|
||||
* Account 本域 Events publish outbox 创建服务。
|
||||
*/
|
||||
public interface AccountEventPublishOutboxService {
|
||||
|
||||
/**
|
||||
* 为 quota adjustment entitlement audit log 创建一条 Account Events publish outbox。
|
||||
*/
|
||||
AccountEventPublishOutboxDO createForQuotaAdjustment(MemberEntitlementAuditLogDO log,
|
||||
String sourceCommandId,
|
||||
String requestHash);
|
||||
|
||||
}
|
||||
@ -0,0 +1,195 @@
|
||||
package cn.iocoder.muse.module.member.application.account;
|
||||
|
||||
import cn.iocoder.muse.framework.common.exception.ServiceException;
|
||||
import cn.iocoder.muse.framework.common.util.json.JsonUtils;
|
||||
import cn.iocoder.muse.module.member.dal.dataobject.account.AccountEventPublishOutboxDO;
|
||||
import cn.iocoder.muse.module.member.dal.dataobject.account.MemberEntitlementAuditLogDO;
|
||||
import cn.iocoder.muse.module.member.dal.mysql.account.AccountEventPublishOutboxMapper;
|
||||
import cn.iocoder.muse.module.member.framework.config.MuseAccountEventsProperties;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.security.MessageDigest;
|
||||
import java.security.NoSuchAlgorithmException;
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* Account quota adjustment 发布 outbox 服务实现。
|
||||
*/
|
||||
@Service
|
||||
@Slf4j
|
||||
public class AccountEventPublishOutboxServiceImpl implements AccountEventPublishOutboxService {
|
||||
|
||||
static final String STATUS_QUEUED = "queued";
|
||||
|
||||
private static final String CHANGE_TYPE_QUOTA_ADJUSTMENT = "quota_adjustment";
|
||||
private static final String SOURCE_REVISION_NONE = "__none__";
|
||||
private static final String EVENT_TYPE_NOTIFICATION = "notification";
|
||||
private static final String NOTIFICATION_QUOTA_ALERT = "quota_alert";
|
||||
private static final String RESOURCE_REF_TYPE_ACCOUNT_QUOTA = "account_quota";
|
||||
static final String ERROR_PAYLOAD_INVALID_CODE = "ACCOUNT_EVENTS_PAYLOAD_INVALID";
|
||||
static final int ERROR_PAYLOAD_INVALID_NUMERIC_CODE = 1_042_000_050;
|
||||
|
||||
private final AccountEventPublishOutboxMapper outboxMapper;
|
||||
private final MuseAccountEventsProperties properties;
|
||||
|
||||
public AccountEventPublishOutboxServiceImpl(AccountEventPublishOutboxMapper outboxMapper,
|
||||
MuseAccountEventsProperties properties) {
|
||||
this.outboxMapper = outboxMapper;
|
||||
this.properties = properties;
|
||||
}
|
||||
|
||||
@Override
|
||||
public AccountEventPublishOutboxDO createForQuotaAdjustment(MemberEntitlementAuditLogDO auditLog,
|
||||
String sourceCommandId,
|
||||
String requestHash) {
|
||||
if (auditLog == null || !CHANGE_TYPE_QUOTA_ADJUSTMENT.equals(auditLog.getChangeType())) {
|
||||
return null;
|
||||
}
|
||||
if (auditLog.getAccountUserId() == null || auditLog.getAccountUserId() <= 0) {
|
||||
log.warn("Account Events outbox 跳过非法 owner 配额审计日志，tenantId={}, auditLogId={}, accountUserId={}",
|
||||
auditLog.getTenantId(), auditLog.getId(), auditLog.getAccountUserId());
|
||||
return null;
|
||||
}
|
||||
if (auditLog.getId() == null || auditLog.getId() <= 0) {
|
||||
log.warn("Account Events outbox 跳过非法 source fact，tenantId={}, accountUserId={}, auditLogId={}",
|
||||
auditLog.getTenantId(), auditLog.getAccountUserId(), auditLog.getId());
|
||||
return null;
|
||||
}
|
||||
AccountEventPublishOutboxDO outbox = buildOutbox(auditLog, sourceCommandId, requestHash);
|
||||
validatePayloadOrThrow(outbox);
|
||||
int inserted = outboxMapper.insertIgnore(outbox);
|
||||
if (inserted > 0) {
|
||||
return outboxMapper.selectByTenantIdAndOutboxId(auditLog.getTenantId(), outbox.getOutboxId());
|
||||
}
|
||||
AccountEventPublishOutboxDO existing = outboxMapper.selectByTenantIdAndAuditLogIdAndNotificationType(
|
||||
auditLog.getTenantId(), auditLog.getId(), NOTIFICATION_QUOTA_ALERT);
|
||||
return existing == null ? outboxMapper.selectByTenantIdAndCommandId(auditLog.getTenantId(), outbox.getCommandId())
|
||||
: existing;
|
||||
}
|
||||
|
||||
private AccountEventPublishOutboxDO buildOutbox(MemberEntitlementAuditLogDO log,
|
||||
String sourceCommandId,
|
||||
String requestHash) {
|
||||
AccountEventPublishOutboxDO outbox = new AccountEventPublishOutboxDO();
|
||||
outbox.setTenantId(log.getTenantId());
|
||||
outbox.setOutboxId(buildOutboxId(log.getTenantId(), log.getId(), log.getAccountUserId()));
|
||||
outbox.setCommandId(buildPublishCommandId(log.getTenantId(), log.getId(), log.getAccountUserId()));
|
||||
outbox.setSourceCommandId(resolveSourceCommandId(sourceCommandId, log.getIdempotencyKey()));
|
||||
outbox.setAuditLogId(log.getId());
|
||||
outbox.setAccountUserId(log.getAccountUserId());
|
||||
outbox.setOperatorUserId(log.getOperatorUserId());
|
||||
outbox.setResourceType(resourceType(log));
|
||||
outbox.setDeltaAmount(deltaAmount(log));
|
||||
outbox.setSourceOwner(log.getSourceOwner());
|
||||
outbox.setAuditEventId(log.getAuditEventId());
|
||||
outbox.setRequestHash(requestHash);
|
||||
outbox.setSourceRevision(SOURCE_REVISION_NONE);
|
||||
outbox.setEventType(EVENT_TYPE_NOTIFICATION);
|
||||
outbox.setNotificationType(NOTIFICATION_QUOTA_ALERT);
|
||||
outbox.setResourceRefType(RESOURCE_REF_TYPE_ACCOUNT_QUOTA);
|
||||
outbox.setResourceRefId(log.getId());
|
||||
LocalDateTime eventTime = eventTimestamp(log);
|
||||
outbox.setPayloadSummary(JsonUtils.toJsonString(notificationPayload(log, eventTime)));
|
||||
outbox.setPublishStatus(STATUS_QUEUED);
|
||||
outbox.setAttemptCount(0);
|
||||
outbox.setMaxAttempt(configuredMaxAttempt());
|
||||
outbox.setNextRetryAt(LocalDateTime.now());
|
||||
outbox.setCreateTime(eventTime);
|
||||
return outbox;
|
||||
}
|
||||
|
||||
private Map<String, Object> notificationPayload(MemberEntitlementAuditLogDO log, LocalDateTime eventTime) {
|
||||
Map<String, Object> payload = new LinkedHashMap<>();
|
||||
payload.put("type", NOTIFICATION_QUOTA_ALERT);
|
||||
payload.put("message", "额度已调整");
|
||||
payload.put("resourceRef", resourceRef(log.getId()));
|
||||
payload.put("timestamp", eventTime.toString());
|
||||
return payload;
|
||||
}
|
||||
|
||||
private LocalDateTime eventTimestamp(MemberEntitlementAuditLogDO log) {
|
||||
return log.getCreateTime() == null ? LocalDateTime.now() : log.getCreateTime();
|
||||
}
|
||||
|
||||
private Map<String, Object> resourceRef(Long auditLogId) {
|
||||
Map<String, Object> resourceRef = new LinkedHashMap<>();
|
||||
resourceRef.put("resourceType", RESOURCE_REF_TYPE_ACCOUNT_QUOTA);
|
||||
resourceRef.put("resourceId", auditLogId);
|
||||
return resourceRef;
|
||||
}
|
||||
|
||||
private void validatePayloadOrThrow(AccountEventPublishOutboxDO outbox) {
|
||||
Map<String, Object> payload = JsonUtils.parseObject(outbox.getPayloadSummary(), Map.class);
|
||||
if (!AccountEventPayloads.isValidQuotaAlertPayload(payload)
|
||||
|| !StringUtils.hasText(outbox.getResourceType())
|
||||
|| outbox.getDeltaAmount() == null) {
|
||||
log.warn("Account Events outbox payload 安全校验失败，tenantId={}, accountUserId={}, auditLogId={}, errorCode={}",
|
||||
outbox.getTenantId(), outbox.getAccountUserId(), outbox.getAuditLogId(),
|
||||
ERROR_PAYLOAD_INVALID_CODE);
|
||||
throw new ServiceException(ERROR_PAYLOAD_INVALID_NUMERIC_CODE, ERROR_PAYLOAD_INVALID_CODE);
|
||||
}
|
||||
}
|
||||
|
||||
private String resolveSourceCommandId(String sourceCommandId, String idempotencyKey) {
|
||||
if (StringUtils.hasText(sourceCommandId)) {
|
||||
return sourceCommandId;
|
||||
}
|
||||
if (!StringUtils.hasText(idempotencyKey)) {
|
||||
return "";
|
||||
}
|
||||
int separator = idempotencyKey.lastIndexOf(':');
|
||||
return separator <= 0 ? idempotencyKey : idempotencyKey.substring(0, separator);
|
||||
}
|
||||
|
||||
private String resourceType(MemberEntitlementAuditLogDO log) {
|
||||
return JsonUtils.parseTree(log.getDeltaValueSnapshot()).path("resourceType").asText(null);
|
||||
}
|
||||
|
||||
private Long deltaAmount(MemberEntitlementAuditLogDO log) {
|
||||
if (JsonUtils.parseTree(log.getDeltaValueSnapshot()).path("delta").isNumber()) {
|
||||
return JsonUtils.parseTree(log.getDeltaValueSnapshot()).path("delta").asLong();
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private int configuredMaxAttempt() {
|
||||
if (properties == null || properties.getPublishWorker() == null
|
||||
|| properties.getPublishWorker().getMaxAttempt() <= 0) {
|
||||
return 5;
|
||||
}
|
||||
return properties.getPublishWorker().getMaxAttempt();
|
||||
}
|
||||
|
||||
private String buildPublishCommandId(Long tenantId, Long auditLogId, Long accountUserId) {
|
||||
String raw = idempotencyKey(tenantId, auditLogId, accountUserId);
|
||||
return "acct_evt:" + sha256Hex(raw).substring(0, 32);
|
||||
}
|
||||
|
||||
private String buildOutboxId(Long tenantId, Long auditLogId, Long accountUserId) {
|
||||
String raw = idempotencyKey(tenantId, auditLogId, accountUserId);
|
||||
return "acct_out:" + sha256Hex(raw).substring(0, 32);
|
||||
}
|
||||
|
||||
private String idempotencyKey(Long tenantId, Long auditLogId, Long accountUserId) {
|
||||
return tenantId + "|" + auditLogId + "|" + accountUserId + "|" + NOTIFICATION_QUOTA_ALERT;
|
||||
}
|
||||
|
||||
private String sha256Hex(String raw) {
|
||||
try {
|
||||
byte[] digest = MessageDigest.getInstance("SHA-256").digest(raw.getBytes(StandardCharsets.UTF_8));
|
||||
StringBuilder hex = new StringBuilder(digest.length * 2);
|
||||
for (byte value : digest) {
|
||||
hex.append(String.format("%02x", value));
|
||||
}
|
||||
return hex.toString();
|
||||
} catch (NoSuchAlgorithmException exception) {
|
||||
throw new IllegalStateException("SHA-256 algorithm is required", exception);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,248 @@
|
||||
package cn.iocoder.muse.module.member.application.account;
|
||||
|
||||
import cn.iocoder.muse.framework.common.pojo.CommonResult;
|
||||
import cn.iocoder.muse.framework.common.util.json.JsonUtils;
|
||||
import cn.iocoder.muse.framework.tenant.core.util.TenantUtils;
|
||||
import cn.iocoder.muse.module.events.api.publish.EventsPublishApi;
|
||||
import cn.iocoder.muse.module.events.api.publish.dto.EventsPublishReqDTO;
|
||||
import cn.iocoder.muse.module.events.api.publish.dto.EventsPublishRespDTO;
|
||||
import cn.iocoder.muse.module.member.dal.dataobject.account.AccountEventPublishOutboxDO;
|
||||
import cn.iocoder.muse.module.member.dal.mysql.account.AccountEventPublishOutboxMapper;
|
||||
import cn.iocoder.muse.module.member.framework.config.MuseAccountEventsProperties;
|
||||
import com.fasterxml.jackson.core.type.TypeReference;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* Account quota adjustment outbox 发布到统一 Events 的 worker。
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
public class AccountEventPublishWorker {
|
||||
|
||||
private static final String PUBLISH_STATUS_ACCEPTED = "accepted";
|
||||
private static final String PUBLISH_STATUS_REJECTED = "rejected";
|
||||
private static final String PUBLISH_STATUS_BLOCKED = "blocked";
|
||||
|
||||
private static final String SOURCE_OWNER_ACCOUNT = "account";
|
||||
private static final String SOURCE_TYPE_ACCOUNT_QUOTA_ADJUSTMENT = "account_quota_adjustment";
|
||||
private static final String EVENT_TYPE_NOTIFICATION = "notification";
|
||||
private static final String SOURCE_REVISION_NONE = "__none__";
|
||||
|
||||
private static final String ERROR_RETRY_EXHAUSTED = "ACCOUNT_EVENTS_PUBLISH_RETRY_EXHAUSTED";
|
||||
private static final String ERROR_API_FAILED = "ACCOUNT_EVENTS_PUBLISH_API_FAILED";
|
||||
private static final String ERROR_TEMPORARY_UNAVAILABLE = "ACCOUNT_EVENTS_PUBLISH_TEMPORARY_UNAVAILABLE";
|
||||
private static final String ERROR_UNKNOWN_STATUS = "ACCOUNT_EVENTS_PUBLISH_UNKNOWN_STATUS";
|
||||
private static final String ERROR_PAYLOAD_INVALID = "ACCOUNT_EVENTS_PAYLOAD_INVALID";
|
||||
private static final String ERROR_OUTBOX_INCOMPLETE = "ACCOUNT_EVENTS_OUTBOX_INCOMPLETE";
|
||||
|
||||
static final long RETRY_BACKOFF_SECONDS = 60L;
|
||||
|
||||
private final AccountEventPublishOutboxMapper outboxMapper;
|
||||
private final EventsPublishApi eventsPublishApi;
|
||||
private final MuseAccountEventsProperties properties;
|
||||
|
||||
public AccountEventPublishWorker(AccountEventPublishOutboxMapper outboxMapper,
|
||||
EventsPublishApi eventsPublishApi,
|
||||
MuseAccountEventsProperties properties) {
|
||||
this.outboxMapper = outboxMapper;
|
||||
this.eventsPublishApi = eventsPublishApi;
|
||||
this.properties = properties;
|
||||
}
|
||||
|
||||
@Scheduled(initialDelayString = "${muse.account.events.publish-worker.initial-delay-ms:1000}",
|
||||
fixedDelayString = "${muse.account.events.publish-worker.fixed-delay-ms:1000}")
|
||||
public void dispatchScheduled() {
|
||||
dispatchOnce();
|
||||
}
|
||||
|
||||
public int dispatchOnce() {
|
||||
if (!isEnabled()) {
|
||||
return 0;
|
||||
}
|
||||
AccountEventPublishOutboxDO outbox = TenantUtils.executeIgnore(
|
||||
() -> outboxMapper.claimNextPublishOutbox(configuredClaimTimeoutSeconds()));
|
||||
if (outbox == null) {
|
||||
return 0;
|
||||
}
|
||||
if (outbox.getTenantId() == null || outbox.getId() == null) {
|
||||
markDeadLetter(outbox, ERROR_OUTBOX_INCOMPLETE, "Account Events publish outbox incomplete");
|
||||
return 0;
|
||||
}
|
||||
return TenantUtils.execute(outbox.getTenantId(), () -> publishClaimedOutbox(outbox));
|
||||
}
|
||||
|
||||
private int publishClaimedOutbox(AccountEventPublishOutboxDO outbox) {
|
||||
if (isRetryExhausted(outbox)) {
|
||||
markDeadLetter(outbox, ERROR_RETRY_EXHAUSTED, "Account Events publish retry exhausted");
|
||||
return 0;
|
||||
}
|
||||
BuildPublishRequestResult buildResult = buildReqDTO(outbox);
|
||||
if (!buildResult.success()) {
|
||||
markDeadLetter(outbox, buildResult.errorCode(), buildResult.errorMessage());
|
||||
return 0;
|
||||
}
|
||||
|
||||
CommonResult<EventsPublishRespDTO> result;
|
||||
try {
|
||||
result = eventsPublishApi.publish(buildResult.reqDTO());
|
||||
} catch (RuntimeException exception) {
|
||||
log.warn("Account Events publish 临时失败，tenantId={}, accountUserId={}, outboxId={}, auditLogId={}, attempt={}, errorType={}",
|
||||
outbox.getTenantId(), outbox.getAccountUserId(), outbox.getId(), outbox.getAuditLogId(),
|
||||
outbox.getAttemptCount(), exception.getClass().getSimpleName());
|
||||
if (isLastAttempt(outbox)) {
|
||||
markDeadLetter(outbox, ERROR_RETRY_EXHAUSTED, "Account Events publish retry exhausted");
|
||||
return 0;
|
||||
}
|
||||
markRetryable(outbox, ERROR_TEMPORARY_UNAVAILABLE, "Account Events publish temporary unavailable");
|
||||
return 0;
|
||||
}
|
||||
if (result == null || result.isError()) {
|
||||
if (isLastAttempt(outbox)) {
|
||||
markDeadLetter(outbox, ERROR_RETRY_EXHAUSTED, "Account Events publish retry exhausted");
|
||||
return 0;
|
||||
}
|
||||
markRetryable(outbox, ERROR_API_FAILED, "Account Events publish failed");
|
||||
return 0;
|
||||
}
|
||||
return handlePublishResponse(outbox, result.getData());
|
||||
}
|
||||
|
||||
private int handlePublishResponse(AccountEventPublishOutboxDO outbox, EventsPublishRespDTO respDTO) {
|
||||
if (respDTO == null || !StringUtils.hasText(respDTO.getPublishStatus())) {
|
||||
markDeadLetter(outbox, ERROR_UNKNOWN_STATUS, "Account Events publish unknown status");
|
||||
return 0;
|
||||
}
|
||||
String publishStatus = respDTO.getPublishStatus();
|
||||
if (PUBLISH_STATUS_ACCEPTED.equals(publishStatus)) {
|
||||
markPublished(outbox, respDTO);
|
||||
return 1;
|
||||
}
|
||||
if (PUBLISH_STATUS_REJECTED.equals(publishStatus) || PUBLISH_STATUS_BLOCKED.equals(publishStatus)) {
|
||||
markDeadLetter(outbox, safeErrorCode(respDTO.getPublishErrorCode()), "Account Events publish rejected");
|
||||
return 1;
|
||||
}
|
||||
markDeadLetter(outbox, ERROR_UNKNOWN_STATUS, "Account Events publish unknown status");
|
||||
return 0;
|
||||
}
|
||||
|
||||
private BuildPublishRequestResult buildReqDTO(AccountEventPublishOutboxDO outbox) {
|
||||
Map<String, Object> payload = parsePayload(outbox);
|
||||
if (!AccountEventPayloads.isValidQuotaAlertPayload(payload) || outbox.getCreateTime() == null
|
||||
|| outbox.getAccountUserId() == null || outbox.getAccountUserId() <= 0
|
||||
|| outbox.getAuditLogId() == null || outbox.getAuditLogId() <= 0
|
||||
|| !StringUtils.hasText(outbox.getCommandId())) {
|
||||
return BuildPublishRequestResult.failed(ERROR_PAYLOAD_INVALID, "Account Events publish payload invalid");
|
||||
}
|
||||
EventsPublishReqDTO reqDTO = new EventsPublishReqDTO();
|
||||
reqDTO.setCommandId(outbox.getCommandId());
|
||||
reqDTO.setTenantId(outbox.getTenantId());
|
||||
reqDTO.setOwnerUserId(outbox.getAccountUserId());
|
||||
reqDTO.setSourceOwner(SOURCE_OWNER_ACCOUNT);
|
||||
reqDTO.setSourceType(SOURCE_TYPE_ACCOUNT_QUOTA_ADJUSTMENT);
|
||||
reqDTO.setSourceId(String.valueOf(outbox.getAuditLogId()));
|
||||
reqDTO.setSourceRevision(StringUtils.hasText(outbox.getSourceRevision())
|
||||
? outbox.getSourceRevision() : SOURCE_REVISION_NONE);
|
||||
reqDTO.setEventType(EVENT_TYPE_NOTIFICATION);
|
||||
reqDTO.setResourceType(outbox.getResourceRefType());
|
||||
reqDTO.setResourceId(String.valueOf(outbox.getResourceRefId()));
|
||||
reqDTO.setPayloadSummary(payload);
|
||||
reqDTO.setEmittedAt(outbox.getCreateTime());
|
||||
return BuildPublishRequestResult.success(reqDTO);
|
||||
}
|
||||
|
||||
private Map<String, Object> parsePayload(AccountEventPublishOutboxDO outbox) {
|
||||
try {
|
||||
Map<String, Object> parsed = JsonUtils.getObjectMapper()
|
||||
.readValue(outbox.getPayloadSummary(), new TypeReference<Map<String, Object>>() {
|
||||
});
|
||||
return parsed == null ? Map.of() : parsed;
|
||||
} catch (Exception exception) {
|
||||
log.warn("Account Events publish payload 解析失败，tenantId={}, accountUserId={}, outboxId={}, auditLogId={}, errorType={}",
|
||||
outbox.getTenantId(), outbox.getAccountUserId(), outbox.getId(), outbox.getAuditLogId(),
|
||||
exception.getClass().getSimpleName());
|
||||
return Map.of();
|
||||
}
|
||||
}
|
||||
|
||||
private boolean isEnabled() {
|
||||
return properties != null
|
||||
&& properties.getPublishWorker() != null
|
||||
&& properties.getPublishWorker().isEnabled();
|
||||
}
|
||||
|
||||
private long configuredClaimTimeoutSeconds() {
|
||||
if (properties == null || properties.getPublishWorker() == null
|
||||
|| properties.getPublishWorker().getClaimTimeoutSeconds() <= 0) {
|
||||
return 60L;
|
||||
}
|
||||
return properties.getPublishWorker().getClaimTimeoutSeconds();
|
||||
}
|
||||
|
||||
private boolean isRetryExhausted(AccountEventPublishOutboxDO outbox) {
|
||||
return outbox.getAttemptCount() != null && outbox.getMaxAttempt() != null
|
||||
&& outbox.getAttemptCount() > outbox.getMaxAttempt();
|
||||
}
|
||||
|
||||
private boolean isLastAttempt(AccountEventPublishOutboxDO outbox) {
|
||||
return outbox.getAttemptCount() != null && outbox.getMaxAttempt() != null
|
||||
&& outbox.getAttemptCount() >= outbox.getMaxAttempt();
|
||||
}
|
||||
|
||||
private void markPublished(AccountEventPublishOutboxDO outbox, EventsPublishRespDTO respDTO) {
|
||||
int updated = outboxMapper.markPublished(outbox.getId(), outbox.getTenantId(), outbox.getAttemptCount(),
|
||||
respDTO.getEventId(), respDTO.getSequenceNo());
|
||||
logSkippedTerminalUpdate(outbox, updated, "published", null);
|
||||
}
|
||||
|
||||
private void markRetryable(AccountEventPublishOutboxDO outbox, String errorCode, String errorMessage) {
|
||||
int updated = outboxMapper.markRetryable(outbox.getId(), outbox.getTenantId(), outbox.getAttemptCount(),
|
||||
LocalDateTime.now().plusSeconds(RETRY_BACKOFF_SECONDS), errorCode, errorMessage);
|
||||
logSkippedTerminalUpdate(outbox, updated, "retryable", errorCode);
|
||||
}
|
||||
|
||||
private void markDeadLetter(AccountEventPublishOutboxDO outbox, String errorCode, String errorMessage) {
|
||||
if (outbox.getId() == null || outbox.getTenantId() == null) {
|
||||
log.error("Account Events publish outbox 无法标记 dead_letter，缺少 id 或 tenantId，errorCode={}", errorCode);
|
||||
return;
|
||||
}
|
||||
int updated = outboxMapper.markDeadLetter(outbox.getId(), outbox.getTenantId(), outbox.getAttemptCount(),
|
||||
errorCode, errorMessage);
|
||||
logSkippedTerminalUpdate(outbox, updated, "dead_letter", errorCode);
|
||||
}
|
||||
|
||||
private String safeErrorCode(String publishErrorCode) {
|
||||
return StringUtils.hasText(publishErrorCode) ? publishErrorCode : ERROR_UNKNOWN_STATUS;
|
||||
}
|
||||
|
||||
private void logSkippedTerminalUpdate(AccountEventPublishOutboxDO outbox, int updated, String targetStatus,
|
||||
String errorCode) {
|
||||
if (updated > 0) {
|
||||
return;
|
||||
}
|
||||
// 原因：0 行通常表示租约过期后被其他 worker 重新领取，必须留下可检索证据，避免重复发布排障断链。
|
||||
log.warn("Account Events publish 终态回写被跳过，tenantId={}, accountUserId={}, outboxId={}, " +
|
||||
"auditLogId={}, attempt={}, targetStatus={}, errorCode={}",
|
||||
outbox.getTenantId(), outbox.getAccountUserId(), outbox.getId(), outbox.getAuditLogId(),
|
||||
outbox.getAttemptCount(), targetStatus, errorCode);
|
||||
}
|
||||
|
||||
private record BuildPublishRequestResult(boolean success, EventsPublishReqDTO reqDTO,
|
||||
String errorCode, String errorMessage) {
|
||||
|
||||
static BuildPublishRequestResult success(EventsPublishReqDTO reqDTO) {
|
||||
return new BuildPublishRequestResult(true, reqDTO, null, null);
|
||||
}
|
||||
|
||||
static BuildPublishRequestResult failed(String errorCode, String errorMessage) {
|
||||
return new BuildPublishRequestResult(false, null, errorCode, errorMessage);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@ -89,6 +89,8 @@ public class AccountQuotaServiceImpl implements AccountQuotaService {
|
||||
private AccountCommandService commandService;
|
||||
@Resource
|
||||
private AccountAuditService auditService;
|
||||
@Resource
|
||||
private AccountEventPublishOutboxService accountEventPublishOutboxService;
|
||||
|
||||
@Override
|
||||
@Transactional(rollbackFor = Exception.class)
|
||||
@ -166,6 +168,9 @@ public class AccountQuotaServiceImpl implements AccountQuotaService {
|
||||
.auditEventId(accountAudit == null ? null : accountAudit.getId())
|
||||
.build();
|
||||
auditLogMapper.insert(log);
|
||||
// 原因：entitlement audit log 是 Account quota 调整的终态事实；outbox 必须在同一事务内创建，
|
||||
// 这样 quota / audit / publish 补偿状态不会出现跨事务断链。
|
||||
accountEventPublishOutboxService.createForQuotaAdjustment(log, reqVO.getCommandId(), requestHash);
|
||||
lastLog = log;
|
||||
}
|
||||
|
||||
|
||||
@ -0,0 +1,56 @@
|
||||
package cn.iocoder.muse.module.member.dal.dataobject.account;
|
||||
|
||||
import cn.iocoder.muse.framework.tenant.core.db.TenantBaseDO;
|
||||
import cn.iocoder.muse.module.member.dal.type.JsonbStringTypeHandler;
|
||||
import com.baomidou.mybatisplus.annotation.IdType;
|
||||
import com.baomidou.mybatisplus.annotation.TableField;
|
||||
import com.baomidou.mybatisplus.annotation.TableId;
|
||||
import com.baomidou.mybatisplus.annotation.TableName;
|
||||
import lombok.Data;
|
||||
import lombok.EqualsAndHashCode;
|
||||
import lombok.ToString;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
/**
|
||||
* Account quota adjustment 发布到统一 Events 的本域 outbox。
|
||||
*/
|
||||
@TableName(value = "muse_account_event_publish_outbox", autoResultMap = true)
|
||||
@Data
|
||||
@EqualsAndHashCode(callSuper = true)
|
||||
@ToString(callSuper = true)
|
||||
public class AccountEventPublishOutboxDO extends TenantBaseDO {
|
||||
|
||||
@TableId(type = IdType.AUTO)
|
||||
private Long id;
|
||||
|
||||
private String outboxId;
|
||||
private String commandId;
|
||||
private String sourceCommandId;
|
||||
private Long auditLogId;
|
||||
private Long accountUserId;
|
||||
private Long operatorUserId;
|
||||
private String resourceType;
|
||||
private Long deltaAmount;
|
||||
private String sourceOwner;
|
||||
private Long auditEventId;
|
||||
private String requestHash;
|
||||
private String sourceRevision;
|
||||
private String eventType;
|
||||
private String notificationType;
|
||||
private String resourceRefType;
|
||||
private Long resourceRefId;
|
||||
@TableField(typeHandler = JsonbStringTypeHandler.class)
|
||||
private String payloadSummary;
|
||||
private String publishStatus;
|
||||
private Integer attemptCount;
|
||||
private Integer maxAttempt;
|
||||
private LocalDateTime nextRetryAt;
|
||||
private LocalDateTime claimedAt;
|
||||
private LocalDateTime claimExpiresAt;
|
||||
private String lastErrorCode;
|
||||
private String lastErrorMessage;
|
||||
private String publishedEventId;
|
||||
private Long publishedSequenceNo;
|
||||
|
||||
}
|
||||
@ -0,0 +1,163 @@
|
||||
package cn.iocoder.muse.module.member.dal.mysql.account;
|
||||
|
||||
import cn.iocoder.muse.framework.mybatis.core.mapper.BaseMapperX;
|
||||
import cn.iocoder.muse.framework.mybatis.core.query.LambdaQueryWrapperX;
|
||||
import cn.iocoder.muse.framework.tenant.core.aop.TenantIgnore;
|
||||
import cn.iocoder.muse.module.member.dal.dataobject.account.AccountEventPublishOutboxDO;
|
||||
import org.apache.ibatis.annotations.Insert;
|
||||
import org.apache.ibatis.annotations.Mapper;
|
||||
import org.apache.ibatis.annotations.Options;
|
||||
import org.apache.ibatis.annotations.Param;
|
||||
import org.apache.ibatis.annotations.Select;
|
||||
import org.apache.ibatis.annotations.Update;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
/**
|
||||
* Account Events publish outbox Mapper。
|
||||
*/
|
||||
@Mapper
|
||||
public interface AccountEventPublishOutboxMapper extends BaseMapperX<AccountEventPublishOutboxDO> {
|
||||
|
||||
default AccountEventPublishOutboxDO selectByTenantIdAndOutboxId(Long tenantId, String outboxId) {
|
||||
return selectOne(new LambdaQueryWrapperX<AccountEventPublishOutboxDO>()
|
||||
.eq(AccountEventPublishOutboxDO::getTenantId, tenantId)
|
||||
.eq(AccountEventPublishOutboxDO::getOutboxId, outboxId));
|
||||
}
|
||||
|
||||
default AccountEventPublishOutboxDO selectByTenantIdAndCommandId(Long tenantId, String commandId) {
|
||||
return selectOne(new LambdaQueryWrapperX<AccountEventPublishOutboxDO>()
|
||||
.eq(AccountEventPublishOutboxDO::getTenantId, tenantId)
|
||||
.eq(AccountEventPublishOutboxDO::getCommandId, commandId));
|
||||
}
|
||||
|
||||
default AccountEventPublishOutboxDO selectByTenantIdAndAuditLogIdAndNotificationType(Long tenantId,
|
||||
Long auditLogId,
|
||||
String notificationType) {
|
||||
return selectOne(new LambdaQueryWrapperX<AccountEventPublishOutboxDO>()
|
||||
.eq(AccountEventPublishOutboxDO::getTenantId, tenantId)
|
||||
.eq(AccountEventPublishOutboxDO::getAuditLogId, auditLogId)
|
||||
.eq(AccountEventPublishOutboxDO::getNotificationType, notificationType));
|
||||
}
|
||||
|
||||
/**
|
||||
* 首次创建 Account Events publish outbox；重复 command 或 audit log notification 时返回 0。
|
||||
*/
|
||||
@Insert("""
|
||||
INSERT INTO muse_account_event_publish_outbox(tenant_id, outbox_id, command_id, source_command_id,
|
||||
audit_log_id, account_user_id, operator_user_id,
|
||||
resource_type, delta_amount, source_owner, audit_event_id,
|
||||
request_hash, source_revision, event_type, notification_type,
|
||||
resource_ref_type, resource_ref_id, payload_summary,
|
||||
publish_status, attempt_count, max_attempt, next_retry_at,
|
||||
claimed_at, claim_expires_at, last_error_code,
|
||||
last_error_message, published_event_id, published_sequence_no,
|
||||
create_time)
|
||||
VALUES (#{tenantId}, #{outboxId}, #{commandId}, #{sourceCommandId}, #{auditLogId}, #{accountUserId},
|
||||
#{operatorUserId}, #{resourceType}, #{deltaAmount}, #{sourceOwner}, #{auditEventId},
|
||||
#{requestHash}, #{sourceRevision}, #{eventType}, #{notificationType}, #{resourceRefType},
|
||||
#{resourceRefId},
|
||||
#{payloadSummary,typeHandler=cn.iocoder.muse.module.member.dal.type.JsonbStringTypeHandler},
|
||||
#{publishStatus}, #{attemptCount}, #{maxAttempt}, #{nextRetryAt}, #{claimedAt},
|
||||
#{claimExpiresAt}, #{lastErrorCode}, #{lastErrorMessage}, #{publishedEventId},
|
||||
#{publishedSequenceNo}, #{createTime})
|
||||
ON CONFLICT DO NOTHING
|
||||
""")
|
||||
int insertIgnore(AccountEventPublishOutboxDO outbox);
|
||||
|
||||
/**
|
||||
* 原子领取下一条待发布 Account Events outbox。
|
||||
*/
|
||||
@TenantIgnore
|
||||
@Options(flushCache = Options.FlushCachePolicy.TRUE, useCache = false)
|
||||
@Select("""
|
||||
UPDATE muse_account_event_publish_outbox
|
||||
SET publish_status = 'running',
|
||||
claimed_at = CURRENT_TIMESTAMP,
|
||||
claim_expires_at = CURRENT_TIMESTAMP + (#{leaseSeconds} * INTERVAL '1 second'),
|
||||
attempt_count = attempt_count + 1,
|
||||
update_time = CURRENT_TIMESTAMP
|
||||
WHERE id = (
|
||||
SELECT id
|
||||
FROM muse_account_event_publish_outbox
|
||||
WHERE deleted = FALSE
|
||||
AND (
|
||||
publish_status = 'queued'
|
||||
OR (publish_status = 'retryable'
|
||||
AND (next_retry_at IS NULL OR next_retry_at <= CURRENT_TIMESTAMP))
|
||||
OR (publish_status = 'running'
|
||||
AND claim_expires_at IS NOT NULL
|
||||
AND claim_expires_at <= CURRENT_TIMESTAMP)
|
||||
)
|
||||
ORDER BY create_time ASC, id ASC
|
||||
FOR UPDATE SKIP LOCKED
|
||||
LIMIT 1
|
||||
)
|
||||
RETURNING *
|
||||
""")
|
||||
AccountEventPublishOutboxDO claimNextPublishOutbox(long leaseSeconds);
|
||||
|
||||
@Update("""
|
||||
UPDATE muse_account_event_publish_outbox
|
||||
SET publish_status = 'published',
|
||||
claimed_at = NULL,
|
||||
claim_expires_at = NULL,
|
||||
next_retry_at = NULL,
|
||||
last_error_code = NULL,
|
||||
last_error_message = NULL,
|
||||
published_event_id = #{publishedEventId},
|
||||
published_sequence_no = #{publishedSequenceNo},
|
||||
update_time = CURRENT_TIMESTAMP
|
||||
WHERE id = #{id}
|
||||
AND tenant_id = #{tenantId}
|
||||
AND deleted = FALSE
|
||||
AND publish_status = 'running'
|
||||
AND attempt_count = #{claimedAttemptCount}
|
||||
""")
|
||||
int markPublished(@Param("id") Long id, @Param("tenantId") Long tenantId,
|
||||
@Param("claimedAttemptCount") Integer claimedAttemptCount,
|
||||
@Param("publishedEventId") String publishedEventId,
|
||||
@Param("publishedSequenceNo") Long publishedSequenceNo);
|
||||
|
||||
@Update("""
|
||||
UPDATE muse_account_event_publish_outbox
|
||||
SET publish_status = 'retryable',
|
||||
claimed_at = NULL,
|
||||
claim_expires_at = NULL,
|
||||
next_retry_at = #{nextRetryAt},
|
||||
last_error_code = #{lastErrorCode},
|
||||
last_error_message = #{lastErrorMessage},
|
||||
update_time = CURRENT_TIMESTAMP
|
||||
WHERE id = #{id}
|
||||
AND tenant_id = #{tenantId}
|
||||
AND deleted = FALSE
|
||||
AND publish_status = 'running'
|
||||
AND attempt_count = #{claimedAttemptCount}
|
||||
""")
|
||||
int markRetryable(@Param("id") Long id, @Param("tenantId") Long tenantId,
|
||||
@Param("claimedAttemptCount") Integer claimedAttemptCount,
|
||||
@Param("nextRetryAt") LocalDateTime nextRetryAt,
|
||||
@Param("lastErrorCode") String lastErrorCode,
|
||||
@Param("lastErrorMessage") String lastErrorMessage);
|
||||
|
||||
@Update("""
|
||||
UPDATE muse_account_event_publish_outbox
|
||||
SET publish_status = 'dead_letter',
|
||||
claimed_at = NULL,
|
||||
claim_expires_at = NULL,
|
||||
next_retry_at = NULL,
|
||||
last_error_code = #{lastErrorCode},
|
||||
last_error_message = #{lastErrorMessage},
|
||||
update_time = CURRENT_TIMESTAMP
|
||||
WHERE id = #{id}
|
||||
AND tenant_id = #{tenantId}
|
||||
AND deleted = FALSE
|
||||
AND publish_status = 'running'
|
||||
AND attempt_count = #{claimedAttemptCount}
|
||||
""")
|
||||
int markDeadLetter(@Param("id") Long id, @Param("tenantId") Long tenantId,
|
||||
@Param("claimedAttemptCount") Integer claimedAttemptCount,
|
||||
@Param("lastErrorCode") String lastErrorCode,
|
||||
@Param("lastErrorMessage") String lastErrorMessage);
|
||||
|
||||
}
|
||||
@ -0,0 +1,12 @@
|
||||
package cn.iocoder.muse.module.member.framework.config;
|
||||
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
/**
|
||||
* Account Events 发布配置装配。
|
||||
*/
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
@EnableConfigurationProperties(MuseAccountEventsProperties.class)
|
||||
public class MuseAccountEventsConfiguration {
|
||||
}
|
||||
@ -0,0 +1,38 @@
|
||||
package cn.iocoder.muse.module.member.framework.config;
|
||||
|
||||
import lombok.Data;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
|
||||
/**
|
||||
* Account 到统一 Events 的本域发布配置。
|
||||
*/
|
||||
@ConfigurationProperties(prefix = "muse.account.events")
|
||||
@Data
|
||||
public class MuseAccountEventsProperties {
|
||||
|
||||
/**
|
||||
* Account quota adjustment 到统一 Events 的本域 worker 配置。
|
||||
*/
|
||||
private PublishWorker publishWorker = new PublishWorker();
|
||||
|
||||
@Data
|
||||
public static class PublishWorker {
|
||||
|
||||
/**
|
||||
* 是否启用 Account Events publish worker；默认关闭，避免未验收链路自动 claim。
|
||||
*/
|
||||
private boolean enabled;
|
||||
|
||||
/**
|
||||
* 单条 outbox 最大发布尝试次数；默认 5 次，便于运行时回滚和调优。
|
||||
*/
|
||||
private int maxAttempt = 5;
|
||||
|
||||
/**
|
||||
* 单次 claim 租约秒数；默认 60 秒，防止 worker 异常退出后长期占用任务。
|
||||
*/
|
||||
private long claimTimeoutSeconds = 60L;
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,232 @@
|
||||
package cn.iocoder.muse.module.member.application.account;
|
||||
|
||||
import cn.iocoder.muse.framework.common.exception.ServiceException;
|
||||
import cn.iocoder.muse.framework.common.util.json.JsonUtils;
|
||||
import cn.iocoder.muse.framework.test.core.ut.BaseMockitoUnitTest;
|
||||
import cn.iocoder.muse.module.events.api.publish.EventsPublishApi;
|
||||
import cn.iocoder.muse.module.member.dal.dataobject.account.AccountEventPublishOutboxDO;
|
||||
import cn.iocoder.muse.module.member.dal.dataobject.account.MemberEntitlementAuditLogDO;
|
||||
import cn.iocoder.muse.module.member.dal.mysql.account.AccountEventPublishOutboxMapper;
|
||||
import cn.iocoder.muse.module.member.framework.config.MuseAccountEventsProperties;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.mockito.Mock;
|
||||
|
||||
import java.lang.reflect.Field;
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
/**
|
||||
* Account quota adjustment Events publish outbox 创建服务测试。
|
||||
*/
|
||||
class AccountEventPublishOutboxServiceTest extends BaseMockitoUnitTest {
|
||||
|
||||
private static final Pattern SHORT_COMMAND_ID = Pattern.compile("acct_evt:[0-9a-f]{32}");
|
||||
|
||||
private AccountEventPublishOutboxServiceImpl outboxService;
|
||||
@Mock
|
||||
private AccountEventPublishOutboxMapper outboxMapper;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
outboxService = outboxServiceWithMaxAttempt(5);
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_createQueuedOutboxWithExactAllowlistAndNumericResourceRef() {
|
||||
MemberEntitlementAuditLogDO log = entitlementLog(8001L, 1001L, "quota_adjustment");
|
||||
when(outboxMapper.insertIgnore(any(AccountEventPublishOutboxDO.class))).thenReturn(1);
|
||||
when(outboxMapper.selectByTenantIdAndOutboxId(any(), any())).thenAnswer(invocation -> {
|
||||
AccountEventPublishOutboxDO inserted = captureInsertedOutbox();
|
||||
inserted.setId(501L);
|
||||
return inserted;
|
||||
});
|
||||
|
||||
AccountEventPublishOutboxDO result =
|
||||
outboxService.createForQuotaAdjustment(log, "cmd-quota:with:colon", "request-hash-secret");
|
||||
|
||||
AccountEventPublishOutboxDO inserted = captureInsertedOutbox();
|
||||
assertEquals(501L, result.getId());
|
||||
assertEquals(100L, inserted.getTenantId());
|
||||
assertEquals("acct_out:be07290313564a1ff50a804c0da78348", inserted.getOutboxId());
|
||||
assertEquals("acct_evt:be07290313564a1ff50a804c0da78348", inserted.getCommandId());
|
||||
assertEquals("cmd-quota:with:colon", inserted.getSourceCommandId());
|
||||
assertEquals(8001L, inserted.getAuditLogId());
|
||||
assertEquals(1001L, inserted.getAccountUserId());
|
||||
assertEquals(9001L, inserted.getOperatorUserId());
|
||||
assertEquals("ai_calls", inserted.getResourceType());
|
||||
assertEquals(25L, inserted.getDeltaAmount());
|
||||
assertEquals("manual", inserted.getSourceOwner());
|
||||
assertEquals(7001L, inserted.getAuditEventId());
|
||||
assertEquals("request-hash-secret", inserted.getRequestHash());
|
||||
assertEquals("__none__", inserted.getSourceRevision());
|
||||
assertEquals("notification", inserted.getEventType());
|
||||
assertEquals("quota_alert", inserted.getNotificationType());
|
||||
assertEquals("account_quota", inserted.getResourceRefType());
|
||||
assertEquals(8001L, inserted.getResourceRefId());
|
||||
assertEquals("queued", inserted.getPublishStatus());
|
||||
assertEquals(0, inserted.getAttemptCount());
|
||||
assertEquals(5, inserted.getMaxAttempt());
|
||||
assertTrue(SHORT_COMMAND_ID.matcher(inserted.getCommandId()).matches());
|
||||
|
||||
Map<String, Object> payload = JsonUtils.parseObject(inserted.getPayloadSummary(), Map.class);
|
||||
assertEquals(Set.of("type", "message", "resourceRef", "timestamp"), payload.keySet());
|
||||
assertEquals("quota_alert", payload.get("type"));
|
||||
assertEquals("额度已调整", payload.get("message"));
|
||||
assertEquals("2026-06-07T12:00", payload.get("timestamp"));
|
||||
Map<?, ?> resourceRef = assertInstanceOf(Map.class, payload.get("resourceRef"));
|
||||
assertEquals(Set.of("resourceType", "resourceId"), resourceRef.keySet());
|
||||
assertEquals("account_quota", resourceRef.get("resourceType"));
|
||||
assertInstanceOf(Number.class, resourceRef.get("resourceId"));
|
||||
assertEquals(8001L, ((Number) resourceRef.get("resourceId")).longValue());
|
||||
assertNoForbiddenPayloadFields(inserted.getPayloadSummary());
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_useLastColonFallbackOnlyWhenSourceCommandIdMissing() {
|
||||
MemberEntitlementAuditLogDO log = entitlementLog(8001L, 1001L, "quota_adjustment");
|
||||
log.setIdempotencyKey("cmd-quota:with:colon:ai_calls");
|
||||
when(outboxMapper.insertIgnore(any(AccountEventPublishOutboxDO.class))).thenReturn(1);
|
||||
when(outboxMapper.selectByTenantIdAndOutboxId(any(), any())).thenAnswer(invocation -> captureInsertedOutbox());
|
||||
|
||||
outboxService.createForQuotaAdjustment(log, null, "request-hash-secret");
|
||||
|
||||
assertEquals("cmd-quota:with:colon", captureInsertedOutbox().getSourceCommandId(),
|
||||
"历史回查 fallback 必须使用 lastIndexOf(':')，不能用第一个冒号截断 commandId");
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_writeConfiguredMaxAttemptIntoOutbox() {
|
||||
outboxService = outboxServiceWithMaxAttempt(9);
|
||||
MemberEntitlementAuditLogDO log = entitlementLog(8001L, 1001L, "quota_adjustment");
|
||||
when(outboxMapper.insertIgnore(any(AccountEventPublishOutboxDO.class))).thenReturn(1);
|
||||
when(outboxMapper.selectByTenantIdAndOutboxId(any(), any())).thenAnswer(invocation -> captureInsertedOutbox());
|
||||
|
||||
outboxService.createForQuotaAdjustment(log, "cmd-quota-1", "request-hash-secret");
|
||||
|
||||
assertEquals(9, captureInsertedOutbox().getMaxAttempt(),
|
||||
"outbox maxAttempt 必须来自 muse.account.events.publish-worker.max-attempt");
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_fallbackTimestampToOutboxCreateTimeWhenAuditLogCreateTimeMissing() {
|
||||
MemberEntitlementAuditLogDO log = entitlementLog(8001L, 1001L, "quota_adjustment");
|
||||
log.setCreateTime(null);
|
||||
when(outboxMapper.insertIgnore(any(AccountEventPublishOutboxDO.class))).thenReturn(1);
|
||||
when(outboxMapper.selectByTenantIdAndOutboxId(any(), any())).thenAnswer(invocation -> captureInsertedOutbox());
|
||||
|
||||
outboxService.createForQuotaAdjustment(log, "cmd-quota-1", "request-hash-secret");
|
||||
|
||||
AccountEventPublishOutboxDO inserted = captureInsertedOutbox();
|
||||
Map<String, Object> payload = JsonUtils.parseObject(inserted.getPayloadSummary(), Map.class);
|
||||
assertEquals(inserted.getCreateTime().toString(), payload.get("timestamp"),
|
||||
"audit log createTime 缺失时，payload timestamp 必须回退到 outbox createTime");
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_reuseExistingOutboxOnDuplicateCreate() {
|
||||
MemberEntitlementAuditLogDO log = entitlementLog(8001L, 1001L, "quota_adjustment");
|
||||
AccountEventPublishOutboxDO existing = new AccountEventPublishOutboxDO();
|
||||
existing.setId(601L);
|
||||
existing.setAuditLogId(8001L);
|
||||
existing.setNotificationType("quota_alert");
|
||||
when(outboxMapper.insertIgnore(any(AccountEventPublishOutboxDO.class))).thenReturn(0);
|
||||
when(outboxMapper.selectByTenantIdAndAuditLogIdAndNotificationType(100L, 8001L,
|
||||
"quota_alert")).thenReturn(existing);
|
||||
|
||||
AccountEventPublishOutboxDO result =
|
||||
outboxService.createForQuotaAdjustment(log, "cmd-quota-1", "request-hash-secret");
|
||||
|
||||
assertEquals(601L, result.getId());
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_notCreateOutboxForUnsupportedChangeTypeOrInvalidOwnerOrMissingAuditId() {
|
||||
assertNull(outboxService.createForQuotaAdjustment(entitlementLog(8001L, 1001L, "subscription_sync"),
|
||||
"cmd-quota-1", "hash"));
|
||||
assertNull(outboxService.createForQuotaAdjustment(entitlementLog(8001L, 0L, "quota_adjustment"),
|
||||
"cmd-quota-1", "hash"));
|
||||
assertNull(outboxService.createForQuotaAdjustment(entitlementLog(null, 1001L, "quota_adjustment"),
|
||||
"cmd-quota-1", "hash"));
|
||||
|
||||
verify(outboxMapper, never()).insertIgnore(any(AccountEventPublishOutboxDO.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_failClosedBeforeInsertWhenPayloadInvalid() {
|
||||
MemberEntitlementAuditLogDO log = entitlementLog(8001L, 1001L, "quota_adjustment");
|
||||
log.setDeltaValueSnapshot("{}");
|
||||
|
||||
ServiceException exception = assertThrows(ServiceException.class,
|
||||
() -> outboxService.createForQuotaAdjustment(log, "cmd-quota-1", "request-hash-secret"));
|
||||
|
||||
assertEquals(1_042_000_050, exception.getCode());
|
||||
assertEquals("ACCOUNT_EVENTS_PAYLOAD_INVALID", exception.getMessage());
|
||||
verify(outboxMapper, never()).insertIgnore(any(AccountEventPublishOutboxDO.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_notDependOnEventsPublishApiInOutboxCreationService() {
|
||||
for (Field field : AccountEventPublishOutboxServiceImpl.class.getDeclaredFields()) {
|
||||
assertFalse(EventsPublishApi.class.equals(field.getType()),
|
||||
"outbox 创建服务只允许写 Account 本域 outbox，不允许持有或调用 EventsPublishApi");
|
||||
}
|
||||
}
|
||||
|
||||
private AccountEventPublishOutboxDO captureInsertedOutbox() {
|
||||
ArgumentCaptor<AccountEventPublishOutboxDO> captor =
|
||||
ArgumentCaptor.forClass(AccountEventPublishOutboxDO.class);
|
||||
verify(outboxMapper).insertIgnore(captor.capture());
|
||||
return captor.getValue();
|
||||
}
|
||||
|
||||
private AccountEventPublishOutboxServiceImpl outboxServiceWithMaxAttempt(int maxAttempt) {
|
||||
MuseAccountEventsProperties properties = new MuseAccountEventsProperties();
|
||||
properties.getPublishWorker().setMaxAttempt(maxAttempt);
|
||||
return new AccountEventPublishOutboxServiceImpl(outboxMapper, properties);
|
||||
}
|
||||
|
||||
private static void assertNoForbiddenPayloadFields(String payloadJson) {
|
||||
for (String forbidden : Set.of("requestHash", "reason", "operatorUserId", "beforeValueSnapshot",
|
||||
"deltaValueSnapshot", "afterValueSnapshot", "correlationId", "external", "errorStack",
|
||||
"credential")) {
|
||||
assertFalse(payloadJson.contains(forbidden), "payload 不得包含敏感字段：" + forbidden);
|
||||
}
|
||||
}
|
||||
|
||||
private static MemberEntitlementAuditLogDO entitlementLog(Long id, Long accountUserId, String changeType) {
|
||||
MemberEntitlementAuditLogDO log = MemberEntitlementAuditLogDO.builder()
|
||||
.id(id)
|
||||
.accountUserId(accountUserId)
|
||||
.changeType(changeType)
|
||||
.sourceOwner("manual")
|
||||
.sourceId(7001L)
|
||||
.idempotencyKey("cmd-quota-1:ai_calls")
|
||||
.beforeValueSnapshot("{\"resourceType\":\"ai_calls\",\"limit\":100,\"used\":40,\"remaining\":60}")
|
||||
.deltaValueSnapshot("{\"resourceType\":\"ai_calls\",\"delta\":25}")
|
||||
.afterValueSnapshot("{\"resourceType\":\"ai_calls\",\"limit\":125,\"used\":40,\"remaining\":85}")
|
||||
.reasonCode("approval-1")
|
||||
.reasonMessage("内部原因不得外发")
|
||||
.operatorUserId(9001L)
|
||||
.auditEventId(7001L)
|
||||
.build();
|
||||
log.setTenantId(100L);
|
||||
log.setCreateTime(LocalDateTime.of(2026, 6, 7, 12, 0));
|
||||
return log;
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,365 @@
|
||||
package cn.iocoder.muse.module.member.application.account;
|
||||
|
||||
import cn.iocoder.muse.framework.common.pojo.CommonResult;
|
||||
import cn.iocoder.muse.framework.common.util.json.JsonUtils;
|
||||
import cn.iocoder.muse.framework.tenant.core.context.TenantContextHolder;
|
||||
import cn.iocoder.muse.module.events.api.publish.EventsPublishApi;
|
||||
import cn.iocoder.muse.module.events.api.publish.dto.EventsPublishReqDTO;
|
||||
import cn.iocoder.muse.module.events.api.publish.dto.EventsPublishRespDTO;
|
||||
import cn.iocoder.muse.module.member.dal.dataobject.account.AccountEventPublishOutboxDO;
|
||||
import cn.iocoder.muse.module.member.dal.mysql.account.AccountEventPublishOutboxMapper;
|
||||
import cn.iocoder.muse.module.member.framework.config.MuseAccountEventsProperties;
|
||||
import ch.qos.logback.classic.Level;
|
||||
import ch.qos.logback.classic.Logger;
|
||||
import ch.qos.logback.classic.spi.ILoggingEvent;
|
||||
import ch.qos.logback.core.read.ListAppender;
|
||||
import org.apache.ibatis.annotations.Update;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.lang.reflect.Method;
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.anyLong;
|
||||
import static org.mockito.ArgumentMatchers.anyString;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
/**
|
||||
* Account Events publish worker 测试。
|
||||
*/
|
||||
class AccountEventPublishWorkerTest {
|
||||
|
||||
private final AccountEventPublishOutboxMapper outboxMapper =
|
||||
mock(AccountEventPublishOutboxMapper.class);
|
||||
private final EventsPublishApi eventsPublishApi = mock(EventsPublishApi.class);
|
||||
|
||||
@AfterEach
|
||||
void clearTenantContext() {
|
||||
TenantContextHolder.clear();
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_notClaimOutboxWhenWorkerDisabled() {
|
||||
AccountEventPublishWorker worker = worker(false);
|
||||
|
||||
int dispatched = worker.dispatchOnce();
|
||||
|
||||
assertEquals(0, dispatched);
|
||||
verify(outboxMapper, never()).claimNextPublishOutbox(anyLong());
|
||||
verify(eventsPublishApi, never()).publish(any());
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_publishAcceptedQuotaAlertForAccountOwnerAndMarkPublished() {
|
||||
AccountEventPublishOutboxDO outbox = runningOutbox();
|
||||
when(outboxMapper.claimNextPublishOutbox(60L)).thenReturn(outbox);
|
||||
EventsPublishRespDTO respDTO = accepted("evt_account_1", 401L);
|
||||
when(eventsPublishApi.publish(any())).thenReturn(CommonResult.success(respDTO));
|
||||
when(outboxMapper.markPublished(501L, 100L, 1, "evt_account_1", 401L)).thenReturn(1);
|
||||
|
||||
int dispatched = worker(true).dispatchOnce();
|
||||
|
||||
assertEquals(1, dispatched);
|
||||
EventsPublishReqDTO reqDTO = capturePublishReq();
|
||||
assertEquals("acct_evt:0123456789abcdef0123456789abcdef", reqDTO.getCommandId());
|
||||
assertEquals(100L, reqDTO.getTenantId());
|
||||
assertEquals(1001L, reqDTO.getOwnerUserId());
|
||||
assertEquals("account", reqDTO.getSourceOwner());
|
||||
assertEquals("account_quota_adjustment", reqDTO.getSourceType());
|
||||
assertEquals("8001", reqDTO.getSourceId());
|
||||
assertEquals("__none__", reqDTO.getSourceRevision());
|
||||
assertEquals("notification", reqDTO.getEventType());
|
||||
assertEquals("account_quota", reqDTO.getResourceType());
|
||||
assertEquals("8001", reqDTO.getResourceId());
|
||||
assertEquals(LocalDateTime.of(2026, 6, 7, 12, 0), reqDTO.getEmittedAt());
|
||||
assertPayload(reqDTO.getPayloadSummary());
|
||||
verify(outboxMapper).markPublished(501L, 100L, 1, "evt_account_1", 401L);
|
||||
assertNull(TenantContextHolder.getTenantId(), "worker 完成后必须清理租户上下文");
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_claimWithConfiguredTimeoutSeconds() {
|
||||
AccountEventPublishOutboxDO outbox = runningOutbox();
|
||||
when(outboxMapper.claimNextPublishOutbox(180L)).thenReturn(outbox);
|
||||
EventsPublishRespDTO respDTO = accepted("evt_account_claim_timeout", 404L);
|
||||
when(eventsPublishApi.publish(any())).thenReturn(CommonResult.success(respDTO));
|
||||
when(outboxMapper.markPublished(501L, 100L, 1, "evt_account_claim_timeout", 404L)).thenReturn(1);
|
||||
MuseAccountEventsProperties properties = new MuseAccountEventsProperties();
|
||||
properties.getPublishWorker().setEnabled(true);
|
||||
properties.getPublishWorker().setClaimTimeoutSeconds(180L);
|
||||
|
||||
int dispatched = new AccountEventPublishWorker(outboxMapper, eventsPublishApi, properties).dispatchOnce();
|
||||
|
||||
assertEquals(1, dispatched);
|
||||
verify(outboxMapper).claimNextPublishOutbox(180L);
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_markDuplicateAcceptedNotificationAsPublished() {
|
||||
AccountEventPublishOutboxDO outbox = runningOutbox();
|
||||
when(outboxMapper.claimNextPublishOutbox(60L)).thenReturn(outbox);
|
||||
EventsPublishRespDTO respDTO = accepted("evt_account_dup", 402L);
|
||||
respDTO.setDuplicate(true);
|
||||
when(eventsPublishApi.publish(any())).thenReturn(CommonResult.success(respDTO));
|
||||
when(outboxMapper.markPublished(501L, 100L, 1, "evt_account_dup", 402L)).thenReturn(1);
|
||||
|
||||
int dispatched = worker(true).dispatchOnce();
|
||||
|
||||
assertEquals(1, dispatched);
|
||||
verify(outboxMapper).markPublished(501L, 100L, 1, "evt_account_dup", 402L);
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_markRejectedAndBlockedPublishAsDeadLetter() {
|
||||
assertDeadLetterForTerminalStatus("rejected");
|
||||
assertDeadLetterForTerminalStatus("blocked");
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_markCommonResultErrorAsRetryableBeforeLastAttemptAndDeadLetterOnLastAttempt() {
|
||||
AccountEventPublishOutboxDO retryable = runningOutbox();
|
||||
retryable.setAttemptCount(2);
|
||||
retryable.setMaxAttempt(5);
|
||||
when(outboxMapper.claimNextPublishOutbox(60L)).thenReturn(retryable);
|
||||
when(eventsPublishApi.publish(any())).thenReturn(CommonResult.error(500, "temporary secret stack"));
|
||||
when(outboxMapper.markRetryable(anyLong(), anyLong(), org.mockito.Mockito.anyInt(),
|
||||
any(LocalDateTime.class), anyString(), anyString())).thenReturn(1);
|
||||
|
||||
assertEquals(0, worker(true).dispatchOnce());
|
||||
verify(outboxMapper).markRetryable(org.mockito.Mockito.eq(501L), org.mockito.Mockito.eq(100L),
|
||||
org.mockito.Mockito.eq(2), any(LocalDateTime.class),
|
||||
org.mockito.Mockito.eq("ACCOUNT_EVENTS_PUBLISH_API_FAILED"),
|
||||
org.mockito.Mockito.eq("Account Events publish failed"));
|
||||
|
||||
AccountEventPublishOutboxMapper lastAttemptMapper = mock(AccountEventPublishOutboxMapper.class);
|
||||
EventsPublishApi lastAttemptApi = mock(EventsPublishApi.class);
|
||||
AccountEventPublishOutboxDO lastAttempt = runningOutbox();
|
||||
lastAttempt.setAttemptCount(5);
|
||||
lastAttempt.setMaxAttempt(5);
|
||||
when(lastAttemptMapper.claimNextPublishOutbox(60L)).thenReturn(lastAttempt);
|
||||
when(lastAttemptApi.publish(any())).thenReturn(CommonResult.error(500, "temporary secret stack"));
|
||||
when(lastAttemptMapper.markDeadLetter(anyLong(), anyLong(), org.mockito.Mockito.anyInt(),
|
||||
anyString(), anyString())).thenReturn(1);
|
||||
|
||||
assertEquals(0, worker(true, lastAttemptMapper, lastAttemptApi).dispatchOnce());
|
||||
verify(lastAttemptMapper).markDeadLetter(501L, 100L, 5,
|
||||
"ACCOUNT_EVENTS_PUBLISH_RETRY_EXHAUSTED", "Account Events publish retry exhausted");
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_markExceptionAsRetryableBeforeLastAttemptAndDeadLetterOnLastAttempt() {
|
||||
AccountEventPublishOutboxDO retryable = runningOutbox();
|
||||
retryable.setAttemptCount(2);
|
||||
retryable.setMaxAttempt(5);
|
||||
when(outboxMapper.claimNextPublishOutbox(60L)).thenReturn(retryable);
|
||||
when(eventsPublishApi.publish(any())).thenThrow(new IllegalStateException("secret stack"));
|
||||
when(outboxMapper.markRetryable(anyLong(), anyLong(), org.mockito.Mockito.anyInt(),
|
||||
any(LocalDateTime.class), anyString(), anyString())).thenReturn(1);
|
||||
|
||||
assertEquals(0, worker(true).dispatchOnce());
|
||||
verify(outboxMapper).markRetryable(org.mockito.Mockito.eq(501L), org.mockito.Mockito.eq(100L),
|
||||
org.mockito.Mockito.eq(2), any(LocalDateTime.class),
|
||||
org.mockito.Mockito.eq("ACCOUNT_EVENTS_PUBLISH_TEMPORARY_UNAVAILABLE"),
|
||||
org.mockito.Mockito.eq("Account Events publish temporary unavailable"));
|
||||
|
||||
AccountEventPublishOutboxMapper lastAttemptMapper = mock(AccountEventPublishOutboxMapper.class);
|
||||
EventsPublishApi lastAttemptApi = mock(EventsPublishApi.class);
|
||||
AccountEventPublishOutboxDO lastAttempt = runningOutbox();
|
||||
lastAttempt.setAttemptCount(5);
|
||||
lastAttempt.setMaxAttempt(5);
|
||||
when(lastAttemptMapper.claimNextPublishOutbox(60L)).thenReturn(lastAttempt);
|
||||
when(lastAttemptApi.publish(any())).thenThrow(new IllegalStateException("secret stack"));
|
||||
when(lastAttemptMapper.markDeadLetter(anyLong(), anyLong(), org.mockito.Mockito.anyInt(),
|
||||
anyString(), anyString())).thenReturn(1);
|
||||
|
||||
assertEquals(0, worker(true, lastAttemptMapper, lastAttemptApi).dispatchOnce());
|
||||
verify(lastAttemptMapper).markDeadLetter(501L, 100L, 5,
|
||||
"ACCOUNT_EVENTS_PUBLISH_RETRY_EXHAUSTED", "Account Events publish retry exhausted");
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_failClosedForInvalidPayloadAndNotPublish() {
|
||||
AccountEventPublishOutboxDO outbox = runningOutbox();
|
||||
outbox.setPayloadSummary(JsonUtils.toJsonString(Map.of(
|
||||
"type", "quota_alert",
|
||||
"message", "额度已调整",
|
||||
"resourceRef", Map.of("resourceType", "account_quota", "resourceId", "8001"),
|
||||
"timestamp", "2026-06-07T10:00:00")));
|
||||
when(outboxMapper.claimNextPublishOutbox(60L)).thenReturn(outbox);
|
||||
when(outboxMapper.markDeadLetter(anyLong(), anyLong(), org.mockito.Mockito.anyInt(),
|
||||
anyString(), anyString())).thenReturn(1);
|
||||
|
||||
int dispatched = worker(true).dispatchOnce();
|
||||
|
||||
assertEquals(0, dispatched);
|
||||
verify(eventsPublishApi, never()).publish(any());
|
||||
verify(outboxMapper).markDeadLetter(501L, 100L, 1,
|
||||
"ACCOUNT_EVENTS_PAYLOAD_INVALID", "Account Events publish payload invalid");
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_failClosedForInvalidOwnerAndNotPublish() {
|
||||
AccountEventPublishOutboxDO outbox = runningOutbox();
|
||||
outbox.setAccountUserId(0L);
|
||||
when(outboxMapper.claimNextPublishOutbox(60L)).thenReturn(outbox);
|
||||
when(outboxMapper.markDeadLetter(anyLong(), anyLong(), org.mockito.Mockito.anyInt(),
|
||||
anyString(), anyString())).thenReturn(1);
|
||||
|
||||
int dispatched = worker(true).dispatchOnce();
|
||||
|
||||
assertEquals(0, dispatched);
|
||||
verify(eventsPublishApi, never()).publish(any());
|
||||
verify(outboxMapper).markDeadLetter(501L, 100L, 1,
|
||||
"ACCOUNT_EVENTS_PAYLOAD_INVALID", "Account Events publish payload invalid");
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_guardTerminalStatusUpdatesWithClaimAttemptOwnership() throws NoSuchMethodException {
|
||||
assertClaimOwnershipGuard("markPublished", Long.class, Long.class, Integer.class, String.class, Long.class);
|
||||
assertClaimOwnershipGuard("markRetryable", Long.class, Long.class, Integer.class, LocalDateTime.class,
|
||||
String.class, String.class);
|
||||
assertClaimOwnershipGuard("markDeadLetter", Long.class, Long.class, Integer.class, String.class,
|
||||
String.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_logWhenAcceptedTerminalUpdateIsSkippedByStaleClaim() {
|
||||
AccountEventPublishOutboxDO outbox = runningOutbox();
|
||||
when(outboxMapper.claimNextPublishOutbox(60L)).thenReturn(outbox);
|
||||
EventsPublishRespDTO respDTO = accepted("evt_stale_claim", 403L);
|
||||
when(eventsPublishApi.publish(any())).thenReturn(CommonResult.success(respDTO));
|
||||
when(outboxMapper.markPublished(501L, 100L, 1, "evt_stale_claim", 403L)).thenReturn(0);
|
||||
ListAppender<ILoggingEvent> appender = attachWorkerLogAppender();
|
||||
|
||||
try {
|
||||
int dispatched = worker(true).dispatchOnce();
|
||||
|
||||
assertEquals(1, dispatched);
|
||||
assertTrue(appender.list.stream().anyMatch(event -> event.getLevel().equals(Level.WARN)
|
||||
&& event.getFormattedMessage().contains("Account Events publish 终态回写被跳过")
|
||||
&& event.getFormattedMessage().contains("outboxId=501")
|
||||
&& event.getFormattedMessage().contains("attempt=1")
|
||||
&& event.getFormattedMessage().contains("targetStatus=published")),
|
||||
"accepted 发布后若终态回写 0 行，必须记录 stale claim 诊断日志");
|
||||
} finally {
|
||||
detachWorkerLogAppender(appender);
|
||||
}
|
||||
}
|
||||
|
||||
private void assertDeadLetterForTerminalStatus(String status) {
|
||||
AccountEventPublishOutboxMapper mapper = mock(AccountEventPublishOutboxMapper.class);
|
||||
EventsPublishApi api = mock(EventsPublishApi.class);
|
||||
AccountEventPublishOutboxDO outbox = runningOutbox();
|
||||
when(mapper.claimNextPublishOutbox(60L)).thenReturn(outbox);
|
||||
EventsPublishRespDTO respDTO = new EventsPublishRespDTO();
|
||||
respDTO.setPublishStatus(status);
|
||||
respDTO.setPublishErrorCode("payload_schema_mismatch");
|
||||
when(api.publish(any())).thenReturn(CommonResult.success(respDTO));
|
||||
when(mapper.markDeadLetter(anyLong(), anyLong(), org.mockito.Mockito.anyInt(),
|
||||
anyString(), anyString())).thenReturn(1);
|
||||
|
||||
int dispatched = worker(true, mapper, api).dispatchOnce();
|
||||
|
||||
assertEquals(1, dispatched);
|
||||
verify(mapper).markDeadLetter(501L, 100L, 1, "payload_schema_mismatch",
|
||||
"Account Events publish rejected");
|
||||
}
|
||||
|
||||
private static void assertClaimOwnershipGuard(String methodName, Class<?>... parameterTypes)
|
||||
throws NoSuchMethodException {
|
||||
Method method = AccountEventPublishOutboxMapper.class.getMethod(methodName, parameterTypes);
|
||||
Update update = method.getAnnotation(Update.class);
|
||||
String sql = String.join(" ", update.value()).toLowerCase().replaceAll("\\s+", " ");
|
||||
assertTrue(sql.contains("publish_status = 'running'"),
|
||||
methodName + " 必须只允许当前 running claim 回写终态");
|
||||
assertTrue(sql.contains("attempt_count = #{claimedattemptcount}"),
|
||||
methodName + " 必须用本次 claim attempt_count 防止过期 worker 覆盖新终态");
|
||||
}
|
||||
|
||||
private static ListAppender<ILoggingEvent> attachWorkerLogAppender() {
|
||||
Logger logger = (Logger) LoggerFactory.getLogger(AccountEventPublishWorker.class);
|
||||
ListAppender<ILoggingEvent> appender = new ListAppender<>();
|
||||
appender.start();
|
||||
logger.addAppender(appender);
|
||||
return appender;
|
||||
}
|
||||
|
||||
private static void detachWorkerLogAppender(ListAppender<ILoggingEvent> appender) {
|
||||
Logger logger = (Logger) LoggerFactory.getLogger(AccountEventPublishWorker.class);
|
||||
logger.detachAppender(appender);
|
||||
appender.stop();
|
||||
}
|
||||
|
||||
private AccountEventPublishWorker worker(boolean enabled) {
|
||||
return worker(enabled, outboxMapper, eventsPublishApi);
|
||||
}
|
||||
|
||||
private AccountEventPublishWorker worker(boolean enabled, AccountEventPublishOutboxMapper mapper,
|
||||
EventsPublishApi api) {
|
||||
MuseAccountEventsProperties properties = new MuseAccountEventsProperties();
|
||||
properties.getPublishWorker().setEnabled(enabled);
|
||||
return new AccountEventPublishWorker(mapper, api, properties);
|
||||
}
|
||||
|
||||
private EventsPublishReqDTO capturePublishReq() {
|
||||
ArgumentCaptor<EventsPublishReqDTO> captor = ArgumentCaptor.forClass(EventsPublishReqDTO.class);
|
||||
verify(eventsPublishApi).publish(captor.capture());
|
||||
return captor.getValue();
|
||||
}
|
||||
|
||||
private static EventsPublishRespDTO accepted(String eventId, Long sequenceNo) {
|
||||
EventsPublishRespDTO respDTO = new EventsPublishRespDTO();
|
||||
respDTO.setPublishStatus("accepted");
|
||||
respDTO.setEventId(eventId);
|
||||
respDTO.setSequenceNo(sequenceNo);
|
||||
return respDTO;
|
||||
}
|
||||
|
||||
private static void assertPayload(Map<String, Object> payload) {
|
||||
assertEquals(Set.of("type", "message", "resourceRef", "timestamp"), payload.keySet());
|
||||
assertEquals("quota_alert", payload.get("type"));
|
||||
assertEquals("额度已调整", payload.get("message"));
|
||||
Map<?, ?> resourceRef = assertInstanceOf(Map.class, payload.get("resourceRef"));
|
||||
assertEquals(Set.of("resourceType", "resourceId"), resourceRef.keySet());
|
||||
assertEquals("account_quota", resourceRef.get("resourceType"));
|
||||
assertInstanceOf(Number.class, resourceRef.get("resourceId"));
|
||||
assertEquals(8001L, ((Number) resourceRef.get("resourceId")).longValue());
|
||||
}
|
||||
|
||||
private static AccountEventPublishOutboxDO runningOutbox() {
|
||||
AccountEventPublishOutboxDO outbox = new AccountEventPublishOutboxDO();
|
||||
outbox.setId(501L);
|
||||
outbox.setTenantId(100L);
|
||||
outbox.setOutboxId("acct_out:0123456789abcdef0123456789abcdef");
|
||||
outbox.setCommandId("acct_evt:0123456789abcdef0123456789abcdef");
|
||||
outbox.setAuditLogId(8001L);
|
||||
outbox.setAccountUserId(1001L);
|
||||
outbox.setSourceRevision("__none__");
|
||||
outbox.setEventType("notification");
|
||||
outbox.setNotificationType("quota_alert");
|
||||
outbox.setResourceRefType("account_quota");
|
||||
outbox.setResourceRefId(8001L);
|
||||
outbox.setPayloadSummary(JsonUtils.toJsonString(Map.of(
|
||||
"type", "quota_alert",
|
||||
"message", "额度已调整",
|
||||
"resourceRef", Map.of("resourceType", "account_quota", "resourceId", 8001L),
|
||||
"timestamp", "2026-06-07T12:00:00")));
|
||||
outbox.setAttemptCount(1);
|
||||
outbox.setMaxAttempt(5);
|
||||
outbox.setCreateTime(LocalDateTime.of(2026, 6, 7, 12, 0));
|
||||
return outbox;
|
||||
}
|
||||
|
||||
}
|
||||
@ -66,6 +66,8 @@ class AccountQuotaServiceTest extends BaseMockitoUnitTest {
|
||||
private AccountCommandService commandService;
|
||||
@Mock
|
||||
private AccountAuditService auditService;
|
||||
@Mock
|
||||
private AccountEventPublishOutboxService accountEventPublishOutboxService;
|
||||
|
||||
@Test
|
||||
void should_createQuotaAdjustment_withCommandAuditAndSnapshots() {
|
||||
@ -114,6 +116,16 @@ class AccountQuotaServiceTest extends BaseMockitoUnitTest {
|
||||
assertEquals(25, JsonUtils.parseTree(log.getDeltaValueSnapshot()).path("delta").asInt());
|
||||
assertEquals(125, JsonUtils.parseTree(log.getAfterValueSnapshot()).path("limit").asInt());
|
||||
assertEquals("corr-quota-1", JsonUtils.parseTree(log.getAfterValueSnapshot()).path("correlationId").asText());
|
||||
verify(accountEventPublishOutboxService, times(2)).createForQuotaAdjustment(
|
||||
any(MemberEntitlementAuditLogDO.class), eq("cmd-quota-1"), eq("hash-quota"));
|
||||
verify(accountEventPublishOutboxService).createForQuotaAdjustment(argThat(outboxLog ->
|
||||
Long.valueOf(8001L).equals(outboxLog.getId())
|
||||
&& "cmd-quota-1:ai_calls".equals(outboxLog.getIdempotencyKey())),
|
||||
eq("cmd-quota-1"), eq("hash-quota"));
|
||||
verify(accountEventPublishOutboxService).createForQuotaAdjustment(argThat(outboxLog ->
|
||||
Long.valueOf(8002L).equals(outboxLog.getId())
|
||||
&& "cmd-quota-1:exports".equals(outboxLog.getIdempotencyKey())),
|
||||
eq("cmd-quota-1"), eq("hash-quota"));
|
||||
verify(auditService).record(argThat(req -> "adminCreateQuotaAdjustment".equals(req.getOperationId())
|
||||
&& Long.valueOf(9001L).equals(req.getActorUserId())
|
||||
&& Long.valueOf(1001L).equals(req.getAccountUserId())
|
||||
@ -205,6 +217,7 @@ class AccountQuotaServiceTest extends BaseMockitoUnitTest {
|
||||
assertEquals("idempotent_hit", result.getStatus());
|
||||
verify(quotaMapper, never()).updateById(any(MemberQuotaDO.class));
|
||||
verify(auditLogMapper, never()).insert(any(MemberEntitlementAuditLogDO.class));
|
||||
verify(accountEventPublishOutboxService, never()).createForQuotaAdjustment(any(), anyString(), anyString());
|
||||
verify(auditService, never()).record(any());
|
||||
verify(commandService, never()).recordSucceeded(any(), anyString());
|
||||
}
|
||||
|
||||
@ -0,0 +1,64 @@
|
||||
-- P1R-7e Account quota adjustment event publish outbox Schema。
|
||||
-- 迁移边界：Account owner 只保存面向 accountUserId 的 quota_alert 摘要；不保存 reason 原文、配额快照、operator 明细或外部凭据。
|
||||
|
||||
CREATE TABLE muse_account_event_publish_outbox (
|
||||
id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
|
||||
tenant_id BIGINT NOT NULL,
|
||||
outbox_id VARCHAR(128) NOT NULL,
|
||||
command_id VARCHAR(128) NOT NULL,
|
||||
source_command_id VARCHAR(128) NOT NULL,
|
||||
audit_log_id BIGINT NOT NULL,
|
||||
account_user_id BIGINT NOT NULL,
|
||||
operator_user_id BIGINT,
|
||||
resource_type VARCHAR(64) NOT NULL,
|
||||
delta_amount BIGINT NOT NULL,
|
||||
source_owner VARCHAR(64) NOT NULL,
|
||||
audit_event_id BIGINT,
|
||||
request_hash VARCHAR(128),
|
||||
source_revision VARCHAR(80) NOT NULL,
|
||||
event_type VARCHAR(32) NOT NULL,
|
||||
notification_type VARCHAR(64) NOT NULL,
|
||||
resource_ref_type VARCHAR(64) NOT NULL,
|
||||
resource_ref_id BIGINT NOT NULL,
|
||||
payload_summary JSONB NOT NULL DEFAULT '{}'::jsonb,
|
||||
publish_status VARCHAR(32) NOT NULL,
|
||||
attempt_count INT NOT NULL DEFAULT 0,
|
||||
max_attempt INT NOT NULL DEFAULT 5,
|
||||
next_retry_at TIMESTAMP,
|
||||
claimed_at TIMESTAMP,
|
||||
claim_expires_at TIMESTAMP,
|
||||
last_error_code VARCHAR(64),
|
||||
last_error_message TEXT,
|
||||
published_event_id VARCHAR(128),
|
||||
published_sequence_no BIGINT,
|
||||
creator VARCHAR(64) NOT NULL DEFAULT '',
|
||||
updater VARCHAR(64) NOT NULL DEFAULT '',
|
||||
create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
deleted BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
CONSTRAINT uk_muse_account_event_publish_outbox_id UNIQUE (tenant_id, outbox_id),
|
||||
CONSTRAINT uk_muse_account_event_publish_outbox_command UNIQUE (tenant_id, command_id),
|
||||
CONSTRAINT uk_muse_account_event_publish_outbox_audit_notification UNIQUE (tenant_id, audit_log_id, notification_type),
|
||||
CONSTRAINT chk_muse_account_event_publish_outbox_event_type CHECK (event_type = 'notification'),
|
||||
CONSTRAINT chk_muse_account_event_publish_outbox_notification_type CHECK (notification_type = 'quota_alert'),
|
||||
CONSTRAINT chk_muse_account_event_publish_outbox_status CHECK (publish_status IN ('queued', 'running', 'retryable', 'published', 'dead_letter')),
|
||||
CONSTRAINT chk_muse_account_event_publish_outbox_account_owner CHECK (account_user_id > 0),
|
||||
CONSTRAINT chk_muse_account_event_publish_outbox_audit_log CHECK (audit_log_id > 0),
|
||||
CONSTRAINT chk_muse_account_event_publish_outbox_resource_ref CHECK (resource_ref_id > 0),
|
||||
CONSTRAINT chk_muse_account_event_publish_outbox_attempt_count CHECK (attempt_count >= 0),
|
||||
CONSTRAINT chk_muse_account_event_publish_outbox_max_attempt CHECK (max_attempt > 0)
|
||||
);
|
||||
|
||||
CREATE INDEX idx_muse_account_event_publish_outbox_claim
|
||||
ON muse_account_event_publish_outbox(tenant_id, publish_status, next_retry_at, id)
|
||||
WHERE deleted = FALSE;
|
||||
|
||||
CREATE INDEX idx_muse_account_event_publish_outbox_owner_status
|
||||
ON muse_account_event_publish_outbox(tenant_id, account_user_id, publish_status, create_time);
|
||||
|
||||
CREATE INDEX idx_muse_account_event_publish_outbox_audit_log
|
||||
ON muse_account_event_publish_outbox(tenant_id, audit_log_id);
|
||||
|
||||
CREATE TRIGGER trg_muse_account_event_publish_outbox_update_time
|
||||
BEFORE UPDATE ON muse_account_event_publish_outbox
|
||||
FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
|
||||
Loading…
x
Reference in New Issue
Block a user