提交

event-bridge-service/src/main/java/com/baiyue/eventbridge/EventBridgeApplication.java

@@ -1,14 +1,19 @@
 package com.baiyue.eventbridge;
 
+import com.baiyue.eventbridge.config.EventDispatcherProperties;
 import org.mybatis.spring.annotation.MapperScan;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.scheduling.annotation.EnableScheduling;
 
 /**
  * 跨区域事件桥启动入口。
  */
 @MapperScan("com.baiyue.eventbridge.mapper")
 @SpringBootApplication
+@EnableScheduling
+@EnableConfigurationProperties(EventDispatcherProperties.class)
 public class EventBridgeApplication {
     public static void main(String[] args) {
         SpringApplication.run(EventBridgeApplication.class, args);

event-bridge-service/src/main/java/com/baiyue/eventbridge/config/EventDispatcherProperties.java

@@ -0,0 +1,34 @@
+package com.baiyue.eventbridge.config;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+import java.net.URI;
+import java.time.Duration;
+
+@ConfigurationProperties(prefix = "baiyue.event-dispatcher")
+public record EventDispatcherProperties(
+        URI consumerUrl,
+        int batchSize,
+        int maxAttempts,
+        Duration leaseDuration,
+        Duration initialBackoff,
+        Duration maxBackoff,
+        Duration connectTimeout,
+        Duration requestTimeout) {
+
+    public EventDispatcherProperties {
+        consumerUrl = consumerUrl == null
+                ? URI.create("http://127.0.0.1:9303/internal/lastleg/events") : consumerUrl;
+        batchSize = batchSize <= 0 ? 50 : batchSize;
+        maxAttempts = maxAttempts <= 0 ? 10 : maxAttempts;
+        leaseDuration = positiveOrDefault(leaseDuration, Duration.ofSeconds(30));
+        initialBackoff = positiveOrDefault(initialBackoff, Duration.ofSeconds(5));
+        maxBackoff = positiveOrDefault(maxBackoff, Duration.ofMinutes(15));
+        connectTimeout = positiveOrDefault(connectTimeout, Duration.ofSeconds(3));
+        requestTimeout = positiveOrDefault(requestTimeout, Duration.ofSeconds(10));
+    }
+
+    private static Duration positiveOrDefault(Duration value, Duration fallback) {
+        return value == null || value.isNegative() || value.isZero() ? fallback : value;
+    }
+}

event-bridge-service/src/main/java/com/baiyue/eventbridge/controller/EventBridgeController.java

@@ -4,13 +4,15 @@ import com.baiyue.common.core.api.ApiResponse;
 import com.baiyue.common.event.EventEnvelope;
 import com.baiyue.eventbridge.dto.BridgeStatus;
 import com.baiyue.eventbridge.dto.EventReceiveResult;
+import com.baiyue.eventbridge.dto.EventReplayRequest;
 import com.baiyue.eventbridge.service.EventInboxService;
+import com.baiyue.eventbridge.service.EventReplayService;
 import com.baiyue.eventbridge.security.InternalServiceTokenGuard;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
 import io.swagger.v3.oas.annotations.responses.ApiResponses;
 import io.swagger.v3.oas.annotations.tags.Tag;
-import lombok.RequiredArgsConstructor;
+import jakarta.annotation.Resource;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PostMapping;
@@ -18,6 +20,7 @@ import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestHeader;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.bind.annotation.PathVariable;
 
 /**
  * 事件桥内部接口。生产环境必须在网关或基础设施层增加 mTLS 与来源白名单。
@@ -25,12 +28,12 @@ import org.springframework.web.bind.annotation.RestController;
 @Tag(name = "事件桥内部接口", description = "跨区域事件的幂等接收与能力状态")
 @RestController
 @RequestMapping("/internal/event-bridge")
-@RequiredArgsConstructor
 public class EventBridgeController {
-    private final EventInboxService eventInboxService;
-    private final InternalServiceTokenGuard internalServiceTokenGuard;
+    @Resource private EventInboxService eventInboxService;
+    @Resource private EventReplayService eventReplayService;
+    @Resource private InternalServiceTokenGuard internalServiceTokenGuard;
 
-    @Value("${baiyue.region:CN}")
+    @Value("${baiyue.region:OVERSEAS}")
     private String region;
 
     @Operation(summary = "幂等接收事件", description = "当前仅完成数据库接收骨架，不代表生产跨区传输和 mTLS 已开通")
@@ -48,6 +51,18 @@ public class EventBridgeController {
         return ApiResponse.success(eventInboxService.receive(envelope));
     }
 
+    @Operation(summary = "人工补投死信事件")
+    @PostMapping("/events/{eventId}/replay")
+    public ApiResponse<Void> replay(
+            @RequestHeader(value = "X-Internal-Service-Token", required = false) String token,
+            @RequestHeader("X-Operator-Id") String operatorId,
+            @PathVariable String eventId,
+            @RequestBody EventReplayRequest request) {
+        internalServiceTokenGuard.verify(token);
+        eventReplayService.replayDead(eventId, operatorId, request.reason());
+        return ApiResponse.success(null);
+    }
+
     @Operation(summary = "查询事件桥能力状态")
     @ApiResponses(@io.swagger.v3.oas.annotations.responses.ApiResponse(responseCode = "200", description = "查询成功"))
     @GetMapping("/status")

event-bridge-service/src/main/java/com/baiyue/eventbridge/domain/BridgeDeliveryAttempt.java

@@ -0,0 +1,24 @@
+package com.baiyue.eventbridge.domain;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+
+import java.time.LocalDateTime;
+
+@Data
+@TableName("bridge_delivery_attempt")
+public class BridgeDeliveryAttempt {
+    @TableId(type = IdType.AUTO)
+    private Long id;
+    private String eventId;
+    private String sourceRegion;
+    private String targetRegion;
+    private Integer attemptNo;
+    private String status;
+    private String responseCode;
+    private String errorMessage;
+    private LocalDateTime startedAt;
+    private LocalDateTime finishedAt;
+}

event-bridge-service/src/main/java/com/baiyue/eventbridge/domain/BridgeInbox.java

@@ -32,5 +32,11 @@ public class BridgeInbox {
     private String payloadHash;
     private String payloadJson;
     private String status;
+    private String leaseOwner;
+    private LocalDateTime leaseUntil;
+    private Integer retryCount;
+    private LocalDateTime nextRetryAt;
+    private String lastError;
     private LocalDateTime receivedAt;
+    private LocalDateTime processedAt;
 }

event-bridge-service/src/main/java/com/baiyue/eventbridge/dto/EventReplayRequest.java

@@ -0,0 +1,4 @@
+package com.baiyue.eventbridge.dto;
+
+public record EventReplayRequest(String reason) {
+}

event-bridge-service/src/main/java/com/baiyue/eventbridge/mapper/BridgeDeliveryAttemptMapper.java

@@ -0,0 +1,9 @@
+package com.baiyue.eventbridge.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.baiyue.eventbridge.domain.BridgeDeliveryAttempt;
+import org.apache.ibatis.annotations.Mapper;
+
+@Mapper
+public interface BridgeDeliveryAttemptMapper extends BaseMapper<BridgeDeliveryAttempt> {
+}

event-bridge-service/src/main/java/com/baiyue/eventbridge/mapper/BridgeInboxMapper.java

@@ -3,10 +3,62 @@ package com.baiyue.eventbridge.mapper;
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
 import com.baiyue.eventbridge.domain.BridgeInbox;
 import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Select;
+import org.apache.ibatis.annotations.Update;
+
+import java.util.List;
 
 /**
  * 入站事件持久化映射。
  */
 @Mapper
 public interface BridgeInboxMapper extends BaseMapper<BridgeInbox> {
+    @Select("""
+            SELECT event_id FROM bridge_inbox
+            WHERE target_region = 'OVERSEAS'
+              AND ((status IN ('RECEIVED','RETRY') AND (next_retry_at IS NULL OR next_retry_at <= CURRENT_TIMESTAMP(3)))
+                OR (status = 'PROCESSING' AND lease_until < CURRENT_TIMESTAMP(3)))
+            ORDER BY received_at LIMIT #{limit}
+            """)
+    List<String> dispatchCandidates(@Param("limit") int limit);
+
+    @Update("""
+            UPDATE bridge_inbox SET status='PROCESSING', lease_owner=#{owner},
+              lease_until=DATE_ADD(CURRENT_TIMESTAMP(3), INTERVAL #{leaseSeconds} SECOND)
+            WHERE event_id=#{eventId} AND target_region='OVERSEAS'
+              AND ((status IN ('RECEIVED','RETRY') AND (next_retry_at IS NULL OR next_retry_at <= CURRENT_TIMESTAMP(3)))
+                OR (status='PROCESSING' AND lease_until < CURRENT_TIMESTAMP(3)))
+            """)
+    int claim(@Param("eventId") String eventId, @Param("owner") String owner,
+              @Param("leaseSeconds") long leaseSeconds);
+
+    @Update("""
+            UPDATE bridge_inbox SET status='PROCESSED', processed_at=CURRENT_TIMESTAMP(3),
+              lease_owner=NULL, lease_until=NULL, next_retry_at=NULL, last_error=NULL
+            WHERE event_id=#{eventId} AND status='PROCESSING' AND lease_owner=#{owner}
+            """)
+    int markProcessed(@Param("eventId") String eventId, @Param("owner") String owner);
+
+    @Update("""
+            UPDATE bridge_inbox SET retry_count=retry_count+1,
+              status=CASE WHEN retry_count+1 >= #{maxAttempts} THEN 'DEAD' ELSE 'RETRY' END,
+              next_retry_at=CASE WHEN retry_count+1 >= #{maxAttempts} THEN NULL
+                ELSE DATE_ADD(CURRENT_TIMESTAMP(3), INTERVAL #{backoffSeconds} SECOND) END,
+              lease_owner=NULL, lease_until=NULL, last_error=#{error}
+            WHERE event_id=#{eventId} AND status='PROCESSING' AND lease_owner=#{owner}
+            """)
+    int markFailed(@Param("eventId") String eventId, @Param("owner") String owner,
+                   @Param("maxAttempts") int maxAttempts,
+                   @Param("backoffSeconds") long backoffSeconds, @Param("error") String error);
+
+    @Select("SELECT COALESCE(MAX(attempt_no), 0) + 1 FROM bridge_delivery_attempt WHERE event_id=#{eventId}")
+    int nextAttemptNo(@Param("eventId") String eventId);
+
+    @Update("""
+            UPDATE bridge_inbox SET status='RETRY', retry_count=0, next_retry_at=CURRENT_TIMESTAMP(3),
+              lease_owner=NULL, lease_until=NULL, last_error=#{reason}
+            WHERE event_id=#{eventId} AND status='DEAD'
+            """)
+    int replayDead(@Param("eventId") String eventId, @Param("reason") String reason);
 }

event-bridge-service/src/main/java/com/baiyue/eventbridge/service/EventInboxService.java

@@ -11,6 +11,7 @@ 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;
 
@@ -19,16 +20,11 @@ import java.time.LocalDateTime;
  */
 @Service
 public class EventInboxService {
-    private final InboxStore inboxStore;
-    private final ObjectMapper objectMapper;
-    private final String localRegion;
+    @Resource private InboxStore inboxStore;
+    @Resource private ObjectMapper objectMapper;
 
-    public EventInboxService(InboxStore inboxStore, ObjectMapper objectMapper,
-                             @Value("${baiyue.region:CN}") String localRegion) {
-        this.inboxStore = inboxStore;
-        this.objectMapper = objectMapper;
-        this.localRegion = localRegion;
-    }
+    @Value("${baiyue.region:CN}")
+    private String localRegion;
 
     @Transactional
     public EventReceiveResult receive(EventEnvelope envelope) {
@@ -36,7 +32,9 @@ public class EventInboxService {
         if (!envelope.targetRegion().name().equalsIgnoreCase(localRegion)) {
             throw new IllegalArgumentException("事件目标区域与当前事件桥区域不一致");
         }
-        if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(envelope.payload()), envelope.payloadHash())) {
+        boolean currentHash = EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(envelope.payload()), envelope.payloadHash());
+        boolean legacyHash = EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalizeLegacy(envelope.payload()), envelope.payloadHash());
+        if (!currentHash && !legacyHash) {
             throw new IllegalArgumentException("payloadHash 与规范化事件负载不一致");
         }
         BridgeInbox existing = inboxStore.findByEventId(envelope.eventId());

event-bridge-service/src/main/java/com/baiyue/eventbridge/service/EventReplayService.java

@@ -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);
+    }
+}

event-bridge-service/src/main/java/com/baiyue/eventbridge/service/HttpLastlegEventClient.java

@@ -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.getTenantId(), 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());
+    }
+}

event-bridge-service/src/main/java/com/baiyue/eventbridge/service/LastlegEventClient.java

@@ -0,0 +1,7 @@
+package com.baiyue.eventbridge.service;
+
+import com.baiyue.eventbridge.domain.BridgeInbox;
+
+public interface LastlegEventClient {
+    String deliver(BridgeInbox event) throws Exception;
+}

event-bridge-service/src/main/java/com/baiyue/eventbridge/service/MybatisInboxStore.java

@@ -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) {

event-bridge-service/src/main/java/com/baiyue/eventbridge/service/OverseasEventDispatcher.java

@@ -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);
+    }
+}