From bb9fb9ffe65bbfb7a67426e9520cd008da509323 Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Wed, 2 Sep 2026 10:44:17 +0800 Subject: [PATCH] fix(firstleg): retain declaration event idempotency history --- .../mapper/OrderDeclarationSyncMapper.java | 9 +++++++- .../service/OrderDeclarationSyncService.java | 23 ++++++++++++++++--- .../resources/db/phase2-inbound-schema.sql | 11 +++++++++ .../OrderDeclarationSyncServiceTest.java | 3 +++ 4 files changed, 42 insertions(+), 4 deletions(-) diff --git a/firstleg/src/main/java/com/baiyue/firstleg/mapper/OrderDeclarationSyncMapper.java b/firstleg/src/main/java/com/baiyue/firstleg/mapper/OrderDeclarationSyncMapper.java index ac4285799..232d3c5b3 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/mapper/OrderDeclarationSyncMapper.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/mapper/OrderDeclarationSyncMapper.java @@ -48,9 +48,16 @@ public interface OrderDeclarationSyncMapper { SnapshotRow lockSnapshot(@Param("tenantId") long tenantId, @Param("orderId") long orderId); @Select("SELECT tenant_id tenantId,order_id orderId,event_id eventId,order_version orderVersion,request_hash requestHash " - + "FROM fl_order_declaration_snapshot WHERE tenant_id=#{tenantId} AND event_id=#{eventId}") + + "FROM fl_order_declaration_event WHERE tenant_id=#{tenantId} AND event_id=#{eventId}") SnapshotRow byEvent(@Param("tenantId") long tenantId, @Param("eventId") String eventId); + @Insert("INSERT INTO fl_order_declaration_event " + + "(tenant_id,event_id,order_id,order_version,request_hash,consumed_at) VALUES " + + "(#{tenantId},#{eventId},#{orderId},#{orderVersion},#{requestHash},CURRENT_TIMESTAMP(3))") + int insertConsumedEvent(@Param("tenantId") long tenantId, @Param("eventId") String eventId, + @Param("orderId") long orderId, @Param("orderVersion") int orderVersion, + @Param("requestHash") String requestHash); + @Insert(""" INSERT INTO fl_order_declaration_snapshot (tenant_id,order_id,event_id,order_version,request_hash,customer_remark,expected_arrival_at, diff --git a/firstleg/src/main/java/com/baiyue/firstleg/service/OrderDeclarationSyncService.java b/firstleg/src/main/java/com/baiyue/firstleg/service/OrderDeclarationSyncService.java index ef66a9966..1230bc021 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/service/OrderDeclarationSyncService.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/service/OrderDeclarationSyncService.java @@ -12,6 +12,7 @@ import jakarta.annotation.Resource; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import org.springframework.util.StringUtils; +import org.springframework.dao.DuplicateKeyException; import java.nio.charset.StandardCharsets; import java.security.MessageDigest; @@ -62,12 +63,16 @@ public class OrderDeclarationSyncService { return; } var current = mapper.lockSnapshot(event.tenantId(), event.orderId()); - if (current != null && event.version() < current.orderVersion()) return; + if (current != null && event.version() < current.orderVersion()) { + recordEvent(event, hash); + return; + } if (current != null && event.version() == current.orderVersion()) { throw conflict("同一订单版本已存在不同申报事件"); } String cargoesJson = write(event.cargoes()); String changesJson = write(event.changes()); + recordEvent(event, hash); int saved = current == null ? mapper.insertSnapshot(event.tenantId(), event.orderId(), event.eventId(), event.version(), hash, event.customerRemark(), event.expectedArrivalAt(), normalizeChannel(event.channelCode()), @@ -83,6 +88,15 @@ public class OrderDeclarationSyncService { } } + private void recordEvent(UpdatedEvent event, String hash) { + try { + if (mapper.insertConsumedEvent(event.tenantId(), event.eventId(), event.orderId(), + event.version(), hash) != 1) throw conflict("订单申报事件幂等记录未保存"); + } catch (DuplicateKeyException duplicate) { + throw conflict("eventId已用于其他申报事件"); + } + } + private static void requireValidationRequest(ChannelChangeValidationRequest request) { if (request == null || request.tenantId() <= 0 || request.orderId() <= 0 || request.orderVersion() < 0 || !StringUtils.hasText(request.channelCode()) || request.channelCode().length() > 64) { @@ -95,8 +109,11 @@ public class OrderDeclarationSyncService { || event.eventId().length() > 128 || event.tenantId() <= 0 || event.orderId() <= 0 || event.version() < 0 || event.operatorId() <= 0 || event.occurredAt() == null || !StringUtils.hasText(event.orderNo()) || !StringUtils.hasText(event.inboundCode()) - || event.cargoes() == null || event.cargoes().isEmpty() - || !StringUtils.hasText(event.channelCode()) || !StringUtils.hasText(event.channelName())) { + || event.cargoes() == null || event.cargoes().isEmpty() || event.cargoes().size() > 500 + || event.changes() == null || !StringUtils.hasText(event.channelCode()) || event.channelCode().length() > 64 + || !StringUtils.hasText(event.channelName()) || event.channelName().length() > 128 + || event.inboundCode().length() > 40 + || event.customerRemark() != null && event.customerRemark().length() > 1000) { throw new BizException(ResultCode.BAD_REQUEST, "订单申报更新事件不完整"); } } diff --git a/firstleg/src/main/resources/db/phase2-inbound-schema.sql b/firstleg/src/main/resources/db/phase2-inbound-schema.sql index a1dca5c1b..d8718e033 100644 --- a/firstleg/src/main/resources/db/phase2-inbound-schema.sql +++ b/firstleg/src/main/resources/db/phase2-inbound-schema.sql @@ -839,6 +839,17 @@ CREATE TABLE IF NOT EXISTS fl_order_declaration_snapshot ( KEY idx_order_declaration_version (tenant_id, order_id, order_version) ) ENGINE=InnoDB COMMENT='Order客户申报版本快照；与箱码、实测、托盘、PDA和计费事实隔离'; +CREATE TABLE IF NOT EXISTS fl_order_declaration_event ( + tenant_id BIGINT UNSIGNED NOT NULL COMMENT '客户ID和隔离键', + event_id VARCHAR(128) NOT NULL COMMENT '永久保留的Order Outbox事件ID', + order_id BIGINT UNSIGNED NOT NULL COMMENT 'Order订单ID', + order_version INT NOT NULL COMMENT '事件携带的Order聚合版本', + request_hash CHAR(64) NOT NULL COMMENT '完整事件SHA-256，用于异参重放拒绝', + consumed_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + PRIMARY KEY (tenant_id, event_id), + KEY idx_order_declaration_event_order (tenant_id, order_id, order_version) +) ENGINE=InnoDB COMMENT='Order客户申报事件永久幂等账本'; + CREATE TABLE IF NOT EXISTS fl_customer_dimension_config ( id BIGINT UNSIGNED NOT NULL COMMENT '配置ID', tenant_id BIGINT UNSIGNED NOT NULL COMMENT '客户ID和数据隔离键', diff --git a/firstleg/src/test/java/com/baiyue/firstleg/service/OrderDeclarationSyncServiceTest.java b/firstleg/src/test/java/com/baiyue/firstleg/service/OrderDeclarationSyncServiceTest.java index b7fe1b347..4d70fa641 100644 --- a/firstleg/src/test/java/com/baiyue/firstleg/service/OrderDeclarationSyncServiceTest.java +++ b/firstleg/src/test/java/com/baiyue/firstleg/service/OrderDeclarationSyncServiceTest.java @@ -69,6 +69,7 @@ class OrderDeclarationSyncServiceTest { when(mapper.lockInbound(10001L, 77L)).thenReturn(row("WAIT_MEASURE", null, null, null, null)); when(mapper.insertSnapshot(anyLong(), anyLong(), anyString(), anyInt(), anyString(), any(), any(), anyString(), anyString(), anyString(), anyString(), any())).thenReturn(1); + when(mapper.insertConsumedEvent(anyLong(), anyString(), anyLong(), anyInt(), anyString())).thenReturn(1); when(mapper.updateAllowedSnapshot(10001L, 88L, 77L, "FL-77", "RK-77", "客户新备注", event.expectedArrivalAt(), "MATSON_NEW", "新渠道", 9L)).thenReturn(1); @@ -84,6 +85,7 @@ class OrderDeclarationSyncServiceTest { when(mapper.lockInbound(10001L, 77L)).thenReturn(row("WAIT_MEASURE", null, null, null, null)); when(mapper.lockSnapshot(10001L, 77L)).thenReturn( new OrderDeclarationSyncMapper.SnapshotRow(10001L, 77L, "event-5", 5, "hash")); + when(mapper.insertConsumedEvent(anyLong(), anyString(), anyLong(), anyInt(), anyString())).thenReturn(1); service.consume(event); @@ -91,6 +93,7 @@ class OrderDeclarationSyncServiceTest { any(), anyString(), anyString(), anyString(), anyString(), any()); verify(mapper, never()).updateAllowedSnapshot(anyLong(), anyLong(), anyLong(), anyString(), anyString(), any(), any(), anyString(), anyString(), anyLong()); + verify(mapper).insertConsumedEvent(anyLong(), anyString(), anyLong(), anyInt(), anyString()); } @Test