Initial commit

2026-08-13 15:57:51 +08:00
commit eb4ab93ab2
102 changed files with 4295 additions and 0 deletions
@@ -0,0 +1,23 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>com.baiyue.lastleg</groupId><artifactId>baiyue-lastleg-platform</artifactId>
+        <version>1.0.0-SNAPSHOT</version><relativePath>../pom.xml</relativePath>
+    </parent>
+    <artifactId>event-bridge-service</artifactId>
+    <name>event-bridge-service</name>
+    <dependencies>
+        <dependency><groupId>com.baiyue.shared</groupId><artifactId>baiyue-common-web</artifactId></dependency>
+        <dependency><groupId>com.baiyue.shared</groupId><artifactId>baiyue-common-openapi</artifactId></dependency>
+        <dependency><groupId>com.baiyue.shared</groupId><artifactId>common-event-contract</artifactId></dependency>
+        <dependency><groupId>com.baomidou</groupId><artifactId>mybatis-plus-spring-boot3-starter</artifactId></dependency>
+        <dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId></dependency>
+        <dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId></dependency>
+        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId></dependency>
+        <dependency><groupId>com.mysql</groupId><artifactId>mysql-connector-j</artifactId><scope>runtime</scope></dependency>
+        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency>
+    </dependencies>
+    <build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build>
+</project>
@@ -0,0 +1,16 @@
+package com.baiyue.eventbridge;
+
+import org.mybatis.spring.annotation.MapperScan;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+/**
+ * 跨区域事件桥启动入口。
+ */
+@MapperScan("com.baiyue.eventbridge.mapper")
+@SpringBootApplication
+public class EventBridgeApplication {
+    public static void main(String[] args) {
+        SpringApplication.run(EventBridgeApplication.class, args);
+    }
+}
@@ -0,0 +1,57 @@
+package com.baiyue.eventbridge.controller;
+
+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.service.EventInboxService;
+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 org.springframework.beans.factory.annotation.Value;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+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;
+
+/**
+ * 事件桥内部接口。生产环境必须在网关或基础设施层增加 mTLS 与来源白名单。
+ */
+@Tag(name = "事件桥内部接口", description = "跨区域事件的幂等接收与能力状态")
+@RestController
+@RequestMapping("/internal/event-bridge")
+@RequiredArgsConstructor
+public class EventBridgeController {
+    private final EventInboxService eventInboxService;
+    private final InternalServiceTokenGuard internalServiceTokenGuard;
+
+    @Value("${baiyue.region:CN}")
+    private String region;
+
+    @Operation(summary = "幂等接收事件", description = "当前仅完成数据库接收骨架，不代表生产跨区传输和 mTLS 已开通")
+    @ApiResponses({
+            @io.swagger.v3.oas.annotations.responses.ApiResponse(responseCode = "200", description = "事件已接收或判定为重复"),
+            @io.swagger.v3.oas.annotations.responses.ApiResponse(responseCode = "401", description = "内部服务令牌无效"),
+            @io.swagger.v3.oas.annotations.responses.ApiResponse(responseCode = "503", description = "内部服务令牌未配置，接收接口关闭")
+    })
+    @PostMapping("/events")
+    public ApiResponse<EventReceiveResult> receive(
+            @Parameter(description = "内部服务共享令牌")
+            @RequestHeader(value = "X-Internal-Service-Token", required = false) String token,
+            @RequestBody EventEnvelope envelope) {
+        internalServiceTokenGuard.verify(token);
+        return ApiResponse.success(eventInboxService.receive(envelope));
+    }
+
+    @Operation(summary = "查询事件桥能力状态")
+    @ApiResponses(@io.swagger.v3.oas.annotations.responses.ApiResponse(responseCode = "200", description = "查询成功"))
+    @GetMapping("/status")
+    public ApiResponse<BridgeStatus> status() {
+        return ApiResponse.success(new BridgeStatus("运行中", region, true, false));
+    }
+}
@@ -0,0 +1,36 @@
+package com.baiyue.eventbridge.domain;
+
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+
+import java.time.LocalDateTime;
+import java.time.Instant;
+
+/**
+ * 事件桥入站幂等记录。
+ */
+@Data
+@TableName("bridge_inbox")
+public class BridgeInbox {
+    @TableId
+    private String eventId;
+    private String idempotencyKey;
+    private String eventType;
+    private Integer eventVersion;
+    private String sourceRegion;
+    private String targetRegion;
+    private String tenantId;
+    private String aggregateType;
+    private String aggregateId;
+    private Long aggregateVersion;
+    private Long handoverVersion;
+    private String traceId;
+    private String correlationId;
+    private String causationId;
+    private Instant occurredAt;
+    private String payloadHash;
+    private String payloadJson;
+    private String status;
+    private LocalDateTime receivedAt;
+}
@@ -0,0 +1,14 @@
+package com.baiyue.eventbridge.dto;
+
+import io.swagger.v3.oas.annotations.media.Schema;
+
+/**
+ * 事件桥能力状态，不代表生产网络链路已经开通。
+ */
+@Schema(description = "事件桥能力状态")
+public record BridgeStatus(
+        @Schema(description = "服务状态") String status,
+        @Schema(description = "当前区域") String region,
+        @Schema(description = "数据库幂等接收是否启用") boolean inboxEnabled,
+        @Schema(description = "生产级跨区传输是否启用") boolean productionTransportEnabled) {
+}
@@ -0,0 +1,15 @@
+package com.baiyue.eventbridge.dto;
+
+import io.swagger.v3.oas.annotations.media.Schema;
+
+import java.time.LocalDateTime;
+
+/**
+ * 事件接收结果。
+ */
+@Schema(description = "事件接收结果")
+public record EventReceiveResult(
+        @Schema(description = "事件标识") String eventId,
+        @Schema(description = "接收结果：已接收或重复") String result,
+        @Schema(description = "接收时间") LocalDateTime receivedAt) {
+}
@@ -0,0 +1,12 @@
+package com.baiyue.eventbridge.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.baiyue.eventbridge.domain.BridgeInbox;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * 入站事件持久化映射。
+ */
+@Mapper
+public interface BridgeInboxMapper extends BaseMapper<BridgeInbox> {
+}
@@ -0,0 +1,32 @@
+package com.baiyue.eventbridge.security;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.http.HttpStatus;
+import org.springframework.stereotype.Component;
+import org.springframework.web.server.ResponseStatusException;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+
+/**
+ * 内部事件接收接口的共享令牌校验器；未配置令牌时默认关闭接收。
+ */
+@Component
+public class InternalServiceTokenGuard {
+    private final String expectedToken;
+
+    public InternalServiceTokenGuard(@Value("${baiyue.security.internal-service-token:}") String expectedToken) {
+        this.expectedToken = expectedToken;
+    }
+
+    public void verify(String actualToken) {
+        if (expectedToken == null || expectedToken.isBlank()) {
+            throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE, "内部事件接收令牌未配置，接口已关闭");
+        }
+        byte[] expected = expectedToken.getBytes(StandardCharsets.UTF_8);
+        byte[] actual = actualToken == null ? new byte[0] : actualToken.getBytes(StandardCharsets.UTF_8);
+        if (!MessageDigest.isEqual(expected, actual)) {
+            throw new ResponseStatusException(HttpStatus.UNAUTHORIZED, "内部服务令牌无效");
+        }
+    }
+}
@@ -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;
+    }
+}
@@ -0,0 +1,11 @@
+package com.baiyue.eventbridge.service;
+
+import com.baiyue.eventbridge.domain.BridgeInbox;
+
+/**
+ * 入站幂等存储边界，便于隔离数据库实现并进行单元测试。
+ */
+public interface InboxStore {
+    BridgeInbox findByEventId(String eventId);
+    void insert(BridgeInbox inbox);
+}
@@ -0,0 +1,25 @@
+package com.baiyue.eventbridge.service;
+
+import com.baiyue.eventbridge.domain.BridgeInbox;
+import com.baiyue.eventbridge.mapper.BridgeInboxMapper;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Repository;
+
+/**
+ * 基于 MyBatis-Plus 的入站幂等存储。
+ */
+@Repository
+@RequiredArgsConstructor
+public class MybatisInboxStore implements InboxStore {
+    private final BridgeInboxMapper mapper;
+
+    @Override
+    public BridgeInbox findByEventId(String eventId) {
+        return mapper.selectById(eventId);
+    }
+
+    @Override
+    public void insert(BridgeInbox inbox) {
+        mapper.insert(inbox);
+    }
+}
@@ -0,0 +1,44 @@
+server:
+  port: ${EVENT_BRIDGE_PORT:9312}
+
+spring:
+  data:
+    redis:
+      host: ${REDIS_HOST:127.0.0.1}
+      port: ${REDIS_PORT:6380}
+  application:
+    name: event-bridge-service
+  config:
+    import:
+      - optional:nacos:baiyue-common.yml?group=BAIYUE&refreshEnabled=true
+      - optional:nacos:${spring.application.name}.yml?group=BAIYUE&refreshEnabled=true
+  datasource:
+    driver-class-name: com.mysql.cj.jdbc.Driver
+    url: ${EVENT_BRIDGE_DB_URL:jdbc:mysql://127.0.0.1:3306/baiyue_event_bridge?useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai&useSSL=false&allowPublicKeyRetrieval=true}
+    username: ${EVENT_BRIDGE_DB_USERNAME:root}
+    password: ${EVENT_BRIDGE_DB_PASSWORD:}
+  cloud:
+    nacos:
+      server-addr: ${NACOS_SERVER_ADDR:127.0.0.1:8858}
+      username: ${NACOS_USERNAME:nacos}
+      password: ${NACOS_PASSWORD:nacos}
+      discovery:
+        group: ${NACOS_GROUP:BAIYUE}
+        namespace: ${NACOS_NAMESPACE:}
+      config:
+        namespace: ${NACOS_NAMESPACE:}
+
+baiyue:
+  region: ${REGION_CODE:CN}
+  security:
+    internal-service-token: ${INTERNAL_SERVICE_TOKEN:}
+
+management:
+  endpoints:
+    web:
+      exposure:
+        include: health,info,prometheus
+
+springdoc:
+  api-docs:
+    path: /v3/api-docs
@@ -0,0 +1,80 @@
+CREATE DATABASE IF NOT EXISTS baiyue_event_bridge DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
+USE baiyue_event_bridge;
+
+CREATE TABLE IF NOT EXISTS bridge_outbox (
+    event_id VARCHAR(128) NOT NULL COMMENT '事件全局标识',
+    idempotency_key VARCHAR(191) NOT NULL COMMENT '发送幂等键',
+    event_type VARCHAR(64) NOT NULL COMMENT '事件类型',
+    event_version INT NOT NULL COMMENT '事件契约版本',
+    source_region VARCHAR(32) NOT NULL COMMENT '来源区域',
+    target_region VARCHAR(32) NOT NULL COMMENT '目标区域',
+    tenant_id VARCHAR(128) NOT NULL COMMENT '租户标识',
+    aggregate_type VARCHAR(64) NOT NULL COMMENT '聚合类型',
+    aggregate_id VARCHAR(128) NOT NULL COMMENT '聚合标识',
+    aggregate_version BIGINT NOT NULL COMMENT '聚合版本',
+    handover_version BIGINT NULL COMMENT '跨区交接版本',
+    trace_id VARCHAR(128) NOT NULL COMMENT '链路标识',
+    correlation_id VARCHAR(128) NOT NULL COMMENT '业务关联标识',
+    causation_id VARCHAR(128) NULL COMMENT '上游事件标识',
+    occurred_at DATETIME(3) NOT NULL COMMENT '业务事实发生时间',
+    payload_hash VARCHAR(71) NOT NULL COMMENT '规范化负载摘要',
+    payload_json JSON NOT NULL COMMENT '事件负载',
+    status VARCHAR(32) NOT NULL DEFAULT 'PENDING' COMMENT '投递状态',
+    retry_count INT NOT NULL DEFAULT 0 COMMENT '已重试次数',
+    next_retry_at DATETIME(3) NULL COMMENT '下次重试时间',
+    lease_owner VARCHAR(128) NULL COMMENT '租约持有者',
+    lease_until DATETIME(3) NULL COMMENT '租约截止时间',
+    last_error VARCHAR(1000) NULL COMMENT '最后错误',
+    created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '创建时间',
+    delivered_at DATETIME(3) NULL COMMENT '投递成功时间',
+    PRIMARY KEY (event_id),
+    UNIQUE KEY uk_bridge_outbox_idempotency (idempotency_key),
+    KEY idx_bridge_outbox_schedule (target_region, status, next_retry_at, lease_until)
+) ENGINE=InnoDB COMMENT='跨区域事件待投递箱';
+
+CREATE TABLE IF NOT EXISTS bridge_inbox (
+    event_id VARCHAR(128) NOT NULL COMMENT '事件全局标识',
+    idempotency_key VARCHAR(191) NOT NULL COMMENT '接收幂等键',
+    event_type VARCHAR(64) NOT NULL COMMENT '事件类型',
+    event_version INT NOT NULL COMMENT '事件契约版本',
+    source_region VARCHAR(32) NOT NULL COMMENT '来源区域',
+    target_region VARCHAR(32) NOT NULL COMMENT '目标区域',
+    tenant_id VARCHAR(128) NOT NULL COMMENT '租户标识',
+    aggregate_type VARCHAR(64) NOT NULL COMMENT '聚合类型',
+    aggregate_id VARCHAR(128) NOT NULL COMMENT '聚合标识',
+    aggregate_version BIGINT NOT NULL COMMENT '聚合版本',
+    handover_version BIGINT NULL COMMENT '跨区交接版本',
+    trace_id VARCHAR(128) NOT NULL COMMENT '链路标识',
+    correlation_id VARCHAR(128) NOT NULL COMMENT '业务关联标识',
+    causation_id VARCHAR(128) NULL COMMENT '上游事件标识',
+    occurred_at DATETIME(3) NOT NULL COMMENT '业务事实发生时间',
+    payload_hash VARCHAR(71) NOT NULL COMMENT '规范化负载摘要',
+    payload_json JSON NOT NULL COMMENT '事件负载',
+    status VARCHAR(32) NOT NULL DEFAULT 'RECEIVED' COMMENT '消费状态',
+    lease_owner VARCHAR(128) NULL COMMENT '消费租约持有者',
+    lease_until DATETIME(3) NULL COMMENT '消费租约截止时间',
+    retry_count INT NOT NULL DEFAULT 0 COMMENT '消费重试次数',
+    next_retry_at DATETIME(3) NULL COMMENT '下次消费时间',
+    last_error VARCHAR(1000) NULL COMMENT '最后错误',
+    received_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '接收时间',
+    processed_at DATETIME(3) NULL COMMENT '消费完成时间',
+    PRIMARY KEY (event_id),
+    UNIQUE KEY uk_bridge_inbox_idempotency (idempotency_key),
+    KEY idx_bridge_inbox_consume (status, next_retry_at, lease_until)
+) ENGINE=InnoDB COMMENT='跨区域事件已接收箱';
+
+CREATE TABLE IF NOT EXISTS bridge_delivery_attempt (
+    id BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键',
+    event_id VARCHAR(128) NOT NULL COMMENT '事件全局标识',
+    source_region VARCHAR(32) NOT NULL COMMENT '来源区域',
+    target_region VARCHAR(32) NOT NULL COMMENT '目标区域',
+    attempt_no INT NOT NULL COMMENT '投递序号',
+    status VARCHAR(32) NOT NULL COMMENT '投递结果',
+    response_code VARCHAR(64) NULL COMMENT '对端响应码',
+    error_message VARCHAR(1000) NULL COMMENT '错误信息',
+    started_at DATETIME(3) NOT NULL COMMENT '开始时间',
+    finished_at DATETIME(3) NULL COMMENT '完成时间',
+    PRIMARY KEY (id),
+    UNIQUE KEY uk_bridge_attempt (event_id, attempt_no),
+    KEY idx_bridge_attempt_event (event_id, started_at)
+) ENGINE=InnoDB COMMENT='跨区域事件投递尝试记录';
@@ -0,0 +1,31 @@
+package com.baiyue.eventbridge.security;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.web.server.ResponseStatusException;
+
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class InternalServiceTokenGuardTest {
+    @Test
+    void 正确令牌允许访问() {
+        InternalServiceTokenGuard guard = new InternalServiceTokenGuard("a-secure-internal-token");
+        assertThatCode(() -> guard.verify("a-secure-internal-token")).doesNotThrowAnyException();
+    }
+
+    @Test
+    void 未配置令牌时默认关闭接收接口() {
+        InternalServiceTokenGuard guard = new InternalServiceTokenGuard("");
+        assertThatThrownBy(() -> guard.verify("anything"))
+                .isInstanceOf(ResponseStatusException.class)
+                .hasMessageContaining("接口已关闭");
+    }
+
+    @Test
+    void 错误令牌被拒绝() {
+        InternalServiceTokenGuard guard = new InternalServiceTokenGuard("expected");
+        assertThatThrownBy(() -> guard.verify("wrong"))
+                .isInstanceOf(ResponseStatusException.class)
+                .hasMessageContaining("令牌无效");
+    }
+}
@@ -0,0 +1,97 @@
+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.common.event.RegionCode;
+import com.baiyue.eventbridge.domain.BridgeInbox;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.Test;
+import org.springframework.dao.DuplicateKeyException;
+
+import java.time.Instant;
+import java.time.LocalDateTime;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class EventInboxServiceTest {
+    private final ObjectMapper objectMapper = new ObjectMapper();
+    private final InboxStore store = mock(InboxStore.class);
+    private final EventInboxService service = new EventInboxService(store, objectMapper, "OVERSEAS");
+
+    @Test
+    void 首次事件落库并返回已接收() {
+        EventEnvelope envelope = envelope("sha256:placeholder");
+        envelope = withHash(envelope);
+
+        assertThat(service.receive(envelope).result()).isEqualTo("已接收");
+        verify(store).insert(any(BridgeInbox.class));
+    }
+
+    @Test
+    void 相同事件和摘要重复接收不再次写入() {
+        EventEnvelope envelope = withHash(envelope("sha256:placeholder"));
+        BridgeInbox existing = existing(envelope.payloadHash());
+        when(store.findByEventId(envelope.eventId())).thenReturn(existing);
+
+        assertThat(service.receive(envelope).result()).isEqualTo("重复");
+    }
+
+    @Test
+    void 相同事件但摘要不同时拒绝覆盖() {
+        EventEnvelope envelope = withHash(envelope("sha256:placeholder"));
+        when(store.findByEventId(envelope.eventId())).thenReturn(existing("sha256:" + "0".repeat(64)));
+
+        assertThatThrownBy(() -> service.receive(envelope))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("拒绝覆盖");
+    }
+
+    @Test
+    void 并发插入冲突后按既有记录判定为重复() {
+        EventEnvelope envelope = withHash(envelope("sha256:placeholder"));
+        when(store.findByEventId(envelope.eventId())).thenReturn(null, existing(envelope.payloadHash()));
+        doThrow(new DuplicateKeyException("并发重复")).when(store).insert(any());
+
+        assertThat(service.receive(envelope).result()).isEqualTo("重复");
+    }
+
+    @Test
+    void 目标区域与当前部署区域不一致时拒绝接收() {
+        EventInboxService cnService = new EventInboxService(store, objectMapper, "CN");
+        EventEnvelope envelope = withHash(envelope("sha256:placeholder"));
+
+        assertThatThrownBy(() -> cnService.receive(envelope))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("目标区域");
+    }
+
+    private EventEnvelope envelope(String hash) {
+        return new EventEnvelope("550e8400-e29b-41d4-a716-446655440000", "OrderApproved", 1,
+                Instant.parse("2026-07-29T08:30:00Z"), RegionCode.CN, RegionCode.OVERSEAS,
+                "20001", "ORDER", "90001", 1, null, "trace-1", "flow-1", null,
+                objectMapper.createObjectNode().put("orderId", "90001"), hash);
+    }
+
+    private EventEnvelope withHash(EventEnvelope source) {
+        String hash = EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(source.payload()));
+        return new EventEnvelope(source.eventId(), source.eventType(), source.eventVersion(), source.occurredAt(),
+                source.sourceRegion(), source.targetRegion(), source.tenantId(), source.aggregateType(),
+                source.aggregateId(), source.aggregateVersion(), source.handoverVersion(), source.traceId(),
+                source.correlationId(), source.causationId(), source.payload(), hash);
+    }
+
+    private BridgeInbox existing(String hash) {
+        BridgeInbox inbox = new BridgeInbox();
+        inbox.setEventId("550e8400-e29b-41d4-a716-446655440000");
+        inbox.setPayloadHash(hash);
+        inbox.setReceivedAt(LocalDateTime.of(2026, 7, 29, 18, 0));
+        return inbox;
+    }
+}