diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/pom.xml b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/pom.xml index d6846abc..adec6d04 100644 --- a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/pom.xml +++ b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/pom.xml @@ -22,6 +22,11 @@ muse-module-knowledge-api ${revision} + + cn.iocoder.cloud + muse-module-events-api + ${revision} + cn.iocoder.cloud muse-module-infra-api diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeEventPublishOutboxService.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeEventPublishOutboxService.java new file mode 100644 index 00000000..7ec7b92e --- /dev/null +++ b/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) { + } + +} diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeEventPublishOutboxServiceImpl.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeEventPublishOutboxServiceImpl.java new file mode 100644 index 00000000..63304db1 --- /dev/null +++ b/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 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 targets = + sourceBindingProjectionMapper.selectFanoutTargetsByLastEventId(event.getKbId(), + event.getSourceEventId()); + Map> grouped = targets.stream() + .collect(Collectors.groupingBy(MuseKnowledgeSourceBindingProjectionDO::getOwnerUserId, + LinkedHashMap::new, Collectors.toList())); + int queued = 0; + for (Map.Entry> 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 notificationPayload(String knowledgeEventType) { + Map payload = new LinkedHashMap<>(); + payload.put("type", NOTIFICATION_SOURCE_STATUS_CHANGE); + payload.put("message", messageFor(knowledgeEventType)); + return payload; + } + + private String knowledgeEventType(MuseKnowledgeSourceEventDO event) { + Map summary = parseMap(event.getEventSummary()); + Object eventType = summary.get("eventType"); + return eventType instanceof String value ? value : ""; + } + + private Map parseMap(String json) { + if (!StringUtils.hasText(json)) { + return Map.of(); + } + try { + Map parsed = JsonUtils.parseObject(json, new TypeReference>() { + }); + 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); + } + } + +} diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeEventPublishWorker.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeEventPublishWorker.java new file mode 100644 index 00000000..318be6b7 --- /dev/null +++ b/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 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 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 parsePayload(MuseKnowledgeEventPublishOutboxDO outbox) { + try { + Map parsed = cn.iocoder.muse.framework.common.util.json.JsonUtils.getObjectMapper() + .readValue(outbox.getPayloadSummary(), new TypeReference>() { + }); + 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 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); + } + + } + +} diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeSourceEventService.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeSourceEventService.java index 32331753..25d7723a 100644 --- a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeSourceEventService.java +++ b/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 projectionTaskSummary(MuseKnowledgeSourceEventDO event, MuseKnowledgeBaseDO kb, AdminKnowledgeTaskVO.SourceEventReqVO reqVO, - int affectedTargets) { + int affectedTargets, + MuseKnowledgeEventPublishOutboxService.OwnerFanoutSummary + ownerFanoutSummary) { Map 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; } diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/dal/dataobject/muse/MuseKnowledgeEventPublishOutboxDO.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/dal/dataobject/muse/MuseKnowledgeEventPublishOutboxDO.java new file mode 100644 index 00000000..42561c1d --- /dev/null +++ b/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; + +} diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/dal/mysql/muse/MuseKnowledgeEventPublishOutboxMapper.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/dal/mysql/muse/MuseKnowledgeEventPublishOutboxMapper.java new file mode 100644 index 00000000..e30cb8e4 --- /dev/null +++ b/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 { + + default MuseKnowledgeEventPublishOutboxDO selectByOutboxId(Long tenantId, String outboxId) { + return selectOne(new LambdaQueryWrapperX() + .eq(MuseKnowledgeEventPublishOutboxDO::getTenantId, tenantId) + .eq(MuseKnowledgeEventPublishOutboxDO::getOutboxId, outboxId)); + } + + default MuseKnowledgeEventPublishOutboxDO selectByCommandId(Long tenantId, String commandId) { + return selectOne(new LambdaQueryWrapperX() + .eq(MuseKnowledgeEventPublishOutboxDO::getTenantId, tenantId) + .eq(MuseKnowledgeEventPublishOutboxDO::getCommandId, commandId)); + } + + default MuseKnowledgeEventPublishOutboxDO selectByOwnerSourceTuple(Long tenantId, String sourceEventId, + Long targetOwnerUserId, + String notificationType) { + return selectOne(new LambdaQueryWrapperX() + .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); + +} diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/dal/mysql/muse/MuseKnowledgeSourceBindingProjectionMapper.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/dal/mysql/muse/MuseKnowledgeSourceBindingProjectionMapper.java index 76ee42a3..15de2cdd 100644 --- a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/dal/mysql/muse/MuseKnowledgeSourceBindingProjectionMapper.java +++ b/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 selectFanoutTargetsByLastEventId(Long kbId, + String lastEventId) { + return selectList(new LambdaQueryWrapperX() + .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 projections = selectList( + new LambdaQueryWrapperX() + .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) { + } + } diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/framework/config/MuseKnowledgeEventsConfiguration.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/framework/config/MuseKnowledgeEventsConfiguration.java new file mode 100644 index 00000000..bbcd4f9d --- /dev/null +++ b/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 { +} diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/framework/config/MuseKnowledgeEventsProperties.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/framework/config/MuseKnowledgeEventsProperties.java new file mode 100644 index 00000000..864bf862 --- /dev/null +++ b/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; + + } + +} diff --git a/muse-cloud/sql/muse/V18__extend_knowledge_events_publish_outbox.sql b/muse-cloud/sql/muse/V18__extend_knowledge_events_publish_outbox.sql new file mode 100644 index 00000000..1371a3e9 --- /dev/null +++ b/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();