优化

firstleg/src/main/java/com/baiyue/firstleg/client/OrderFulfillmentCompletionClient.java

@@ -0,0 +1,21 @@
+package com.baiyue.firstleg.client;
+
+import com.baiyue.common.core.api.ApiResponse;
+import org.springframework.cloud.openfeign.FeignClient;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestHeader;
+
+import java.time.Instant;
+
+@FeignClient(name = "order-service", contextId = "firstlegOrderFulfillmentCompletionClient",
+    url = "${baiyue.clients.order.url:}")
+public interface OrderFulfillmentCompletionClient {
+    @PostMapping("/internal/order/v1/fulfillment-completions")
+    ApiResponse<Result> complete(@RequestHeader("X-Internal-Service-Token") String token,
+                                 @RequestBody Request request);
+
+    record Request(String tenantId, String orderId, String orderNo, String inboundCode,
+                   String deliveryNo, Instant completedAt, String eventId, String payloadHash) { }
+    record Result(Long orderId, String status, boolean replayed) { }
+}

firstleg/src/main/java/com/baiyue/firstleg/controller/InternalOverseasContainerEventController.java

@@ -7,6 +7,7 @@ import com.baiyue.common.event.EventEnvelope;
 import com.baiyue.common.event.EventTypes;
 import com.baiyue.firstleg.service.OverseasContainerEventConsumer;
 import com.baiyue.firstleg.service.OverseasInboundOrderDetainedConsumer;
+import com.baiyue.firstleg.service.OverseasDeliveryCompletedConsumer;
 import lombok.RequiredArgsConstructor;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.web.bind.annotation.PostMapping;
