feat(p1r): 接入 Market 事件传播真实链路

muse-cloud/muse-module-market/muse-module-market-server/pom.xml

@@ -22,6 +22,11 @@
             <artifactId>muse-module-market-api</artifactId>
             <version>${revision}</version>
         </dependency>
+        <dependency>
+            <groupId>cn.iocoder.cloud</groupId>
+            <artifactId>muse-module-events-api</artifactId>
+            <version>${revision}</version>
+        </dependency>
         <dependency>
             <groupId>cn.iocoder.cloud</groupId>
             <artifactId>muse-module-member-server</artifactId>

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<AdminMarketAssetSummary> 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,

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<String> PAYLOAD_KEYS = Set.of("type", "message", "resourceRef", "timestamp");
+    private static final Set<String> RESOURCE_REF_KEYS = Set.of("resourceType", "resourceId");
+    private static final Set<String> 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<String, Object> 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;
+    }
+
+}

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);
+
+}

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<String> 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<String, Object> notificationPayload(MuseMarketGovernanceActionDO action) {
+        Map<String, Object> 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<String, Object> resourceRef(Long assetId) {
+        Map<String, Object> resourceRef = new LinkedHashMap<>();
+        resourceRef.put("resourceType", RESOURCE_TYPE_MARKET_ASSET);
+        resourceRef.put("resourceId", assetId);
+        return resourceRef;
+    }
+
+    private void validatePayloadOrThrow(MuseMarketEventPublishOutboxDO outbox) {
+        Map<String, Object> 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);
+        }
+    }
+
+}

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<EventsPublishRespDTO> 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<String, Object> 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<String, Object> parsePayload(MuseMarketEventPublishOutboxDO outbox) {
+        try {
+            Map<String, Object> parsed = JsonUtils.getObjectMapper()
+                    .readValue(outbox.getPayloadSummary(), new TypeReference<Map<String, Object>>() {
+                    });
+            return parsed == null ? Map.of() : parsed;
+        } catch (Exception exception) {
+            log.warn("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);
+        }
+
+    }
+
+}

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;
+
+}

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<MuseMarketEventPublishOutboxDO> {
+
+    default MuseMarketEventPublishOutboxDO selectByTenantIdAndOutboxId(Long tenantId, String outboxId) {
+        return selectOne(new LambdaQueryWrapperX<MuseMarketEventPublishOutboxDO>()
+                .eq(MuseMarketEventPublishOutboxDO::getTenantId, tenantId)
+                .eq(MuseMarketEventPublishOutboxDO::getOutboxId, outboxId));
+    }
+
+    default MuseMarketEventPublishOutboxDO selectByTenantIdAndCommandId(Long tenantId, String commandId) {
+        return selectOne(new LambdaQueryWrapperX<MuseMarketEventPublishOutboxDO>()
+                .eq(MuseMarketEventPublishOutboxDO::getTenantId, tenantId)
+                .eq(MuseMarketEventPublishOutboxDO::getCommandId, commandId));
+    }
+
+    default MuseMarketEventPublishOutboxDO selectByTenantIdAndActionIdAndNotificationType(Long tenantId,
+                                                                                          String actionId,
+                                                                                          String notificationType) {
+        return selectOne(new LambdaQueryWrapperX<MuseMarketEventPublishOutboxDO>()
+                .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);
+
+}

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 {
+}

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;
+
+    }
+
+}

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();