提交

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

@@ -16,8 +16,7 @@ import java.util.List;
 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)))
+            WHERE ((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}
             """)
@@ -26,7 +25,7 @@ public interface BridgeInboxMapper extends BaseMapper<BridgeInbox> {
     @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'
+            WHERE event_id=#{eventId}
               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)))
             """)

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

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

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

@@ -0,0 +1,52 @@
+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 jakarta.annotation.PostConstruct;
+import jakarta.annotation.Resource;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+
+@Component
+public class HttpFirstlegEventClient implements FirstlegEventClient {
+    @Resource private ObjectMapper objectMapper;
+    @Resource private EventDispatcherProperties properties;
+    @Value("${baiyue.event-dispatcher.firstleg-consumer-url:http://127.0.0.1:9203/internal/firstleg/v1/container-events}")
+    private URI consumerUrl;
+    @Value("${baiyue.security.internal-service-token:}") private String serviceToken;
+    private HttpClient httpClient;
+
+    @PostConstruct
+    void initializeHttpClient() {
+        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());
+        HttpRequest request = HttpRequest.newBuilder(consumerUrl).timeout(properties.requestTimeout())
+            .header("Content-Type", "application/json")
+            .header("X-Internal-Service-Token", serviceToken)
+            .POST(HttpRequest.BodyPublishers.ofString(objectMapper.writeValueAsString(envelope))).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/OverseasEventDispatcher.java

@@ -20,6 +20,7 @@ public class OverseasEventDispatcher {
     @Resource private BridgeInboxMapper inboxMapper;
     @Resource private BridgeDeliveryAttemptMapper attemptMapper;
     @Resource private LastlegEventClient eventClient;
+    @Resource private FirstlegEventClient firstlegEventClient;
     @Resource private EventDispatcherProperties properties;
     private final String leaseOwner = ManagementFactory.getRuntimeMXBean().getName() + ":" + UUID.randomUUID();
 
@@ -40,7 +41,7 @@ public class OverseasEventDispatcher {
         int deliveryAttemptNo = inboxMapper.nextAttemptNo(eventId);
         BridgeDeliveryAttempt attempt = startAttempt(event, deliveryAttemptNo);
         try {
-            String responseCode = eventClient.deliver(event);
+            String responseCode = deliver(event);
             if (inboxMapper.markProcessed(eventId, leaseOwner) != 1) {
                 throw new IllegalStateException("事件租约已失效，无法确认投递结果");
             }
@@ -49,11 +50,16 @@ public class OverseasEventDispatcher {
             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={}",
+            log.warn("跨区事件投递失败 eventId={}, retryAttempt={}, deliveryAttempt={}, reason={}",
                     eventId, retryAttempt, deliveryAttemptNo, error);
         }
     }
 
+    private String deliver(BridgeInbox event) throws Exception {
+        return "CN".equals(event.getTargetRegion())
+            ? firstlegEventClient.deliver(event) : eventClient.deliver(event);
+    }
+
     private BridgeDeliveryAttempt startAttempt(BridgeInbox event, int attemptNo) {
         BridgeDeliveryAttempt attempt = new BridgeDeliveryAttempt();
         attempt.setEventId(event.getEventId());

event-bridge-service/src/main/resources/application.yml

@@ -32,6 +32,7 @@ baiyue:
   region: ${REGION_CODE:OVERSEAS}
   event-dispatcher:
     consumer-url: ${LASTLEG_CONSUMER_URL:http://127.0.0.1:9303/internal/lastleg/events}
+    firstleg-consumer-url: ${FIRSTLEG_CONTAINER_EVENT_CONSUMER_URL:http://127.0.0.1:9203/internal/firstleg/v1/container-events}
     fixed-delay: ${EVENT_DISPATCH_FIXED_DELAY:3000}
     batch-size: ${EVENT_DISPATCH_BATCH_SIZE:50}
     max-attempts: ${EVENT_DISPATCH_MAX_ATTEMPTS:10}

event-bridge-service/src/test/java/com/baiyue/eventbridge/service/OverseasEventDispatcherTest.java

@@ -21,6 +21,7 @@ class OverseasEventDispatcherTest {
     private final BridgeInboxMapper inboxMapper = mock(BridgeInboxMapper.class);
     private final BridgeDeliveryAttemptMapper attemptMapper = mock(BridgeDeliveryAttemptMapper.class);
     private final LastlegEventClient eventClient = mock(LastlegEventClient.class);
+    private final FirstlegEventClient firstlegEventClient = mock(FirstlegEventClient.class);
     private OverseasEventDispatcher dispatcher;
     private BridgeInbox event;
 
@@ -34,6 +35,7 @@ class OverseasEventDispatcherTest {
         ReflectionTestUtils.setField(dispatcher, "inboxMapper", inboxMapper);
         ReflectionTestUtils.setField(dispatcher, "attemptMapper", attemptMapper);
         ReflectionTestUtils.setField(dispatcher, "eventClient", eventClient);
+        ReflectionTestUtils.setField(dispatcher, "firstlegEventClient", firstlegEventClient);
         ReflectionTestUtils.setField(dispatcher, "properties", properties);
         event = new BridgeInbox();
         event.setEventId("event-1");
@@ -83,9 +85,24 @@ class OverseasEventDispatcherTest {
         dispatcher.dispatchOne("event-1");
 
         verifyNoInteractions(eventClient);
+        verifyNoInteractions(firstlegEventClient);
         verify(attemptMapper, never()).insert(any(BridgeDeliveryAttempt.class));
     }
 
+    @Test
+    void cnTargetIsDeliveredToFirstleg() throws Exception {
+        event.setSourceRegion("OVERSEAS");
+        event.setTargetRegion("CN");
+        when(firstlegEventClient.deliver(event)).thenReturn("200");
+        when(inboxMapper.markProcessed(eq("event-1"), anyString())).thenReturn(1);
+
+        dispatcher.dispatchOne("event-1");
+
+        verify(firstlegEventClient).deliver(event);
+        verifyNoInteractions(eventClient);
+        assertThat(updatedAttempt().getStatus()).isEqualTo("DELIVERED");
+    }
+
     private BridgeDeliveryAttempt updatedAttempt() {
         ArgumentCaptor<BridgeDeliveryAttempt> captor = ArgumentCaptor.forClass(BridgeDeliveryAttempt.class);
         verify(attemptMapper).updateById(captor.capture());