@@ -22,6 +23,7 @@ import java.security.MessageDigest;
 public class InternalOverseasContainerEventController {
     private final OverseasContainerEventConsumer consumer;
     private final OverseasInboundOrderDetainedConsumer detainedConsumer;
+    private final OverseasDeliveryCompletedConsumer deliveryCompletedConsumer;
     @Value("${baiyue.security.internal-service-token}") private String expectedToken;
 
     @PostMapping("/internal/firstleg/v1/container-events")
@@ -30,7 +32,9 @@ public class InternalOverseasContainerEventController {
         @RequestBody EventEnvelope event) {
         authenticate(token);
         Object result = EventTypes.OVERSEAS_INBOUND_ORDER_DETAINED_CHANGED.equals(event.eventType())
-            ? detainedConsumer.consume(event) : consumer.consume(event);
+            ? detainedConsumer.consume(event)
+            : EventTypes.OVERSEAS_DELIVERY_COMPLETED.equals(event.eventType())
+                ? deliveryCompletedConsumer.consume(event) : consumer.consume(event);
         return ApiResponse.success(result);
     }
 

firstleg/src/main/java/com/baiyue/firstleg/mapper/ContainerBookingMapper.java

@@ -11,8 +11,8 @@ import org.apache.ibatis.annotations.Update;
 
 @Mapper
 public interface ContainerBookingMapper extends BaseMapper<ContainerBookingEntity> {
-    @Update("UPDATE fl_container_booking SET container_no=#{containerNo},ocean_booking_id=#{oceanBookingId},ocean_booking_requirement_id=#{requirementId},record_status='ACTIVE',bill_of_lading_no=#{billOfLadingNo},vessel_name=#{vesselName},voyage_no=#{voyageNo},shipping_company_id=#{shippingCompanyId},shipping_company_name=#{shippingCompanyName},departure_port_code=#{departurePortCode},departure_port_name=#{departurePortName},destination_port_code=#{destinationPortCode},destination_port_name=#{destinationPortName},channel_code=#{channelCode},channel_name=#{channelName},customer_id=#{customerId},customer_name=#{customerName},estimated_sailing_at=#{estimatedSailingAt},estimated_port_arrival_at=#{estimatedPortArrivalAt},updated_by=#{operatorId},updated_by_name=#{operatorName},updated_at=NOW(),version=version+1 WHERE id=#{id} AND version=#{version} AND source_type='SIMULATION' AND record_status='ACTIVE' AND ocean_booking_id IS NULL")
-    int linkReleasedBooking(@Param("id") long id, @Param("oceanBookingId") long oceanBookingId, @Param("requirementId") long requirementId, @Param("containerNo") String containerNo, @Param("billOfLadingNo") String billOfLadingNo, @Param("vesselName") String vesselName, @Param("voyageNo") String voyageNo, @Param("shippingCompanyId") Long shippingCompanyId, @Param("shippingCompanyName") String shippingCompanyName, @Param("departurePortCode") String departurePortCode, @Param("departurePortName") String departurePortName, @Param("destinationPortCode") String destinationPortCode, @Param("destinationPortName") String destinationPortName, @Param("channelCode") String channelCode, @Param("channelName") String channelName, @Param("customerId") Long customerId, @Param("customerName") String customerName, @Param("estimatedSailingAt") java.time.LocalDateTime estimatedSailingAt, @Param("estimatedPortArrivalAt") java.time.LocalDateTime estimatedPortArrivalAt, @Param("version") int version, @Param("operatorId") long operatorId, @Param("operatorName") String operatorName);
+    @Update("UPDATE fl_container_booking SET container_no=#{containerNo},ocean_booking_id=#{oceanBookingId},ocean_booking_requirement_id=#{requirementId},record_status='ACTIVE',bill_of_lading_no=#{billOfLadingNo},vessel_name=#{vesselName},voyage_no=#{voyageNo},shipping_company_id=#{shippingCompanyId},shipping_company_name=#{shippingCompanyName},departure_port_code=#{departurePortCode},departure_port_name=#{departurePortName},destination_port_code=#{destinationPortCode},destination_port_name=#{destinationPortName},customer_id=#{customerId},customer_name=#{customerName},estimated_sailing_at=#{estimatedSailingAt},estimated_port_arrival_at=#{estimatedPortArrivalAt},updated_by=#{operatorId},updated_by_name=#{operatorName},updated_at=NOW(),version=version+1 WHERE id=#{id} AND version=#{version} AND source_type='SIMULATION' AND record_status='ACTIVE' AND ocean_booking_id IS NULL")
+    int linkReleasedBooking(@Param("id") long id, @Param("oceanBookingId") long oceanBookingId, @Param("requirementId") long requirementId, @Param("containerNo") String containerNo, @Param("billOfLadingNo") String billOfLadingNo, @Param("vesselName") String vesselName, @Param("voyageNo") String voyageNo, @Param("shippingCompanyId") Long shippingCompanyId, @Param("shippingCompanyName") String shippingCompanyName, @Param("departurePortCode") String departurePortCode, @Param("departurePortName") String departurePortName, @Param("destinationPortCode") String destinationPortCode, @Param("destinationPortName") String destinationPortName, @Param("customerId") Long customerId, @Param("customerName") String customerName, @Param("estimatedSailingAt") java.time.LocalDateTime estimatedSailingAt, @Param("estimatedPortArrivalAt") java.time.LocalDateTime estimatedPortArrivalAt, @Param("version") int version, @Param("operatorId") long operatorId, @Param("operatorName") String operatorName);
     @Update("""
         UPDATE fl_container_booking SET container_no=#{containerNo},updated_by=#{operatorId},
           updated_by_name=#{operatorName},updated_at=CURRENT_TIMESTAMP(3),version=version+1

firstleg/src/main/java/com/baiyue/firstleg/mapper/InboundOrderMapper.java

@@ -24,6 +24,29 @@ public interface InboundOrderMapper extends BaseMapper<InboundOrderEntity> {
                           @Param("detained") boolean detained, @Param("reason") String reason,
                           @Param("sourceVersion") long sourceVersion, @Param("operatorId") long operatorId);
 
+    @Update("""
+        UPDATE fl_inbound_order
+           SET process_status='DELIVERED', updated_by=0, updated_at=#{completedAt}, version=version+1
+         WHERE tenant_id=#{tenantId} AND id=#{inboundId} AND deleted=0 AND process_status='LOADED'
+        """)
+    int completeOverseasDelivery(@Param("tenantId") long tenantId, @Param("inboundId") long inboundId,
+                                 @Param("completedAt") java.time.LocalDateTime completedAt);
+
+    @Select("""
+        SELECT COUNT(*) FROM fl_inbound_order
+         WHERE tenant_id=#{tenantId} AND order_id=#{orderId} AND deleted=0 AND process_status!='DELIVERED'
+        """)
+    int countNotDeliveredByOrder(@Param("tenantId") long tenantId, @Param("orderId") long orderId);
+
+    @Update("""
+        UPDATE fl_inbound_order
+           SET order_status_snapshot='COMPLETED', updated_by=0, updated_at=#{completedAt}, version=version+1
+         WHERE tenant_id=#{tenantId} AND order_id=#{orderId} AND deleted=0
+           AND (order_status_snapshot IS NULL OR order_status_snapshot!='COMPLETED')
+        """)
+    int markOrderSnapshotCompleted(@Param("tenantId") long tenantId, @Param("orderId") long orderId,
+                                   @Param("completedAt") java.time.LocalDateTime completedAt);
+
     @Update("""
         UPDATE fl_inbound_order
            SET hold_flag=#{holdFlag}, current_hold_reason=#{reason},

firstleg/src/main/java/com/baiyue/firstleg/oceanbooking/mapper/OceanBookingMapper.java

@@ -76,16 +76,14 @@ public interface OceanBookingMapper extends BaseMapper<OceanBookingEntity> {
           AND allocated.requirement_id=requirement.id
         WHERE booking.status='RELEASED'
           AND booking.warehouse_id=#{query.warehouseId}
-          AND booking.channel_code IN
-  <foreach collection="channelCodes" item="channelCode" open="(" separator="," close=")">#{channelCode}</foreach>
+          AND booking.channel_code=#{query.channelGroupCode}
           AND requirement.container_type=#{query.containerType}
           AND requirement.quantity>COALESCE(allocated.allocated_quantity,0)
         ORDER BY booking.updated_at DESC,booking.id DESC,requirement.sort_no ASC,requirement.id ASC
         </script>
         """)
     Page<LinkCandidate> selectLinkCandidates(Page<LinkCandidate> page,
-                                              @Param("query") LinkCandidateQuery query,
-                                              @Param("channelCodes") java.util.Set<String> channelCodes);
+                                              @Param("query") LinkCandidateQuery query);
 
     @Update("""
         UPDATE fl_ocean_booking SET status=#{toStatus},

firstleg/src/main/java/com/baiyue/firstleg/oceanbooking/repository/OceanBookingRepository.java

@@ -33,8 +33,8 @@ public class OceanBookingRepository {
     public PageResult<OceanBookingEntity> page(Query query) {
         return PageResult.from(bookings.selectPage(new Page<>(query.page(), query.size()), query));
     }
-    public PageResult<LinkCandidate> linkCandidates(LinkCandidateQuery query, java.util.Set<String> channelCodes) {
-        return PageResult.from(bookings.selectLinkCandidates(new Page<>(query.page(), query.size()), query, channelCodes));
+    public PageResult<LinkCandidate> linkCandidates(LinkCandidateQuery query) {
+        return PageResult.from(bookings.selectLinkCandidates(new Page<>(query.page(), query.size()), query));
     }
     public OceanBookingEntity find(long id) { return bookings.selectById(id); }
     public OceanBookingEntity findForUpdate(long id) { return bookings.selectForUpdate(id); }

firstleg/src/main/java/com/baiyue/firstleg/oceanbooking/service/OceanBookingService.java

@@ -77,7 +77,7 @@ public class OceanBookingService {
         catch (IllegalArgumentException exception) { throw badRequest(exception.getMessage()); }
         String containerType = ContainerSpecificationCatalog.require(query.containerType()).containerType();
         return repository.linkCandidates(new LinkCandidateQuery(query.warehouseId(),
-            group.code(), containerType, query.page(), query.size()), group.channelCodes());
+            group.code(), containerType, query.page(), query.size()));
     }
 
     public List<ChannelOption> channels(long originWarehouseId) {
@@ -239,16 +239,18 @@ public class OceanBookingService {
             || requirements == null || requirements.isEmpty()) throw badRequest("订舱资料尚未填写完整，无法发布");
         validateRange(booking.getEstimatedSailingAt(), booking.getEstimatedArrivalAt());
     }
-    private ChannelOption requireActiveChannel(Long originWarehouseId, String channelCode) {
+    private SmartLoadingChannelGroupCatalog.Definition requireActiveChannelGroup(Long originWarehouseId, String channelGroupCode) {
         if (originWarehouseId == null || originWarehouseId <= 0) throw badRequest("请选择有效起运仓");
-        String normalized = normalizeChannel(channelCode);
+        SmartLoadingChannelGroupCatalog.Definition group;
+        try { group = SmartLoadingChannelGroupCatalog.require(normalizeChannel(channelGroupCode)); }
+        catch (IllegalArgumentException exception) { throw badRequest(exception.getMessage()); }
         Long versionId = priceBooks.activeVersionId();
         if (versionId == null) throw conflict("当前没有 ACTIVE 价格版本");
-        ChannelRow row = priceBooks.channels(versionId, originWarehouseId).stream()
-            .filter(value -> normalized.equals(normalizeChannel(value.getChannelCode())))
-            .findFirst().orElseThrow(() -> conflict("所选物流渠道已失效，请重新选择"));
-        if (!StringUtils.hasText(row.getChannelName())) throw conflict("所选物流渠道名称配置无效");
-        return channelOption(row, java.util.Map.of());
+        boolean active = priceBooks.channels(versionId, originWarehouseId).stream()
+            .map(ChannelRow::getChannelCode).filter(StringUtils::hasText).map(String::trim)
+            .anyMatch(group.channelCodes()::contains);
+        if (!active) throw conflict("所选物流渠道分组在当前仓库已失效，请重新选择");
+        return group;
     }
 
     private static ChannelOption channelOption(ChannelRow row, java.util.Map<String, List<String>> taxModes) {
@@ -269,7 +271,7 @@ public class OceanBookingService {
         if (d.warehouseId() != null && StringUtils.hasText(d.channelCode())) {
             WarehouseMasterDataClient.Warehouse warehouse = requireDomesticTransitWarehouse(d.warehouseId());
             warehouseName = warehouse.warehouseName();
-            channelName = requireActiveChannel(warehouse.id(), d.channelCode()).channelName();
+            channelName = requireActiveChannelGroup(warehouse.id(), d.channelCode()).name();
         }
         var carrier = resolveCarrier(d.shippingCompanyId(), d.shippingCompanyName(), d.carrierCode());
         var agent = resolveBookingAgent(d.bookingAgentId(), d.bookingAgentName(), carrier == null ? null : carrier.carrierCode());

firstleg/src/main/java/com/baiyue/firstleg/repository/ContainerRepository.java

@@ -37,7 +37,7 @@ public class ContainerRepository {
     public int linkReleasedBooking(long id, long oceanBookingId, long requirementId, String containerNo,
                                    com.baiyue.firstleg.oceanbooking.entity.OceanBookingEntity booking,
                                    int version, long operatorId, String operatorName) {
-        return mapper.linkReleasedBooking(id, oceanBookingId, requirementId, containerNo, booking.getBillOfLadingNo(), booking.getVesselName(), booking.getVoyageNo(), booking.getShippingCompanyId(), booking.getShippingCompanyName(), booking.getDeparturePortCode(), booking.getDeparturePortName(), booking.getDestinationPortCode(), booking.getDestinationPortName(), booking.getChannelCode(), booking.getChannelName(), booking.getCustomerId(), booking.getCustomerName(), booking.getEstimatedSailingAt(), booking.getEstimatedArrivalAt(), version, operatorId, operatorName);
+        return mapper.linkReleasedBooking(id, oceanBookingId, requirementId, containerNo, booking.getBillOfLadingNo(), booking.getVesselName(), booking.getVoyageNo(), booking.getShippingCompanyId(), booking.getShippingCompanyName(), booking.getDeparturePortCode(), booking.getDeparturePortName(), booking.getDestinationPortCode(), booking.getDestinationPortName(), booking.getCustomerId(), booking.getCustomerName(), booking.getEstimatedSailingAt(), booking.getEstimatedArrivalAt(), version, operatorId, operatorName);
     }
     public int assignRealContainerNo(long id, String containerNo, int version,
                                      long operatorId, String operatorName) {

firstleg/src/main/java/com/baiyue/firstleg/service/ContainerService.java

@@ -89,7 +89,7 @@ public class ContainerService {
         String containerNo = requestedContainerNo == null ? null : Iso6346ContainerNumber.require(requestedContainerNo);
         if (containerNo != null && repository.containerNoExists(containerNo, id)) throw conflict("集装箱号已存在");
         if (!java.util.Objects.equals(container.getWarehouseId(), booking.getWarehouseId())
-            || !SmartLoadingChannelGroupCatalog.contains(container.getChannelGroupCode(), booking.getChannelCode())
+            || !java.util.Objects.equals(container.getChannelGroupCode(), booking.getChannelCode())
             || !java.util.Objects.equals(container.getContainerType(), requirement.getContainerType())) {
             throw conflict("模拟柜与真实订舱柜仓库、渠道分组或柜型不一致");
         }

firstleg/src/main/java/com/baiyue/firstleg/service/OverseasDeliveryCompletedConsumer.java

@@ -0,0 +1,148 @@
+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.OverseasDeliveryCompletedV1;
+import com.baiyue.common.event.RegionCode;
+import com.baiyue.firstleg.client.OrderFulfillmentCompletionClient;
+import com.baiyue.firstleg.entity.ContainerEventInboxEntity;
+import com.baiyue.firstleg.entity.InboundOrderEntity;
+import com.baiyue.firstleg.entity.InboundStatusHistoryEntity;
+import com.baiyue.firstleg.mapper.ContainerEventInboxMapper;
+import com.baiyue.firstleg.mapper.InboundOrderMapper;
+import com.baiyue.firstleg.mapper.InboundStatusHistoryMapper;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.RequiredArgsConstructor;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.util.StringUtils;
+
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.util.Locale;
+import java.util.Objects;
+
+@Service
+@RequiredArgsConstructor
+public class OverseasDeliveryCompletedConsumer {
+    private final InboundOrderMapper inboundOrders;
+    private final InboundStatusHistoryMapper histories;
+    private final ContainerEventInboxMapper inbox;
+    private final OrderFulfillmentCompletionClient orderClient;
+    private final ObjectMapper objectMapper;
+    @Value("${baiyue.security.internal-service-token}") private String internalToken;
+
+    @Transactional
+    public InboundOrderEntity consume(EventEnvelope event) {
+        validate(event);
+        OverseasDeliveryCompletedV1 payload = payload(event);
+        long inboundId = positive(payload.sourceInboundId(), "来源入仓单 ID 无效");
+        long tenantId = positive(event.tenantId(), "客户 ID 无效");
+        if (!Objects.equals(event.aggregateId(), payload.sourceInboundId())) {
+            throw new IllegalArgumentException("事件 aggregateId 与来源入仓单 ID 不一致");
+        }
+        ContainerEventInboxEntity existing = inbox.selectById(event.eventId());
+        if (existing != null) {
+            if (!existing.getPayloadHash().equalsIgnoreCase(event.payloadHash())) {
+                throw new IllegalArgumentException("相同 eventId 对应的事件摘要冲突");
+            }
+            return required(inboundId, tenantId, payload);
+        }
+        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 required(inboundId, tenantId, payload);
+        }
+        InboundOrderEntity inbound = required(inboundId, tenantId, payload);
+        if (!"LOADED".equals(inbound.getProcessStatus()) && !"DELIVERED".equals(inbound.getProcessStatus())) {
+            throw new IllegalArgumentException("来源头程入仓单尚未装柜完成");
+        }
+        if ("LOADED".equals(inbound.getProcessStatus())) {
+            LocalDateTime completedAt = LocalDateTime.ofInstant(payload.completedAt(), ZoneOffset.UTC);
+            if (inboundOrders.completeOverseasDelivery(tenantId, inboundId, completedAt) != 1) {
+                throw new IllegalStateException("来源头程入仓单状态并发变化");
+            }
+            InboundStatusHistoryEntity history = new InboundStatusHistoryEntity();
+            history.setId(IdWorker.getId()); history.setTenantId(tenantId); history.setInboundId(inboundId);
+            history.setFromStatus("LOADED"); history.setToStatus("DELIVERED");
+            history.setAction("COMPLETE_DELIVERY");
+            history.setReason("海外派送单 " + payload.deliveryNo() + " 已签收");
+            history.setOperatorId(0L); history.setOperatorName("OVERSEAS_EVENT_BRIDGE");
+            history.setOccurredAt(completedAt); histories.insert(history);
+        }
+        long sourceOrderId = positive(payload.sourceOrderId(), "来源订单 ID 无效");
+        if (inboundOrders.countNotDeliveredByOrder(tenantId, sourceOrderId) == 0) {
+            var response = orderClient.complete(internalToken, new OrderFulfillmentCompletionClient.Request(
+                event.tenantId(), payload.sourceOrderId(), payload.sourceOrderNo(), payload.inboundCode(),
+                payload.deliveryNo(), payload.completedAt(), event.eventId(), event.payloadHash()));
+            if (response == null || !response.successful() || response.data() == null
+                || !"COMPLETED".equals(response.data().status())) {
+                throw new IllegalStateException("Order 服务未确认来源订单完成");
+            }
+            inboundOrders.markOrderSnapshotCompleted(tenantId, sourceOrderId,
+                LocalDateTime.ofInstant(payload.completedAt(), ZoneOffset.UTC));
+        }
+        inboxRow.setStatus("PROCESSED"); inboxRow.setProcessedAt(LocalDateTime.now(ZoneOffset.UTC));
+        inbox.updateById(inboxRow);
+        return required(inboundId, tenantId, payload);
+    }
+
+    private void validate(EventEnvelope event) {
+        event.requireValid();
+        if (!EventTypes.OVERSEAS_DELIVERY_COMPLETED.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 OverseasDeliveryCompletedV1 payload(EventEnvelope event) {
+        try {
+            OverseasDeliveryCompletedV1 value = objectMapper.treeToValue(event.payload(), OverseasDeliveryCompletedV1.class);
+            if (value.completedAt() == null || !StringUtils.hasText(value.deliveryNo())
+                || !StringUtils.hasText(value.sourceOrderId()) || !StringUtils.hasText(value.sourceOrderNo())
+                || !StringUtils.hasText(value.inboundCode())) throw new IllegalArgumentException("字段不完整");
+            return value;
+        } catch (Exception exception) {
+            throw new IllegalArgumentException("海外派送完成事件负载无效", exception);
+        }
+    }
+
+    private InboundOrderEntity required(long inboundId, long tenantId, OverseasDeliveryCompletedV1 payload) {
+        InboundOrderEntity inbound = inboundOrders.selectActiveByIdForUpdate(inboundId);
+        if (inbound == null || !Objects.equals(inbound.getTenantId(), tenantId)
+            || !Objects.equals(inbound.getInboundCode(), payload.inboundCode())
+            || !Objects.equals(String.valueOf(inbound.getOrderId()), payload.sourceOrderId())
+            || !Objects.equals(inbound.getOrderNo(), payload.sourceOrderNo())
+            || (StringUtils.hasText(payload.fbaWarehouseCode()) && StringUtils.hasText(inbound.getDestinationCode())
+                && !Objects.equals(inbound.getDestinationCode(), payload.fbaWarehouseCode()))) {
+            throw new IllegalArgumentException("事件与来源头程入仓单归属不一致");
+        }
+        return inbound;
+    }
+
+    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 static long positive(String value, String message) {
+        try { long parsed = Long.parseLong(value); if (parsed > 0) return parsed; }
+        catch (RuntimeException ignored) { }
+        throw new IllegalArgumentException(message);
+    }
+}

firstleg/src/main/resources/db/phase3-ocean-booking-schema.sql

@@ -7,8 +7,8 @@ CREATE TABLE IF NOT EXISTS fl_ocean_booking (
   status VARCHAR(24) NOT NULL DEFAULT 'DRAFT' COMMENT '订舱状态：DRAFT草稿、RELEASED已发布且永久只读',
   customer_id BIGINT UNSIGNED NULL COMMENT '可选客户ID快照',
   customer_name VARCHAR(128) NULL COMMENT '可选客户名称快照',
-  channel_code VARCHAR(64) NULL COMMENT '订舱渠道编码快照',
-  channel_name VARCHAR(128) NULL COMMENT '订舱渠道名称快照',
+  channel_code VARCHAR(64) NULL COMMENT '订舱计划物流渠道组编码快照',
+  channel_name VARCHAR(128) NULL COMMENT '订舱计划物流渠道组名称快照',
   warehouse_id BIGINT UNSIGNED NULL COMMENT '国内中转仓ID快照',
   warehouse_name VARCHAR(128) NULL COMMENT '国内中转仓名称快照',
   bill_of_lading_no VARCHAR(128) NULL COMMENT '提单号',

firstleg/src/main/resources/openapi/ocean-bookings-v1.yaml

@@ -165,7 +165,7 @@ components:
       properties:
         customerId: {type: integer, format: int64, minimum: 1, nullable: true, description: 可选启用客户ID；多客户订舱留空，空值不是公开访问授权}
         customerName: {type: string, nullable: true, description: 可选名称提示；有客户ID时由服务端校验并保存客户主数据名称，清空客户时同时置null}
-        channelCode: {type: string, nullable: true}
+        channelCode: {type: string, nullable: true, description: 固定计划物流渠道组编码}
         warehouseId: {type: integer, format: int64, nullable: true}
         billOfLadingNo: {type: string, nullable: true}
         shippingCompanyId: {type: integer, format: int64, nullable: true}

firstleg/src/test/java/com/baiyue/firstleg/oceanbooking/repository/OceanBookingMapperContractTest.java

@@ -49,12 +49,12 @@ class OceanBookingMapperContractTest {
     @Test
     void linkCandidatesUsesOneBoundedSetQueryAndExactRemainingSemantics() throws Exception {
         String sql = String.join("\n", OceanBookingMapper.class.getDeclaredMethod("selectLinkCandidates",
-            com.baomidou.mybatisplus.extension.plugins.pagination.Page.class, LinkCandidateQuery.class, java.util.Set.class)
+            com.baomidou.mybatisplus.extension.plugins.pagination.Page.class, LinkCandidateQuery.class)
             .getAnnotation(Select.class).value());
         assertThat(sql)
             .contains("booking.status='RELEASED'")
             .contains("booking.warehouse_id=#{query.warehouseId}")
-            .contains("<foreach collection=\"channelCodes\"", "#{channelCode}")
+            .contains("booking.channel_code=#{query.channelGroupCode}")
             .contains("requirement.container_type=#{query.containerType}")
             .contains("FROM fl_ocean_booking_container_allocation")
             .contains("container.record_status='ACTIVE'")

firstleg/src/test/java/com/baiyue/firstleg/oceanbooking/service/OceanBookingServiceTest.java

@@ -150,7 +150,7 @@ class OceanBookingServiceTest {
         Detail result = service.create(9L, "admin", "draft-complete", new CreateRequest(0, completeDraft()));
 
         assertThat(result.oceanBooking().getStatus()).isEqualTo("DRAFT");
-        assertThat(inserted.get().getChannelName()).isEqualTo("美西海运");
+        assertThat(inserted.get().getChannelName()).isEqualTo("美森正班");
         assertThat(inserted.get().getWarehouseName()).isEqualTo("深圳中转仓");
         assertThat(inserted.get().getBookingAgentName()).isEqualTo("订舱代理");
         assertThat(inserted.get().getCustomerId()).isEqualTo(100L);
@@ -328,8 +328,8 @@ class OceanBookingServiceTest {
     void linkCandidatesDefaultsPaginationAndDelegatesOneNormalizedSetQuery() {
         LinkCandidate candidate = new LinkCandidate(101L, "OB-260818-0001", 201L, "40HQ",
             new BigDecimal("68"), new BigDecimal("19500"), 3, 1, 2,
-            "客户A", "深圳中转仓", "美西海运");
-        when(repository.linkCandidates(any(), anySet())).thenReturn(new PageResult<>(List.of(candidate), 1, 20, 1, 1));
+            "客户A", "深圳中转仓", "美森正班");
+        when(repository.linkCandidates(any())).thenReturn(new PageResult<>(List.of(candidate), 1, 20, 1, 1));
 
         PageResult<LinkCandidate> result = service.linkCandidates(new LinkCandidateQuery(701L,
             " MATSON_CLX ", " 40HQ ", null, null));
@@ -337,13 +337,11 @@ class OceanBookingServiceTest {
         assertThat(result.records()).containsExactly(candidate);
         assertThat(result.records().get(0).remainingQuantity()).isEqualTo(2);
         ArgumentCaptor<LinkCandidateQuery> query = ArgumentCaptor.forClass(LinkCandidateQuery.class);
-        @SuppressWarnings("unchecked") ArgumentCaptor<java.util.Set<String>> channels = ArgumentCaptor.forClass(java.util.Set.class);
-        verify(repository).linkCandidates(query.capture(), channels.capture());
+        verify(repository).linkCandidates(query.capture());
         assertThat(query.getValue().channelGroupCode()).isEqualTo("MATSON_CLX");
         assertThat(query.getValue().containerType()).isEqualTo("40HQ");
         assertThat(query.getValue().page()).isEqualTo(1L);
         assertThat(query.getValue().size()).isEqualTo(20L);
-        assertThat(channels.getValue()).contains("MATSON_CLX_TRUCK", "MATSON_CLX_PARCEL", "ZBYHK");
         verify(repository, never()).find(anyLong());
         verify(repository, never()).requirements(anyLong());
         verify(repository, never()).containers(anyLong());
@@ -351,7 +349,7 @@ class OceanBookingServiceTest {
 
     @Test
     void linkCandidatesReturnsNormalEmptyPageWithOneSetQueryAndNoDiagnosticFields() {
-        when(repository.linkCandidates(any(), anySet())).thenReturn(new PageResult<>(List.of(), 1, 20, 0, 0));
+        when(repository.linkCandidates(any())).thenReturn(new PageResult<>(List.of(), 1, 20, 0, 0));
 
         PageResult<LinkCandidate> result = service.linkCandidates(
             new LinkCandidateQuery(701L, "MATSON_CLX", "45HQ", null, null));
@@ -359,7 +357,7 @@ class OceanBookingServiceTest {
         assertThat(result.records()).isEmpty();
         assertThat(result.total()).isZero();
         assertThat(result.pages()).isZero();
-        verify(repository, times(1)).linkCandidates(any(), anySet());
+        verify(repository, times(1)).linkCandidates(any());
         assertThat(LinkCandidate.class.getRecordComponents()).extracting(java.lang.reflect.RecordComponent::getName)
             .doesNotContain("mismatchReasons", "disabledReason", "warehouseId", "channelCode");
     }
@@ -386,7 +384,7 @@ class OceanBookingServiceTest {
 
     private void stubMasterData() {
         ChannelRow row = new ChannelRow();
-        row.setPriceBookVersionId(51L); row.setChannelCode("SEA-US"); row.setChannelName("美西海运");
+        row.setPriceBookVersionId(51L); row.setChannelCode("MATSON_CLX_TRUCK"); row.setChannelName("美森正班海卡");
         row.setPriceModel("FBA_WAREHOUSE_RATE");
         when(priceBooks.activeVersionId()).thenReturn(51L);
         when(priceBooks.channels(51L, 701L)).thenReturn(List.of(row));
@@ -406,7 +404,7 @@ class OceanBookingServiceTest {
     }
 
     private static DraftData completeDraft() {
-        return new DraftData(100L, "客户", "SEA-US", 701L, "BL-001", 71L, "船公司", "MATSON",
+        return new DraftData(100L, "客户", "MATSON_CLX", 701L, "BL-001", 71L, "船公司", "MATSON",
             81L, null, "VESSEL", "V001", "CNSZX", "深圳", "USLAX", "洛杉矶",
             "LAX", LocalDateTime.of(2026, 9, 1, 0, 0),
             LocalDateTime.of(2026, 9, 20, 0, 0), null,
@@ -419,7 +417,7 @@ class OceanBookingServiceTest {
 
     private static OceanBookingEntity completeBooking(String status, int version) {
         OceanBookingEntity value = booking(status, version);
-        value.setCustomerId(100L); value.setCustomerName("客户"); value.setChannelCode("SEA-US");
+        value.setCustomerId(100L); value.setCustomerName("客户"); value.setChannelCode("MATSON_CLX");
         value.setWarehouseId(701L); value.setBillOfLadingNo("BL-001"); value.setShippingCompanyId(71L);
         value.setCarrierCode("MATSON");
         value.setBookingAgentId(81L); value.setVesselName("VESSEL"); value.setVoyageNo("V001");

firstleg/src/test/java/com/baiyue/firstleg/repository/ContainerBookingMapperContractTest.java

@@ -64,6 +64,7 @@ class ContainerBookingMapperContractTest {
             .filter(candidate -> candidate.getName().equals("linkReleasedBooking"))
             .findFirst().orElseThrow();
         String sql = String.join("\n", method.getAnnotation(Update.class).value());
+        assertThat(sql).doesNotContain("channel_code=#{channelCode}", "channel_name=#{channelName}");
         var parameterNames = Arrays.stream(method.getParameters())
             .map(parameter -> parameter.getAnnotation(Param.class).value())
             .toList();

firstleg/src/test/java/com/baiyue/firstleg/service/ContainerServiceTest.java

@@ -207,7 +207,7 @@ class ContainerServiceTest {
         container.setWarehouseId(101L); container.setChannelGroupCode("EXX"); container.setChannelCode(null); container.setContainerType("40HQ");
         container.setCapacityCbm(new BigDecimal("85.5000")); container.setMaxWeightKg(new BigDecimal("27500.000"));
         OceanBookingEntity booking = new OceanBookingEntity();
-        booking.setId(21L); booking.setStatus("RELEASED"); booking.setWarehouseId(101L); booking.setChannelCode("EXX_RAIL_NY");
+        booking.setId(21L); booking.setStatus("RELEASED"); booking.setWarehouseId(101L); booking.setChannelCode("EXX");
         OceanBookingContainerRequirementEntity requirement = new OceanBookingContainerRequirementEntity();
         requirement.setId(31L); requirement.setQuantity(1); requirement.setContainerType("40HQ"); requirement.setCapacityCbm(new BigDecimal("68.0000")); requirement.setMaxWeightKg(new BigDecimal("19500.000"));
         when(repository.find(11L)).thenReturn(container, container);
@@ -319,7 +319,7 @@ class ContainerServiceTest {
         second.setChannelGroupCode("MATSON_CLX"); second.setChannelCode(null); second.setContainerType("40HQ");
         OceanBookingEntity oceanBooking = new OceanBookingEntity();
         oceanBooking.setId(21L); oceanBooking.setStatus("RELEASED");
-        oceanBooking.setWarehouseId(101L); oceanBooking.setChannelCode("MATSON_CLX_TRUCK");
+        oceanBooking.setWarehouseId(101L); oceanBooking.setChannelCode("MATSON_CLX");
         OceanBookingContainerRequirementEntity requirement = new OceanBookingContainerRequirementEntity();
         requirement.setId(31L); requirement.setQuantity(2); requirement.setContainerType("40HQ");
         when(repository.find(11L)).thenReturn(first);
@@ -362,19 +362,19 @@ class ContainerServiceTest {
     @Test
     @DisplayName("首次关联拒绝装柜仓库不一致")
     void rejectsLinkWhenWarehouseDiffers() {
-        assertThreeFieldMismatch(102L, "MATSON_CLX_TRUCK", "40HQ");
+        assertThreeFieldMismatch(102L, "MATSON_CLX", "40HQ");
     }
 
     @Test
     @DisplayName("首次关联拒绝物流渠道不一致")
     void rejectsLinkWhenChannelDiffers() {
-        assertThreeFieldMismatch(101L, "HEDE_TRUCK", "40HQ");
+        assertThreeFieldMismatch(101L, "HEDE", "40HQ");
     }
 
     @Test
     @DisplayName("首次关联拒绝柜型不一致")
     void rejectsLinkWhenContainerTypeDiffers() {
-        assertThreeFieldMismatch(101L, "MATSON_CLX_TRUCK", "45HQ");
+        assertThreeFieldMismatch(101L, "MATSON_CLX", "45HQ");
     }
 
     @Test
@@ -443,7 +443,7 @@ class ContainerServiceTest {
         container.setContainerType("40HQ");
         OceanBookingEntity oceanBooking = new OceanBookingEntity();
         oceanBooking.setId(21L); oceanBooking.setStatus("RELEASED");
-        oceanBooking.setWarehouseId(101L); oceanBooking.setChannelCode("MATSON_CLX_TRUCK");
+        oceanBooking.setWarehouseId(101L); oceanBooking.setChannelCode("MATSON_CLX");
         OceanBookingContainerRequirementEntity requirement = new OceanBookingContainerRequirementEntity();
         requirement.setId(31L); requirement.setQuantity(2); requirement.setContainerType("40HQ");
         when(repository.find(11L)).thenReturn(container);

firstleg/src/test/java/com/baiyue/firstleg/service/OverseasDeliveryCompletedConsumerTest.java

@@ -0,0 +1,86 @@
+package com.baiyue.firstleg.service;
+
+import com.baiyue.common.core.api.ApiResponse;
+import com.baiyue.common.event.*;
+import com.baiyue.firstleg.client.OrderFulfillmentCompletionClient;
+import com.baiyue.firstleg.entity.InboundOrderEntity;
+import com.baiyue.firstleg.mapper.ContainerEventInboxMapper;
+import com.baiyue.firstleg.mapper.InboundOrderMapper;
+import com.baiyue.firstleg.mapper.InboundStatusHistoryMapper;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import org.junit.jupiter.api.Test;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.time.Instant;
+
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+class OverseasDeliveryCompletedConsumerTest {
+    @Test
+    void completesOnlyEventSourceInboundAndOrder() {
+        InboundOrderMapper inbounds = mock(InboundOrderMapper.class);
+        InboundStatusHistoryMapper histories = mock(InboundStatusHistoryMapper.class);
+        ContainerEventInboxMapper inbox = mock(ContainerEventInboxMapper.class);
+        OrderFulfillmentCompletionClient orders = mock(OrderFulfillmentCompletionClient.class);
+        ObjectMapper json = new ObjectMapper().registerModule(new JavaTimeModule());
+        OverseasDeliveryCompletedConsumer consumer = new OverseasDeliveryCompletedConsumer(
+            inbounds, histories, inbox, orders, json);
+        ReflectionTestUtils.setField(consumer, "internalToken", "token");
+        InboundOrderEntity inbound = new InboundOrderEntity();
+        inbound.setId(101L); inbound.setTenantId(88L); inbound.setOrderId(201L);
+        inbound.setOrderNo("FL-201"); inbound.setInboundCode("IN-1"); inbound.setDestinationCode("ONT8");
+        inbound.setProcessStatus("LOADED");
+        when(inbounds.selectActiveByIdForUpdate(101L)).thenReturn(inbound);
+        when(inbounds.completeOverseasDelivery(eq(88L), eq(101L), any())).thenReturn(1);
+        when(inbounds.countNotDeliveredByOrder(88L, 201L)).thenReturn(0);
+        when(orders.complete(eq("token"), any())).thenReturn(ApiResponse.success(
+            new OrderFulfillmentCompletionClient.Result(201L, "COMPLETED", false)));
+
+        consumer.consume(event(json));
+
+        verify(inbounds).completeOverseasDelivery(eq(88L), eq(101L), any());
+        verify(orders).complete(eq("token"), argThat(request ->
+            "201".equals(request.orderId()) && "IN-1".equals(request.inboundCode())
+                && "DL-ONT8".equals(request.deliveryNo())));
+        verify(inbounds).markOrderSnapshotCompleted(eq(88L), eq(201L), any());
+        verify(inbounds, never()).completeOverseasDelivery(eq(88L), eq(102L), any());
+    }
+
+    @Test
+    void leavesOrderOpenWhileAnotherSourceInboundIsNotDelivered() {
+        InboundOrderMapper inbounds = mock(InboundOrderMapper.class);
+        InboundStatusHistoryMapper histories = mock(InboundStatusHistoryMapper.class);
+        ContainerEventInboxMapper inbox = mock(ContainerEventInboxMapper.class);
+        OrderFulfillmentCompletionClient orders = mock(OrderFulfillmentCompletionClient.class);
+        ObjectMapper json = new ObjectMapper().registerModule(new JavaTimeModule());
+        OverseasDeliveryCompletedConsumer consumer = new OverseasDeliveryCompletedConsumer(
+            inbounds, histories, inbox, orders, json);
+        ReflectionTestUtils.setField(consumer, "internalToken", "token");
+        InboundOrderEntity inbound = new InboundOrderEntity();
+        inbound.setId(101L); inbound.setTenantId(88L); inbound.setOrderId(201L);
+        inbound.setOrderNo("FL-201"); inbound.setInboundCode("IN-1"); inbound.setDestinationCode("ONT8");
+        inbound.setProcessStatus("LOADED");
+        when(inbounds.selectActiveByIdForUpdate(101L)).thenReturn(inbound);
+        when(inbounds.completeOverseasDelivery(eq(88L), eq(101L), any())).thenReturn(1);
+        when(inbounds.countNotDeliveredByOrder(88L, 201L)).thenReturn(1);
+
+        consumer.consume(event(json));
+
+        verifyNoInteractions(orders);
+        verify(inbounds, never()).markOrderSnapshotCompleted(anyLong(), anyLong(), any());
+    }
+
+    private EventEnvelope event(ObjectMapper json) {
+        Instant time = Instant.parse("2026-09-15T08:00:00Z");
+        var payload = new OverseasDeliveryCompletedV1("1", "DL-ONT8", "11", "LM-11",
+            "101", "IN-1", "201", "FL-201", "201", "ONT8", time);
+        var node = json.valueToTree(payload);
+        return new EventEnvelope("00000000-0000-4000-8000-000000000001",
+            EventTypes.OVERSEAS_DELIVERY_COMPLETED, 1, time, RegionCode.OVERSEAS, RegionCode.CN,
+            "88", "INBOUND_ORDER", "101", 2L, 1L,
+            "00000000-0000-4000-8000-000000000099", "delivery-101", null, node,
+            EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(node)));
+    }
+}