diff --git a/muse-cloud/muse-module-market/muse-module-market-server/pom.xml b/muse-cloud/muse-module-market/muse-module-market-server/pom.xml index cf315ece..3c67b0c8 100644 --- a/muse-cloud/muse-module-market/muse-module-market-server/pom.xml +++ b/muse-cloud/muse-module-market/muse-module-market-server/pom.xml @@ -22,6 +22,11 @@ muse-module-market-api ${revision} + + cn.iocoder.cloud + muse-module-events-api + ${revision} + cn.iocoder.cloud muse-module-member-server diff --git a/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/application/muse/AdminMarketGovernanceServiceImpl.java b/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/application/muse/AdminMarketGovernanceServiceImpl.java index 99834755..305b4120 100644 --- a/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/application/muse/AdminMarketGovernanceServiceImpl.java +++ b/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/application/muse/AdminMarketGovernanceServiceImpl.java @@ -91,6 +91,7 @@ public class AdminMarketGovernanceServiceImpl implements AdminMarketGovernanceSe private final MuseMarketGovernanceImpactMapper governanceImpactMapper; private final MuseMarketSourceStatusEventMapper sourceStatusEventMapper; private final MarketCommandService commandService; + private final MarketEventPublishOutboxService eventPublishOutboxService; @Override public PageResult listMarketAssets(Long adminUserId, String apiVersion, Integer pageNo, @@ -349,6 +350,8 @@ public class AdminMarketGovernanceServiceImpl implements AdminMarketGovernanceSe action.setReason(reason); action.setTenantId(tenantId); governanceActionMapper.insert(action); + // 治理事实和 Events 发布意图必须同事务落地；任一失败都回滚，避免用户可见通知与事实审计断链。 + eventPublishOutboxService.createForGovernanceAction(action); writeActionImpact(tenantId, asset, actionId, commandId, requestHash, actionType, STATUS_APPLIED, linkedMap("scope", scope, "actionPolicy", actionPolicy, "source", "market_owned_projection")); return new GovernanceActionResult(String.valueOf(asset.getId()), listingStatus, diff --git a/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/application/muse/MarketEventPayloads.java b/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/application/muse/MarketEventPayloads.java new file mode 100644 index 00000000..9e8a86b4 --- /dev/null +++ b/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/application/muse/MarketEventPayloads.java @@ -0,0 +1,37 @@ +package cn.iocoder.muse.module.market.application.muse; + +import java.util.Map; +import java.util.Set; + +/** + * Market Events payload 安全边界校验。 + */ +final class MarketEventPayloads { + + private static final Set PAYLOAD_KEYS = Set.of("type", "message", "resourceRef", "timestamp"); + private static final Set RESOURCE_REF_KEYS = Set.of("resourceType", "resourceId"); + private static final Set SAFE_MESSAGES = Set.of("资产已下架", "资产已召回"); + private static final String NOTIFICATION_GOVERNANCE_ACTION = "governance_action"; + private static final String RESOURCE_TYPE_MARKET_ASSET = "market_asset"; + + private MarketEventPayloads() { + } + + static boolean isValidGovernancePayload(Map payload) { + if (payload == null || !payload.keySet().equals(PAYLOAD_KEYS) + || !NOTIFICATION_GOVERNANCE_ACTION.equals(payload.get("type")) + || !(payload.get("message") instanceof String message) + || !SAFE_MESSAGES.contains(message) + || !(payload.get("timestamp") instanceof String) + || !(payload.get("resourceRef") instanceof Map resourceRef)) { + return false; + } + if (!resourceRef.keySet().equals(RESOURCE_REF_KEYS) + || !RESOURCE_TYPE_MARKET_ASSET.equals(resourceRef.get("resourceType"))) { + return false; + } + Object resourceId = resourceRef.get("resourceId"); + return resourceId instanceof Number; + } + +} diff --git a/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/application/muse/MarketEventPublishOutboxService.java b/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/application/muse/MarketEventPublishOutboxService.java new file mode 100644 index 00000000..31f4b735 --- /dev/null +++ b/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/application/muse/MarketEventPublishOutboxService.java @@ -0,0 +1,19 @@ +package cn.iocoder.muse.module.market.application.muse; + +import cn.iocoder.muse.module.market.dal.dataobject.muse.MuseMarketEventPublishOutboxDO; +import cn.iocoder.muse.module.market.dal.dataobject.muse.MuseMarketGovernanceActionDO; + +/** + * Market 治理事实到统一 Events 的本域 outbox 服务。 + */ +public interface MarketEventPublishOutboxService { + + /** + * 为治理动作创建待发布 outbox。 + * + * @param action 已落库的 Market governance action 事实 + * @return 新建或幂等回查到的 outbox；非法 owner 会跳过并返回 null + */ + MuseMarketEventPublishOutboxDO createForGovernanceAction(MuseMarketGovernanceActionDO action); + +} diff --git a/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/application/muse/MarketEventPublishOutboxServiceImpl.java b/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/application/muse/MarketEventPublishOutboxServiceImpl.java new file mode 100644 index 00000000..7b5d3532 --- /dev/null +++ b/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/application/muse/MarketEventPublishOutboxServiceImpl.java @@ -0,0 +1,175 @@ +package cn.iocoder.muse.module.market.application.muse; + +import cn.iocoder.muse.framework.common.exception.ServiceException; +import cn.iocoder.muse.framework.common.util.json.JsonUtils; +import cn.iocoder.muse.module.market.dal.dataobject.muse.MuseMarketEventPublishOutboxDO; +import cn.iocoder.muse.module.market.dal.dataobject.muse.MuseMarketGovernanceActionDO; +import cn.iocoder.muse.module.market.dal.mysql.muse.MuseMarketEventPublishOutboxMapper; +import cn.iocoder.muse.module.market.framework.config.MuseMarketEventsProperties; +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; +import java.util.Set; + +/** + * Market 治理动作发布 outbox 服务实现。 + */ +@Service +@Slf4j +public class MarketEventPublishOutboxServiceImpl implements MarketEventPublishOutboxService { + + static final String STATUS_QUEUED = "queued"; + + private static final String SOURCE_REVISION_NONE = "__none__"; + private static final String EVENT_TYPE_NOTIFICATION = "notification"; + private static final String NOTIFICATION_GOVERNANCE_ACTION = "governance_action"; + private static final String RESOURCE_TYPE_MARKET_ASSET = "market_asset"; + static final String ERROR_PAYLOAD_INVALID_CODE = "MARKET_EVENTS_PAYLOAD_INVALID"; + static final int ERROR_PAYLOAD_INVALID_NUMERIC_CODE = 1_044_000_030; + private static final Set PUBLISHABLE_ACTION_TYPES = Set.of("delist", "recall"); + + private final MuseMarketEventPublishOutboxMapper outboxMapper; + private final MuseMarketEventsProperties properties; + + public MarketEventPublishOutboxServiceImpl(MuseMarketEventPublishOutboxMapper outboxMapper, + MuseMarketEventsProperties properties) { + this.outboxMapper = outboxMapper; + this.properties = properties; + } + + @Override + public MuseMarketEventPublishOutboxDO createForGovernanceAction(MuseMarketGovernanceActionDO action) { + if (action == null || !StringUtils.hasText(action.getActionId()) + || !PUBLISHABLE_ACTION_TYPES.contains(action.getActionType())) { + return null; + } + if (action.getOwnerUserId() == null || action.getOwnerUserId() <= 0) { + log.warn("Market Events outbox 跳过非法 owner 治理动作，tenantId={}, actionId={}, assetId={}, ownerUserId={}", + action.getTenantId(), action.getActionId(), action.getAssetId(), action.getOwnerUserId()); + return null; + } + MuseMarketEventPublishOutboxDO outbox = buildOutbox(action); + validatePayloadOrThrow(outbox); + int inserted = outboxMapper.insertIgnore(outbox); + if (inserted > 0) { + return outboxMapper.selectByTenantIdAndOutboxId(action.getTenantId(), outbox.getOutboxId()); + } + MuseMarketEventPublishOutboxDO existing = outboxMapper.selectByTenantIdAndActionIdAndNotificationType( + action.getTenantId(), action.getActionId(), NOTIFICATION_GOVERNANCE_ACTION); + return existing == null ? outboxMapper.selectByTenantIdAndCommandId(action.getTenantId(), outbox.getCommandId()) + : existing; + } + + private MuseMarketEventPublishOutboxDO buildOutbox(MuseMarketGovernanceActionDO action) { + MuseMarketEventPublishOutboxDO outbox = new MuseMarketEventPublishOutboxDO(); + outbox.setTenantId(action.getTenantId()); + outbox.setOutboxId(buildOutboxId(action.getTenantId(), action.getActionId(), action.getOwnerUserId())); + outbox.setCommandId(buildPublishCommandId(action.getTenantId(), action.getActionId(), + action.getOwnerUserId())); + outbox.setSourceCommandId(action.getCommandId()); + outbox.setActionId(action.getActionId()); + outbox.setActionType(action.getActionType()); + outbox.setAssetId(action.getAssetId()); + outbox.setOwnerUserId(action.getOwnerUserId()); + outbox.setActorUserId(action.getActorUserId()); + outbox.setPreviewId(action.getPreviewId()); + outbox.setRequestHash(action.getRequestHash()); + outbox.setSourceRevision(normalizeSourceRevision( + action.getRevision() == null ? null : String.valueOf(action.getRevision()))); + outbox.setEventType(EVENT_TYPE_NOTIFICATION); + outbox.setNotificationType(NOTIFICATION_GOVERNANCE_ACTION); + outbox.setResourceType(RESOURCE_TYPE_MARKET_ASSET); + outbox.setResourceId(String.valueOf(action.getAssetId())); + outbox.setPayloadSummary(JsonUtils.toJsonString(notificationPayload(action))); + outbox.setPublishStatus(STATUS_QUEUED); + outbox.setAttemptCount(0); + outbox.setMaxAttempt(configuredMaxAttempt()); + outbox.setNextRetryAt(LocalDateTime.now()); + return outbox; + } + + private Map notificationPayload(MuseMarketGovernanceActionDO action) { + Map payload = new LinkedHashMap<>(); + payload.put("type", NOTIFICATION_GOVERNANCE_ACTION); + payload.put("message", messageFor(action.getActionType())); + payload.put("resourceRef", resourceRef(action.getAssetId())); + payload.put("timestamp", eventTimestamp(action).toString()); + return payload; + } + + private LocalDateTime eventTimestamp(MuseMarketGovernanceActionDO action) { + return action.getCreateTime() == null ? LocalDateTime.now() : action.getCreateTime(); + } + + private Map resourceRef(Long assetId) { + Map resourceRef = new LinkedHashMap<>(); + resourceRef.put("resourceType", RESOURCE_TYPE_MARKET_ASSET); + resourceRef.put("resourceId", assetId); + return resourceRef; + } + + private void validatePayloadOrThrow(MuseMarketEventPublishOutboxDO outbox) { + Map payload = JsonUtils.parseObject(outbox.getPayloadSummary(), Map.class); + if (!MarketEventPayloads.isValidGovernancePayload(payload)) { + log.warn("Market Events outbox payload 安全校验失败，tenantId={}, actionId={}, assetId={}, errorCode={}", + outbox.getTenantId(), outbox.getActionId(), outbox.getAssetId(), + ERROR_PAYLOAD_INVALID_CODE); + throw new ServiceException(ERROR_PAYLOAD_INVALID_NUMERIC_CODE, ERROR_PAYLOAD_INVALID_CODE); + } + } + + private String messageFor(String actionType) { + return switch (actionType) { + case "delist" -> "资产已下架"; + case "recall" -> "资产已召回"; + default -> "资产治理状态已变化"; + }; + } + + private String normalizeSourceRevision(String sourceRevision) { + return StringUtils.hasText(sourceRevision) ? sourceRevision : SOURCE_REVISION_NONE; + } + + private int configuredMaxAttempt() { + if (properties == null || properties.getPublishWorker() == null + || properties.getPublishWorker().getMaxAttempt() <= 0) { + return 5; + } + return properties.getPublishWorker().getMaxAttempt(); + } + + private String buildPublishCommandId(Long tenantId, String actionId, Long ownerUserId) { + String raw = idempotencyKey(tenantId, actionId, ownerUserId); + return "mk_evt:" + sha256Hex(raw).substring(0, 32); + } + + private String buildOutboxId(Long tenantId, String actionId, Long ownerUserId) { + String raw = idempotencyKey(tenantId, actionId, ownerUserId); + return "mk_out:" + sha256Hex(raw).substring(0, 32); + } + + private String idempotencyKey(Long tenantId, String actionId, Long ownerUserId) { + return tenantId + "|" + actionId + "|" + ownerUserId + "|" + NOTIFICATION_GOVERNANCE_ACTION; + } + + 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-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/application/muse/MarketEventPublishWorker.java b/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/application/muse/MarketEventPublishWorker.java new file mode 100644 index 00000000..616e0956 --- /dev/null +++ b/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/application/muse/MarketEventPublishWorker.java @@ -0,0 +1,246 @@ +package cn.iocoder.muse.module.market.application.muse; + +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.market.dal.dataobject.muse.MuseMarketEventPublishOutboxDO; +import cn.iocoder.muse.module.market.dal.mysql.muse.MuseMarketEventPublishOutboxMapper; +import cn.iocoder.muse.module.market.framework.config.MuseMarketEventsProperties; +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; + +/** + * Market governance action outbox 发布到统一 Events 的 worker。 + */ +@Slf4j +@Component +public class MarketEventPublishWorker { + + 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_MARKET = "market"; + private static final String SOURCE_TYPE_MARKET_GOVERNANCE_ACTION = "market_governance_action"; + private static final String EVENT_TYPE_NOTIFICATION = "notification"; + + private static final String ERROR_RETRY_EXHAUSTED = "MARKET_EVENTS_PUBLISH_RETRY_EXHAUSTED"; + private static final String ERROR_API_FAILED = "MARKET_EVENTS_PUBLISH_API_FAILED"; + private static final String ERROR_TEMPORARY_UNAVAILABLE = "MARKET_EVENTS_PUBLISH_TEMPORARY_UNAVAILABLE"; + private static final String ERROR_UNKNOWN_STATUS = "MARKET_EVENTS_PUBLISH_UNKNOWN_STATUS"; + private static final String ERROR_PAYLOAD_INVALID = "MARKET_EVENTS_PAYLOAD_INVALID"; + private static final String ERROR_OUTBOX_INCOMPLETE = "MARKET_EVENTS_OUTBOX_INCOMPLETE"; + + static final long RETRY_BACKOFF_SECONDS = 60L; + + private final MuseMarketEventPublishOutboxMapper outboxMapper; + private final EventsPublishApi eventsPublishApi; + private final MuseMarketEventsProperties properties; + + public MarketEventPublishWorker(MuseMarketEventPublishOutboxMapper outboxMapper, + EventsPublishApi eventsPublishApi, + MuseMarketEventsProperties properties) { + this.outboxMapper = outboxMapper; + this.eventsPublishApi = eventsPublishApi; + this.properties = properties; + } + + @Scheduled(initialDelayString = "${muse.market.events.publish-worker.initial-delay-ms:1000}", + fixedDelayString = "${muse.market.events.publish-worker.fixed-delay-ms:1000}") + public void dispatchScheduled() { + dispatchOnce(); + } + + public int dispatchOnce() { + if (!isEnabled()) { + return 0; + } + MuseMarketEventPublishOutboxDO outbox = TenantUtils.executeIgnore( + () -> outboxMapper.claimNextPublishOutbox(configuredClaimTimeoutSeconds())); + if (outbox == null) { + return 0; + } + if (outbox.getTenantId() == null || outbox.getId() == null) { + markDeadLetter(outbox, ERROR_OUTBOX_INCOMPLETE, "Market Events publish outbox incomplete"); + return 0; + } + return TenantUtils.execute(outbox.getTenantId(), () -> publishClaimedOutbox(outbox)); + } + + private int publishClaimedOutbox(MuseMarketEventPublishOutboxDO outbox) { + if (isRetryExhausted(outbox)) { + markDeadLetter(outbox, ERROR_RETRY_EXHAUSTED, "Market 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("Market Events publish 临时失败，tenantId={}, ownerUserId={}, outboxId={}, actionId={}, attempt={}, errorType={}", + outbox.getTenantId(), outbox.getOwnerUserId(), outbox.getId(), outbox.getActionId(), + outbox.getAttemptCount(), exception.getClass().getSimpleName()); + if (isLastAttempt(outbox)) { + markDeadLetter(outbox, ERROR_RETRY_EXHAUSTED, "Market Events publish retry exhausted"); + return 0; + } + markRetryable(outbox, ERROR_TEMPORARY_UNAVAILABLE, "Market Events publish temporary unavailable"); + return 0; + } + if (result == null || result.isError()) { + if (isLastAttempt(outbox)) { + markDeadLetter(outbox, ERROR_RETRY_EXHAUSTED, "Market Events publish retry exhausted"); + return 0; + } + markRetryable(outbox, ERROR_API_FAILED, "Market Events publish failed"); + return 0; + } + return handlePublishResponse(outbox, result.getData()); + } + + private int handlePublishResponse(MuseMarketEventPublishOutboxDO outbox, EventsPublishRespDTO respDTO) { + if (respDTO == null || !StringUtils.hasText(respDTO.getPublishStatus())) { + markDeadLetter(outbox, ERROR_UNKNOWN_STATUS, "Market 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()), "Market Events publish rejected"); + return 1; + } + markDeadLetter(outbox, ERROR_UNKNOWN_STATUS, "Market Events publish unknown status"); + return 0; + } + + private BuildPublishRequestResult buildReqDTO(MuseMarketEventPublishOutboxDO outbox) { + Map payload = parsePayload(outbox); + if (!MarketEventPayloads.isValidGovernancePayload(payload) || outbox.getCreateTime() == null + || outbox.getOwnerUserId() == null || outbox.getOwnerUserId() <= 0 + || !StringUtils.hasText(outbox.getActionId()) + || !StringUtils.hasText(outbox.getCommandId())) { + return BuildPublishRequestResult.failed(ERROR_PAYLOAD_INVALID, "Market Events publish payload invalid"); + } + EventsPublishReqDTO reqDTO = new EventsPublishReqDTO(); + reqDTO.setCommandId(outbox.getCommandId()); + reqDTO.setTenantId(outbox.getTenantId()); + reqDTO.setOwnerUserId(outbox.getOwnerUserId()); + reqDTO.setSourceOwner(SOURCE_OWNER_MARKET); + reqDTO.setSourceType(SOURCE_TYPE_MARKET_GOVERNANCE_ACTION); + reqDTO.setSourceId(outbox.getActionId()); + reqDTO.setSourceRevision(outbox.getSourceRevision()); + reqDTO.setEventType(EVENT_TYPE_NOTIFICATION); + reqDTO.setResourceType(outbox.getResourceType()); + reqDTO.setResourceId(outbox.getResourceId()); + reqDTO.setPayloadSummary(payload); + reqDTO.setEmittedAt(outbox.getCreateTime()); + return BuildPublishRequestResult.success(reqDTO); + } + + private Map parsePayload(MuseMarketEventPublishOutboxDO outbox) { + try { + Map parsed = JsonUtils.getObjectMapper() + .readValue(outbox.getPayloadSummary(), new TypeReference>() { + }); + return parsed == null ? Map.of() : parsed; + } catch (Exception exception) { + log.warn("Market Events publish payload 解析失败，tenantId={}, ownerUserId={}, outboxId={}, actionId={}, errorType={}", + outbox.getTenantId(), outbox.getOwnerUserId(), outbox.getId(), outbox.getActionId(), + 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(MuseMarketEventPublishOutboxDO outbox) { + return outbox.getAttemptCount() != null && outbox.getMaxAttempt() != null + && outbox.getAttemptCount() > outbox.getMaxAttempt(); + } + + private boolean isLastAttempt(MuseMarketEventPublishOutboxDO outbox) { + return outbox.getAttemptCount() != null && outbox.getMaxAttempt() != null + && outbox.getAttemptCount() >= outbox.getMaxAttempt(); + } + + private void markPublished(MuseMarketEventPublishOutboxDO 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(MuseMarketEventPublishOutboxDO 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(MuseMarketEventPublishOutboxDO outbox, String errorCode, String errorMessage) { + if (outbox.getId() == null || outbox.getTenantId() == null) { + log.error("Market 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(MuseMarketEventPublishOutboxDO outbox, int updated, String targetStatus, + String errorCode) { + if (updated > 0) { + return; + } + // WHY：0 行通常表示租约过期后被其他 worker 重新领取，必须留下可检索证据，避免重复发布排障断链。 + log.warn("Market Events publish 终态回写被跳过，tenantId={}, ownerUserId={}, outboxId={}, " + + "actionId={}, attempt={}, targetStatus={}, errorCode={}", + outbox.getTenantId(), outbox.getOwnerUserId(), outbox.getId(), outbox.getActionId(), + 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-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/dal/dataobject/muse/MuseMarketEventPublishOutboxDO.java b/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/dal/dataobject/muse/MuseMarketEventPublishOutboxDO.java new file mode 100644 index 00000000..45aed02b --- /dev/null +++ b/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/dal/dataobject/muse/MuseMarketEventPublishOutboxDO.java @@ -0,0 +1,55 @@ +package cn.iocoder.muse.module.market.dal.dataobject.muse; + +import cn.iocoder.muse.framework.tenant.core.db.TenantBaseDO; +import cn.iocoder.muse.module.market.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; + +/** + * Market 治理动作发布到统一 Events 的本域 outbox。 + */ +@TableName(value = "muse_market_event_publish_outbox", autoResultMap = true) +@Data +@EqualsAndHashCode(callSuper = true) +@ToString(callSuper = true) +public class MuseMarketEventPublishOutboxDO extends TenantBaseDO { + + @TableId(type = IdType.AUTO) + private Long id; + + private String outboxId; + private String commandId; + private String sourceCommandId; + private String actionId; + private String actionType; + private Long assetId; + private Long ownerUserId; + private Long actorUserId; + private String previewId; + private String requestHash; + private String sourceRevision; + private String eventType; + private String notificationType; + private String resourceType; + private String resourceId; + @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-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/dal/mysql/muse/MuseMarketEventPublishOutboxMapper.java b/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/dal/mysql/muse/MuseMarketEventPublishOutboxMapper.java new file mode 100644 index 00000000..170f311f --- /dev/null +++ b/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/dal/mysql/muse/MuseMarketEventPublishOutboxMapper.java @@ -0,0 +1,161 @@ +package cn.iocoder.muse.module.market.dal.mysql.muse; + +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.market.dal.dataobject.muse.MuseMarketEventPublishOutboxDO; +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; + +/** + * Market Events publish outbox Mapper。 + */ +@Mapper +public interface MuseMarketEventPublishOutboxMapper extends BaseMapperX { + + default MuseMarketEventPublishOutboxDO selectByTenantIdAndOutboxId(Long tenantId, String outboxId) { + return selectOne(new LambdaQueryWrapperX() + .eq(MuseMarketEventPublishOutboxDO::getTenantId, tenantId) + .eq(MuseMarketEventPublishOutboxDO::getOutboxId, outboxId)); + } + + default MuseMarketEventPublishOutboxDO selectByTenantIdAndCommandId(Long tenantId, String commandId) { + return selectOne(new LambdaQueryWrapperX() + .eq(MuseMarketEventPublishOutboxDO::getTenantId, tenantId) + .eq(MuseMarketEventPublishOutboxDO::getCommandId, commandId)); + } + + default MuseMarketEventPublishOutboxDO selectByTenantIdAndActionIdAndNotificationType(Long tenantId, + String actionId, + String notificationType) { + return selectOne(new LambdaQueryWrapperX() + .eq(MuseMarketEventPublishOutboxDO::getTenantId, tenantId) + .eq(MuseMarketEventPublishOutboxDO::getActionId, actionId) + .eq(MuseMarketEventPublishOutboxDO::getNotificationType, notificationType)); + } + + /** + * 首次创建 Market Events publish outbox；重复 command 或 action notification 时返回 0。 + */ + @Insert(""" + INSERT INTO muse_market_event_publish_outbox(tenant_id, outbox_id, command_id, source_command_id, + action_id, action_type, asset_id, owner_user_id, + actor_user_id, preview_id, request_hash, source_revision, + event_type, notification_type, resource_type, resource_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) + VALUES (#{tenantId}, #{outboxId}, #{commandId}, #{sourceCommandId}, #{actionId}, #{actionType}, + #{assetId}, #{ownerUserId}, #{actorUserId}, #{previewId}, #{requestHash}, #{sourceRevision}, + #{eventType}, #{notificationType}, #{resourceType}, #{resourceId}, + #{payloadSummary,typeHandler=cn.iocoder.muse.module.market.dal.type.JsonbStringTypeHandler}, + #{publishStatus}, #{attemptCount}, #{maxAttempt}, #{nextRetryAt}, #{claimedAt}, + #{claimExpiresAt}, #{lastErrorCode}, #{lastErrorMessage}, #{publishedEventId}, + #{publishedSequenceNo}) + ON CONFLICT DO NOTHING + """) + int insertIgnore(MuseMarketEventPublishOutboxDO outbox); + + /** + * 原子领取下一条待发布 Market Events outbox。 + */ + @TenantIgnore + @Options(flushCache = Options.FlushCachePolicy.TRUE, useCache = false) + @Select(""" + UPDATE muse_market_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_market_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 * + """) + MuseMarketEventPublishOutboxDO claimNextPublishOutbox(long leaseSeconds); + + @Update(""" + UPDATE muse_market_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_market_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_market_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-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/framework/config/MuseMarketEventsConfiguration.java b/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/framework/config/MuseMarketEventsConfiguration.java new file mode 100644 index 00000000..4a041a99 --- /dev/null +++ b/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/framework/config/MuseMarketEventsConfiguration.java @@ -0,0 +1,12 @@ +package cn.iocoder.muse.module.market.framework.config; + +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Configuration; + +/** + * Market Events 发布配置装配。 + */ +@Configuration(proxyBeanMethods = false) +@EnableConfigurationProperties(MuseMarketEventsProperties.class) +public class MuseMarketEventsConfiguration { +} diff --git a/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/framework/config/MuseMarketEventsProperties.java b/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/framework/config/MuseMarketEventsProperties.java new file mode 100644 index 00000000..ddd70de6 --- /dev/null +++ b/muse-cloud/muse-module-market/muse-module-market-server/src/main/java/cn/iocoder/muse/module/market/framework/config/MuseMarketEventsProperties.java @@ -0,0 +1,38 @@ +package cn.iocoder.muse.module.market.framework.config; + +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; + +/** + * Market 到统一 Events 的本域发布配置。 + */ +@ConfigurationProperties(prefix = "muse.market.events") +@Data +public class MuseMarketEventsProperties { + + /** + * Market governance action 到统一 Events 的本域 worker 配置。 + */ + private PublishWorker publishWorker = new PublishWorker(); + + @Data + public static class PublishWorker { + + /** + * 是否启用 Market 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/sql/muse/V19__extend_market_events_publish_outbox.sql b/muse-cloud/sql/muse/V19__extend_market_events_publish_outbox.sql new file mode 100644 index 00000000..63f946f2 --- /dev/null +++ b/muse-cloud/sql/muse/V19__extend_market_events_publish_outbox.sql @@ -0,0 +1,62 @@ +-- P1R-7d Market governance action event publish outbox Schema。 +-- 迁移边界：Market owner 只保存面向资产发布者的安全 notification 摘要；不保存治理原因、完整快照、授权摘要或 handoff token。 + +CREATE TABLE muse_market_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, + action_id VARCHAR(128) NOT NULL, + action_type VARCHAR(32) NOT NULL, + asset_id BIGINT NOT NULL, + owner_user_id BIGINT NOT NULL, + actor_user_id BIGINT NOT NULL, + preview_id VARCHAR(128) NOT NULL, + request_hash VARCHAR(128) NOT NULL, + source_revision VARCHAR(80) NOT NULL, + event_type VARCHAR(32) NOT NULL, + notification_type VARCHAR(64) NOT NULL, + resource_type VARCHAR(64) NOT NULL, + resource_id VARCHAR(128) 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_market_event_publish_outbox_id UNIQUE (tenant_id, outbox_id), + CONSTRAINT uk_muse_market_event_publish_outbox_command UNIQUE (tenant_id, command_id), + CONSTRAINT uk_muse_market_event_publish_outbox_action_notification UNIQUE (tenant_id, action_id, notification_type), + CONSTRAINT chk_muse_market_event_publish_outbox_event_type CHECK (event_type = 'notification'), + CONSTRAINT chk_muse_market_event_publish_outbox_notification_type CHECK (notification_type = 'governance_action'), + CONSTRAINT chk_muse_market_event_publish_outbox_action_type CHECK (action_type IN ('delist', 'recall')), + CONSTRAINT chk_muse_market_event_publish_outbox_status CHECK (publish_status IN ('queued', 'running', 'retryable', 'published', 'dead_letter')), + CONSTRAINT chk_muse_market_event_publish_outbox_owner CHECK (owner_user_id > 0), + CONSTRAINT chk_muse_market_event_publish_outbox_attempt_count CHECK (attempt_count >= 0), + CONSTRAINT chk_muse_market_event_publish_outbox_max_attempt CHECK (max_attempt > 0) +); + +CREATE INDEX idx_muse_market_event_publish_outbox_claim + ON muse_market_event_publish_outbox(tenant_id, publish_status, next_retry_at, id) + WHERE deleted = FALSE; + +CREATE INDEX idx_muse_market_event_publish_outbox_owner_status + ON muse_market_event_publish_outbox(tenant_id, owner_user_id, publish_status, create_time); + +CREATE INDEX idx_muse_market_event_publish_outbox_action + ON muse_market_event_publish_outbox(tenant_id, action_id); + +CREATE TRIGGER trg_muse_market_event_publish_outbox_update_time + BEFORE UPDATE ON muse_market_event_publish_outbox + FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();