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

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

@@ -22,6 +22,11 @@
             <artifactId>muse-module-knowledge-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-infra-api</artifactId>

muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeEventPublishOutboxService.java

@@ -0,0 +1,22 @@
+package cn.iocoder.muse.module.knowledge.application.muse;
+
+import cn.iocoder.muse.module.knowledge.dal.dataobject.muse.MuseKnowledgeSourceEventDO;
+
+/**
+ * Knowledge source status 发布 outbox 服务。
+ */
+public interface MuseKnowledgeEventPublishOutboxService {
+
+    /**
+     * 为 canonical Knowledge source event 创建 per-owner Events publish outbox。
+     */
+    OwnerFanoutSummary createForSourceEvent(MuseKnowledgeSourceEventDO event);
+
+    record OwnerFanoutSummary(long totalProjectionCount,
+                              long validOwnerProjectionCount,
+                              long invalidOwnerProjectionCount,
+                              long distinctTargetOwnerCount,
+                              int queuedOutboxCount) {
+    }
+
+}

muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeEventPublishOutboxServiceImpl.java

@@ -0,0 +1,192 @@
+package cn.iocoder.muse.module.knowledge.application.muse;
+
+import cn.iocoder.muse.framework.common.util.json.JsonUtils;
+import cn.iocoder.muse.module.knowledge.dal.dataobject.muse.MuseKnowledgeEventPublishOutboxDO;
+import cn.iocoder.muse.module.knowledge.dal.dataobject.muse.MuseKnowledgeSourceBindingProjectionDO;
+import cn.iocoder.muse.module.knowledge.dal.dataobject.muse.MuseKnowledgeSourceEventDO;
+import cn.iocoder.muse.module.knowledge.dal.mysql.muse.MuseKnowledgeEventPublishOutboxMapper;
+import cn.iocoder.muse.module.knowledge.dal.mysql.muse.MuseKnowledgeSourceBindingProjectionMapper;
+import com.fasterxml.jackson.core.type.TypeReference;
+import jakarta.annotation.Resource;
+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.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Knowledge source status 发布 outbox 服务实现。
+ */
+@Service
+@Slf4j
+public class MuseKnowledgeEventPublishOutboxServiceImpl implements MuseKnowledgeEventPublishOutboxService {
+
+    static final String STATUS_QUEUED = "queued";
+    static final int DEFAULT_MAX_ATTEMPT = 5;
+
+    private static final String SOURCE_REVISION_NONE = "__none__";
+    private static final String EVENT_TYPE_NOTIFICATION = "notification";
+    private static final String NOTIFICATION_SOURCE_STATUS_CHANGE = "source_status_change";
+    private static final Set<String> PUBLISHABLE_KNOWLEDGE_EVENT_TYPES = Set.of(
+            "version_changed", "policy_changed", "kb_disabled", "kb_enabled", "status_changed");
+
+    @Resource
+    private MuseKnowledgeEventPublishOutboxMapper outboxMapper;
+    @Resource
+    private MuseKnowledgeSourceBindingProjectionMapper sourceBindingProjectionMapper;
+
+    @Override
+    public OwnerFanoutSummary createForSourceEvent(MuseKnowledgeSourceEventDO event) {
+        if (event == null || !StringUtils.hasText(event.getSourceEventId()) || event.getKbId() == null) {
+            return new OwnerFanoutSummary(0L, 0L, 0L, 0L, 0);
+        }
+        String knowledgeEventType = knowledgeEventType(event);
+        MuseKnowledgeSourceBindingProjectionMapper.FanoutOwnerAuditSummary auditSummary =
+                sourceBindingProjectionMapper.selectFanoutOwnerAuditSummaryByLastEventId(event.getKbId(),
+                        event.getSourceEventId());
+        if (!PUBLISHABLE_KNOWLEDGE_EVENT_TYPES.contains(knowledgeEventType)) {
+            log.info("Knowledge Events outbox 跳过非首批 source event, tenantId={}, kbId={}, sourceEventId={}, eventType={}",
+                    event.getTenantId(), event.getKbId(), event.getSourceEventId(), knowledgeEventType);
+            return ownerFanoutSummary(auditSummary, 0);
+        }
+
+        List<MuseKnowledgeSourceBindingProjectionDO> targets =
+                sourceBindingProjectionMapper.selectFanoutTargetsByLastEventId(event.getKbId(),
+                        event.getSourceEventId());
+        Map<Long, List<MuseKnowledgeSourceBindingProjectionDO>> grouped = targets.stream()
+                .collect(Collectors.groupingBy(MuseKnowledgeSourceBindingProjectionDO::getOwnerUserId,
+                        LinkedHashMap::new, Collectors.toList()));
+        int queued = 0;
+        for (Map.Entry<Long, List<MuseKnowledgeSourceBindingProjectionDO>> entry : grouped.entrySet()) {
+            MuseKnowledgeSourceBindingProjectionDO representative = entry.getValue().getFirst();
+            MuseKnowledgeEventPublishOutboxDO outbox = buildOutbox(event, knowledgeEventType, representative,
+                    entry.getValue().size());
+            int inserted = outboxMapper.insertIgnore(outbox);
+            if (inserted > 0) {
+                queued++;
+            }
+        }
+        if (auditSummary.invalidOwnerProjectionCount() > 0) {
+            // 非法 owner 不写 outbox，保留安全计数用于 projection task summary 和排障。
+            log.warn("Knowledge Events outbox 发现非法 owner projection，tenantId={}, kbId={}, sourceEventId={}, invalidOwnerProjectionCount={}",
+                    event.getTenantId(), event.getKbId(), event.getSourceEventId(),
+                    auditSummary.invalidOwnerProjectionCount());
+        }
+        return ownerFanoutSummary(auditSummary, queued);
+    }
+
+    private MuseKnowledgeEventPublishOutboxDO buildOutbox(MuseKnowledgeSourceEventDO event, String knowledgeEventType,
+                                                          MuseKnowledgeSourceBindingProjectionDO projection,
+                                                          int affectedOwnerProjectionCount) {
+        Long targetOwnerUserId = projection.getOwnerUserId();
+        String commandId = buildCommandId(event.getTenantId(), event.getCommandId(), event.getSourceEventId(),
+                targetOwnerUserId);
+        MuseKnowledgeEventPublishOutboxDO outbox = new MuseKnowledgeEventPublishOutboxDO();
+        outbox.setTenantId(event.getTenantId());
+        outbox.setOutboxId(buildOutboxId(event.getTenantId(), event.getSourceEventId(), targetOwnerUserId));
+        outbox.setSourceEventId(event.getSourceEventId());
+        outbox.setSourceFactOwnerUserId(event.getOwnerUserId());
+        outbox.setTargetOwnerUserId(targetOwnerUserId);
+        outbox.setKbId(event.getKbId());
+        outbox.setSourceType(event.getSourceType());
+        outbox.setSourceId(event.getSourceId());
+        outbox.setSourceRevision(normalizeSourceRevision(event.getSourceRevision()));
+        outbox.setSourceStatus(event.getStatus());
+        outbox.setActionPolicy(event.getActionPolicy());
+        outbox.setKnowledgeEventType(knowledgeEventType);
+        outbox.setEventType(EVENT_TYPE_NOTIFICATION);
+        outbox.setNotificationType(NOTIFICATION_SOURCE_STATUS_CHANGE);
+        outbox.setTargetWorkId(projection.getWorkId());
+        outbox.setBindingId(projection.getBindingId());
+        outbox.setProjectionId(projection.getProjectionId());
+        outbox.setAffectedOwnerProjectionCount(affectedOwnerProjectionCount);
+        outbox.setPayloadSummary(JsonUtils.toJsonString(notificationPayload(knowledgeEventType)));
+        outbox.setCommandId(commandId);
+        outbox.setPublishStatus(STATUS_QUEUED);
+        outbox.setAttemptCount(0);
+        outbox.setMaxAttempt(DEFAULT_MAX_ATTEMPT);
+        return outbox;
+    }
+
+    private Map<String, Object> notificationPayload(String knowledgeEventType) {
+        Map<String, Object> payload = new LinkedHashMap<>();
+        payload.put("type", NOTIFICATION_SOURCE_STATUS_CHANGE);
+        payload.put("message", messageFor(knowledgeEventType));
+        return payload;
+    }
+
+    private String knowledgeEventType(MuseKnowledgeSourceEventDO event) {
+        Map<String, Object> summary = parseMap(event.getEventSummary());
+        Object eventType = summary.get("eventType");
+        return eventType instanceof String value ? value : "";
+    }
+
+    private Map<String, Object> parseMap(String json) {
+        if (!StringUtils.hasText(json)) {
+            return Map.of();
+        }
+        try {
+            Map<String, Object> parsed = JsonUtils.parseObject(json, new TypeReference<Map<String, Object>>() {
+            });
+            return parsed == null ? Map.of() : parsed;
+        } catch (RuntimeException exception) {
+            log.warn("Knowledge Events outbox event_summary 解析失败，按不可发布处理");
+            return Map.of();
+        }
+    }
+
+    private String messageFor(String knowledgeEventType) {
+        return switch (knowledgeEventType) {
+            case "version_changed" -> "知识库版本变化，需要重新检查相关使用方";
+            case "policy_changed" -> "知识库访问策略变化，需要重新确认使用影响";
+            case "kb_disabled" -> "知识库已禁用，相关能力不可用";
+            case "kb_enabled" -> "知识库已恢复可用";
+            default -> "知识库状态已变化";
+        };
+    }
+
+    private String normalizeSourceRevision(String sourceRevision) {
+        return StringUtils.hasText(sourceRevision) ? sourceRevision : SOURCE_REVISION_NONE;
+    }
+
+    private OwnerFanoutSummary ownerFanoutSummary(
+            MuseKnowledgeSourceBindingProjectionMapper.FanoutOwnerAuditSummary auditSummary,
+            int queuedOutboxCount) {
+        return new OwnerFanoutSummary(auditSummary.totalProjectionCount(), auditSummary.validOwnerProjectionCount(),
+                auditSummary.invalidOwnerProjectionCount(), auditSummary.distinctTargetOwnerCount(),
+                queuedOutboxCount);
+    }
+
+    private String buildCommandId(Long tenantId, String sourceCommandId, String sourceEventId, Long targetOwnerUserId) {
+        String raw = String.valueOf(tenantId) + "|" + sourceCommandId + "|" + sourceEventId + "|"
+                + targetOwnerUserId + "|" + NOTIFICATION_SOURCE_STATUS_CHANGE;
+        return "kn_evt:" + sha256Hex(raw).substring(0, 32);
+    }
+
+    private String buildOutboxId(Long tenantId, String sourceEventId, Long targetOwnerUserId) {
+        String raw = String.valueOf(tenantId) + "|" + sourceEventId + "|" + targetOwnerUserId + "|"
+                + NOTIFICATION_SOURCE_STATUS_CHANGE;
+        return "kn_out:" + sha256Hex(raw).substring(0, 32);
+    }
+
+    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-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeEventPublishWorker.java

@@ -0,0 +1,251 @@
+package cn.iocoder.muse.module.knowledge.application.muse;
+
+import cn.iocoder.muse.framework.common.pojo.CommonResult;
+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.knowledge.dal.dataobject.muse.MuseKnowledgeEventPublishOutboxDO;
+import cn.iocoder.muse.module.knowledge.dal.mysql.muse.MuseKnowledgeEventPublishOutboxMapper;
+import cn.iocoder.muse.module.knowledge.framework.config.MuseKnowledgeEventsProperties;
+import com.fasterxml.jackson.core.type.TypeReference;
+import jakarta.annotation.Resource;
+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;
+
+/**
+ * Knowledge source status outbox 发布到统一 Events 的 worker。
+ */
+@Slf4j
+@Component
+public class MuseKnowledgeEventPublishWorker {
+
+    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_KNOWLEDGE = "knowledge";
+    private static final String SOURCE_TYPE_KNOWLEDGE_SOURCE_STATUS_OWNER = "knowledge_source_status_owner";
+    private static final String EVENT_TYPE_NOTIFICATION = "notification";
+    private static final String NOTIFICATION_TYPE_SOURCE_STATUS_CHANGE = "source_status_change";
+
+    private static final String ERROR_RETRY_EXHAUSTED = "KNOWLEDGE_EVENTS_PUBLISH_RETRY_EXHAUSTED";
+    private static final String ERROR_API_FAILED = "KNOWLEDGE_EVENTS_PUBLISH_API_FAILED";
+    private static final String ERROR_TEMPORARY_UNAVAILABLE = "KNOWLEDGE_EVENTS_PUBLISH_TEMPORARY_UNAVAILABLE";
+    private static final String ERROR_UNKNOWN_STATUS = "KNOWLEDGE_EVENTS_PUBLISH_UNKNOWN_STATUS";
+    private static final String ERROR_PAYLOAD_INVALID = "KNOWLEDGE_EVENTS_PUBLISH_PAYLOAD_INVALID";
+    private static final String ERROR_OUTBOX_INCOMPLETE = "KNOWLEDGE_EVENTS_OUTBOX_INCOMPLETE";
+
+    static final long CLAIM_LEASE_SECONDS = 60L;
+    static final long RETRY_BACKOFF_SECONDS = 60L;
+
+    @Resource
+    private MuseKnowledgeEventPublishOutboxMapper outboxMapper;
+    @Resource
+    private EventsPublishApi eventsPublishApi;
+    @Resource
+    private MuseKnowledgeEventsProperties properties;
+
+    public MuseKnowledgeEventPublishWorker() {
+    }
+
+    MuseKnowledgeEventPublishWorker(MuseKnowledgeEventPublishOutboxMapper outboxMapper,
+                                    EventsPublishApi eventsPublishApi,
+                                    MuseKnowledgeEventsProperties properties) {
+        this.outboxMapper = outboxMapper;
+        this.eventsPublishApi = eventsPublishApi;
+        this.properties = properties;
+    }
+
+    @Scheduled(initialDelayString = "${muse.knowledge.events.publish-worker.initial-delay-ms:1000}",
+            fixedDelayString = "${muse.knowledge.events.publish-worker.fixed-delay-ms:1000}")
+    public void dispatchScheduled() {
+        dispatchOnce();
+    }
+
+    public int dispatchOnce() {
+        if (!isEnabled()) {
+            return 0;
+        }
+        MuseKnowledgeEventPublishOutboxDO outbox = TenantUtils.executeIgnore(
+                () -> outboxMapper.claimNextPublishOutbox(CLAIM_LEASE_SECONDS));
+        if (outbox == null) {
+            return 0;
+        }
+        if (outbox.getTenantId() == null || outbox.getId() == null) {
+            markDeadLetter(outbox, ERROR_OUTBOX_INCOMPLETE, "Knowledge Events publish outbox incomplete");
+            return 0;
+        }
+        return TenantUtils.execute(outbox.getTenantId(), () -> publishClaimedOutbox(outbox));
+    }
+
+    private int publishClaimedOutbox(MuseKnowledgeEventPublishOutboxDO outbox) {
+        if (isRetryExhausted(outbox)) {
+            markDeadLetter(outbox, ERROR_RETRY_EXHAUSTED, "Knowledge 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("Knowledge Events publish 临时失败，tenantId={}, targetOwnerUserId={}, outboxId={}, sourceEventId={}, attempt={}, errorType={}",
+                    outbox.getTenantId(), outbox.getTargetOwnerUserId(), outbox.getId(), outbox.getSourceEventId(),
+                    outbox.getAttemptCount(), exception.getClass().getSimpleName());
+            if (isLastAttempt(outbox)) {
+                markDeadLetter(outbox, ERROR_RETRY_EXHAUSTED, "Knowledge Events publish retry exhausted");
+                return 0;
+            }
+            markRetryable(outbox, ERROR_TEMPORARY_UNAVAILABLE, "Knowledge Events publish temporary unavailable");
+            return 0;
+        }
+        if (result == null || result.isError()) {
+            if (isLastAttempt(outbox)) {
+                markDeadLetter(outbox, ERROR_RETRY_EXHAUSTED, "Knowledge Events publish retry exhausted");
+                return 0;
+            }
+            markRetryable(outbox, ERROR_API_FAILED, "Knowledge Events publish failed");
+            return 0;
+        }
+        return handlePublishResponse(outbox, result.getData());
+    }
+
+    private int handlePublishResponse(MuseKnowledgeEventPublishOutboxDO outbox, EventsPublishRespDTO respDTO) {
+        if (respDTO == null || !StringUtils.hasText(respDTO.getPublishStatus())) {
+            markDeadLetter(outbox, ERROR_UNKNOWN_STATUS, "Knowledge 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()),
+                    "Knowledge Events publish rejected");
+            return 1;
+        }
+        markDeadLetter(outbox, ERROR_UNKNOWN_STATUS, "Knowledge Events publish unknown status");
+        return 0;
+    }
+
+    private BuildPublishRequestResult buildReqDTO(MuseKnowledgeEventPublishOutboxDO outbox) {
+        Map<String, Object> payload = parsePayload(outbox);
+        if (!validPayload(payload) || outbox.getCreateTime() == null
+                || outbox.getTargetOwnerUserId() == null || outbox.getTargetOwnerUserId() <= 0
+                || !StringUtils.hasText(outbox.getSourceEventId())
+                || !StringUtils.hasText(outbox.getCommandId())) {
+            return BuildPublishRequestResult.failed(ERROR_PAYLOAD_INVALID, "Knowledge Events publish payload invalid");
+        }
+        EventsPublishReqDTO reqDTO = new EventsPublishReqDTO();
+        reqDTO.setCommandId(outbox.getCommandId());
+        reqDTO.setTenantId(outbox.getTenantId());
+        reqDTO.setOwnerUserId(outbox.getTargetOwnerUserId());
+        reqDTO.setSourceOwner(SOURCE_OWNER_KNOWLEDGE);
+        reqDTO.setSourceType(SOURCE_TYPE_KNOWLEDGE_SOURCE_STATUS_OWNER);
+        reqDTO.setSourceId(outbox.getSourceEventId() + ":owner:" + outbox.getTargetOwnerUserId());
+        reqDTO.setSourceRevision(outbox.getSourceRevision());
+        reqDTO.setEventType(EVENT_TYPE_NOTIFICATION);
+        reqDTO.setPayloadSummary(payload);
+        reqDTO.setEmittedAt(outbox.getCreateTime());
+        return BuildPublishRequestResult.success(reqDTO);
+    }
+
+    private Map<String, Object> parsePayload(MuseKnowledgeEventPublishOutboxDO outbox) {
+        try {
+            Map<String, Object> parsed = cn.iocoder.muse.framework.common.util.json.JsonUtils.getObjectMapper()
+                    .readValue(outbox.getPayloadSummary(), new TypeReference<Map<String, Object>>() {
+                    });
+            return parsed == null ? Map.of() : parsed;
+        } catch (Exception exception) {
+            log.warn("Knowledge Events publish payload 解析失败，tenantId={}, targetOwnerUserId={}, outboxId={}, sourceEventId={}, errorType={}",
+                    outbox.getTenantId(), outbox.getTargetOwnerUserId(), outbox.getId(), outbox.getSourceEventId(),
+                    exception.getClass().getSimpleName());
+            return Map.of();
+        }
+    }
+
+    private boolean validPayload(Map<String, Object> payload) {
+        return payload != null && payload.size() == 2
+                && NOTIFICATION_TYPE_SOURCE_STATUS_CHANGE.equals(payload.get("type"))
+                && payload.get("message") instanceof String;
+    }
+
+    private boolean isEnabled() {
+        return properties != null
+                && properties.getPublishWorker() != null
+                && properties.getPublishWorker().isEnabled();
+    }
+
+    private boolean isRetryExhausted(MuseKnowledgeEventPublishOutboxDO outbox) {
+        return outbox.getAttemptCount() != null && outbox.getMaxAttempt() != null
+                && outbox.getAttemptCount() > outbox.getMaxAttempt();
+    }
+
+    private boolean isLastAttempt(MuseKnowledgeEventPublishOutboxDO outbox) {
+        return outbox.getAttemptCount() != null && outbox.getMaxAttempt() != null
+                && outbox.getAttemptCount() >= outbox.getMaxAttempt();
+    }
+
+    private void markPublished(MuseKnowledgeEventPublishOutboxDO 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(MuseKnowledgeEventPublishOutboxDO 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(MuseKnowledgeEventPublishOutboxDO outbox, String errorCode, String errorMessage) {
+        if (outbox.getId() == null || outbox.getTenantId() == null) {
+            log.error("Knowledge 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(MuseKnowledgeEventPublishOutboxDO outbox, int updated, String targetStatus,
+                                          String errorCode) {
+        if (updated > 0) {
+            return;
+        }
+        // WHY：0 行通常表示租约过期后被其他 worker 重新领取，必须留下可检索证据，避免重复发布排障断链。
+        log.warn("Knowledge Events publish 终态回写被跳过，tenantId={}, targetOwnerUserId={}, outboxId={}, " +
+                        "sourceEventId={}, attempt={}, targetStatus={}, errorCode={}",
+                outbox.getTenantId(), outbox.getTargetOwnerUserId(), outbox.getId(), outbox.getSourceEventId(),
+                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-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeSourceEventService.java

@@ -57,6 +57,8 @@ public class MuseKnowledgeSourceEventService {
     private MuseKnowledgeProjectionTaskMapper projectionTaskMapper;
     @Resource
     private MuseKnowledgeCommandService commandService;
+    @Resource
+    private MuseKnowledgeEventPublishOutboxService eventPublishOutboxService;
 
     @Transactional(rollbackFor = Exception.class)
     public AdminKnowledgeTaskVO.SourceEventRespVO triggerGlobalKBSourceEvent(Long actorUserId, String apiVersion,
@@ -80,7 +82,9 @@ public class MuseKnowledgeSourceEventService {
         sourceEventMapper.insert(event);
         int affectedTargets = sourceBindingProjectionMapper.updateStatusByKbId(kbId, sourceStatus, actionPolicy,
                 event.getSourceEventId(), event.getSourceRevision(), projectionSummaryPatch(reqVO, event));
-        createProjectionTask(event, kb, reqVO, affectedTargets);
+        MuseKnowledgeEventPublishOutboxService.OwnerFanoutSummary ownerFanoutSummary =
+                eventPublishOutboxService.createForSourceEvent(event);
+        createProjectionTask(event, kb, reqVO, affectedTargets, ownerFanoutSummary);
 
         AdminKnowledgeTaskVO.SourceEventRespVO respVO = new AdminKnowledgeTaskVO.SourceEventRespVO();
         respVO.setEventId(event.getSourceEventId());
@@ -122,7 +126,8 @@ public class MuseKnowledgeSourceEventService {
     }
 
     private void createProjectionTask(MuseKnowledgeSourceEventDO event, MuseKnowledgeBaseDO kb,
-                                      AdminKnowledgeTaskVO.SourceEventReqVO reqVO, int affectedTargets) {
+                                      AdminKnowledgeTaskVO.SourceEventReqVO reqVO, int affectedTargets,
+                                      MuseKnowledgeEventPublishOutboxService.OwnerFanoutSummary ownerFanoutSummary) {
         MuseKnowledgeProjectionTaskDO task = new MuseKnowledgeProjectionTaskDO();
         task.setTaskId("projection-task-" + event.getSourceEventId());
         task.setSourceEventId(event.getSourceEventId());
@@ -138,7 +143,8 @@ public class MuseKnowledgeSourceEventService {
         task.setRetryable(false);
         task.setStartedAt(LocalDateTime.now());
         task.setFinishedAt(LocalDateTime.now());
-        task.setResultSummary(JsonUtils.toJsonString(projectionTaskSummary(event, kb, reqVO, affectedTargets)));
+        task.setResultSummary(JsonUtils.toJsonString(projectionTaskSummary(event, kb, reqVO, affectedTargets,
+                ownerFanoutSummary)));
         task.setTenantId(TenantContextHolder.getRequiredTenantId());
         projectionTaskMapper.insert(task);
     }
@@ -178,7 +184,9 @@ public class MuseKnowledgeSourceEventService {
 
     private Map<String, Object> projectionTaskSummary(MuseKnowledgeSourceEventDO event, MuseKnowledgeBaseDO kb,
                                                       AdminKnowledgeTaskVO.SourceEventReqVO reqVO,
-                                                      int affectedTargets) {
+                                                      int affectedTargets,
+                                                      MuseKnowledgeEventPublishOutboxService.OwnerFanoutSummary
+                                                              ownerFanoutSummary) {
         Map<String, Object> summary = new LinkedHashMap<>();
         summary.put("eventType", reqVO.getEventType());
         summary.put("reason", reqVO.getReason());
@@ -189,6 +197,14 @@ public class MuseKnowledgeSourceEventService {
         }
         summary.put("affectedTargets", affectedTargets);
         summary.put("affectedProjectionCount", affectedTargets);
+        if (ownerFanoutSummary != null) {
+            summary.put("ownerFanoutSummary", Map.of(
+                    "totalProjectionCount", ownerFanoutSummary.totalProjectionCount(),
+                    "validOwnerProjectionCount", ownerFanoutSummary.validOwnerProjectionCount(),
+                    "invalidOwnerProjectionCount", ownerFanoutSummary.invalidOwnerProjectionCount(),
+                    "distinctTargetOwnerCount", ownerFanoutSummary.distinctTargetOwnerCount(),
+                    "queuedOutboxCount", ownerFanoutSummary.queuedOutboxCount()));
+        }
         return summary;
     }
 

muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/dal/dataobject/muse/MuseKnowledgeEventPublishOutboxDO.java

@@ -0,0 +1,58 @@
+package cn.iocoder.muse.module.knowledge.dal.dataobject.muse;
+
+import cn.iocoder.muse.framework.tenant.core.db.TenantBaseDO;
+import cn.iocoder.muse.module.knowledge.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;
+
+/**
+ * Knowledge source status 发布到统一 Events 的本域 outbox。
+ */
+@TableName(value = "muse_knowledge_event_publish_outbox", autoResultMap = true)
+@Data
+@EqualsAndHashCode(callSuper = true)
+@ToString(callSuper = true)
+public class MuseKnowledgeEventPublishOutboxDO extends TenantBaseDO {
+
+    @TableId(type = IdType.AUTO)
+    private Long id;
+
+    private String outboxId;
+    private String sourceEventId;
+    private Long sourceFactOwnerUserId;
+    private Long targetOwnerUserId;
+    private Long kbId;
+    private String sourceType;
+    private String sourceId;
+    private String sourceRevision;
+    private String sourceStatus;
+    private String actionPolicy;
+    private String knowledgeEventType;
+    private String eventType;
+    private String notificationType;
+    private Long targetWorkId;
+    private Long bindingId;
+    private String projectionId;
+    private Integer affectedOwnerProjectionCount;
+    @TableField(typeHandler = JsonbStringTypeHandler.class)
+    private String payloadSummary;
+    private String commandId;
+    private String publishStatus;
+    private Integer attemptCount;
+    private Integer maxAttempt;
+    private LocalDateTime claimedAt;
+    private LocalDateTime claimExpiresAt;
+    private LocalDateTime nextRetryAt;
+    private String lastErrorCode;
+    private String lastErrorMessage;
+    private String publishedEventId;
+    private Long publishedSequenceNo;
+
+}

muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/dal/mysql/muse/MuseKnowledgeEventPublishOutboxMapper.java

@@ -0,0 +1,166 @@
+package cn.iocoder.muse.module.knowledge.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.knowledge.dal.dataobject.muse.MuseKnowledgeEventPublishOutboxDO;
+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;
+
+/**
+ * Knowledge Events publish outbox Mapper。
+ */
+@Mapper
+public interface MuseKnowledgeEventPublishOutboxMapper extends BaseMapperX<MuseKnowledgeEventPublishOutboxDO> {
+
+    default MuseKnowledgeEventPublishOutboxDO selectByOutboxId(Long tenantId, String outboxId) {
+        return selectOne(new LambdaQueryWrapperX<MuseKnowledgeEventPublishOutboxDO>()
+                .eq(MuseKnowledgeEventPublishOutboxDO::getTenantId, tenantId)
+                .eq(MuseKnowledgeEventPublishOutboxDO::getOutboxId, outboxId));
+    }
+
+    default MuseKnowledgeEventPublishOutboxDO selectByCommandId(Long tenantId, String commandId) {
+        return selectOne(new LambdaQueryWrapperX<MuseKnowledgeEventPublishOutboxDO>()
+                .eq(MuseKnowledgeEventPublishOutboxDO::getTenantId, tenantId)
+                .eq(MuseKnowledgeEventPublishOutboxDO::getCommandId, commandId));
+    }
+
+    default MuseKnowledgeEventPublishOutboxDO selectByOwnerSourceTuple(Long tenantId, String sourceEventId,
+                                                                       Long targetOwnerUserId,
+                                                                       String notificationType) {
+        return selectOne(new LambdaQueryWrapperX<MuseKnowledgeEventPublishOutboxDO>()
+                .eq(MuseKnowledgeEventPublishOutboxDO::getTenantId, tenantId)
+                .eq(MuseKnowledgeEventPublishOutboxDO::getSourceEventId, sourceEventId)
+                .eq(MuseKnowledgeEventPublishOutboxDO::getTargetOwnerUserId, targetOwnerUserId)
+                .eq(MuseKnowledgeEventPublishOutboxDO::getNotificationType, notificationType));
+    }
+
+    /**
+     * 首次创建 Knowledge Events publish outbox；重复 command 或 source/owner tuple 时返回 0。
+     */
+    @Insert("""
+            INSERT INTO muse_knowledge_event_publish_outbox(tenant_id, outbox_id, source_event_id,
+                                                            source_fact_owner_user_id, target_owner_user_id, kb_id,
+                                                            source_type, source_id, source_revision, source_status,
+                                                            action_policy, knowledge_event_type, event_type,
+                                                            notification_type, target_work_id, binding_id,
+                                                            projection_id, affected_owner_projection_count,
+                                                            payload_summary, command_id, publish_status,
+                                                            attempt_count, max_attempt, claimed_at,
+                                                            claim_expires_at, next_retry_at, last_error_code,
+                                                            last_error_message, published_event_id,
+                                                            published_sequence_no)
+            VALUES (#{tenantId}, #{outboxId}, #{sourceEventId}, #{sourceFactOwnerUserId}, #{targetOwnerUserId},
+                    #{kbId}, #{sourceType}, #{sourceId}, #{sourceRevision}, #{sourceStatus}, #{actionPolicy},
+                    #{knowledgeEventType}, #{eventType}, #{notificationType}, #{targetWorkId}, #{bindingId},
+                    #{projectionId}, #{affectedOwnerProjectionCount},
+                    #{payloadSummary,typeHandler=cn.iocoder.muse.module.knowledge.dal.type.JsonbStringTypeHandler},
+                    #{commandId}, #{publishStatus}, #{attemptCount}, #{maxAttempt}, #{claimedAt}, #{claimExpiresAt},
+                    #{nextRetryAt}, #{lastErrorCode}, #{lastErrorMessage}, #{publishedEventId},
+                    #{publishedSequenceNo})
+            ON CONFLICT DO NOTHING
+            """)
+    int insertIgnore(MuseKnowledgeEventPublishOutboxDO outbox);
+
+    /**
+     * 原子领取下一条待发布 Knowledge Events outbox。
+     */
+    @TenantIgnore
+    @Options(flushCache = Options.FlushCachePolicy.TRUE, useCache = false)
+    @Select("""
+            UPDATE muse_knowledge_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_knowledge_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 *
+            """)
+    MuseKnowledgeEventPublishOutboxDO claimNextPublishOutbox(long leaseSeconds);
+
+    @Update("""
+            UPDATE muse_knowledge_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_knowledge_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_knowledge_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-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/dal/mysql/muse/MuseKnowledgeSourceBindingProjectionMapper.java

@@ -48,6 +48,36 @@ public interface MuseKnowledgeSourceBindingProjectionMapper extends BaseMapperX<
                 .orderByDesc(MuseKnowledgeSourceBindingProjectionDO::getUpdateTime));
     }
 
+    default List<MuseKnowledgeSourceBindingProjectionDO> selectFanoutTargetsByLastEventId(Long kbId,
+                                                                                          String lastEventId) {
+        return selectList(new LambdaQueryWrapperX<MuseKnowledgeSourceBindingProjectionDO>()
+                .eq(MuseKnowledgeSourceBindingProjectionDO::getKbId, kbId)
+                .eq(MuseKnowledgeSourceBindingProjectionDO::getLastEventId, lastEventId)
+                .eq(MuseKnowledgeSourceBindingProjectionDO::getDeleted, false)
+                .gt(MuseKnowledgeSourceBindingProjectionDO::getOwnerUserId, 0L)
+                .orderByAsc(MuseKnowledgeSourceBindingProjectionDO::getOwnerUserId)
+                .orderByAsc(MuseKnowledgeSourceBindingProjectionDO::getBindingId)
+                .orderByAsc(MuseKnowledgeSourceBindingProjectionDO::getProjectionId));
+    }
+
+    default FanoutOwnerAuditSummary selectFanoutOwnerAuditSummaryByLastEventId(Long kbId, String lastEventId) {
+        List<MuseKnowledgeSourceBindingProjectionDO> projections = selectList(
+                new LambdaQueryWrapperX<MuseKnowledgeSourceBindingProjectionDO>()
+                        .eq(MuseKnowledgeSourceBindingProjectionDO::getKbId, kbId)
+                        .eq(MuseKnowledgeSourceBindingProjectionDO::getLastEventId, lastEventId)
+                        .eq(MuseKnowledgeSourceBindingProjectionDO::getDeleted, false));
+        long total = projections.size();
+        long valid = projections.stream()
+                .filter(projection -> projection.getOwnerUserId() != null && projection.getOwnerUserId() > 0)
+                .count();
+        long distinctOwners = projections.stream()
+                .map(MuseKnowledgeSourceBindingProjectionDO::getOwnerUserId)
+                .filter(ownerUserId -> ownerUserId != null && ownerUserId > 0)
+                .distinct()
+                .count();
+        return new FanoutOwnerAuditSummary(total, valid, total - valid, distinctOwners);
+    }
+
     default int updateStatusByKbId(Long kbId, String status, String actionPolicy, String lastEventId) {
         return updateStatusByKbId(kbId, status, actionPolicy, lastEventId, null, null);
     }
@@ -92,4 +122,10 @@ public interface MuseKnowledgeSourceBindingProjectionMapper extends BaseMapperX<
                 .eq(MuseKnowledgeSourceBindingProjectionDO::getStatus, status));
     }
 
+    record FanoutOwnerAuditSummary(long totalProjectionCount,
+                                   long validOwnerProjectionCount,
+                                   long invalidOwnerProjectionCount,
+                                   long distinctTargetOwnerCount) {
+    }
+
 }

muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/framework/config/MuseKnowledgeEventsConfiguration.java

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

muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/framework/config/MuseKnowledgeEventsProperties.java

@@ -0,0 +1,28 @@
+package cn.iocoder.muse.module.knowledge.framework.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * Knowledge 到统一 Events 的本域发布配置。
+ */
+@ConfigurationProperties(prefix = "muse.knowledge.events")
+@Data
+public class MuseKnowledgeEventsProperties {
+
+    /**
+     * Knowledge source status 到统一 Events 的本域 worker 配置。
+     */
+    private PublishWorker publishWorker = new PublishWorker();
+
+    @Data
+    public static class PublishWorker {
+
+        /**
+         * 是否启用 Knowledge Events publish worker；默认关闭，避免未验收链路自动 claim。
+         */
+        private boolean enabled;
+
+    }
+
+}

muse-cloud/sql/muse/V18__extend_knowledge_events_publish_outbox.sql

@@ -0,0 +1,71 @@
+-- P1R-7c Knowledge source owner event publish outbox Schema。
+-- 迁移边界：Knowledge owner 只保存面向目标 owner 的安全 notification 摘要；不保存 KB 内部原因、授权快照明文或外部服务凭据。
+
+CREATE TABLE muse_knowledge_event_publish_outbox (
+    id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+    tenant_id BIGINT NOT NULL,
+    outbox_id VARCHAR(128) NOT NULL,
+    -- source_event_id 引用 muse_knowledge_source_event.source_event_id 语义；引用完整性由 Knowledge 应用层在同事务内保证，避免跨切片数据库外键阻塞后续演进。
+    source_event_id VARCHAR(128) NOT NULL,
+    source_fact_owner_user_id BIGINT NOT NULL,
+    target_owner_user_id BIGINT NOT NULL,
+    kb_id BIGINT NOT NULL,
+    source_type VARCHAR(64) NOT NULL,
+    source_id VARCHAR(128) NOT NULL,
+    source_revision VARCHAR(80) NOT NULL,
+    source_status VARCHAR(32),
+    action_policy VARCHAR(64),
+    knowledge_event_type VARCHAR(64) NOT NULL,
+    event_type VARCHAR(32) NOT NULL,
+    notification_type VARCHAR(64) NOT NULL,
+    target_work_id BIGINT,
+    binding_id BIGINT,
+    projection_id VARCHAR(128),
+    affected_owner_projection_count INT NOT NULL DEFAULT 0,
+    payload_summary JSONB NOT NULL DEFAULT '{}'::jsonb,
+    command_id VARCHAR(128) NOT NULL,
+    publish_status VARCHAR(32) NOT NULL,
+    attempt_count INT NOT NULL DEFAULT 0,
+    max_attempt INT NOT NULL DEFAULT 5,
+    claimed_at TIMESTAMP,
+    claim_expires_at TIMESTAMP,
+    next_retry_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 '',
+    create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updater VARCHAR(64) NOT NULL DEFAULT '',
+    update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    deleted BOOLEAN NOT NULL DEFAULT FALSE,
+    CONSTRAINT uk_muse_knowledge_event_publish_outbox_id UNIQUE (tenant_id, outbox_id),
+    CONSTRAINT uk_muse_knowledge_event_publish_outbox_command UNIQUE (tenant_id, command_id),
+    CONSTRAINT uk_muse_knowledge_event_publish_outbox_owner_source UNIQUE (tenant_id, source_event_id, target_owner_user_id, notification_type),
+    CONSTRAINT chk_muse_knowledge_event_publish_outbox_event_type CHECK (event_type = 'notification'),
+    CONSTRAINT chk_muse_knowledge_event_publish_outbox_notification_type CHECK (notification_type = 'source_status_change'),
+    CONSTRAINT chk_muse_knowledge_event_publish_outbox_status CHECK (publish_status IN ('queued', 'running', 'published', 'retryable', 'dead_letter')),
+    CONSTRAINT chk_muse_knowledge_event_publish_outbox_attempt_count CHECK (attempt_count >= 0),
+    CONSTRAINT chk_muse_knowledge_event_publish_outbox_max_attempt CHECK (max_attempt > 0),
+    CONSTRAINT chk_muse_knowledge_event_publish_outbox_target_owner CHECK (target_owner_user_id > 0),
+    CONSTRAINT chk_muse_knowledge_event_publish_outbox_source_fact_owner CHECK (source_fact_owner_user_id >= 0),
+    CONSTRAINT chk_muse_knowledge_event_publish_outbox_projection_count CHECK (affected_owner_projection_count >= 0)
+);
+
+CREATE INDEX idx_muse_knowledge_event_publish_outbox_claim
+    ON muse_knowledge_event_publish_outbox(publish_status, claim_expires_at, next_retry_at, create_time, id)
+    WHERE deleted = FALSE;
+
+CREATE INDEX idx_muse_knowledge_event_publish_outbox_owner_status
+    ON muse_knowledge_event_publish_outbox(tenant_id, target_owner_user_id, publish_status, create_time);
+
+CREATE INDEX idx_muse_knowledge_event_publish_outbox_source_event
+    ON muse_knowledge_event_publish_outbox(tenant_id, source_event_id, publish_status);
+
+CREATE INDEX idx_muse_knowledge_source_projection_last_event
+    ON muse_knowledge_source_binding_projection(tenant_id, kb_id, last_event_id)
+    WHERE last_event_id IS NOT NULL;
+
+CREATE TRIGGER trg_muse_knowledge_event_publish_outbox_update_time
+    BEFORE UPDATE ON muse_knowledge_event_publish_outbox
+    FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();