尾程暂扣回传

2026-08-31 15:22:32 +08:00
parent a4189a0074
commit a213be67c5
22 changed files with 438 additions and 17 deletions
@@ -4,8 +4,9 @@ import com.baiyue.common.core.api.ApiResponse;
 import com.baiyue.common.core.api.ResultCode;
 import com.baiyue.common.core.exception.BizException;
 import com.baiyue.common.event.EventEnvelope;
-import com.baiyue.firstleg.entity.ContainerBookingEntity;
+import com.baiyue.common.event.EventTypes;
 import com.baiyue.firstleg.service.OverseasContainerEventConsumer;
+import com.baiyue.firstleg.service.OverseasInboundOrderDetainedConsumer;
 import lombok.RequiredArgsConstructor;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.web.bind.annotation.PostMapping;
@@ -20,14 +21,17 @@ import java.security.MessageDigest;
@RequiredArgsConstructor
 public class InternalOverseasContainerEventController {
    private final OverseasContainerEventConsumer consumer;
+    private final OverseasInboundOrderDetainedConsumer detainedConsumer;
    @Value("${baiyue.security.internal-service-token}") private String expectedToken;

    @PostMapping("/internal/firstleg/v1/container-events")
-    public ApiResponse<ContainerBookingEntity> receive(
+    public ApiResponse<Object> receive(
        @RequestHeader(value = "X-Internal-Service-Token", required = false) String token,
        @RequestBody EventEnvelope event) {
        authenticate(token);
-        return ApiResponse.success(consumer.consume(event));
+        Object result = EventTypes.OVERSEAS_INBOUND_ORDER_DETAINED_CHANGED.equals(event.eventType())
+            ? detainedConsumer.consume(event) : consumer.consume(event);
+        return ApiResponse.success(result);
    }

    private void authenticate(String token) {
@@ -0,0 +1,24 @@
+package com.baiyue.firstleg.entity;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Getter;
+import lombok.Setter;
+
+import java.time.LocalDateTime;
+
+@Getter
+@Setter
+@TableName("fl_inbound_hold_record")
+public class InboundHoldRecordEntity {
+    @TableId(type = IdType.INPUT) private Long id;
+    private Long tenantId;
+    private Long inboundId;
+    private String holdAction;
+    private String reason;
+    private Long evidenceFileId;
+    private Long operatorId;
+    private String operatorName;
+    private LocalDateTime operatedAt;
+}
@@ -53,6 +53,7 @@ public class InboundOrderEntity extends BaseEntity {
    private String destinationName;
    private Boolean holdFlag;
    private String currentHoldReason;
+    private Long lastOverseasHoldVersion;
    private Integer totalPackageCount;
    private BigDecimal actualGrossWeightKg;
    private BigDecimal chargeableWeightKg;
@@ -0,0 +1,9 @@
+package com.baiyue.firstleg.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.baiyue.firstleg.entity.InboundHoldRecordEntity;
+import org.apache.ibatis.annotations.Mapper;
+
+@Mapper
+public interface InboundHoldRecordMapper extends BaseMapper<InboundHoldRecordEntity> {
+}
@@ -5,10 +5,25 @@ import com.baiyue.firstleg.entity.InboundOrderEntity;
 import org.apache.ibatis.annotations.Mapper;
 import org.apache.ibatis.annotations.Param;
 import org.apache.ibatis.annotations.Update;
+import org.apache.ibatis.annotations.Select;

 /** 入库聚合根条件更新 Mapper，所有推进均同时校验租户、前置状态和版本。 */
@Mapper
 public interface InboundOrderMapper extends BaseMapper<InboundOrderEntity> {
+    @Select("SELECT * FROM fl_inbound_order WHERE id=#{inboundId} AND deleted=0 FOR UPDATE")
+    InboundOrderEntity selectActiveByIdForUpdate(@Param("inboundId") long inboundId);
+
+    @Update("""
+        UPDATE fl_inbound_order
+           SET hold_flag=#{detained}, current_hold_reason=#{reason},
+               last_overseas_hold_version=#{sourceVersion}, updated_by=#{operatorId}, version=version+1
+         WHERE id=#{inboundId} AND tenant_id=#{tenantId} AND deleted=0
+           AND (last_overseas_hold_version IS NULL OR last_overseas_hold_version < #{sourceVersion})
+        """)
+    int applyOverseasHold(@Param("tenantId") long tenantId, @Param("inboundId") long inboundId,
+                          @Param("detained") boolean detained, @Param("reason") String reason,
+                          @Param("sourceVersion") long sourceVersion, @Param("operatorId") long operatorId);
+
    @Update("""
        UPDATE fl_inbound_order
           SET process_status = 'WAIT_SPLIT', measure_review_status = 'APPROVED',
@@ -0,0 +1,135 @@
+package com.baiyue.firstleg.service;
+
+import com.baomidou.mybatisplus.core.toolkit.IdWorker;
+import com.baiyue.common.event.EventEnvelope;
+import com.baiyue.common.event.EventPayloadCanonicalizer;
+import com.baiyue.common.event.EventPayloadHash;
+import com.baiyue.common.event.EventTypes;
+import com.baiyue.common.event.OverseasInboundOrderDetainedChangedV1;
+import com.baiyue.common.event.RegionCode;
+import com.baiyue.firstleg.entity.ContainerEventInboxEntity;
+import com.baiyue.firstleg.entity.InboundHoldRecordEntity;
+import com.baiyue.firstleg.entity.InboundOrderEntity;
+import com.baiyue.firstleg.mapper.ContainerEventInboxMapper;
+import com.baiyue.firstleg.mapper.InboundHoldRecordMapper;
+import com.baiyue.firstleg.mapper.InboundOrderMapper;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.RequiredArgsConstructor;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.util.Locale;
+import java.util.Objects;
+
+/** 将海外暂扣事实按来源入仓单绑定回写到国内头程入仓单。 */
+@Service
+@RequiredArgsConstructor
+public class OverseasInboundOrderDetainedConsumer {
+    private final InboundOrderMapper inboundOrders;
+    private final InboundHoldRecordMapper holdRecords;
+    private final ContainerEventInboxMapper inbox;
+    private final ObjectMapper objectMapper;
+
+    @Transactional
+    public InboundOrderEntity consume(EventEnvelope event) {
+        validateEnvelope(event);
+        OverseasInboundOrderDetainedChangedV1 payload = payload(event);
+        long inboundId = parsePositive(payload.sourceInboundId(), "来源入仓单 ID 无效");
+        long tenantId = parsePositive(event.tenantId(), "客户 ID 无效");
+        if (!payload.sourceInboundId().equals(event.aggregateId())) {
+            throw new IllegalArgumentException("事件 aggregateId 与来源入仓单 ID 不一致");
+        }
+        if (payload.changedAt() == null || payload.inboundCode() == null || payload.inboundCode().isBlank()
+            || payload.reason() == null || payload.reason().isBlank()) {
+            throw new IllegalArgumentException("暂扣事件缺少入仓码、原因或变更时间");
+        }
+
+        ContainerEventInboxEntity existing = inbox.selectById(event.eventId());
+        if (existing != null) {
+            if (!existing.getPayloadHash().equalsIgnoreCase(event.payloadHash())) {
+                throw new IllegalArgumentException("相同 eventId 对应的事件摘要冲突");
+            }
+            return requiredInbound(inboundId, tenantId, payload.inboundCode());
+        }
+        ContainerEventInboxEntity inboxRow = inbox(event);
+        try {
+            inbox.insert(inboxRow);
+        } catch (DuplicateKeyException duplicate) {
+            ContainerEventInboxEntity raced = inbox.selectById(event.eventId());
+            if (raced == null || !raced.getPayloadHash().equalsIgnoreCase(event.payloadHash())) throw duplicate;
+            return requiredInbound(inboundId, tenantId, payload.inboundCode());
+        }
+
+        InboundOrderEntity inbound = requiredInbound(inboundId, tenantId, payload.inboundCode());
+        if (inbound.getLastOverseasHoldVersion() == null
+            || event.aggregateVersion() > inbound.getLastOverseasHoldVersion()) {
+            int updated = inboundOrders.applyOverseasHold(tenantId, inboundId, payload.detained(),
+                payload.detained() ? payload.reason() : null, event.aggregateVersion(),
+                parseOperatorId(payload.operatorId()));
+            if (updated == 1) insertHistory(event, payload, tenantId, inboundId);
+        }
+        inboxRow.setStatus("PROCESSED");
+        inboxRow.setProcessedAt(LocalDateTime.now(ZoneOffset.UTC));
+        inbox.updateById(inboxRow);
+        return requiredInbound(inboundId, tenantId, payload.inboundCode());
+    }
+
+    private void validateEnvelope(EventEnvelope event) {
+        event.requireValid();
+        if (!EventTypes.OVERSEAS_INBOUND_ORDER_DETAINED_CHANGED.equals(event.eventType())
+            || event.eventVersion() != 1 || event.sourceRegion() != RegionCode.OVERSEAS
+            || event.targetRegion() != RegionCode.CN || !"INBOUND_ORDER".equals(event.aggregateType())) {
+            throw new IllegalArgumentException("不支持的海外入仓单暂扣事件");
+        }
+        if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(event.payload()), event.payloadHash())) {
+            throw new IllegalArgumentException("payloadHash 与事件负载不一致");
+        }
+    }
+
+    private OverseasInboundOrderDetainedChangedV1 payload(EventEnvelope event) {
+        try { return objectMapper.treeToValue(event.payload(), OverseasInboundOrderDetainedChangedV1.class); }
+        catch (Exception exception) { throw new IllegalArgumentException("海外入仓单暂扣事件负载无效", exception); }
+    }
+
+    private InboundOrderEntity requiredInbound(long inboundId, long tenantId, String inboundCode) {
+        InboundOrderEntity value = inboundOrders.selectActiveByIdForUpdate(inboundId);
+        if (value == null) throw new IllegalArgumentException("来源头程入仓单不存在");
+        if (!Objects.equals(value.getTenantId(), tenantId) || !Objects.equals(value.getInboundCode(), inboundCode)) {
+            throw new IllegalArgumentException("事件与来源头程入仓单归属不一致");
+        }
+        return value;
+    }
+
+    private void insertHistory(EventEnvelope event, OverseasInboundOrderDetainedChangedV1 payload,
+                               long tenantId, long inboundId) {
+        InboundHoldRecordEntity row = new InboundHoldRecordEntity();
+        row.setId(IdWorker.getId()); row.setTenantId(tenantId); row.setInboundId(inboundId);
+        row.setHoldAction(payload.detained() ? "HOLD" : "RELEASE"); row.setReason(payload.reason());
+        row.setOperatorId(parseOperatorId(payload.operatorId())); row.setOperatorName(payload.operatorName());
+        row.setOperatedAt(LocalDateTime.ofInstant(payload.changedAt(), ZoneOffset.UTC));
+        holdRecords.insert(row);
+    }
+
+    private ContainerEventInboxEntity inbox(EventEnvelope event) {
+        ContainerEventInboxEntity row = new ContainerEventInboxEntity();
+        row.setEventId(event.eventId()); row.setEventType(event.eventType()); row.setEventVersion(event.eventVersion());
+        row.setAggregateId(event.aggregateId()); row.setAggregateVersion(event.aggregateVersion());
+        row.setPayloadHash(event.payloadHash().toLowerCase(Locale.ROOT)); row.setPayloadJson(event.payload().toString());
+        row.setStatus("RECEIVED"); row.setReceivedAt(LocalDateTime.now(ZoneOffset.UTC));
+        return row;
+    }
+
+    private long parsePositive(String value, String message) {
+        try { long parsed = Long.parseLong(value); if (parsed > 0) return parsed; }
+        catch (RuntimeException ignored) { }
+        throw new IllegalArgumentException(message);
+    }
+
+    private long parseOperatorId(String value) {
+        try { return value == null ? 0L : Long.parseLong(value); }
+        catch (NumberFormatException ignored) { return 0L; }
+    }
+}
@@ -42,6 +42,7 @@ CREATE TABLE IF NOT EXISTS fl_inbound_order (
  destination_name VARCHAR(128) NULL COMMENT '目的仓名称快照',
  hold_flag TINYINT(1) NOT NULL DEFAULT 0 COMMENT '当前是否扣货',
  current_hold_reason VARCHAR(500) NULL COMMENT '当前扣货理由快照',
+  last_overseas_hold_version BIGINT UNSIGNED NULL COMMENT '海外暂扣事件最新聚合版本，用于阻止乱序覆盖',
  total_package_count INT UNSIGNED NULL COMMENT '确认后的总箱数',
  actual_gross_weight_kg DECIMAL(18,3) NULL COMMENT '已确认称重图片含托盘毛重汇总kg',
  chargeable_weight_kg DECIMAL(18,3) NULL COMMENT '当前费用快照的计费重量kg',
@@ -102,8 +102,8 @@ CREATE TABLE IF NOT EXISTS fl_container_event_inbox (
  event_id VARCHAR(64) NOT NULL COMMENT '跨区事件全局唯一ID',
  event_type VARCHAR(64) NOT NULL COMMENT '事件类型',
  event_version INT UNSIGNED NOT NULL COMMENT '事件契约版本',
-  aggregate_id VARCHAR(64) NOT NULL COMMENT '货柜聚合ID',
-  aggregate_version BIGINT UNSIGNED NOT NULL COMMENT '货柜聚合版本',
+  aggregate_id VARCHAR(64) NOT NULL COMMENT '跨区业务聚合ID',
+  aggregate_version BIGINT UNSIGNED NOT NULL COMMENT '跨区业务聚合版本',
  payload_hash VARCHAR(80) NOT NULL COMMENT '规范化负载SHA-256摘要',
  payload_json JSON NOT NULL COMMENT '原始事件负载',
  status VARCHAR(16) NOT NULL COMMENT 'RECEIVED或PROCESSED',
@@ -111,7 +111,7 @@ CREATE TABLE IF NOT EXISTS fl_container_event_inbox (
  processed_at DATETIME(3) NULL COMMENT '处理完成时间（UTC）',
  PRIMARY KEY (event_id),
  KEY idx_container_event_aggregate (aggregate_id, aggregate_version)
-) ENGINE=InnoDB COMMENT='海外货柜事件消费Inbox';
+) ENGINE=InnoDB COMMENT='海外到国内业务事件消费Inbox';

 DROP PROCEDURE IF EXISTS phase3_container_real_unit_columns;
 DELIMITER $$
@@ -0,0 +1,87 @@
+package com.baiyue.firstleg.service;
+
+import com.baiyue.common.event.EventEnvelope;
+import com.baiyue.common.event.EventPayloadCanonicalizer;
+import com.baiyue.common.event.EventPayloadHash;
+import com.baiyue.common.event.EventTypes;
+import com.baiyue.common.event.OverseasInboundOrderDetainedChangedV1;
+import com.baiyue.common.event.RegionCode;
+import com.baiyue.firstleg.entity.InboundHoldRecordEntity;
+import com.baiyue.firstleg.entity.InboundOrderEntity;
+import com.baiyue.firstleg.mapper.ContainerEventInboxMapper;
+import com.baiyue.firstleg.mapper.InboundHoldRecordMapper;
+import com.baiyue.firstleg.mapper.InboundOrderMapper;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.time.Instant;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+@ExtendWith(MockitoExtension.class)
+class OverseasInboundOrderDetainedConsumerTest {
+    @Mock InboundOrderMapper inboundOrders;
+    @Mock InboundHoldRecordMapper holdRecords;
+    @Mock ContainerEventInboxMapper inbox;
+    ObjectMapper objectMapper;
+    OverseasInboundOrderDetainedConsumer consumer;
+
+    @BeforeEach
+    void setUp() {
+        objectMapper = new ObjectMapper().registerModule(new JavaTimeModule());
+        consumer = new OverseasInboundOrderDetainedConsumer(inboundOrders, holdRecords, inbox, objectMapper);
+    }
+
+    @Test
+    void appliesHoldBySourceInboundIdAndWritesHistory() {
+        InboundOrderEntity inbound = inbound(null);
+        when(inboundOrders.selectActiveByIdForUpdate(1001L)).thenReturn(inbound);
+        when(inboundOrders.applyOverseasHold(88L, 1001L, true, "等待查验", 7L, 66L)).thenReturn(1);
+
+        consumer.consume(event(7, true, "等待查验"));
+
+        verify(inboundOrders).applyOverseasHold(88L, 1001L, true, "等待查验", 7L, 66L);
+        ArgumentCaptor<InboundHoldRecordEntity> history = ArgumentCaptor.forClass(InboundHoldRecordEntity.class);
+        verify(holdRecords).insert(history.capture());
+        assertThat(history.getValue().getInboundId()).isEqualTo(1001L);
+        assertThat(history.getValue().getHoldAction()).isEqualTo("HOLD");
+        verify(inbox).updateById(argThat(row -> "PROCESSED".equals(row.getStatus())));
+    }
+
+    @Test
+    void staleReleaseCannotOverwriteNewerHoldState() {
+        when(inboundOrders.selectActiveByIdForUpdate(1001L)).thenReturn(inbound(9L));
+
+        consumer.consume(event(7, false, "解除"));
+
+        verify(inboundOrders, never()).applyOverseasHold(anyLong(), anyLong(), anyBoolean(), any(), anyLong(), anyLong());
+        verifyNoInteractions(holdRecords);
+        verify(inbox).updateById(argThat(row -> "PROCESSED".equals(row.getStatus())));
+    }
+
+    private InboundOrderEntity inbound(Long sourceVersion) {
+        InboundOrderEntity value = new InboundOrderEntity();
+        value.setId(1001L); value.setTenantId(88L); value.setInboundCode("IN-1001");
+        value.setLastOverseasHoldVersion(sourceVersion);
+        return value;
+    }
+
+    private EventEnvelope event(long version, boolean detained, String reason) {
+        var payload = new OverseasInboundOrderDetainedChangedV1("1001", "IN-1001", "9001",
+            "LM-9001", detained, reason, "66", "海外操作员", Instant.parse("2026-08-31T06:00:00Z"));
+        var node = objectMapper.valueToTree(payload);
+        return new EventEnvelope(String.format("00000000-0000-4000-8000-%012d", version),
+            EventTypes.OVERSEAS_INBOUND_ORDER_DETAINED_CHANGED, 1, payload.changedAt(),
+            RegionCode.OVERSEAS, RegionCode.CN, "88", "INBOUND_ORDER", "1001", version, 1L,
+            "00000000-0000-4000-8000-000000000099", "hold-1001-" + version, null, node,
+            EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(node)));
+    }
+}