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

将 saveBlock 首次成功后的 active source attribution 写入 Content 本域 outbox，并通过默认关闭的 worker 发布 Events notification/source_status_change。V21 增加 outbox 表、幂等约束、active-only check 与 claim/retry/dead-letter 状态字段。

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

@@ -27,6 +27,11 @@
             <artifactId>muse-module-system-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-spring-boot-starter-biz-tenant</artifactId>

muse-cloud/muse-module-content/muse-module-content-server/src/main/java/cn/iocoder/muse/module/content/application/ContentAppServiceImpl.java

@@ -44,6 +44,8 @@ public class ContentAppServiceImpl implements ContentAppService {
     private ContentCommandService commandService;
     @Resource
     private ContentAuditService auditService;
+    @Resource
+    private ContentEventPublishOutboxService eventPublishOutboxService;
 
     @Override
     public PageResult<WorkSummaryRespVO> listWorks(Long userId, Integer pageNo, Integer pageSize, String status) {
@@ -293,7 +295,7 @@ public class ContentAppServiceImpl implements ContentAppService {
 
         // 保存正文后立即写来源归因，确保 Canonical Block 与来源快照可审计追溯。
         ContentSourceSnapshotVO sourceSnapshot = reqVO.getSourceSnapshot();
-        sourceAttributionMapper.insert(BlockSourceAttributionDO.builder()
+        BlockSourceAttributionDO sourceAttribution = BlockSourceAttributionDO.builder()
                 .workId(workId)
                 .blockId(blockId)
                 .revision(block.getRevision())
@@ -302,7 +304,11 @@ public class ContentAppServiceImpl implements ContentAppService {
                 .sourceVersion(sourceSnapshot.getSourceVersion())
                 .authorizationSnapshotId(parseLongQuietly(sourceSnapshot.getAuthorizationSnapshotId()))
                 .sourceStatus("active")
-                .build());
+                .build();
+        sourceAttributionMapper.insert(sourceAttribution);
+        // 只把 source attribution 的 active 终端事实写入 Content 本域 outbox；正文、授权快照和 lineage 均不外发。
+        eventPublishOutboxService.createForBlockSourceAttribution(userId, workId, blockId, block.getRevision(),
+                reqVO.getCommandId(), sourceAttribution);
         SaveBlockRespVO respVO = new SaveBlockRespVO(block.getRevision());
         auditService.recordSucceededWithAudit(reqVO.getCommandId(), "save_block", userId,
                 "app", "block", blockId, requestHash, JsonUtils.toJsonString(respVO),

muse-cloud/muse-module-content/muse-module-content-server/src/main/java/cn/iocoder/muse/module/content/application/ContentEventPayloads.java

@@ -0,0 +1,34 @@
+package cn.iocoder.muse.module.content.application;
+
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Content Events payload 安全白名单。
+ */
+public final class ContentEventPayloads {
+
+    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 ContentEventPayloads() {
+    }
+
+    public static boolean isValidSourceStatusChangePayload(Map<String, Object> payload) {
+        if (payload == null || !PAYLOAD_KEYS.equals(payload.keySet())) {
+            return false;
+        }
+        if (!"source_status_change".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 "content_block".equals(resourceRef.get("resourceType"))
+                && resourceRef.get("resourceId") instanceof Number;
+    }
+
+}

muse-cloud/muse-module-content/muse-module-content-server/src/main/java/cn/iocoder/muse/module/content/application/ContentEventPublishOutboxService.java

@@ -0,0 +1,21 @@
+package cn.iocoder.muse.module.content.application;
+
+import cn.iocoder.muse.module.content.dal.dataobject.BlockSourceAttributionDO;
+import cn.iocoder.muse.module.content.dal.dataobject.ContentEventPublishOutboxDO;
+
+/**
+ * Content source attribution 到统一 Events 的本域 outbox 服务。
+ */
+public interface ContentEventPublishOutboxService {
+
+    /**
+     * 为 saveBlock 首次成功写入的 active source attribution 创建发布 outbox。
+     */
+    ContentEventPublishOutboxDO createForBlockSourceAttribution(Long ownerUserId,
+                                                                Long workId,
+                                                                Long blockId,
+                                                                Integer blockRevision,
+                                                                String sourceCommandId,
+                                                                BlockSourceAttributionDO sourceAttribution);
+
+}

muse-cloud/muse-module-content/muse-module-content-server/src/main/java/cn/iocoder/muse/module/content/application/ContentEventPublishOutboxServiceImpl.java

@@ -0,0 +1,189 @@
+package cn.iocoder.muse.module.content.application;
+
+import cn.iocoder.muse.framework.common.exception.ServiceException;
+import cn.iocoder.muse.framework.common.util.json.JsonUtils;
+import cn.iocoder.muse.framework.tenant.core.context.TenantContextHolder;
+import cn.iocoder.muse.module.content.dal.dataobject.BlockSourceAttributionDO;
+import cn.iocoder.muse.module.content.dal.dataobject.ContentEventPublishOutboxDO;
+import cn.iocoder.muse.module.content.dal.mysql.ContentEventPublishOutboxMapper;
+import cn.iocoder.muse.module.content.framework.config.MuseContentEventsProperties;
+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;
+
+/**
+ * Content source attribution 发布 outbox 服务实现。
+ */
+@Service
+@Slf4j
+public class ContentEventPublishOutboxServiceImpl implements ContentEventPublishOutboxService {
+
+    static final String STATUS_QUEUED = "queued";
+
+    private static final String SOURCE_STATUS_ACTIVE = "active";
+    private static final String EVENT_TYPE_NOTIFICATION = "notification";
+    private static final String NOTIFICATION_SOURCE_STATUS_CHANGE = "source_status_change";
+    private static final String RESOURCE_REF_TYPE_CONTENT_BLOCK = "content_block";
+    static final String ERROR_PAYLOAD_INVALID_CODE = "CONTENT_EVENTS_PAYLOAD_INVALID";
+    static final int ERROR_PAYLOAD_INVALID_NUMERIC_CODE = 1_041_000_050;
+
+    private final ContentEventPublishOutboxMapper outboxMapper;
+    private final MuseContentEventsProperties properties;
+
+    public ContentEventPublishOutboxServiceImpl(ContentEventPublishOutboxMapper outboxMapper,
+                                                MuseContentEventsProperties properties) {
+        this.outboxMapper = outboxMapper;
+        this.properties = properties;
+    }
+
+    @Override
+    public ContentEventPublishOutboxDO createForBlockSourceAttribution(Long ownerUserId,
+                                                                       Long workId,
+                                                                       Long blockId,
+                                                                       Integer blockRevision,
+                                                                       String sourceCommandId,
+                                                                       BlockSourceAttributionDO sourceAttribution) {
+        if (sourceAttribution == null || sourceAttribution.getId() == null || sourceAttribution.getId() <= 0) {
+            log.warn("Content Events outbox 跳过缺失 source attribution，ownerUserId={}, workId={}, blockId={}, blockRevision={}",
+                    ownerUserId, workId, blockId, blockRevision);
+            return null;
+        }
+        if (ownerUserId == null || ownerUserId <= 0 || blockRevision == null || blockRevision <= 0) {
+            log.warn("Content Events outbox 跳过非法 owner 或 revision，tenantId={}, ownerUserId={}, workId={}, blockId={}, blockRevision={}",
+                    sourceAttribution.getTenantId(), ownerUserId, workId, blockId, blockRevision);
+            return null;
+        }
+        if (!SOURCE_STATUS_ACTIVE.equals(sourceAttribution.getSourceStatus())) {
+            log.warn("Content Events outbox 跳过非 active 来源状态，tenantId={}, ownerUserId={}, sourceAttributionId={}, blockId={}, revision={}, sourceStatus={}",
+                    sourceAttribution.getTenantId(), ownerUserId, sourceAttribution.getId(), blockId,
+                    blockRevision, sourceAttribution.getSourceStatus());
+            return null;
+        }
+        ContentEventPublishOutboxDO outbox = buildOutbox(ownerUserId, workId, blockId, blockRevision,
+                sourceCommandId, sourceAttribution);
+        validatePayloadOrThrow(outbox);
+        int inserted = outboxMapper.insertIgnore(outbox);
+        if (inserted > 0) {
+            return outboxMapper.selectByTenantIdAndOutboxId(outbox.getTenantId(), outbox.getOutboxId());
+        }
+        ContentEventPublishOutboxDO existing = outboxMapper.selectByTenantIdAndBlockIdAndBlockRevisionAndNotificationType(
+                outbox.getTenantId(), outbox.getBlockId(), outbox.getBlockRevision(), NOTIFICATION_SOURCE_STATUS_CHANGE);
+        return existing == null ? outboxMapper.selectByTenantIdAndCommandId(outbox.getTenantId(), outbox.getCommandId())
+                : existing;
+    }
+
+    private ContentEventPublishOutboxDO buildOutbox(Long ownerUserId,
+                                                    Long workId,
+                                                    Long blockId,
+                                                    Integer blockRevision,
+                                                    String sourceCommandId,
+                                                    BlockSourceAttributionDO sourceAttribution) {
+        Long revision = blockRevision.longValue();
+        Long tenantId = resolveTenantId(sourceAttribution);
+        LocalDateTime eventTime = eventTimestamp(sourceAttribution);
+        ContentEventPublishOutboxDO outbox = new ContentEventPublishOutboxDO();
+        outbox.setTenantId(tenantId);
+        outbox.setOutboxId(buildOutboxId(tenantId, blockId, revision));
+        outbox.setCommandId(buildPublishCommandId(tenantId, blockId, revision));
+        outbox.setSourceCommandId(StringUtils.hasText(sourceCommandId) ? sourceCommandId : "");
+        outbox.setSourceAttributionId(sourceAttribution.getId());
+        outbox.setOwnerUserId(ownerUserId);
+        outbox.setWorkId(workId);
+        outbox.setBlockId(blockId);
+        outbox.setBlockRevision(revision);
+        outbox.setSourceStatus(SOURCE_STATUS_ACTIVE);
+        outbox.setSourceRevision(String.valueOf(revision));
+        outbox.setEventType(EVENT_TYPE_NOTIFICATION);
+        outbox.setNotificationType(NOTIFICATION_SOURCE_STATUS_CHANGE);
+        outbox.setResourceRefType(RESOURCE_REF_TYPE_CONTENT_BLOCK);
+        outbox.setResourceRefId(blockId);
+        outbox.setPayloadSummary(JsonUtils.toJsonString(notificationPayload(blockId, eventTime)));
+        outbox.setPublishStatus(STATUS_QUEUED);
+        outbox.setAttemptCount(0);
+        outbox.setMaxAttempt(configuredMaxAttempt());
+        outbox.setNextRetryAt(LocalDateTime.now());
+        outbox.setCreateTime(eventTime);
+        return outbox;
+    }
+
+    private Map<String, Object> notificationPayload(Long blockId, LocalDateTime eventTime) {
+        Map<String, Object> payload = new LinkedHashMap<>();
+        payload.put("type", NOTIFICATION_SOURCE_STATUS_CHANGE);
+        payload.put("message", "内容来源状态已更新");
+        payload.put("resourceRef", resourceRef(blockId));
+        payload.put("timestamp", eventTime.toString());
+        return payload;
+    }
+
+    private Map<String, Object> resourceRef(Long blockId) {
+        Map<String, Object> resourceRef = new LinkedHashMap<>();
+        resourceRef.put("resourceType", RESOURCE_REF_TYPE_CONTENT_BLOCK);
+        resourceRef.put("resourceId", blockId);
+        return resourceRef;
+    }
+
+    private void validatePayloadOrThrow(ContentEventPublishOutboxDO outbox) {
+        Map<String, Object> payload = JsonUtils.parseObject(outbox.getPayloadSummary(), Map.class);
+        if (!ContentEventPayloads.isValidSourceStatusChangePayload(payload)
+                || outbox.getOwnerUserId() == null || outbox.getOwnerUserId() <= 0
+                || outbox.getBlockId() == null || outbox.getBlockId() <= 0
+                || outbox.getBlockRevision() == null || outbox.getBlockRevision() <= 0
+                || outbox.getResourceRefId() == null || outbox.getResourceRefId() <= 0) {
+            log.warn("Content Events outbox payload 安全校验失败，tenantId={}, ownerUserId={}, blockId={}, blockRevision={}, errorCode={}",
+                    outbox.getTenantId(), outbox.getOwnerUserId(), outbox.getBlockId(), outbox.getBlockRevision(),
+                    ERROR_PAYLOAD_INVALID_CODE);
+            throw new ServiceException(ERROR_PAYLOAD_INVALID_NUMERIC_CODE, ERROR_PAYLOAD_INVALID_CODE);
+        }
+    }
+
+    private Long resolveTenantId(BlockSourceAttributionDO sourceAttribution) {
+        return sourceAttribution.getTenantId() == null ? TenantContextHolder.getTenantId() : sourceAttribution.getTenantId();
+    }
+
+    private LocalDateTime eventTimestamp(BlockSourceAttributionDO sourceAttribution) {
+        return sourceAttribution.getCreateTime() == null ? LocalDateTime.now() : sourceAttribution.getCreateTime();
+    }
+
+    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 blockId, Long blockRevision) {
+        String raw = idempotencyKey(tenantId, blockId, blockRevision);
+        return "content_evt:" + sha256Hex(raw).substring(0, 32);
+    }
+
+    private String buildOutboxId(Long tenantId, Long blockId, Long blockRevision) {
+        String raw = idempotencyKey(tenantId, blockId, blockRevision);
+        return "content_out:" + sha256Hex(raw).substring(0, 32);
+    }
+
+    private String idempotencyKey(Long tenantId, Long blockId, Long blockRevision) {
+        return tenantId + "|" + blockId + "|" + blockRevision + "|" + NOTIFICATION_SOURCE_STATUS_CHANGE;
+    }
+
+    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-content/muse-module-content-server/src/main/java/cn/iocoder/muse/module/content/application/ContentEventPublishWorker.java

@@ -0,0 +1,254 @@
+package cn.iocoder.muse.module.content.application;
+
+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.content.dal.dataobject.ContentEventPublishOutboxDO;
+import cn.iocoder.muse.module.content.dal.mysql.ContentEventPublishOutboxMapper;
+import cn.iocoder.muse.module.content.framework.config.MuseContentEventsProperties;
+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 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;
+
+/**
+ * Content source attribution outbox 发布到统一 Events 的 worker。
+ */
+@Slf4j
+@Component
+public class ContentEventPublishWorker {
+
+    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_CONTENT = "content";
+    private static final String SOURCE_TYPE_CONTENT_BLOCK_SOURCE_ATTRIBUTION = "content_block_source_attribution";
+    private static final String EVENT_TYPE_NOTIFICATION = "notification";
+    private static final String NOTIFICATION_TYPE_SOURCE_STATUS_CHANGE = "source_status_change";
+    private static final String SOURCE_STATUS_ACTIVE = "active";
+
+    private static final String ERROR_RETRY_EXHAUSTED = "CONTENT_EVENTS_PUBLISH_RETRY_EXHAUSTED";
+    private static final String ERROR_API_FAILED = "CONTENT_EVENTS_PUBLISH_API_FAILED";
+    private static final String ERROR_TEMPORARY_UNAVAILABLE = "CONTENT_EVENTS_PUBLISH_TEMPORARY_UNAVAILABLE";
+    private static final String ERROR_UNKNOWN_STATUS = "CONTENT_EVENTS_PUBLISH_UNKNOWN_STATUS";
+    private static final String ERROR_PAYLOAD_INVALID = "CONTENT_EVENTS_PAYLOAD_INVALID";
+    private static final String ERROR_OUTBOX_INCOMPLETE = "CONTENT_EVENTS_OUTBOX_INCOMPLETE";
+
+    static final long RETRY_BACKOFF_SECONDS = 60L;
+
+    private final ContentEventPublishOutboxMapper outboxMapper;
+    private final EventsPublishApi eventsPublishApi;
+    private final MuseContentEventsProperties properties;
+
+    public ContentEventPublishWorker(ContentEventPublishOutboxMapper outboxMapper,
+                                     EventsPublishApi eventsPublishApi,
+                                     MuseContentEventsProperties properties) {
+        this.outboxMapper = outboxMapper;
+        this.eventsPublishApi = eventsPublishApi;
+        this.properties = properties;
+    }
+
+    @Scheduled(initialDelayString = "${muse.content.events.publish-worker.initial-delay-ms:1000}",
+            fixedDelayString = "${muse.content.events.publish-worker.fixed-delay-ms:1000}")
+    public void dispatchScheduled() {
+        dispatchOnce();
+    }
+
+    public int dispatchOnce() {
+        if (!isEnabled()) {
+            return 0;
+        }
+        ContentEventPublishOutboxDO outbox = TenantUtils.executeIgnore(
+                () -> outboxMapper.claimNextPublishOutbox(configuredClaimTimeoutSeconds()));
+        if (outbox == null) {
+            return 0;
+        }
+        if (outbox.getTenantId() == null || outbox.getId() == null) {
+            markDeadLetter(outbox, ERROR_OUTBOX_INCOMPLETE, "Content Events publish outbox incomplete");
+            return 0;
+        }
+        return TenantUtils.execute(outbox.getTenantId(), () -> publishClaimedOutbox(outbox));
+    }
+
+    private int publishClaimedOutbox(ContentEventPublishOutboxDO outbox) {
+        if (isRetryExhausted(outbox)) {
+            markDeadLetter(outbox, ERROR_RETRY_EXHAUSTED, "Content 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("Content Events publish 临时失败，tenantId={}, ownerUserId={}, outboxId={}, blockId={}, blockRevision={}, attempt={}, errorType={}",
+                    outbox.getTenantId(), outbox.getOwnerUserId(), outbox.getId(), outbox.getBlockId(),
+                    outbox.getBlockRevision(), outbox.getAttemptCount(), exception.getClass().getSimpleName());
+            if (isLastAttempt(outbox)) {
+                markDeadLetter(outbox, ERROR_RETRY_EXHAUSTED, "Content Events publish retry exhausted");
+                return 0;
+            }
+            markRetryable(outbox, ERROR_TEMPORARY_UNAVAILABLE, "Content Events publish temporary unavailable");
+            return 0;
+        }
+        if (result == null || result.isError()) {
+            if (isLastAttempt(outbox)) {
+                markDeadLetter(outbox, ERROR_RETRY_EXHAUSTED, "Content Events publish retry exhausted");
+                return 0;
+            }
+            markRetryable(outbox, ERROR_API_FAILED, "Content Events publish failed");
+            return 0;
+        }
+        return handlePublishResponse(outbox, result.getData());
+    }
+
+    private int handlePublishResponse(ContentEventPublishOutboxDO outbox, EventsPublishRespDTO respDTO) {
+        if (respDTO == null || !StringUtils.hasText(respDTO.getPublishStatus())) {
+            markDeadLetter(outbox, ERROR_UNKNOWN_STATUS, "Content 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()), "Content Events publish rejected");
+            return 1;
+        }
+        markDeadLetter(outbox, ERROR_UNKNOWN_STATUS, "Content Events publish unknown status");
+        return 0;
+    }
+
+    private BuildPublishRequestResult buildReqDTO(ContentEventPublishOutboxDO outbox) {
+        Map<String, Object> payload = parsePayload(outbox);
+        if (!ContentEventPayloads.isValidSourceStatusChangePayload(payload)
+                || !EVENT_TYPE_NOTIFICATION.equals(outbox.getEventType())
+                || !NOTIFICATION_TYPE_SOURCE_STATUS_CHANGE.equals(outbox.getNotificationType())
+                || !SOURCE_STATUS_ACTIVE.equals(outbox.getSourceStatus())
+                || outbox.getCreateTime() == null
+                || outbox.getOwnerUserId() == null || outbox.getOwnerUserId() <= 0
+                || outbox.getBlockId() == null || outbox.getBlockId() <= 0
+                || outbox.getBlockRevision() == null || outbox.getBlockRevision() <= 0
+                || !StringUtils.hasText(outbox.getCommandId())) {
+            return BuildPublishRequestResult.failed(ERROR_PAYLOAD_INVALID, "Content Events publish payload invalid");
+        }
+        EventsPublishReqDTO reqDTO = new EventsPublishReqDTO();
+        reqDTO.setCommandId(outbox.getCommandId());
+        reqDTO.setTenantId(outbox.getTenantId());
+        reqDTO.setOwnerUserId(outbox.getOwnerUserId());
+        reqDTO.setSourceOwner(SOURCE_OWNER_CONTENT);
+        reqDTO.setSourceType(SOURCE_TYPE_CONTENT_BLOCK_SOURCE_ATTRIBUTION);
+        reqDTO.setSourceId(String.valueOf(outbox.getBlockId()));
+        reqDTO.setSourceRevision(StringUtils.hasText(outbox.getSourceRevision())
+                ? outbox.getSourceRevision() : String.valueOf(outbox.getBlockRevision()));
+        reqDTO.setEventType(EVENT_TYPE_NOTIFICATION);
+        reqDTO.setResourceType(outbox.getResourceRefType());
+        reqDTO.setResourceId(String.valueOf(outbox.getResourceRefId()));
+        reqDTO.setPayloadSummary(payload);
+        reqDTO.setEmittedAt(outbox.getCreateTime());
+        return BuildPublishRequestResult.success(reqDTO);
+    }
+
+    private Map<String, Object> parsePayload(ContentEventPublishOutboxDO 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("Content Events publish payload 解析失败，tenantId={}, ownerUserId={}, outboxId={}, blockId={}, blockRevision={}, errorType={}",
+                    outbox.getTenantId(), outbox.getOwnerUserId(), outbox.getId(), outbox.getBlockId(),
+                    outbox.getBlockRevision(), 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(ContentEventPublishOutboxDO outbox) {
+        return outbox.getAttemptCount() != null && outbox.getMaxAttempt() != null
+                && outbox.getAttemptCount() > outbox.getMaxAttempt();
+    }
+
+    private boolean isLastAttempt(ContentEventPublishOutboxDO outbox) {
+        return outbox.getAttemptCount() != null && outbox.getMaxAttempt() != null
+                && outbox.getAttemptCount() >= outbox.getMaxAttempt();
+    }
+
+    private void markPublished(ContentEventPublishOutboxDO 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(ContentEventPublishOutboxDO 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(ContentEventPublishOutboxDO outbox, String errorCode, String errorMessage) {
+        if (outbox.getId() == null || outbox.getTenantId() == null) {
+            log.error("Content 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(ContentEventPublishOutboxDO outbox, int updated, String targetStatus,
+                                          String errorCode) {
+        if (updated > 0) {
+            return;
+        }
+        // 原因：0 行通常表示租约过期后被其他 worker 重新领取，必须留下可检索证据，避免重复发布排障断链。
+        log.warn("Content Events publish 终态回写被跳过，tenantId={}, ownerUserId={}, outboxId={}, " +
+                        "blockId={}, blockRevision={}, attempt={}, targetStatus={}, errorCode={}",
+                outbox.getTenantId(), outbox.getOwnerUserId(), outbox.getId(), outbox.getBlockId(),
+                outbox.getBlockRevision(), 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-content/muse-module-content-server/src/main/java/cn/iocoder/muse/module/content/dal/dataobject/ContentEventPublishOutboxDO.java

@@ -0,0 +1,54 @@
+package cn.iocoder.muse.module.content.dal.dataobject;
+
+import cn.iocoder.muse.framework.tenant.core.db.TenantBaseDO;
+import cn.iocoder.muse.module.content.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;
+
+/**
+ * Content source attribution 发布到统一 Events 的本域 outbox。
+ */
+@TableName(value = "muse_content_event_publish_outbox", autoResultMap = true)
+@Data
+@EqualsAndHashCode(callSuper = true)
+@ToString(callSuper = true)
+public class ContentEventPublishOutboxDO extends TenantBaseDO {
+
+    @TableId(type = IdType.AUTO)
+    private Long id;
+
+    private String outboxId;
+    private String commandId;
+    private String sourceCommandId;
+    private Long sourceAttributionId;
+    private Long ownerUserId;
+    private Long workId;
+    private Long blockId;
+    private Long blockRevision;
+    private String sourceStatus;
+    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;
+
+}

muse-cloud/muse-module-content/muse-module-content-server/src/main/java/cn/iocoder/muse/module/content/dal/mysql/ContentEventPublishOutboxMapper.java

@@ -0,0 +1,161 @@
+package cn.iocoder.muse.module.content.dal.mysql;
+
+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.content.dal.dataobject.ContentEventPublishOutboxDO;
+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;
+
+/**
+ * Content Events publish outbox Mapper。
+ */
+@Mapper
+public interface ContentEventPublishOutboxMapper extends BaseMapperX<ContentEventPublishOutboxDO> {
+
+    default ContentEventPublishOutboxDO selectByTenantIdAndOutboxId(Long tenantId, String outboxId) {
+        return selectOne(new LambdaQueryWrapperX<ContentEventPublishOutboxDO>()
+                .eq(ContentEventPublishOutboxDO::getTenantId, tenantId)
+                .eq(ContentEventPublishOutboxDO::getOutboxId, outboxId));
+    }
+
+    default ContentEventPublishOutboxDO selectByTenantIdAndCommandId(Long tenantId, String commandId) {
+        return selectOne(new LambdaQueryWrapperX<ContentEventPublishOutboxDO>()
+                .eq(ContentEventPublishOutboxDO::getTenantId, tenantId)
+                .eq(ContentEventPublishOutboxDO::getCommandId, commandId));
+    }
+
+    default ContentEventPublishOutboxDO selectByTenantIdAndBlockIdAndBlockRevisionAndNotificationType(
+            Long tenantId, Long blockId, Long blockRevision, String notificationType) {
+        return selectOne(new LambdaQueryWrapperX<ContentEventPublishOutboxDO>()
+                .eq(ContentEventPublishOutboxDO::getTenantId, tenantId)
+                .eq(ContentEventPublishOutboxDO::getBlockId, blockId)
+                .eq(ContentEventPublishOutboxDO::getBlockRevision, blockRevision)
+                .eq(ContentEventPublishOutboxDO::getNotificationType, notificationType));
+    }
+
+    /**
+     * 首次创建 Content Events publish outbox；重复 command 或 block revision notification 时返回 0。
+     */
+    @Insert("""
+            INSERT INTO muse_content_event_publish_outbox(tenant_id, outbox_id, command_id, source_command_id,
+                                                          source_attribution_id, owner_user_id, work_id, block_id,
+                                                          block_revision, source_status, 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}, #{sourceAttributionId},
+                    #{ownerUserId}, #{workId}, #{blockId}, #{blockRevision}, #{sourceStatus}, #{sourceRevision},
+                    #{eventType}, #{notificationType}, #{resourceRefType}, #{resourceRefId},
+                    #{payloadSummary,typeHandler=cn.iocoder.muse.module.content.dal.type.JsonbStringTypeHandler},
+                    #{publishStatus}, #{attemptCount}, #{maxAttempt}, #{nextRetryAt}, #{claimedAt},
+                    #{claimExpiresAt}, #{lastErrorCode}, #{lastErrorMessage}, #{publishedEventId},
+                    #{publishedSequenceNo}, #{createTime})
+            ON CONFLICT DO NOTHING
+            """)
+    int insertIgnore(ContentEventPublishOutboxDO outbox);
+
+    /**
+     * 原子领取下一条待发布 Content Events outbox。
+     */
+    @TenantIgnore
+    @Options(flushCache = Options.FlushCachePolicy.TRUE, useCache = false)
+    @Select("""
+            UPDATE muse_content_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_content_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 *
+            """)
+    ContentEventPublishOutboxDO claimNextPublishOutbox(long leaseSeconds);
+
+    @Update("""
+            UPDATE muse_content_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_content_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_content_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-content/muse-module-content-server/src/main/java/cn/iocoder/muse/module/content/framework/config/MuseContentEventsConfiguration.java

@@ -0,0 +1,12 @@
+package cn.iocoder.muse.module.content.framework.config;
+
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * Content Events 发布配置装配。
+ */
+@Configuration(proxyBeanMethods = false)
+@EnableConfigurationProperties(MuseContentEventsProperties.class)
+public class MuseContentEventsConfiguration {
+}

muse-cloud/muse-module-content/muse-module-content-server/src/main/java/cn/iocoder/muse/module/content/framework/config/MuseContentEventsProperties.java

@@ -0,0 +1,38 @@
+package cn.iocoder.muse.module.content.framework.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * Content 到统一 Events 的本域发布配置。
+ */
+@ConfigurationProperties(prefix = "muse.content.events")
+@Data
+public class MuseContentEventsProperties {
+
+    /**
+     * Content source attribution 到统一 Events 的本域 worker 配置。
+     */
+    private PublishWorker publishWorker = new PublishWorker();
+
+    @Data
+    public static class PublishWorker {
+
+        /**
+         * 是否启用 Content Events publish worker；默认关闭，避免未验收链路自动 claim。
+         */
+        private boolean enabled;
+
+        /**
+         * 单条 outbox 最大发布尝试次数；默认 5 次，便于运行时回滚和调优。
+         */
+        private int maxAttempt = 5;
+
+        /**
+         * 单次 claim 租约秒数；默认 60 秒，防止 worker 异常退出后长期占用任务。
+         */
+        private long claimTimeoutSeconds = 60L;
+
+    }
+
+}

muse-cloud/sql/muse/V21__extend_content_events_publish_outbox.sql

@@ -0,0 +1,64 @@
+-- P1R-7f Content source attribution event publish outbox Schema。
+-- 迁移边界：Content owner 只保存面向作品 owner 的 source_status_change 摘要；不保存正文、sourceSnapshot、授权快照、lineage、license 限制或错误栈。
+
+CREATE TABLE muse_content_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,
+    source_attribution_id BIGINT NOT NULL,
+    owner_user_id BIGINT NOT NULL,
+    work_id BIGINT NOT NULL,
+    block_id BIGINT NOT NULL,
+    block_revision BIGINT NOT NULL,
+    source_status VARCHAR(32) NOT NULL,
+    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_content_event_publish_outbox_id UNIQUE (tenant_id, outbox_id),
+    CONSTRAINT uk_muse_content_event_publish_outbox_command UNIQUE (tenant_id, command_id),
+    CONSTRAINT uk_muse_content_event_publish_outbox_block_notification UNIQUE (tenant_id, block_id, block_revision, notification_type),
+    CONSTRAINT chk_muse_content_event_publish_outbox_event_type CHECK (event_type = 'notification'),
+    CONSTRAINT chk_muse_content_event_publish_outbox_notification_type CHECK (notification_type = 'source_status_change'),
+    CONSTRAINT chk_muse_content_event_publish_outbox_source_status CHECK (source_status = 'active'),
+    CONSTRAINT chk_muse_content_event_publish_outbox_status CHECK (publish_status IN ('queued', 'running', 'retryable', 'published', 'dead_letter')),
+    CONSTRAINT chk_muse_content_event_publish_outbox_owner CHECK (owner_user_id > 0),
+    CONSTRAINT chk_muse_content_event_publish_outbox_block CHECK (block_id > 0),
+    CONSTRAINT chk_muse_content_event_publish_outbox_block_revision CHECK (block_revision > 0),
+    CONSTRAINT chk_muse_content_event_publish_outbox_resource_ref CHECK (resource_ref_id > 0),
+    CONSTRAINT chk_muse_content_event_publish_outbox_attempt_count CHECK (attempt_count >= 0),
+    CONSTRAINT chk_muse_content_event_publish_outbox_max_attempt CHECK (max_attempt > 0)
+);
+
+CREATE INDEX idx_muse_content_event_publish_outbox_claim
+    ON muse_content_event_publish_outbox(tenant_id, publish_status, next_retry_at, id)
+    WHERE deleted = FALSE;
+
+CREATE INDEX idx_muse_content_event_publish_outbox_owner_status
+    ON muse_content_event_publish_outbox(tenant_id, owner_user_id, publish_status, create_time);
+
+CREATE INDEX idx_muse_content_event_publish_outbox_block_revision
+    ON muse_content_event_publish_outbox(tenant_id, block_id, block_revision);
+
+CREATE TRIGGER trg_muse_content_event_publish_outbox_update_time
+    BEFORE UPDATE ON muse_content_event_publish_outbox
+    FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();