暂存提交

2026-07-30 18:01:36 +08:00
commit 2f4e159356
368 changed files with 20370 additions and 0 deletions
@@ -0,0 +1,93 @@
+package com.baiyue.eventbridge.service;
+
+import com.baiyue.common.event.EventEnvelope;
+import com.baiyue.common.event.EventPayloadCanonicalizer;
+import com.baiyue.common.event.EventPayloadHash;
+import com.baiyue.eventbridge.domain.BridgeInbox;
+import com.baiyue.eventbridge.dto.EventReceiveResult;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.LocalDateTime;
+
+/**
+ * 入站事件校验与幂等落库服务。
+ */
+@Service
+public class EventInboxService {
+    private final InboxStore inboxStore;
+    private final ObjectMapper objectMapper;
+    private final String localRegion;
+
+    public EventInboxService(InboxStore inboxStore, ObjectMapper objectMapper,
+                             @Value("${baiyue.region:CN}") String localRegion) {
+        this.inboxStore = inboxStore;
+        this.objectMapper = objectMapper;
+        this.localRegion = localRegion;
+    }
+
+    @Transactional
+    public EventReceiveResult receive(EventEnvelope envelope) {
+        envelope.requireValid();
+        if (!envelope.targetRegion().name().equalsIgnoreCase(localRegion)) {
+            throw new IllegalArgumentException("事件目标区域与当前事件桥区域不一致");
+        }
+        if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(envelope.payload()), envelope.payloadHash())) {
+            throw new IllegalArgumentException("payloadHash 与规范化事件负载不一致");
+        }
+        BridgeInbox existing = inboxStore.findByEventId(envelope.eventId());
+        if (existing != null) {
+            return duplicateOrConflict(existing, envelope);
+        }
+        BridgeInbox inbox = toInbox(envelope);
+        try {
+            inboxStore.insert(inbox);
+            return new EventReceiveResult(envelope.eventId(), "已接收", inbox.getReceivedAt());
+        } catch (DuplicateKeyException exception) {
+            BridgeInbox concurrent = inboxStore.findByEventId(envelope.eventId());
+            if (concurrent == null) {
+                throw exception;
+            }
+            return duplicateOrConflict(concurrent, envelope);
+        }
+    }
+
+    private EventReceiveResult duplicateOrConflict(BridgeInbox existing, EventEnvelope envelope) {
+        if (!existing.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) {
+            throw new IllegalArgumentException("相同 eventId 对应的 payloadHash 不一致，拒绝覆盖既有事件");
+        }
+        return new EventReceiveResult(envelope.eventId(), "重复", existing.getReceivedAt());
+    }
+
+    private BridgeInbox toInbox(EventEnvelope envelope) {
+        BridgeInbox inbox = new BridgeInbox();
+        inbox.setEventId(envelope.eventId());
+        inbox.setIdempotencyKey(envelope.sourceRegion() + ":" + envelope.eventId());
+        inbox.setEventType(envelope.eventType());
+        inbox.setEventVersion(envelope.eventVersion());
+        inbox.setSourceRegion(envelope.sourceRegion().name());
+        inbox.setTargetRegion(envelope.targetRegion().name());
+        inbox.setTenantId(envelope.tenantId());
+        inbox.setAggregateType(envelope.aggregateType());
+        inbox.setAggregateId(envelope.aggregateId());
+        inbox.setAggregateVersion(envelope.aggregateVersion());
+        inbox.setHandoverVersion(envelope.handoverVersion());
+        inbox.setTraceId(envelope.traceId());
+        inbox.setCorrelationId(envelope.correlationId());
+        inbox.setCausationId(envelope.causationId());
+        inbox.setOccurredAt(envelope.occurredAt());
+        inbox.setPayloadHash(envelope.payloadHash().toLowerCase(java.util.Locale.ROOT));
+        try {
+            inbox.setPayloadJson(objectMapper.writeValueAsString(envelope.payload()));
+        } catch (JsonProcessingException exception) {
+            throw new IllegalArgumentException("事件负载无法序列化", exception);
+        }
+        inbox.setStatus("RECEIVED");
+        inbox.setReceivedAt(LocalDateTime.now());
+        return inbox;
+    }
+}