From a213be67c5ef218042085af5407d2286bbbd8151 Mon Sep 17 00:00:00 2001 From: liuyihe <2513762445@qq.com> Date: Mon, 31 Aug 2026 15:22:32 +0800 Subject: [PATCH] =?UTF-8?q?=E5=B0=BE=E7=A8=8B=E6=9A=82=E6=89=A3=E5=9B=9E?= =?UTF-8?q?=E4=BC=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- common-event-contract/AGENTS.md | 1 + .../com/baiyue/common/event/EventTypes.java | 1 + ...OverseasInboundOrderDetainedChangedV1.java | 17 +++ deploy/sql/头程数据库统一增量-MySQL8.sql | 31 +++- .../metrics/EventBridgeMetrics.java | 1 + firstleg/AGENTS.md | 1 + ...ernalOverseasContainerEventController.java | 10 +- .../entity/InboundHoldRecordEntity.java | 24 ++++ .../firstleg/entity/InboundOrderEntity.java | 1 + .../mapper/InboundHoldRecordMapper.java | 9 ++ .../firstleg/mapper/InboundOrderMapper.java | 15 ++ .../OverseasInboundOrderDetainedConsumer.java | 135 ++++++++++++++++++ .../resources/db/phase2-inbound-schema.sql | 1 + .../db/phase3-smart-loading-schema.sql | 6 +- ...rseasInboundOrderDetainedConsumerTest.java | 87 +++++++++++ lastleg-service/AGENTS.md | 1 + .../controller/LastlegOrderController.java | 3 +- .../InboundUnloadCompletedOutboxRelay.java | 6 +- .../UnloadCompletedOutboxRepository.java | 11 +- .../InboundOrderDetainedOutboxService.java | 73 ++++++++++ .../lastleg/service/LastlegOrderService.java | 12 ++ .../service/LastlegOrderServiceTest.java | 9 ++ 22 files changed, 438 insertions(+), 17 deletions(-) create mode 100644 common-event-contract/src/main/java/com/baiyue/common/event/OverseasInboundOrderDetainedChangedV1.java create mode 100644 firstleg/src/main/java/com/baiyue/firstleg/entity/InboundHoldRecordEntity.java create mode 100644 firstleg/src/main/java/com/baiyue/firstleg/mapper/InboundHoldRecordMapper.java create mode 100644 firstleg/src/main/java/com/baiyue/firstleg/service/OverseasInboundOrderDetainedConsumer.java create mode 100644 firstleg/src/test/java/com/baiyue/firstleg/service/OverseasInboundOrderDetainedConsumerTest.java create mode 100644 lastleg-service/src/main/java/com/baiyue/lastleg/service/InboundOrderDetainedOutboxService.java diff --git a/common-event-contract/AGENTS.md b/common-event-contract/AGENTS.md index bf9e87321..6e1702ccc 100644 --- a/common-event-contract/AGENTS.md +++ b/common-event-contract/AGENTS.md @@ -17,6 +17,7 @@ | 头程装柜完成 V2（当前生产） | `FirstLegLoadingPlanCompletedV2.java`；订单级冻结 `destinationType=FBA|OVERSEAS`，FBA 可按箱数接收，OVERSEAS 逐明细冻结与实装箱数一致的 SKU/箱号；同时冻结来源/根订单与入仓标识、客户、业务类型、目的仓/备注/托盘及一键代发托盘 SKU 快照；国内生产者见 `../firstleg/src/main/java/com/baiyue/firstleg/service/LoadingCompletedOutboxService.java`，海外消费者见 `../lastleg-service/src/main/java/com/baiyue/lastleg/service/LoadingCompletedConsumer.java` | | 头程装柜完成 V1 解析模型 | `FirstLegLoadingPlanCompletedV1.java`；用于消费端冻结字段映射，新增生产不得回退为 V1 | | 海外卸柜完成 V1 | `OverseasContainerUnloadedV1.java`；按来源柜预订 ID 做柜级聚合/幂等，同一柜只发布一次；`unloadingCompletedAt` 必须是全柜 complete 时刻，不得用签收拍照时刻代替；海外生产者见 `../lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxService.java`，国内消费者见 `../firstleg/src/main/java/com/baiyue/firstleg/service/OverseasContainerEventConsumer.java` | +| 海外入仓单暂扣变更 V1 | `OverseasInboundOrderDetainedChangedV1.java`；聚合键固定为来源头程入仓单 ID，暂扣与解除暂扣均由尾程 Outbox 回传，国内按入仓单绑定、客户归属及聚合版本单调消费 | | 国内船节点记录 V1 | `FirstLegContainerShippingNodeRecordedV1.java`；只接受十节点中带实际时间和完整供应商审计字段的事实，国内生产者 `../firstleg/src/main/java/com/baiyue/firstleg/portschedule/service/CarrierScheduleNodeOutboxService.java`，海外消费者 `../lastleg-service/src/main/java/com/baiyue/lastleg/shipping/service/ContainerShippingNodeConsumer.java` | | IAM 安全事件 V1 | `IamSecurityEventsV1.java`；账号、授权快照等安全事件的冻结模型 | | 契约说明 | `package-info.java` | diff --git a/common-event-contract/src/main/java/com/baiyue/common/event/EventTypes.java b/common-event-contract/src/main/java/com/baiyue/common/event/EventTypes.java index 029056ada..eac3a335a 100644 --- a/common-event-contract/src/main/java/com/baiyue/common/event/EventTypes.java +++ b/common-event-contract/src/main/java/com/baiyue/common/event/EventTypes.java @@ -13,6 +13,7 @@ public final class EventTypes { public static final String FIRST_LEG_CONTAINER_SHIPPING_NODE_RECORDED = "FirstLegContainerShippingNodeRecorded"; public static final String OVERSEAS_CONTAINER_UNLOADED = "OverseasContainerUnloaded"; + public static final String OVERSEAS_INBOUND_ORDER_DETAINED_CHANGED = "OverseasInboundOrderDetainedChanged"; public static final String IAM_USER_STATUS_CHANGED = "IamUserStatusChanged"; public static final String IAM_TENANT_STATUS_CHANGED = "IamTenantStatusChanged"; public static final String IAM_PERMISSION_VERSION_CHANGED = "IamPermissionVersionChanged"; diff --git a/common-event-contract/src/main/java/com/baiyue/common/event/OverseasInboundOrderDetainedChangedV1.java b/common-event-contract/src/main/java/com/baiyue/common/event/OverseasInboundOrderDetainedChangedV1.java new file mode 100644 index 000000000..88cf9681b --- /dev/null +++ b/common-event-contract/src/main/java/com/baiyue/common/event/OverseasInboundOrderDetainedChangedV1.java @@ -0,0 +1,17 @@ +package com.baiyue.common.event; + +import java.time.Instant; + +/** 海外尾程按来源入仓单回传的暂扣状态事实。 */ +public record OverseasInboundOrderDetainedChangedV1( + String sourceInboundId, + String inboundCode, + String lastlegOrderId, + String lastlegOrderNo, + boolean detained, + String reason, + String operatorId, + String operatorName, + Instant changedAt +) { +} diff --git a/deploy/sql/头程数据库统一增量-MySQL8.sql b/deploy/sql/头程数据库统一增量-MySQL8.sql index 005c544bd..cdcd06ad8 100644 --- a/deploy/sql/头程数据库统一增量-MySQL8.sql +++ b/deploy/sql/头程数据库统一增量-MySQL8.sql @@ -2502,6 +2502,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', @@ -13504,6 +13505,7 @@ CALL queue_column_comment('baiyue_firstleg', 'fl_inbound_order', 'created_at', ' CALL queue_column_comment('baiyue_firstleg', 'fl_inbound_order', 'created_by', '创建人ID'); CALL queue_column_comment('baiyue_firstleg', 'fl_inbound_order', 'created_by_name', '创建人名称快照'); CALL queue_column_comment('baiyue_firstleg', 'fl_inbound_order', 'current_hold_reason', '当前扣货理由快照'); +CALL queue_column_comment('baiyue_firstleg', 'fl_inbound_order', 'last_overseas_hold_version', '海外暂扣事件最新聚合版本，用于阻止乱序覆盖'); CALL queue_column_comment('baiyue_firstleg', 'fl_inbound_order', 'customer_id', '客户逻辑ID'); CALL queue_column_comment('baiyue_firstleg', 'fl_inbound_order', 'customer_name', '客户名称快照'); CALL queue_column_comment('baiyue_firstleg', 'fl_inbound_order', 'customs_inspection', '是否商检'); @@ -16298,12 +16300,35 @@ DROP PROCEDURE IF EXISTS add_container_empty_report_fields; 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 '货柜聚合版本', payload_hash VARCHAR(80) NOT NULL COMMENT '规范化负载SHA-256摘要', + event_version INT 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', received_at DATETIME(3) NOT NULL COMMENT '接收时间（UTC）', 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 add_inbound_overseas_hold_version; +DELIMITER $$ +CREATE PROCEDURE add_inbound_overseas_hold_version() +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM information_schema.columns + WHERE table_schema=DATABASE() AND table_name='fl_inbound_order' + AND column_name='last_overseas_hold_version' + ) THEN + ALTER TABLE fl_inbound_order + ADD COLUMN last_overseas_hold_version BIGINT UNSIGNED NULL + COMMENT '海外暂扣事件最新聚合版本，用于阻止乱序覆盖' + AFTER current_hold_reason; + END IF; + ALTER TABLE fl_container_event_inbox + MODIFY COLUMN aggregate_id VARCHAR(64) NOT NULL COMMENT '跨区业务聚合ID', + MODIFY COLUMN aggregate_version BIGINT UNSIGNED NOT NULL COMMENT '跨区业务聚合版本', + COMMENT='海外到国内业务事件消费Inbox'; +END$$ +DELIMITER ; +CALL add_inbound_overseas_hold_version(); +DROP PROCEDURE IF EXISTS add_inbound_overseas_hold_version; -- 标准跨区事件摘要包含 sha256: 前缀，共 71 字符；兼容既有 Finance 64 位摘要。 ALTER TABLE fl_inbound_outbox diff --git a/event-bridge-service/src/main/java/com/baiyue/eventbridge/metrics/EventBridgeMetrics.java b/event-bridge-service/src/main/java/com/baiyue/eventbridge/metrics/EventBridgeMetrics.java index 85756ae0e..156b7b28a 100644 --- a/event-bridge-service/src/main/java/com/baiyue/eventbridge/metrics/EventBridgeMetrics.java +++ b/event-bridge-service/src/main/java/com/baiyue/eventbridge/metrics/EventBridgeMetrics.java @@ -30,6 +30,7 @@ public class EventBridgeMetrics { EventTypes.FIRST_LEG_LOADING_PLAN_COMPLETED, EventTypes.FIRST_LEG_CONTAINER_SHIPPING_NODE_RECORDED, EventTypes.OVERSEAS_CONTAINER_UNLOADED, + EventTypes.OVERSEAS_INBOUND_ORDER_DETAINED_CHANGED, EventTypes.IAM_USER_STATUS_CHANGED, EventTypes.IAM_TENANT_STATUS_CHANGED, EventTypes.IAM_PERMISSION_VERSION_CHANGED, diff --git a/firstleg/AGENTS.md b/firstleg/AGENTS.md index a6ff072d0..c0a3d8543 100644 --- a/firstleg/AGENTS.md +++ b/firstleg/AGENTS.md @@ -33,6 +33,7 @@ | PDA 拣货与装柜 | `controller/PlatformLoadingPdaController.java`、`loading/controller/PlatformLoadingDeliveryController.java` | | 装柜完成跨区事件 | `service/LoadingCompletedOutboxService.java`、`LoadingCompletedOutboxRelay.java`、`repository/LoadingCompletedOutboxRepository.java`、`mapper/LoadingCartonManifestMapper.java`、`client/EventBridgeClient.java`；从 `fl_inbound_order.destination_type` 冻结 V2 `orders[].destinationType=FBA|OVERSEAS`，OVERSEAS 按最终实装与稳定箱码冻结，FBA 可仅冻结箱数；真实生产验收 `src/test/java/com/baiyue/firstleg/service/TaillegRealProducerAcceptanceTest.java`；事件模型见 `../common-event-contract/AGENTS.md` | | 海外卸柜完成事件回写 | 内部入口 `controller/InternalOverseasContainerEventController.java`、消费者 `service/OverseasContainerEventConsumer.java`、Inbox `entity/ContainerEventInboxEntity.java` 与 `mapper/ContainerEventInboxMapper.java`；事件模型见 `../common-event-contract/AGENTS.md` | +| 海外暂扣状态回写 | 同一内部事件入口、消费者 `service/OverseasInboundOrderDetainedConsumer.java`；按来源入仓单 ID 更新 `fl_inbound_order.hold_flag/current_hold_reason`，复用跨区 Inbox 并写 `fl_inbound_hold_record` | | 三期海运订舱、关联候选与出仓前船期资料 | `oceanbooking/controller/PlatformOceanBookingController.java`、`oceanbooking/service/OceanBookingService.java`、`oceanbooking/mapper/OceanBookingMapper.java`、`portschedule/controller/PlatformPortScheduleController.java`；关联候选入口 `/api/platform/v1/ocean-bookings/link-candidates`，订舱契约见 `src/main/resources/openapi/ocean-bookings-v1.yaml`，表结构见 `src/main/resources/db/phase3-ocean-booking-schema.sql` | | 四期国内出仓后船节点定时查询 | `outbound/dispatch/service/OutboundDispatchService.java` 在确认出仓事务内幂等创建现有 `CarrierScheduleQuery`；编排与租约 `portschedule/service/CarrierScheduleQueryScheduler.java`，任务/十节点事实/异常事实/不可变 attempt `CarrierScheduleQueryService.java`，平台查询与人工补录入口 `PlatformCarrierScheduleQueryController.java` 的 `/{queryId}/manual-nodes`；停止只由实际 `EMPTY_RETURNED` 节点触发，不提供无节点停止入口；外部船司/轨迹调用经 `integration-service/`；标准见 `../../docs/四期/船节点标准化契约.md` | | 船节点跨区交接事件 | `portschedule/service/CarrierScheduleNodeOutboxService.java`；实际节点落库后发布 `FIRST_LEG_CONTAINER_SHIPPING_NODE_RECORDED` V1，复用 `LoadingCompletedOutboxRepository`/Relay 投 Event Bridge；海外消费见 `../lastleg-service/src/main/java/com/baiyue/lastleg/shipping/service/ContainerShippingNodeConsumer.java` | diff --git a/firstleg/src/main/java/com/baiyue/firstleg/controller/InternalOverseasContainerEventController.java b/firstleg/src/main/java/com/baiyue/firstleg/controller/InternalOverseasContainerEventController.java index 54f1a67be..095e376c8 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/controller/InternalOverseasContainerEventController.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/controller/InternalOverseasContainerEventController.java @@ -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 receive( + public ApiResponse 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) { diff --git a/firstleg/src/main/java/com/baiyue/firstleg/entity/InboundHoldRecordEntity.java b/firstleg/src/main/java/com/baiyue/firstleg/entity/InboundHoldRecordEntity.java new file mode 100644 index 000000000..931ac782f --- /dev/null +++ b/firstleg/src/main/java/com/baiyue/firstleg/entity/InboundHoldRecordEntity.java @@ -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; +} diff --git a/firstleg/src/main/java/com/baiyue/firstleg/entity/InboundOrderEntity.java b/firstleg/src/main/java/com/baiyue/firstleg/entity/InboundOrderEntity.java index 75f165e1b..816a28e11 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/entity/InboundOrderEntity.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/entity/InboundOrderEntity.java @@ -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; diff --git a/firstleg/src/main/java/com/baiyue/firstleg/mapper/InboundHoldRecordMapper.java b/firstleg/src/main/java/com/baiyue/firstleg/mapper/InboundHoldRecordMapper.java new file mode 100644 index 000000000..690bac3c6 --- /dev/null +++ b/firstleg/src/main/java/com/baiyue/firstleg/mapper/InboundHoldRecordMapper.java @@ -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 { +} diff --git a/firstleg/src/main/java/com/baiyue/firstleg/mapper/InboundOrderMapper.java b/firstleg/src/main/java/com/baiyue/firstleg/mapper/InboundOrderMapper.java index c5663cf46..7ec6a1a22 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/mapper/InboundOrderMapper.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/mapper/InboundOrderMapper.java @@ -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 { + @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', diff --git a/firstleg/src/main/java/com/baiyue/firstleg/service/OverseasInboundOrderDetainedConsumer.java b/firstleg/src/main/java/com/baiyue/firstleg/service/OverseasInboundOrderDetainedConsumer.java new file mode 100644 index 000000000..df37c939f --- /dev/null +++ b/firstleg/src/main/java/com/baiyue/firstleg/service/OverseasInboundOrderDetainedConsumer.java @@ -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; } + } +} diff --git a/firstleg/src/main/resources/db/phase2-inbound-schema.sql b/firstleg/src/main/resources/db/phase2-inbound-schema.sql index 902aede82..7d5e63c1d 100644 --- a/firstleg/src/main/resources/db/phase2-inbound-schema.sql +++ b/firstleg/src/main/resources/db/phase2-inbound-schema.sql @@ -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', diff --git a/firstleg/src/main/resources/db/phase3-smart-loading-schema.sql b/firstleg/src/main/resources/db/phase3-smart-loading-schema.sql index ff90ac11e..1b98ea460 100644 --- a/firstleg/src/main/resources/db/phase3-smart-loading-schema.sql +++ b/firstleg/src/main/resources/db/phase3-smart-loading-schema.sql @@ -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 $$ diff --git a/firstleg/src/test/java/com/baiyue/firstleg/service/OverseasInboundOrderDetainedConsumerTest.java b/firstleg/src/test/java/com/baiyue/firstleg/service/OverseasInboundOrderDetainedConsumerTest.java new file mode 100644 index 000000000..315e104cb --- /dev/null +++ b/firstleg/src/test/java/com/baiyue/firstleg/service/OverseasInboundOrderDetainedConsumerTest.java @@ -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 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))); + } +} diff --git a/lastleg-service/AGENTS.md b/lastleg-service/AGENTS.md index bec72d461..11708fa2c 100644 --- a/lastleg-service/AGENTS.md +++ b/lastleg-service/AGENTS.md @@ -27,6 +27,7 @@ | PDA 入仓、实收、上架与完成 | `inboundplan/pda/controller/ReceivingContainerController.java`、`inboundplan/pda/service/ReceivingContainerService.java`、内部编排 `InboundPdaWorkflowService.java`；公网入口固定为 `/api/pda/v1/last-leg/receiving-containers/**`，`/warehouse-profile` 只按 Token 绑定仓解析共享仓库中文名称与编码，旧 planNo/allocationId Controller 已移除 | | PDA 卸柜照片上传、归属校验与预览 | `ReceivingContainerController.java` 的 `/current/unload-photos/files`、`/current/unload-photos/files/{fileId}/download`，网关 `inboundplan/pda/gateway/UnloadPhotoFileGateway.java`、`HttpUnloadPhotoFileGateway.java`；File 类型固定 `OVERSEAS_UNLOAD_PHOTO`，下载只允许当前会话已签收的 fileId 并返回 `no-store` | | 卸柜照片与海外卸柜完成 Outbox | `ReceivingContainerController.java`、`ReceivingContainerService.java`、`inboundplan/pda/service/InboundUnloadCompletedOutboxService.java`、`entity/LastlegEventOutbox.java`、`mapper/LastlegEventOutboxMapper.java`；多计划同柜只能在全柜完成后调用 `publishContainerCompleted`，按来源柜预订 ID 幂等发布一次，完成时刻与拍照时刻分离；事件模型见 `../common-event-contract/AGENTS.md` | +| 尾程暂扣状态回传 | `service/InboundOrderDetainedOutboxService.java`；暂扣与解除暂扣在订单更新事务内按 `sourceInboundId` 写入海外到国内 Outbox，国内按来源入仓单绑定更新 | | 入仓后 SKU 库存余额、库位和流水 | `inventory/`；平台只读查询与非出库库存命令分别从 `inventory/query/controller/InventoryQueryController.java`、`inventory/command/controller/InventoryCommandController.java` 进入 | | 海外PDA盘点、同仓整托移库、托盘组拆换绑与换标现场作业 | `pda/stocktake/`、`pda/move/`、`palletops/`、`relabel/pda/`；版本化契约 `src/main/resources/openapi/overseas-pda-operations-v1.yaml`；全部范围取自PDA Token，不含出库和派送 | | 船节点接收与海外 PDA 可收货门禁 | `shipping/service/ContainerShippingNodeConsumer.java`、`shipping/entity/ContainerShippingGate.java`、`shipping/mapper/ContainerShippingGateMapper.java`；只单调接受实际标准节点，`PICKUP_READY` 及以后才允许对应柜进入 PDA 收货候选 | diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/controller/LastlegOrderController.java b/lastleg-service/src/main/java/com/baiyue/lastleg/controller/LastlegOrderController.java index 201e61a28..77972386d 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/controller/LastlegOrderController.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/controller/LastlegOrderController.java @@ -101,7 +101,8 @@ public class LastlegOrderController { @RequestParam(required = false) String customerId, @PathVariable long id, @Valid @RequestBody LastlegOrderRequests.SetDetained request) { - orderService.setDetained(platformScope(user, customerId), id, request, operatorName(user)); + orderService.setDetained(platformScope(user, customerId), id, request, + String.valueOf(user.userId()), operatorName(user)); return ApiResponse.success(null); } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxRelay.java b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxRelay.java index e9a5e50c0..b7988f67d 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxRelay.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxRelay.java @@ -15,7 +15,7 @@ import org.springframework.stereotype.Component; import java.time.LocalDateTime; import java.util.UUID; -/** 至少一次投递海外卸柜完成事件；国内事件桥按 eventId 幂等接收。 */ +/** 至少一次投递海外到国内业务事件；国内事件桥按 eventId 幂等接收。 */ @Component @Slf4j public class InboundUnloadCompletedOutboxRelay { @@ -44,7 +44,7 @@ public class InboundUnloadCompletedOutboxRelay { } catch (Exception exception) { repository.retry(row.getId(), workerId, LocalDateTime.now().plusSeconds(Math.max(retrySeconds, 1)), exception.getMessage()); - log.warn("海外卸柜完成事件投递失败: eventKey={}, reason={}", row.getEventKey(), exception.getMessage()); + log.warn("海外到国内事件投递失败: eventKey={}, reason={}", row.getEventKey(), exception.getMessage()); } } } @@ -53,7 +53,7 @@ public class InboundUnloadCompletedOutboxRelay { com.baiyue.common.core.api.ApiResponse response) { if (response == null) throw new IllegalStateException("国内事件桥返回空响应"); if (!response.successful()) { - throw new IllegalStateException("国内事件桥拒绝海外卸柜完成事件: code=" + response.code() + throw new IllegalStateException("国内事件桥拒绝海外事件: code=" + response.code() + ", message=" + response.message() + ", requestId=" + response.requestId()); } if (response.data() == null) throw new IllegalStateException("国内事件桥成功响应缺少确认数据"); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/repository/UnloadCompletedOutboxRepository.java b/lastleg-service/src/main/java/com/baiyue/lastleg/repository/UnloadCompletedOutboxRepository.java index 53e110c8a..92863e47a 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/repository/UnloadCompletedOutboxRepository.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/repository/UnloadCompletedOutboxRepository.java @@ -13,11 +13,14 @@ import java.util.List; @Repository public class UnloadCompletedOutboxRepository { + private static final List OVERSEAS_TO_CN_TYPES = List.of( + EventTypes.OVERSEAS_CONTAINER_UNLOADED, + EventTypes.OVERSEAS_INBOUND_ORDER_DETAINED_CHANGED); @Resource private LastlegEventOutboxMapper mapper; public List pending(int limit, LocalDateTime now, LocalDateTime expiredLeaseBefore) { return mapper.selectList(new LambdaQueryWrapper() - .eq(LastlegEventOutbox::getEventType, EventTypes.OVERSEAS_CONTAINER_UNLOADED) + .in(LastlegEventOutbox::getEventType, OVERSEAS_TO_CN_TYPES) .and(status -> status.eq(LastlegEventOutbox::getStatus, "PENDING") .and(retry -> retry.isNull(LastlegEventOutbox::getNextRetryAt) .or().le(LastlegEventOutbox::getNextRetryAt, now)) @@ -30,7 +33,7 @@ public class UnloadCompletedOutboxRepository { public boolean claim(long id, String worker, LocalDateTime now, LocalDateTime expiredLeaseBefore) { return mapper.update(null, new LambdaUpdateWrapper() .eq(LastlegEventOutbox::getId, id) - .eq(LastlegEventOutbox::getEventType, EventTypes.OVERSEAS_CONTAINER_UNLOADED) + .in(LastlegEventOutbox::getEventType, OVERSEAS_TO_CN_TYPES) .and(status -> status.eq(LastlegEventOutbox::getStatus, "PENDING") .and(retry -> retry.isNull(LastlegEventOutbox::getNextRetryAt) .or().le(LastlegEventOutbox::getNextRetryAt, now)) @@ -44,7 +47,7 @@ public class UnloadCompletedOutboxRepository { public LastlegEventOutbox claimed(long id, String worker) { return mapper.selectOne(new LambdaQueryWrapper() .eq(LastlegEventOutbox::getId, id) - .eq(LastlegEventOutbox::getEventType, EventTypes.OVERSEAS_CONTAINER_UNLOADED) + .in(LastlegEventOutbox::getEventType, OVERSEAS_TO_CN_TYPES) .eq(LastlegEventOutbox::getStatus, "PROCESSING") .eq(LastlegEventOutbox::getLockedBy, worker)); } @@ -72,7 +75,7 @@ public class UnloadCompletedOutboxRepository { private LambdaUpdateWrapper claimedBase(long id, String worker) { return new LambdaUpdateWrapper() .eq(LastlegEventOutbox::getId, id) - .eq(LastlegEventOutbox::getEventType, EventTypes.OVERSEAS_CONTAINER_UNLOADED) + .in(LastlegEventOutbox::getEventType, OVERSEAS_TO_CN_TYPES) .eq(LastlegEventOutbox::getStatus, "PROCESSING") .eq(LastlegEventOutbox::getLockedBy, worker); } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/service/InboundOrderDetainedOutboxService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/service/InboundOrderDetainedOutboxService.java new file mode 100644 index 000000000..f3bcdc4f0 --- /dev/null +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/service/InboundOrderDetainedOutboxService.java @@ -0,0 +1,73 @@ +package com.baiyue.lastleg.service; + +import com.baomidou.mybatisplus.core.toolkit.IdWorker; +import com.baiyue.common.event.EventEnvelope; +import com.baiyue.common.event.EventIdentifiers; +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.lastleg.entity.LastlegEventOutbox; +import com.baiyue.lastleg.entity.LastlegOrder; +import com.baiyue.lastleg.mapper.LastlegEventOutboxMapper; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import jakarta.annotation.Resource; +import org.springframework.stereotype.Service; + +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneOffset; + +/** 在尾程业务事务内冻结按来源入仓单回传国内的暂扣状态事件。 */ +@Service +public class InboundOrderDetainedOutboxService { + @Resource private LastlegEventOutboxMapper mapper; + @Resource private ObjectMapper objectMapper; + + public void publish(LastlegOrder order, boolean detained, String reason, String operatorId, + String operatorName, Instant changedAt, long changeVersion) { + requireText(order.getSourceInboundId(), "sourceInboundId"); + requireText(order.getInboundCode(), "inboundCode"); + requireText(order.getCustomerId(), "customerId"); + var payload = new OverseasInboundOrderDetainedChangedV1(order.getSourceInboundId(), + order.getInboundCode(), String.valueOf(order.getId()), order.getOrderNo(), detained, + reason, operatorId, operatorName, changedAt); + JsonNode payloadNode = objectMapper.valueToTree(payload); + String eventKey = EventTypes.OVERSEAS_INBOUND_ORDER_DETAINED_CHANGED + ':' + + order.getSourceInboundId() + ':' + changeVersion; + EventEnvelope envelope = new EventEnvelope(EventIdentifiers.newUuid(), + EventTypes.OVERSEAS_INBOUND_ORDER_DETAINED_CHANGED, 1, changedAt, + RegionCode.OVERSEAS, RegionCode.CN, order.getCustomerId(), "INBOUND_ORDER", + order.getSourceInboundId(), changeVersion, 1L, EventIdentifiers.newUuid(), + eventKey, order.getSourceEventId(), payloadNode, + EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(payloadNode))).requireValid(); + + LastlegEventOutbox event = new LastlegEventOutbox(); + event.setId(IdWorker.getId()); + event.setCustomerId(order.getCustomerId()); + event.setAggregateType(envelope.aggregateType()); + event.setAggregateId(envelope.aggregateId()); + event.setEventType(envelope.eventType()); + event.setEventVersion(envelope.eventVersion()); + event.setEventKey(eventKey); + event.setPayloadJson(json(envelope)); + event.setStatus("PENDING"); + event.setRetryCount(0); + event.setCreatedAt(LocalDateTime.ofInstant(changedAt, ZoneOffset.UTC)); + mapper.insert(event); + } + + private static void requireText(String value, String field) { + if (value == null || value.isBlank()) throw new IllegalStateException(field + " is required"); + } + + private String json(Object value) { + try { return objectMapper.writeValueAsString(value); } + catch (JsonProcessingException exception) { + throw new IllegalStateException("Cannot serialize inbound detained event", exception); + } + } +} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/service/LastlegOrderService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/service/LastlegOrderService.java index b882994a7..03789b5ad 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/service/LastlegOrderService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/service/LastlegOrderService.java @@ -33,6 +33,7 @@ import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.OutputStream; import java.time.LocalDateTime; +import java.time.ZoneOffset; import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; @@ -54,6 +55,7 @@ public class LastlegOrderService { @Resource private LastlegOrderOperationLogMapper operationLogMapper; @Resource private InboundPlanCartonMapper cartonMapper; @Resource private InventoryCommandService inventory; + @Resource private InboundOrderDetainedOutboxService detainedOutbox; public PageResult page(String customerId, LastlegOrderRequests.Query query) { Page page = orderMapper.selectPage( @@ -123,6 +125,12 @@ public class LastlegOrderService { @Transactional public void setDetained(String customerId, long orderId, LastlegOrderRequests.SetDetained request, String operatorName) { + setDetained(customerId, orderId, request, null, operatorName); + } + + @Transactional + public void setDetained(String customerId, long orderId, LastlegOrderRequests.SetDetained request, + String operatorId, String operatorName) { LastlegOrder current = findOwned(customerId, orderId); LastlegOrder order = orderMapper.lockById(current.getCustomerId(), orderId); if (order == null) throw new ResponseStatusException(HttpStatus.NOT_FOUND, "Tail-leg order not found"); @@ -149,6 +157,10 @@ public class LastlegOrderService { operation.setResult(request.remark().trim()); operation.setCreatedAt(now); operationLogMapper.insert(operation); + if (StringUtils.hasText(order.getSourceInboundId())) { + detainedOutbox.publish(order, detained, request.remark().trim(), operatorId, operatorName, + now.toInstant(ZoneOffset.UTC), operation.getId()); + } } @Transactional diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/service/LastlegOrderServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/service/LastlegOrderServiceTest.java index 98ce8674a..74abbfe37 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/service/LastlegOrderServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/service/LastlegOrderServiceTest.java @@ -29,6 +29,8 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; @@ -43,6 +45,7 @@ class LastlegOrderServiceTest { @Mock LastlegOrderOperationLogMapper operations; @Mock InboundPlanCartonMapper cartons; @Mock InventoryCommandService inventory; + @Mock InboundOrderDetainedOutboxService detainedOutbox; LastlegOrderService service; @BeforeEach @@ -55,6 +58,7 @@ class LastlegOrderServiceTest { ReflectionTestUtils.setField(service, "operationLogMapper", operations); ReflectionTestUtils.setField(service, "cartonMapper", cartons); ReflectionTestUtils.setField(service, "inventory", inventory); + ReflectionTestUtils.setField(service, "detainedOutbox", detainedOutbox); } @Test @@ -228,6 +232,7 @@ class LastlegOrderServiceTest { verify(operations).insert(operation.capture()); assertEquals("暂扣", operation.getValue().getOperationNode()); assertEquals("等待质检", operation.getValue().getResult()); + verify(detainedOutbox).publish(eq(order), eq(true), eq("等待质检"), isNull(), eq("operator"), any(), anyLong()); verify(inventory, never()).releaseInboundOrderHold(any(), any(), anyLong(), any(), any()); } @@ -243,6 +248,7 @@ class LastlegOrderServiceTest { verify(inventory).releaseInboundOrderHold("100", "operator", 11L, "LM-1", "质检通过"); assertEquals(false, order.getDetained()); + verify(detainedOutbox).publish(eq(order), eq(false), eq("质检通过"), isNull(), eq("operator"), any(), anyLong()); } private static LastlegOrder order(long id, String customerId, String status, String inboundOrderNo) { @@ -251,6 +257,9 @@ class LastlegOrderServiceTest { order.setCustomerId(customerId); order.setCurrentStatus(status); order.setInboundOrderNo(inboundOrderNo); + order.setOrderNo("LM-" + id); + order.setSourceInboundId(String.valueOf(1000 + id)); + order.setInboundCode("IN-" + id); return order; }