暂存提交

2026-08-19 16:24:02 +08:00
parent f5c4bf130a
commit 6acb6178da
354 changed files with 25000 additions and 198 deletions
@@ -1,7 +1,5 @@
 package com.baiyue.eventbridge.service;

-import jakarta.annotation.Resource;
-
 import com.baiyue.common.event.EventEnvelope;
 import com.baiyue.common.event.EventPayloadCanonicalizer;
 import com.baiyue.common.event.EventPayloadHash;
@@ -13,21 +11,20 @@ import org.springframework.beans.factory.annotation.Value;
 import org.springframework.dao.DuplicateKeyException;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
+import jakarta.annotation.Resource;

 import java.time.LocalDateTime;
-import java.util.Objects;

 /**
 * 入站事件校验与幂等落库服务。
 */
@Service
 public class EventInboxService {
-    @Resource
-    private InboxStore inboxStore;
-    @Resource
-    private ObjectMapper objectMapper;
+    @Resource private InboxStore inboxStore;
+    @Resource private ObjectMapper objectMapper;
+
    @Value("${baiyue.region:CN}")
-    private  String localRegion;
+    private String localRegion;

    @Transactional
    public EventReceiveResult receive(EventEnvelope envelope) {
@@ -61,24 +58,9 @@ public class EventInboxService {
        if (!existing.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) {
            throw new IllegalArgumentException("相同 eventId 对应的 payloadHash 不一致，拒绝覆盖既有事件");
        }
-        if (!hasSameEnvelopeIdentity(existing, envelope)) {
-            throw new IllegalArgumentException("相同 eventId 对应的客户或事件信封身份不一致，拒绝覆盖既有事件");
-        }
        return new EventReceiveResult(envelope.eventId(), "重复", existing.getReceivedAt());
    }

-    private boolean hasSameEnvelopeIdentity(BridgeInbox existing, EventEnvelope envelope) {
-        return Objects.equals(existing.getTenantId(), envelope.tenantId())
-                && Objects.equals(existing.getEventType(), envelope.eventType())
-                && Objects.equals(existing.getEventVersion(), envelope.eventVersion())
-                && Objects.equals(existing.getSourceRegion(), envelope.sourceRegion().name())
-                && Objects.equals(existing.getTargetRegion(), envelope.targetRegion().name())
-                && Objects.equals(existing.getAggregateType(), envelope.aggregateType())
-                && Objects.equals(existing.getAggregateId(), envelope.aggregateId())
-                && Objects.equals(existing.getAggregateVersion(), envelope.aggregateVersion())
-                && Objects.equals(existing.getHandoverVersion(), envelope.handoverVersion());
-    }
-
    private BridgeInbox toInbox(EventEnvelope envelope) {
        BridgeInbox inbox = new BridgeInbox();
        inbox.setEventId(envelope.eventId());
@@ -87,7 +69,7 @@ public class EventInboxService {
        inbox.setEventVersion(envelope.eventVersion());
        inbox.setSourceRegion(envelope.sourceRegion().name());
        inbox.setTargetRegion(envelope.targetRegion().name());
-        inbox.setTenantId(envelope.tenantId());
+        inbox.setCustomerId(envelope.tenantId());
        inbox.setAggregateType(envelope.aggregateType());
        inbox.setAggregateId(envelope.aggregateId());
        inbox.setAggregateVersion(envelope.aggregateVersion());
@@ -0,0 +1,27 @@
+package com.baiyue.eventbridge.service;
+
+import com.baiyue.eventbridge.mapper.BridgeInboxMapper;
+import jakarta.annotation.Resource;
+import org.springframework.stereotype.Service;
+
+@Service
+public class EventReplayService {
+    @Resource private BridgeInboxMapper inboxMapper;
+
+    public void replayDead(String eventId, String operatorId, String reason) {
+        if (operatorId == null || operatorId.isBlank()) {
+            throw new IllegalArgumentException("人工补投必须提供操作人");
+        }
+        if (reason == null || reason.isBlank()) {
+            throw new IllegalArgumentException("人工补投必须提供原因");
+        }
+        String auditReason = abbreviate("manual replay by " + operatorId + ": " + reason, 1000);
+        if (inboxMapper.replayDead(eventId, auditReason) != 1) {
+            throw new IllegalStateException("事件不存在或未处于 DEAD 状态");
+        }
+    }
+
+    private static String abbreviate(String value, int maxLength) {
+        return value.length() <= maxLength ? value : value.substring(0, maxLength);
+    }
+}
@@ -0,0 +1,55 @@
+package com.baiyue.eventbridge.service;
+
+import com.baiyue.common.event.EventEnvelope;
+import com.baiyue.common.event.RegionCode;
+import com.baiyue.eventbridge.config.EventDispatcherProperties;
+import com.baiyue.eventbridge.domain.BridgeInbox;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.Resource;
+
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+
+@Component
+public class HttpLastlegEventClient implements LastlegEventClient {
+    @Resource private ObjectMapper objectMapper;
+    @Resource private EventDispatcherProperties properties;
+
+    @Value("${baiyue.security.internal-service-token:}")
+    private String serviceToken;
+
+    private HttpClient httpClient;
+
+    @PostConstruct
+    void initializeHttpClient() {
+        this.httpClient = HttpClient.newBuilder().connectTimeout(properties.connectTimeout()).build();
+    }
+
+    @Override
+    public String deliver(BridgeInbox row) throws Exception {
+        EventEnvelope envelope = new EventEnvelope(row.getEventId(), row.getEventType(), row.getEventVersion(),
+                row.getOccurredAt(), RegionCode.valueOf(row.getSourceRegion()), RegionCode.valueOf(row.getTargetRegion()),
+                row.getCustomerId(), row.getAggregateType(), row.getAggregateId(), row.getAggregateVersion(),
+                row.getHandoverVersion(), row.getTraceId(), row.getCorrelationId(), row.getCausationId(),
+                objectMapper.readTree(row.getPayloadJson()), row.getPayloadHash());
+        String body = objectMapper.writeValueAsString(envelope);
+        HttpRequest request = HttpRequest.newBuilder(properties.consumerUrl())
+                .timeout(properties.requestTimeout())
+                .header("Content-Type", "application/json")
+                .header("X-Internal-Service-Token", serviceToken)
+                .POST(HttpRequest.BodyPublishers.ofString(body))
+                .build();
+        HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
+        JsonNode responseBody = objectMapper.readTree(response.body());
+        if (response.statusCode() / 100 != 2 || responseBody == null
+                || !"200".equals(responseBody.path("code").asText())) {
+            throw new IllegalStateException("尾程消费失败: HTTP " + response.statusCode());
+        }
+        return String.valueOf(response.statusCode());
+    }
+}
@@ -0,0 +1,7 @@
+package com.baiyue.eventbridge.service;
+
+import com.baiyue.eventbridge.domain.BridgeInbox;
+
+public interface LastlegEventClient {
+    String deliver(BridgeInbox event) throws Exception;
+}
@@ -2,16 +2,15 @@ package com.baiyue.eventbridge.service;

 import com.baiyue.eventbridge.domain.BridgeInbox;
 import com.baiyue.eventbridge.mapper.BridgeInboxMapper;
-import lombok.RequiredArgsConstructor;
+import jakarta.annotation.Resource;
 import org.springframework.stereotype.Repository;

 /**
 * 基于 MyBatis-Plus 的入站幂等存储。
 */
@Repository
-@RequiredArgsConstructor
 public class MybatisInboxStore implements InboxStore {
-    private final BridgeInboxMapper mapper;
+    @Resource private BridgeInboxMapper mapper;

    @Override
    public BridgeInbox findByEventId(String eventId) {
@@ -0,0 +1,88 @@
+package com.baiyue.eventbridge.service;
+
+import com.baiyue.eventbridge.config.EventDispatcherProperties;
+import com.baiyue.eventbridge.domain.BridgeDeliveryAttempt;
+import com.baiyue.eventbridge.domain.BridgeInbox;
+import com.baiyue.eventbridge.mapper.BridgeDeliveryAttemptMapper;
+import com.baiyue.eventbridge.mapper.BridgeInboxMapper;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import java.lang.management.ManagementFactory;
+import java.time.LocalDateTime;
+import java.util.UUID;
+
+@Component
+@Slf4j
+public class OverseasEventDispatcher {
+    @Resource private BridgeInboxMapper inboxMapper;
+    @Resource private BridgeDeliveryAttemptMapper attemptMapper;
+    @Resource private LastlegEventClient eventClient;
+    @Resource private EventDispatcherProperties properties;
+    private final String leaseOwner = ManagementFactory.getRuntimeMXBean().getName() + ":" + UUID.randomUUID();
+
+    @Scheduled(fixedDelayString = "${baiyue.event-dispatcher.fixed-delay:3s}")
+    public void dispatch() {
+        for (String eventId : inboxMapper.dispatchCandidates(properties.batchSize())) {
+            dispatchOne(eventId);
+        }
+    }
+
+    void dispatchOne(String eventId) {
+        if (inboxMapper.claim(eventId, leaseOwner, properties.leaseDuration().toSeconds()) != 1) {
+            return;
+        }
+
+        BridgeInbox event = inboxMapper.selectById(eventId);
+        int retryAttempt = event.getRetryCount() == null ? 1 : event.getRetryCount() + 1;
+        int deliveryAttemptNo = inboxMapper.nextAttemptNo(eventId);
+        BridgeDeliveryAttempt attempt = startAttempt(event, deliveryAttemptNo);
+        try {
+            String responseCode = eventClient.deliver(event);
+            if (inboxMapper.markProcessed(eventId, leaseOwner) != 1) {
+                throw new IllegalStateException("事件租约已失效，无法确认投递结果");
+            }
+            finishAttempt(attempt, "DELIVERED", responseCode, null);
+        } catch (Exception exception) {
+            String error = abbreviate(exception.getMessage(), 1000);
+            inboxMapper.markFailed(eventId, leaseOwner, properties.maxAttempts(), backoffSeconds(retryAttempt), error);
+            finishAttempt(attempt, retryAttempt >= properties.maxAttempts() ? "DEAD" : "RETRY", null, error);
+            log.warn("尾程事件投递失败 eventId={}, retryAttempt={}, deliveryAttempt={}, reason={}",
+                    eventId, retryAttempt, deliveryAttemptNo, error);
+        }
+    }
+
+    private BridgeDeliveryAttempt startAttempt(BridgeInbox event, int attemptNo) {
+        BridgeDeliveryAttempt attempt = new BridgeDeliveryAttempt();
+        attempt.setEventId(event.getEventId());
+        attempt.setSourceRegion(event.getSourceRegion());
+        attempt.setTargetRegion(event.getTargetRegion());
+        attempt.setAttemptNo(attemptNo);
+        attempt.setStatus("PROCESSING");
+        attempt.setStartedAt(LocalDateTime.now());
+        attemptMapper.insert(attempt);
+        return attempt;
+    }
+
+    private void finishAttempt(BridgeDeliveryAttempt attempt, String status, String responseCode, String error) {
+        attempt.setStatus(status);
+        attempt.setResponseCode(responseCode);
+        attempt.setErrorMessage(error);
+        attempt.setFinishedAt(LocalDateTime.now());
+        attemptMapper.updateById(attempt);
+    }
+
+    private long backoffSeconds(int attemptNo) {
+        long initial = properties.initialBackoff().toSeconds();
+        int shift = Math.min(Math.max(attemptNo - 1, 0), 20);
+        long delay = initial > Long.MAX_VALUE >> shift ? Long.MAX_VALUE : initial << shift;
+        return Math.min(delay, properties.maxBackoff().toSeconds());
+    }
+
+    private static String abbreviate(String value, int maxLength) {
+        String safe = value == null || value.isBlank() ? "未知投递错误" : value;
+        return safe.length() <= maxLength ? safe : safe.substring(0, maxLength);
+    }
+}