From 0e10caf324166989a26e8bfac36da88847d35f76 Mon Sep 17 00:00:00 2001 From: zizi Date: Mon, 8 Jun 2026 02:49:02 +0800 Subject: [PATCH] =?UTF-8?q?feat(p1r):=20=E6=8E=A5=E5=85=A5=20Account=20?= =?UTF-8?q?=E4=BA=8B=E4=BB=B6=E4=BC=A0=E6=92=AD=E7=9C=9F=E5=AE=9E=E9=93=BE?= =?UTF-8?q?=E8=B7=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../muse-module-member-server/pom.xml | 5 + .../account/AccountEventPayloads.java | 33 ++ .../AccountEventPublishOutboxService.java | 18 + .../AccountEventPublishOutboxServiceImpl.java | 195 ++++++++++ .../account/AccountEventPublishWorker.java | 248 ++++++++++++ .../account/AccountQuotaServiceImpl.java | 5 + .../account/AccountEventPublishOutboxDO.java | 56 +++ .../AccountEventPublishOutboxMapper.java | 163 ++++++++ .../MuseAccountEventsConfiguration.java | 12 + .../config/MuseAccountEventsProperties.java | 38 ++ .../AccountEventPublishOutboxServiceTest.java | 232 +++++++++++ .../AccountEventPublishWorkerTest.java | 365 ++++++++++++++++++ .../account/AccountQuotaServiceTest.java | 13 + ...__extend_account_events_publish_outbox.sql | 64 +++ 14 files changed, 1447 insertions(+) create mode 100644 muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/application/account/AccountEventPayloads.java create mode 100644 muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/application/account/AccountEventPublishOutboxService.java create mode 100644 muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/application/account/AccountEventPublishOutboxServiceImpl.java create mode 100644 muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/application/account/AccountEventPublishWorker.java create mode 100644 muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/dal/dataobject/account/AccountEventPublishOutboxDO.java create mode 100644 muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/dal/mysql/account/AccountEventPublishOutboxMapper.java create mode 100644 muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/framework/config/MuseAccountEventsConfiguration.java create mode 100644 muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/framework/config/MuseAccountEventsProperties.java create mode 100644 muse-cloud/muse-module-member/muse-module-member-server/src/test/java/cn/iocoder/muse/module/member/application/account/AccountEventPublishOutboxServiceTest.java create mode 100644 muse-cloud/muse-module-member/muse-module-member-server/src/test/java/cn/iocoder/muse/module/member/application/account/AccountEventPublishWorkerTest.java create mode 100644 muse-cloud/sql/muse/V20__extend_account_events_publish_outbox.sql diff --git a/muse-cloud/muse-module-member/muse-module-member-server/pom.xml b/muse-cloud/muse-module-member/muse-module-member-server/pom.xml index 8dacb5b2..195420e0 100644 --- a/muse-cloud/muse-module-member/muse-module-member-server/pom.xml +++ b/muse-cloud/muse-module-member/muse-module-member-server/pom.xml @@ -40,6 +40,11 @@ muse-module-infra-api ${revision} + + cn.iocoder.cloud + muse-module-events-api + ${revision} + diff --git a/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/application/account/AccountEventPayloads.java b/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/application/account/AccountEventPayloads.java new file mode 100644 index 00000000..b2936102 --- /dev/null +++ b/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/application/account/AccountEventPayloads.java @@ -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 PAYLOAD_KEYS = Set.of("type", "message", "resourceRef", "timestamp"); + private static final Set RESOURCE_REF_KEYS = Set.of("resourceType", "resourceId"); + + private AccountEventPayloads() { + } + + public static boolean isValidQuotaAlertPayload(Map 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; + } + +} diff --git a/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/application/account/AccountEventPublishOutboxService.java b/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/application/account/AccountEventPublishOutboxService.java new file mode 100644 index 00000000..243a3ea2 --- /dev/null +++ b/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/application/account/AccountEventPublishOutboxService.java @@ -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); + +} diff --git a/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/application/account/AccountEventPublishOutboxServiceImpl.java b/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/application/account/AccountEventPublishOutboxServiceImpl.java new file mode 100644 index 00000000..49ebfe09 --- /dev/null +++ b/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/application/account/AccountEventPublishOutboxServiceImpl.java @@ -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 notificationPayload(MemberEntitlementAuditLogDO log, LocalDateTime eventTime) { + Map 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 resourceRef(Long auditLogId) { + Map resourceRef = new LinkedHashMap<>(); + resourceRef.put("resourceType", RESOURCE_REF_TYPE_ACCOUNT_QUOTA); + resourceRef.put("resourceId", auditLogId); + return resourceRef; + } + + private void validatePayloadOrThrow(AccountEventPublishOutboxDO outbox) { + Map 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); + } + } + +} diff --git a/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/application/account/AccountEventPublishWorker.java b/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/application/account/AccountEventPublishWorker.java new file mode 100644 index 00000000..c47f1e34 --- /dev/null +++ b/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/application/account/AccountEventPublishWorker.java @@ -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 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 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 parsePayload(AccountEventPublishOutboxDO outbox) { + try { + Map parsed = JsonUtils.getObjectMapper() + .readValue(outbox.getPayloadSummary(), new TypeReference>() { + }); + 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); + } + + } + +} diff --git a/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/application/account/AccountQuotaServiceImpl.java b/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/application/account/AccountQuotaServiceImpl.java index da224cd7..023f05ac 100644 --- a/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/application/account/AccountQuotaServiceImpl.java +++ b/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/application/account/AccountQuotaServiceImpl.java @@ -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; } diff --git a/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/dal/dataobject/account/AccountEventPublishOutboxDO.java b/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/dal/dataobject/account/AccountEventPublishOutboxDO.java new file mode 100644 index 00000000..aa8a4443 --- /dev/null +++ b/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/dal/dataobject/account/AccountEventPublishOutboxDO.java @@ -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; + +} diff --git a/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/dal/mysql/account/AccountEventPublishOutboxMapper.java b/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/dal/mysql/account/AccountEventPublishOutboxMapper.java new file mode 100644 index 00000000..19dc8932 --- /dev/null +++ b/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/dal/mysql/account/AccountEventPublishOutboxMapper.java @@ -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 { + + default AccountEventPublishOutboxDO selectByTenantIdAndOutboxId(Long tenantId, String outboxId) { + return selectOne(new LambdaQueryWrapperX() + .eq(AccountEventPublishOutboxDO::getTenantId, tenantId) + .eq(AccountEventPublishOutboxDO::getOutboxId, outboxId)); + } + + default AccountEventPublishOutboxDO selectByTenantIdAndCommandId(Long tenantId, String commandId) { + return selectOne(new LambdaQueryWrapperX() + .eq(AccountEventPublishOutboxDO::getTenantId, tenantId) + .eq(AccountEventPublishOutboxDO::getCommandId, commandId)); + } + + default AccountEventPublishOutboxDO selectByTenantIdAndAuditLogIdAndNotificationType(Long tenantId, + Long auditLogId, + String notificationType) { + return selectOne(new LambdaQueryWrapperX() + .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); + +} diff --git a/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/framework/config/MuseAccountEventsConfiguration.java b/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/framework/config/MuseAccountEventsConfiguration.java new file mode 100644 index 00000000..0ff96d8a --- /dev/null +++ b/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/framework/config/MuseAccountEventsConfiguration.java @@ -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 { +} diff --git a/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/framework/config/MuseAccountEventsProperties.java b/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/framework/config/MuseAccountEventsProperties.java new file mode 100644 index 00000000..8a17f7d4 --- /dev/null +++ b/muse-cloud/muse-module-member/muse-module-member-server/src/main/java/cn/iocoder/muse/module/member/framework/config/MuseAccountEventsProperties.java @@ -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; + + } + +} diff --git a/muse-cloud/muse-module-member/muse-module-member-server/src/test/java/cn/iocoder/muse/module/member/application/account/AccountEventPublishOutboxServiceTest.java b/muse-cloud/muse-module-member/muse-module-member-server/src/test/java/cn/iocoder/muse/module/member/application/account/AccountEventPublishOutboxServiceTest.java new file mode 100644 index 00000000..42bd9bf6 --- /dev/null +++ b/muse-cloud/muse-module-member/muse-module-member-server/src/test/java/cn/iocoder/muse/module/member/application/account/AccountEventPublishOutboxServiceTest.java @@ -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 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 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 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; + } + +} diff --git a/muse-cloud/muse-module-member/muse-module-member-server/src/test/java/cn/iocoder/muse/module/member/application/account/AccountEventPublishWorkerTest.java b/muse-cloud/muse-module-member/muse-module-member-server/src/test/java/cn/iocoder/muse/module/member/application/account/AccountEventPublishWorkerTest.java new file mode 100644 index 00000000..bc84c852 --- /dev/null +++ b/muse-cloud/muse-module-member/muse-module-member-server/src/test/java/cn/iocoder/muse/module/member/application/account/AccountEventPublishWorkerTest.java @@ -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 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 attachWorkerLogAppender() { + Logger logger = (Logger) LoggerFactory.getLogger(AccountEventPublishWorker.class); + ListAppender appender = new ListAppender<>(); + appender.start(); + logger.addAppender(appender); + return appender; + } + + private static void detachWorkerLogAppender(ListAppender 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 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 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; + } + +} diff --git a/muse-cloud/muse-module-member/muse-module-member-server/src/test/java/cn/iocoder/muse/module/member/application/account/AccountQuotaServiceTest.java b/muse-cloud/muse-module-member/muse-module-member-server/src/test/java/cn/iocoder/muse/module/member/application/account/AccountQuotaServiceTest.java index adcabdce..0de37557 100644 --- a/muse-cloud/muse-module-member/muse-module-member-server/src/test/java/cn/iocoder/muse/module/member/application/account/AccountQuotaServiceTest.java +++ b/muse-cloud/muse-module-member/muse-module-member-server/src/test/java/cn/iocoder/muse/module/member/application/account/AccountQuotaServiceTest.java @@ -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()); } diff --git a/muse-cloud/sql/muse/V20__extend_account_events_publish_outbox.sql b/muse-cloud/sql/muse/V20__extend_account_events_publish_outbox.sql new file mode 100644 index 00000000..b9438d79 --- /dev/null +++ b/muse-cloud/sql/muse/V20__extend_account_events_publish_outbox.sql @@ -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();