From 4893542ec40cf5df37e28173c64cedc29c7da16d Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Mon, 24 Aug 2026 17:48:08 +0800 Subject: [PATCH] 1 --- common-event-contract/AGENTS.md | 2 +- .../event/OverseasContainerUnloadedV1.java | 5 +- .../OverseasContainerUnloadedV1Test.java | 2 + deploy/sql/海外区域统一增量-MySQL8.sql | 19 ++- lastleg-service/AGENTS.md | 2 +- .../pda/dto/ReceivingContainerRequests.java | 10 +- .../gateway/HttpWarehouseResourceGateway.java | 13 ++ .../pda/gateway/WarehouseResourceGateway.java | 1 + .../service/InboundPdaWorkflowService.java | 108 ++++++++++++-- .../InboundUnloadCompletedOutboxService.java | 77 ++++++++-- .../service/ReceivingContainerService.java | 47 ++++-- .../mapper/LastlegEventOutboxMapper.java | 4 + .../src/main/resources/db/lastleg.sql | 19 ++- .../openapi/overseas-business-v1.yaml | 24 ++-- ...boundUnloadCompletedOutboxServiceTest.java | 136 ++++++++++++++++++ 15 files changed, 410 insertions(+), 59 deletions(-) create mode 100644 lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxServiceTest.java diff --git a/common-event-contract/AGENTS.md b/common-event-contract/AGENTS.md index 0a675effd..c35c4997c 100644 --- a/common-event-contract/AGENTS.md +++ b/common-event-contract/AGENTS.md @@ -16,7 +16,7 @@ | 事件类型与区域 | `EventTypes.java`、`RegionCode.java` | | 头程装柜完成 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`；海外生产者见 `../lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxService.java`，国内消费者见 `../firstleg/src/main/java/com/baiyue/firstleg/service/OverseasContainerEventConsumer.java` | +| 海外卸柜完成 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` | | IAM 安全事件 V1 | `IamSecurityEventsV1.java`；账号、授权快照等安全事件的冻结模型 | | 契约说明 | `package-info.java` | | 契约测试 | `src/test/java/com/baiyue/common/event/EventEnvelopeTest.java`、`FirstLegLoadingPlanCompletedV1Test.java`、`FirstLegLoadingPlanCompletedV2Test.java`、`OverseasContainerUnloadedV1Test.java` | diff --git a/common-event-contract/src/main/java/com/baiyue/common/event/OverseasContainerUnloadedV1.java b/common-event-contract/src/main/java/com/baiyue/common/event/OverseasContainerUnloadedV1.java index 6841c9815..a919d58c5 100644 --- a/common-event-contract/src/main/java/com/baiyue/common/event/OverseasContainerUnloadedV1.java +++ b/common-event-contract/src/main/java/com/baiyue/common/event/OverseasContainerUnloadedV1.java @@ -2,7 +2,10 @@ package com.baiyue.common.event; import java.time.Instant; -/** 海外仓卸柜完成事实；国内订舱 ID 是稳定聚合标识，containerNo 仅用于核对物理柜号。 */ +/** + * 海外仓柜级卸柜完成事实；国内订舱 ID 是稳定聚合与幂等标识，containerNo 用于核对物理柜号。 + * unloadingCompletedAt 是全柜 complete 时刻，不得使用签收拍照上传时刻代替。 + */ public record OverseasContainerUnloadedV1( String sourceContainerBookingId, String containerNo, diff --git a/common-event-contract/src/test/java/com/baiyue/common/event/OverseasContainerUnloadedV1Test.java b/common-event-contract/src/test/java/com/baiyue/common/event/OverseasContainerUnloadedV1Test.java index 70f843856..0942ac072 100644 --- a/common-event-contract/src/test/java/com/baiyue/common/event/OverseasContainerUnloadedV1Test.java +++ b/common-event-contract/src/test/java/com/baiyue/common/event/OverseasContainerUnloadedV1Test.java @@ -20,5 +20,7 @@ class OverseasContainerUnloadedV1Test { assertThat(mapper.readValue(json, OverseasContainerUnloadedV1.class)).isEqualTo(payload); assertThat(json).contains("sourceContainerBookingId", "1001", "MSCU1234567", "unloadingPhotoUploadedAt"); + assertThat(payload.unloadingCompletedAt()).isAfter(payload.unloadingPhotoUploadedAt()); + assertThat(payload.unloadingCompletedAt()).isNotEqualTo(payload.unloadingPhotoUploadedAt()); } } diff --git a/deploy/sql/海外区域统一增量-MySQL8.sql b/deploy/sql/海外区域统一增量-MySQL8.sql index bbce2f48d..3c8be6dd6 100644 --- a/deploy/sql/海外区域统一增量-MySQL8.sql +++ b/deploy/sql/海外区域统一增量-MySQL8.sql @@ -341,7 +341,7 @@ SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE tab 'SELECT 1', 'ALTER TABLE ll_inbound_plan_line ADD COLUMN customer_id VARCHAR(128) NOT NULL DEFAULT '''' COMMENT ''客户账号标识'' AFTER id')); PREPARE phase4_receiving_line_stmt FROM @ddl; EXECUTE phase4_receiving_line_stmt; DEALLOCATE PREPARE phase4_receiving_line_stmt; -UPDATE ll_inbound_plan_line l JOIN ll_inbound_plan p ON p.id=l.plan_id SET l.customer_id=p.customer_id WHERE l.customer_id=''; +UPDATE ll_inbound_plan_line l JOIN ll_inbound_plan p ON p.id=l.plan_id SET l.customer_id=p.customer_id WHERE l.customer_id IS NULL OR l.customer_id=''; ALTER TABLE ll_inbound_plan_line MODIFY COLUMN customer_id VARCHAR(128) NOT NULL COMMENT '客户账号标识'; SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name='ll_inbound_plan_line' AND index_name='idx_ll_inbound_line_customer_plan'), 'SELECT 1', 'ALTER TABLE ll_inbound_plan_line ADD KEY idx_ll_inbound_line_customer_plan(customer_id,plan_id)')); PREPARE phase4_receiving_scope_stmt FROM @ddl; EXECUTE phase4_receiving_scope_stmt; DEALLOCATE PREPARE phase4_receiving_scope_stmt; @@ -349,7 +349,7 @@ SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE tab 'SELECT 1', 'ALTER TABLE ll_inbound_plan_pallet ADD COLUMN customer_id VARCHAR(128) NOT NULL DEFAULT '''' COMMENT ''客户账号标识'' AFTER id')); PREPARE phase4_receiving_scope_stmt FROM @ddl; EXECUTE phase4_receiving_scope_stmt; DEALLOCATE PREPARE phase4_receiving_scope_stmt; -UPDATE ll_inbound_plan_pallet v JOIN ll_inbound_plan p ON p.id=v.plan_id SET v.customer_id=p.customer_id WHERE v.customer_id=''; +UPDATE ll_inbound_plan_pallet v JOIN ll_inbound_plan p ON p.id=v.plan_id SET v.customer_id=p.customer_id WHERE v.customer_id IS NULL OR v.customer_id=''; ALTER TABLE ll_inbound_plan_pallet MODIFY COLUMN customer_id VARCHAR(128) NOT NULL COMMENT '客户账号标识'; SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name='ll_inbound_plan_pallet' AND index_name='idx_ll_inbound_pallet_customer_plan'), 'SELECT 1', 'ALTER TABLE ll_inbound_plan_pallet ADD KEY idx_ll_inbound_pallet_customer_plan(customer_id,plan_id)')); PREPARE phase4_receiving_scope_stmt FROM @ddl; EXECUTE phase4_receiving_scope_stmt; DEALLOCATE PREPARE phase4_receiving_scope_stmt; @@ -357,10 +357,23 @@ SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE tab 'SELECT 1', 'ALTER TABLE ll_inbound_plan_pallet_allocation ADD COLUMN customer_id VARCHAR(128) NOT NULL DEFAULT '''' COMMENT ''客户账号标识'' AFTER id')); PREPARE phase4_receiving_scope_stmt FROM @ddl; EXECUTE phase4_receiving_scope_stmt; DEALLOCATE PREPARE phase4_receiving_scope_stmt; -UPDATE ll_inbound_plan_pallet_allocation a JOIN ll_inbound_plan p ON p.id=a.plan_id SET a.customer_id=p.customer_id WHERE a.customer_id=''; +UPDATE ll_inbound_plan_pallet_allocation a JOIN ll_inbound_plan p ON p.id=a.plan_id SET a.customer_id=p.customer_id WHERE a.customer_id IS NULL OR a.customer_id=''; ALTER TABLE ll_inbound_plan_pallet_allocation MODIFY COLUMN customer_id VARCHAR(128) NOT NULL COMMENT '客户账号标识'; SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name='ll_inbound_plan_pallet_allocation' AND index_name='idx_ll_inbound_allocation_customer_plan'), 'SELECT 1', 'ALTER TABLE ll_inbound_plan_pallet_allocation ADD KEY idx_ll_inbound_allocation_customer_plan(customer_id,plan_id)')); PREPARE phase4_receiving_scope_stmt FROM @ddl; EXECUTE phase4_receiving_scope_stmt; DEALLOCATE PREPARE phase4_receiving_scope_stmt; +DROP PROCEDURE IF EXISTS assert_phase4_inbound_plan_customer_scope_ready; +DELIMITER $$ +CREATE PROCEDURE assert_phase4_inbound_plan_customer_scope_ready() +BEGIN + IF EXISTS (SELECT 1 FROM ll_inbound_plan_line l LEFT JOIN ll_inbound_plan p ON p.id=l.plan_id WHERE l.customer_id IS NULL OR l.customer_id='' OR p.id IS NULL OR l.customer_id<>p.customer_id) + OR EXISTS (SELECT 1 FROM ll_inbound_plan_pallet v LEFT JOIN ll_inbound_plan p ON p.id=v.plan_id WHERE v.customer_id IS NULL OR v.customer_id='' OR p.id IS NULL OR v.customer_id<>p.customer_id) + OR EXISTS (SELECT 1 FROM ll_inbound_plan_pallet_allocation a LEFT JOIN ll_inbound_plan p ON p.id=a.plan_id WHERE a.customer_id IS NULL OR a.customer_id='' OR p.id IS NULL OR a.customer_id<>p.customer_id) THEN + SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='Inbound plan child customer scope migration is incomplete'; + END IF; +END$$ +DELIMITER ; +CALL assert_phase4_inbound_plan_customer_scope_ready(); +DROP PROCEDURE assert_phase4_inbound_plan_customer_scope_ready; SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='ll_inbound_plan_line' AND column_name='fba_warehouse_code'), 'ALTER TABLE ll_inbound_plan_line MODIFY COLUMN fba_warehouse_code VARCHAR(128) NULL COMMENT ''FBA仓库编码''', 'ALTER TABLE ll_inbound_plan_line ADD COLUMN fba_warehouse_code VARCHAR(128) NULL COMMENT ''FBA仓库编码'' AFTER inbound_code')); diff --git a/lastleg-service/AGENTS.md b/lastleg-service/AGENTS.md index 3ea28089c..b6d2cc413 100644 --- a/lastleg-service/AGENTS.md +++ b/lastleg-service/AGENTS.md @@ -24,7 +24,7 @@ | FBA 预约创建、详情与编辑 | `InboundPlanController.java` 的 `/fba-appointments/**`、`inboundplan/service/InboundFbaAppointmentService.java`、`BatchFbaAppointmentRequest.java`、`UpdateFbaAppointmentRequest.java` | | PDA 入仓、实收、上架、合托、完成 | `inboundplan/pda/controller/`、`inboundplan/pda/service/InboundPdaWorkflowService.java`；入口固定为 `/api/pda/v1/last-leg/**`，不得挂入平台浏览器 API | | PDA 卸柜照片文件上传与归属校验 | `InboundPdaController.java` 的 `/unload-photos/files`、`inboundplan/pda/gateway/UnloadPhotoFileGateway.java`、`HttpUnloadPhotoFileGateway.java`；File 类型固定 `OVERSEAS_UNLOAD_PHOTO` | -| 卸柜照片与海外卸柜完成 Outbox | `inboundplan/pda/controller/InboundPdaController.java`、`inboundplan/pda/service/InboundUnloadCompletedOutboxService.java`、`entity/LastlegEventOutbox.java`、`mapper/LastlegEventOutboxMapper.java`；事件模型见 `../common-event-contract/AGENTS.md` | +| 卸柜照片与海外卸柜完成 Outbox | `inboundplan/pda/controller/InboundPdaController.java`、`inboundplan/pda/service/InboundUnloadCompletedOutboxService.java`、`entity/LastlegEventOutbox.java`、`mapper/LastlegEventOutboxMapper.java`；多计划同柜只能在全柜完成后调用 `publishContainerCompleted`，按来源柜预订 ID 幂等发布一次，完成时刻与拍照时刻分离；事件模型见 `../common-event-contract/AGENTS.md` | | 入仓后 SKU 库存余额、库位和流水 | `inventory/`；平台只读查询与非出库库存命令分别从 `inventory/query/controller/InventoryQueryController.java`、`inventory/command/controller/InventoryCommandController.java` 进入 | | 平台排车列表/详情与智能预排 | `dispatch/controller/DispatchQueryController.java`、`SmartDispatchController.java`、`dispatch/mapper/DispatchCandidateMapper.java`、`dispatch/service/DispatchService.java`；草稿柜 `entity/DispatchDraftContainer.java`、`mapper/DispatchDraftContainerMapper.java`，托盘锁 `DispatchPalletLock.java`；入口 `/api/platform/v1/last-leg/dispatches`，范围开关与禁用写操作见“四期开发范围路由” | | 仓储资源调用 | `inboundplan/pda/gateway/WarehouseResourceGateway.java`、`HttpWarehouseResourceGateway.java` | diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/dto/ReceivingContainerRequests.java b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/dto/ReceivingContainerRequests.java index 191cad025..093a03cce 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/dto/ReceivingContainerRequests.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/dto/ReceivingContainerRequests.java @@ -3,18 +3,20 @@ package com.baiyue.lastleg.inboundplan.pda.dto; import jakarta.validation.constraints.NotBlank; import jakarta.validation.constraints.NotEmpty; import jakarta.validation.constraints.Positive; +import jakarta.validation.constraints.Size; import java.util.List; public final class ReceivingContainerRequests { private ReceivingContainerRequests() { } - public record Action(@NotBlank String idempotencyKey) { } - public record Sign(@NotEmpty List<@Positive Long> photoFileIds, @NotBlank String idempotencyKey) { } + public record Action(@NotBlank @Size(max = 255) String idempotencyKey) { } + public record Sign(@NotEmpty List<@Positive Long> photoFileIds, + @NotBlank @Size(max = 255) String idempotencyKey) { } public record CartonResolve(@NotBlank String cartonMark) { } public record PalletResolve(@NotBlank String palletCode, @NotBlank String inboundCode) { } public record BulkReceive(@NotBlank String inboundCode, @NotBlank String palletCode, - @Positive int packageCount, @NotBlank String idempotencyKey) { } + @Positive int packageCount, @NotBlank @Size(max = 255) String idempotencyKey) { } public record CartonReceive(@NotBlank String cartonMark, @NotBlank String palletCode, - @NotBlank String idempotencyKey) { } + @NotBlank @Size(max = 255) String idempotencyKey) { } } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/gateway/HttpWarehouseResourceGateway.java b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/gateway/HttpWarehouseResourceGateway.java index c23317f40..c28a0ab89 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/gateway/HttpWarehouseResourceGateway.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/gateway/HttpWarehouseResourceGateway.java @@ -71,6 +71,19 @@ public class HttpWarehouseResourceGateway implements WarehouseResourceGateway { .toBodilessEntity(); } + @Override + public void release(String customerId, long resourceLockId) { + client.post() + .uri("/internal/v1/warehouse-resource-locks/{lockId}/release", resourceLockId) + .header("X-Internal-Service-Token", internalToken) + .body(Map.of("customerId", Long.parseLong(customerId))) + .retrieve() + .onStatus(HttpStatusCode::isError, (request, response) -> { + throw new IllegalStateException("Warehouse rejected pallet release: HTTP " + response.getStatusCode()); + }) + .toBodilessEntity(); + } + @Override public void scrapPallet(String customerId, long palletId) { client.post() diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/gateway/WarehouseResourceGateway.java b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/gateway/WarehouseResourceGateway.java index a1f5f2419..0350073c4 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/gateway/WarehouseResourceGateway.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/gateway/WarehouseResourceGateway.java @@ -4,6 +4,7 @@ public interface WarehouseResourceGateway { WarehousePalletReference ensureDropshipPallet(String customerId, long warehouseId, String businessId); long reserve(String customerId, long warehouseId, long palletId, long locationId, String businessType, String businessId, String reservationKey); + void release(String customerId, long resourceLockId); void scrap(String customerId, long resourceLockId); void scrapPallet(String customerId, long palletId); WarehousePalletReference resolvePallet(String palletCode); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/InboundPdaWorkflowService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/InboundPdaWorkflowService.java index 85427550c..9321d0511 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/InboundPdaWorkflowService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/InboundPdaWorkflowService.java @@ -154,10 +154,10 @@ public class InboundPdaWorkflowService { } @Transactional - public void assignResolvedPallet(String customerId, InboundPlan plan, InboundPlanPallet pallet, + public Long assignResolvedPallet(String customerId, InboundPlan plan, InboundPlanPallet pallet, WarehouseResourceGateway.WarehousePalletReference resource) { if (pallet.getWarehouseResourceLockId() != null) { - if (Objects.equals(pallet.getWarehousePalletId(), resource.palletId())) return; + if (Objects.equals(pallet.getWarehousePalletId(), resource.palletId())) return null; throw conflict("Destination pallet is already bound to another physical pallet"); } String destinationKey = destinationKey(pallet); @@ -165,9 +165,53 @@ public class InboundPdaWorkflowService { String reservationKey = "INBOUND_CONTAINER_DESTINATION_PALLET:" + businessId + ":" + resource.palletId(); long lockId = warehouse.reserve(customerId, resource.warehouseId(), resource.palletId(), resource.locationId(), "INBOUND_CONTAINER_DESTINATION_PALLET", businessId, reservationKey); - pallet.setWarehousePalletId(resource.palletId()); pallet.setWarehouseLocationId(resource.locationId()); - pallet.setWarehouseResourceLockId(lockId); pallet.setAssignmentStatus("ASSIGNED"); - pallet.setLockedAt(LocalDateTime.now()); repository.update(pallet); + try { + pallet.setWarehousePalletId(resource.palletId()); pallet.setWarehouseLocationId(resource.locationId()); + pallet.setWarehouseResourceLockId(lockId); pallet.setAssignmentStatus("ASSIGNED"); + pallet.setLockedAt(LocalDateTime.now()); repository.update(pallet); + return lockId; + } catch (RuntimeException failure) { + compensateReservation(customerId, lockId, failure); + throw failure; + } + } + + public void validateBulkResolved(String customerId, InboundPlan plan, InboundPlanLine line, int packageCount) { + requireActiveTask(customerId, plan); + int received = repository.allocations(plan.getId()).stream() + .filter(value -> Objects.equals(value.getLineId(), line.getId())) + .mapToInt(value -> value(value.getPickedPackageCount())).sum(); + int pending = value(line.getRequiredPackageCount()) - received; + if (packageCount <= 0 || packageCount > pending) { + throw conflict("Received package count exceeds the inbound pending total"); + } + } + + public void validateCartonResolved(String customerId, InboundPlan plan, InboundPlanLine line, + InboundPlanPallet pallet, String cartonMark) { + requireActiveTask(customerId, plan); + InboundPlanCarton carton = repository.lockCarton(customerId, cartonMark); + if (carton == null || !("UNASSIGNED".equals(carton.getStatus()) || "EXPECTED".equals(carton.getStatus()))) { + throw conflict("Carton mark has already been scanned or is unavailable"); + } + InboundPlanLine resolvedLine = carton.getLineId() == null + ? repository.lineForCarton(plan.getId(), carton) : requireLine(plan.getId(), carton.getLineId()); + if (resolvedLine == null || !Objects.equals(resolvedLine.getId(), line.getId())) { + throw conflict("Carton mark belongs to another inbound code or container"); + } + if (!sameWarehouse(carton, pallet)) throw conflict("Carton and pallet destination warehouses must match"); + if (repository.scannedCartons(plan.getId(), line.getId()) >= value(line.getRequiredPackageCount())) { + throw conflict("This inbound plan line has reached its required carton count"); + } + } + + public void compensateReservation(String customerId, Long resourceLockId, RuntimeException originalFailure) { + if (resourceLockId == null) return; + try { + warehouse.release(customerId, resourceLockId); + } catch (RuntimeException releaseFailure) { + originalFailure.addSuppressed(releaseFailure); + } } @Transactional @@ -444,13 +488,42 @@ public class InboundPdaWorkflowService { @Transactional public PdaResponses.Task complete(String customerId, String planNo) { InboundPlan plan = requirePlan(customerId, planNo); + return completeContainer(customerId, List.of(plan)).get(0); + } + + /** Completes every plan of one physical container and emits exactly one container-level outbox event. */ + @Transactional + public List completeContainer(String customerId, List containerPlans) { + if (containerPlans == null || containerPlans.isEmpty()) throw conflict("Receiving container has no inbound plans"); + List orderedPlans = containerPlans.stream().sorted(Comparator.comparing(InboundPlan::getId)).toList(); + String containerNo = orderedPlans.get(0).getContainerNo(); + if (containerNo == null || containerNo.isBlank() || orderedPlans.stream().anyMatch(plan -> + !customerId.equals(plan.getCustomerId()) || !containerNo.equals(plan.getContainerNo()))) { + throw conflict("Inbound plans must belong to one customer and one physical container"); + } + List contexts = orderedPlans.stream().map(plan -> completionContext(customerId, plan)).toList(); + boolean anyCompleted = contexts.stream().anyMatch(context -> "COMPLETED".equals(context.task().getStatus())); + boolean allCompleted = contexts.stream().allMatch(context -> "COMPLETED".equals(context.task().getStatus())); + if (allCompleted) return contexts.stream().map(context -> task(context.task(), context.plan().getPlanNo())).toList(); + if (anyCompleted) throw conflict("Container plans are only partially completed; manual reconciliation is required"); + + LocalDateTime now = LocalDateTime.now(); + for (CompletionContext context : contexts) completePlan(customerId, context, now); + CompletionContext eventSource = contexts.get(0); + unloadCompletedOutbox.publish(eventSource.plan(), eventSource.task(), eventSource.photoFileIds()); + return contexts.stream().map(context -> task(context.task(), context.plan().getPlanNo())).toList(); + } + + private CompletionContext completionContext(String customerId, InboundPlan plan) { InboundPdaTask task = repository.lockTask(customerId, plan.getId()); - if (task == null) throw conflict("Push the inbound plan to PDA before operating"); - if ("COMPLETED".equals(task.getStatus())) return task(task, planNo); + if (task == null) throw conflict("Push every inbound plan in the container to PDA before completing"); + if ("COMPLETED".equals(task.getStatus())) { + return new CompletionContext(plan, task, List.of(), List.of(), List.of(), List.of()); + } List unloadPhotoFileIds = unloadPhotoFileIds(task); if (unloadPhotoFileIds.isEmpty() || task.getUnloadPhotoUploadedAt() == null || task.getEstimatedEmptyReturnAt() == null) { - throw conflict("Upload unload photos before completing the PDA inbound task"); + throw conflict("Upload unload photos for every inbound plan before completing the container"); } Map pickedByLine = new HashMap<>(); List allocations = repository.allocations(plan.getId()); @@ -471,9 +544,15 @@ public class InboundPdaWorkflowService { && !"ON_PALLET".equals(carton.getStatus()))) { throw conflict("All expected carton marks must be scanned onto a pallet before stock-in"); } - LocalDateTime now = LocalDateTime.now(); - stockReceipts.stockIn(customerId, plan, task, lines, allocations, repository.pallets(plan.getId()), now); - for (InboundPlanCarton carton : cartons) { + return new CompletionContext(plan, task, unloadPhotoFileIds, lines, allocations, cartons); + } + + private void completePlan(String customerId, CompletionContext context, LocalDateTime now) { + InboundPlan plan = context.plan(); + InboundPdaTask task = context.task(); + stockReceipts.stockIn(customerId, plan, task, context.lines(), context.allocations(), + repository.pallets(plan.getId()), now); + for (InboundPlanCarton carton : context.cartons()) { carton.setStatus("ON_PALLET".equals(carton.getStatus()) ? "AVAILABLE" : "BULK_RECEIVED"); carton.setReceivedAt(now); carton.setUpdatedAt(now); carton.setVersion(value(carton.getVersion()) + 1); repository.update(carton); @@ -486,8 +565,6 @@ public class InboundPdaWorkflowService { plan.setStatus("COMPLETED"); plan.setUpdatedAt(now); repository.update(plan); - unloadCompletedOutbox.publish(plan, task, unloadPhotoFileIds); - return task(task, planNo); } @Transactional @@ -840,6 +917,11 @@ public class InboundPdaWorkflowService { allocation.setVersion(value(allocation.getVersion()) + 1); } + private record CompletionContext(InboundPlan plan, InboundPdaTask task, List photoFileIds, + List lines, + List allocations, + List cartons) { } + private static int value(Integer value) { return value == null ? 0 : value; } private static BigDecimal volume(BigDecimal value) { return value == null ? ZERO_VOLUME : value; } private static BigDecimal nonNegative(BigDecimal value) { return value.signum() < 0 ? ZERO_VOLUME : value.setScale(4, RoundingMode.HALF_UP); } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxService.java index 5a502dab5..535f62079 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxService.java @@ -19,11 +19,13 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import jakarta.annotation.Resource; import org.springframework.stereotype.Service; +import org.springframework.dao.DuplicateKeyException; import java.time.Instant; import java.time.LocalDateTime; import java.time.ZoneOffset; import java.util.List; +import java.util.Objects; /** Writes the overseas-to-CN handover to the local transactional outbox. */ @Service @@ -32,31 +34,88 @@ public class InboundUnloadCompletedOutboxService { @Resource private LastlegOrderMapper orders; @Resource private ObjectMapper objectMapper; + /** 兼容单计划入仓完成；多计划同柜必须改调 publishContainerCompleted，并传柜级完成时刻。 */ public void publish(InboundPlan plan, InboundPdaTask task, List photoFileIds) { LastlegOrder sourceOrder = orders.selectById(plan.getSourceOrderId()); if (sourceOrder == null || sourceOrder.getSourceContainerBookingId() == null || sourceOrder.getSourceContainerBookingId().isBlank()) { throw new IllegalStateException("Inbound plan has no source container booking ID"); } - Instant photoUploadedAt = task.getUnloadPhotoUploadedAt().toInstant(ZoneOffset.UTC); - var payload = new OverseasContainerUnloadedV1(sourceOrder.getSourceContainerBookingId(), - plan.getContainerNo(), photoUploadedAt, photoUploadedAt, null); + if (task.getCompletedAt() == null) { + throw new IllegalStateException("Inbound task has no completion time"); + } + if (task.getUnloadPhotoUploadedAt() == null) { + throw new IllegalStateException("Inbound task has no unload photo upload time"); + } + publishContainerCompleted(task.getCustomerId(), sourceOrder.getSourceContainerBookingId(), + plan.getContainerNo(), task.getCompletedAt().toInstant(ZoneOffset.UTC), + task.getUnloadPhotoUploadedAt().toInstant(ZoneOffset.UTC), + Math.max(1, task.getVersion() == null ? 1 : task.getVersion())); + } + + /** + * 柜级唯一发布入口。调用方只能在同一物理柜的全部计划完成后调用一次；重复调用保留首次冻结事实。 + */ + public void publishContainerCompleted(String customerId, String sourceContainerBookingId, + String containerNo, Instant unloadingCompletedAt, + Instant unloadingPhotoUploadedAt, long aggregateVersion) { + requireText(customerId, "customerId"); + requireText(sourceContainerBookingId, "sourceContainerBookingId"); + requireText(containerNo, "containerNo"); + Objects.requireNonNull(unloadingCompletedAt, "unloadingCompletedAt"); + Objects.requireNonNull(unloadingPhotoUploadedAt, "unloadingPhotoUploadedAt"); + if (unloadingCompletedAt.isBefore(unloadingPhotoUploadedAt)) { + throw new IllegalArgumentException("unloadingCompletedAt cannot be before unloadingPhotoUploadedAt"); + } + + String eventKey = EventTypes.OVERSEAS_CONTAINER_UNLOADED + ':' + sourceContainerBookingId; + LastlegEventOutbox existing = mapper.findByEventKey(eventKey); + if (existing != null) { + requireSameContainer(existing, customerId, sourceContainerBookingId, containerNo); + return; + } + var payload = new OverseasContainerUnloadedV1(sourceContainerBookingId, + containerNo, unloadingCompletedAt, unloadingPhotoUploadedAt, null); JsonNode payloadNode = objectMapper.valueToTree(payload); - long aggregateVersion = Math.max(1, task.getVersion() == null ? 1 : task.getVersion()); - String eventKey = EventTypes.OVERSEAS_CONTAINER_UNLOADED + ':' + task.getId(); String traceId = EventIdentifiers.newUuid(); EventEnvelope envelope = new EventEnvelope(EventIdentifiers.newUuid(), EventTypes.OVERSEAS_CONTAINER_UNLOADED, - 1, Instant.now(), RegionCode.OVERSEAS, RegionCode.CN, task.getCustomerId(), "CONTAINER_BOOKING", - sourceOrder.getSourceContainerBookingId(), aggregateVersion, 1L, traceId, eventKey, null, payloadNode, + 1, Instant.now(), RegionCode.OVERSEAS, RegionCode.CN, customerId, "CONTAINER_BOOKING", + sourceContainerBookingId, Math.max(1, aggregateVersion), 1L, traceId, eventKey, null, payloadNode, EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(payloadNode))).requireValid(); LastlegEventOutbox event = new LastlegEventOutbox(); - event.setId(IdWorker.getId()); event.setCustomerId(task.getCustomerId()); + event.setId(IdWorker.getId()); event.setCustomerId(customerId); 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.now(ZoneOffset.UTC)); - mapper.insert(event); + try { + mapper.insert(event); + } catch (DuplicateKeyException exception) { + LastlegEventOutbox raced = mapper.findByEventKey(eventKey); + if (raced == null) throw exception; + requireSameContainer(raced, customerId, sourceContainerBookingId, containerNo); + } + } + + private void requireSameContainer(LastlegEventOutbox existing, String customerId, + String sourceContainerBookingId, String containerNo) { + try { + EventEnvelope envelope = objectMapper.readValue(existing.getPayloadJson(), EventEnvelope.class); + OverseasContainerUnloadedV1 frozen = objectMapper.treeToValue( + envelope.payload(), OverseasContainerUnloadedV1.class); + if (!Objects.equals(existing.getCustomerId(), customerId) + || !Objects.equals(existing.getAggregateId(), sourceContainerBookingId) + || !Objects.equals(frozen.containerNo(), containerNo)) { + throw new IllegalStateException("Unload-completed event key conflicts with another container"); + } + } catch (JsonProcessingException exception) { + throw new IllegalStateException("Cannot read existing unload-completed event", exception); + } + } + + private void requireText(String value, String field) { + if (value == null || value.isBlank()) throw new IllegalArgumentException(field + " is required"); } private String json(Object value) { diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/ReceivingContainerService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/ReceivingContainerService.java index 5f92fb9d6..c34323c29 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/ReceivingContainerService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/ReceivingContainerService.java @@ -92,8 +92,8 @@ public class ReceivingContainerService { public ReceivingContainerResponses.Session detail(String customerId, String operatorKey, String containerNo) { List containerPlans = requirePlans(customerId, containerNo); ReceivingContainerSession current = sessions.findActiveByOperator(customerId, activeOperatorKey(customerId, operatorKey)); - if (current != null && !containerNo.equals(current.getContainerNo()) && !"WAIT_SIGN".equals(current.getStatus())) { - throw conflict("Current account is locked to another signed container"); + if (current != null && !containerNo.equals(current.getContainerNo())) { + throw conflict("Current account is locked to another selected container"); } return snapshot(current != null && containerNo.equals(current.getContainerNo()) ? current : waitingView( customerId, operatorKey, containerNo), containerPlans); @@ -102,7 +102,8 @@ public class ReceivingContainerService { @Transactional public ReceivingContainerResponses.Session prepare(String customerId, String operatorKey, String operatorName, String containerNo, ReceivingContainerRequests.Action request) { - ReceivingContainerOperation replay = replay(customerId, operatorKey, "PREPARE", request.idempotencyKey(), request); + PrepareCommand command = new PrepareCommand(containerNo, request.idempotencyKey()); + ReceivingContainerOperation replay = replay(customerId, operatorKey, "PREPARE", request.idempotencyKey(), command); if (replay != null) return read(replay, ReceivingContainerResponses.Session.class); List containerPlans = requirePlans(customerId, containerNo); String activeOperatorKey = activeOperatorKey(customerId, operatorKey); @@ -110,7 +111,7 @@ public class ReceivingContainerService { if (active != null) { if (!containerNo.equals(active.getContainerNo())) throw conflict("Finish the current container before selecting another one"); ReceivingContainerResponses.Session result = snapshot(active, containerPlans); - save(active, "PREPARE", request.idempotencyKey(), request, result); return result; + save(active, "PREPARE", request.idempotencyKey(), command, result); return result; } if (sessions.lockActiveByContainer(customerId, activeContainerKey(customerId, containerNo)) != null) { throw conflict("Container is already signed or prepared by another account"); @@ -125,7 +126,7 @@ public class ReceivingContainerService { try { sessions.insert(session); } catch (DuplicateKeyException duplicate) { throw conflict("Account or container already has an active receiving session"); } ReceivingContainerResponses.Session result = snapshot(session, containerPlans); - save(session, "PREPARE", request.idempotencyKey(), request, result); return result; + save(session, "PREPARE", request.idempotencyKey(), command, result); return result; } public ReceivingContainerResponses.Photo uploadPhoto(String customerId, String operatorKey, long operatorId, @@ -203,9 +204,16 @@ public class ReceivingContainerService { WarehouseResourceGateway.WarehousePalletReference resource = warehouse.resolvePallet(request.palletCode()); validatePallet(ref, resource); InboundPlanPallet pallet = choosePallet(ref, resource); - workflow.assignResolvedPallet(customerId, ref.plan(), pallet, resource); - int accepted = workflow.receiveBulkResolved(customerId, ref.plan(), ref.line(), pallet, - request.packageCount(), operatorKey, operatorName); + workflow.validateBulkResolved(customerId, ref.plan(), ref.line(), request.packageCount()); + Long resourceLockId = workflow.assignResolvedPallet(customerId, ref.plan(), pallet, resource); + final int accepted; + try { + accepted = workflow.receiveBulkResolved(customerId, ref.plan(), ref.line(), pallet, + request.packageCount(), operatorKey, operatorName); + } catch (RuntimeException failure) { + workflow.compensateReservation(customerId, resourceLockId, failure); + throw failure; + } markReceiving(session); ReceivingContainerResponses.Inbound inbound = inbound(ref); ReceivingContainerResponses.Receipt result = new ReceivingContainerResponses.Receipt("RECEIVED", @@ -232,9 +240,15 @@ public class ReceivingContainerService { WarehouseResourceGateway.WarehousePalletReference resource = warehouse.resolvePallet(request.palletCode()); validatePallet(ref, resource); InboundPlanPallet pallet = choosePallet(ref, resource); - workflow.assignResolvedPallet(customerId, ref.plan(), pallet, resource); - workflow.scanCarton(customerId, ref.plan().getPlanNo(), new PdaRequests.ScanCarton( - request.cartonMark(), pallet.getId(), request.idempotencyKey() + ":SCAN", operatorKey, operatorName)); + workflow.validateCartonResolved(customerId, ref.plan(), ref.line(), pallet, request.cartonMark()); + Long resourceLockId = workflow.assignResolvedPallet(customerId, ref.plan(), pallet, resource); + try { + workflow.scanCarton(customerId, ref.plan().getPlanNo(), new PdaRequests.ScanCarton( + request.cartonMark(), pallet.getId(), request.idempotencyKey() + ":SCAN", operatorKey, operatorName)); + } catch (RuntimeException failure) { + workflow.compensateReservation(customerId, resourceLockId, failure); + throw failure; + } markReceiving(session); ReceivingContainerResponses.Inbound inbound = inbound(ref); ReceivingContainerResponses.Receipt result = new ReceivingContainerResponses.Receipt("RECEIVED", @@ -252,7 +266,7 @@ public class ReceivingContainerService { List containerPlans = requirePlans(customerId, session.getContainerNo()); Totals totals = totals(containerPlans); if (totals.total() != totals.received()) throw conflict("Every inbound code must be fully received before completing the container"); - for (InboundPlan plan : containerPlans) workflow.complete(customerId, plan.getPlanNo()); + workflow.completeContainer(customerId, containerPlans); LocalDateTime now = LocalDateTime.now(); session.setStatus("COMPLETED"); session.setCompletedAt(now); session.setUpdatedAt(now); session.setActiveOperatorKey(null); session.setActiveContainerKey(null); update(session); @@ -420,6 +434,7 @@ public class ReceivingContainerService { } private ReceivingContainerOperation replay(String customerId, String operatorKey, String action, String key, Object request) { + key = idempotencyKey(key); ReceivingContainerOperation value = operations.find(customerId, operatorKey, key); if (value == null) return null; if (!action.equals(value.getAction()) || !hash(request).equals(value.getRequestHash())) { @@ -429,6 +444,7 @@ public class ReceivingContainerService { } private void save(ReceivingContainerSession session, String action, String key, Object request, Object result) { + key = idempotencyKey(key); ReceivingContainerOperation value = new ReceivingContainerOperation(); value.setId(IdWorker.getId()); value.setSessionId(session.getId()); value.setCustomerId(session.getCustomerId()); value.setOperatorKey(session.getOperatorKey()); value.setAction(action); value.setIdempotencyKey(key); @@ -460,10 +476,17 @@ public class ReceivingContainerService { private static String activeOperatorKey(String customerId, String operatorKey) { return customerId + ":" + operatorKey; } private static String activeContainerKey(String customerId, String containerNo) { return customerId + ":" + containerNo; } + private static String idempotencyKey(String value) { + if (value == null || value.isBlank()) throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "Idempotency key is required"); + value = value.trim(); + if (value.length() > 255) throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "Idempotency key exceeds 255 characters"); + return value; + } private static int value(Integer value) { return value == null ? 0 : value; } private static ResponseStatusException conflict(String message) { return new ResponseStatusException(HttpStatus.CONFLICT, message); } private static ResponseStatusException notFound(String message) { return new ResponseStatusException(HttpStatus.NOT_FOUND, message); } private record InboundRef(InboundPlan plan, InboundPlanLine line, LastlegOrder order) { } private record Totals(int inboundCount, int total, int received) { } + private record PrepareCommand(String containerNo, String idempotencyKey) { } } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/mapper/LastlegEventOutboxMapper.java b/lastleg-service/src/main/java/com/baiyue/lastleg/mapper/LastlegEventOutboxMapper.java index c7871d177..767551233 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/mapper/LastlegEventOutboxMapper.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/mapper/LastlegEventOutboxMapper.java @@ -3,7 +3,11 @@ package com.baiyue.lastleg.mapper; import com.baomidou.mybatisplus.core.mapper.BaseMapper; import com.baiyue.lastleg.entity.LastlegEventOutbox; import org.apache.ibatis.annotations.Mapper; +import org.apache.ibatis.annotations.Param; +import org.apache.ibatis.annotations.Select; @Mapper public interface LastlegEventOutboxMapper extends BaseMapper { + @Select("SELECT * FROM ll_event_outbox WHERE event_key=#{eventKey} LIMIT 1") + LastlegEventOutbox findByEventKey(@Param("eventKey") String eventKey); } diff --git a/lastleg-service/src/main/resources/db/lastleg.sql b/lastleg-service/src/main/resources/db/lastleg.sql index 0f3c1503e..c02cca9d0 100644 --- a/lastleg-service/src/main/resources/db/lastleg.sql +++ b/lastleg-service/src/main/resources/db/lastleg.sql @@ -640,22 +640,35 @@ SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE tab PREPARE inbound_schema_stmt FROM @ddl; EXECUTE inbound_schema_stmt; DEALLOCATE PREPARE inbound_schema_stmt; SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='ll_inbound_plan_line' AND column_name='customer_id'), 'SELECT 1', 'ALTER TABLE ll_inbound_plan_line ADD COLUMN customer_id VARCHAR(128) NOT NULL DEFAULT '''' COMMENT ''客户账号标识'' AFTER id')); PREPARE receiving_container_line_stmt FROM @ddl; EXECUTE receiving_container_line_stmt; DEALLOCATE PREPARE receiving_container_line_stmt; -UPDATE ll_inbound_plan_line l JOIN ll_inbound_plan p ON p.id=l.plan_id SET l.customer_id=p.customer_id WHERE l.customer_id=''; +UPDATE ll_inbound_plan_line l JOIN ll_inbound_plan p ON p.id=l.plan_id SET l.customer_id=p.customer_id WHERE l.customer_id IS NULL OR l.customer_id=''; ALTER TABLE ll_inbound_plan_line MODIFY COLUMN customer_id VARCHAR(128) NOT NULL COMMENT '客户账号标识'; SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name='ll_inbound_plan_line' AND index_name='idx_ll_inbound_line_customer_plan'), 'SELECT 1', 'ALTER TABLE ll_inbound_plan_line ADD KEY idx_ll_inbound_line_customer_plan(customer_id,plan_id)')); PREPARE inbound_customer_scope_stmt FROM @ddl; EXECUTE inbound_customer_scope_stmt; DEALLOCATE PREPARE inbound_customer_scope_stmt; SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='ll_inbound_plan_pallet' AND column_name='customer_id'), 'SELECT 1', 'ALTER TABLE ll_inbound_plan_pallet ADD COLUMN customer_id VARCHAR(128) NOT NULL DEFAULT '''' COMMENT ''客户账号标识'' AFTER id')); PREPARE inbound_customer_scope_stmt FROM @ddl; EXECUTE inbound_customer_scope_stmt; DEALLOCATE PREPARE inbound_customer_scope_stmt; -UPDATE ll_inbound_plan_pallet v JOIN ll_inbound_plan p ON p.id=v.plan_id SET v.customer_id=p.customer_id WHERE v.customer_id=''; +UPDATE ll_inbound_plan_pallet v JOIN ll_inbound_plan p ON p.id=v.plan_id SET v.customer_id=p.customer_id WHERE v.customer_id IS NULL OR v.customer_id=''; ALTER TABLE ll_inbound_plan_pallet MODIFY COLUMN customer_id VARCHAR(128) NOT NULL COMMENT '客户账号标识'; SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name='ll_inbound_plan_pallet' AND index_name='idx_ll_inbound_pallet_customer_plan'), 'SELECT 1', 'ALTER TABLE ll_inbound_plan_pallet ADD KEY idx_ll_inbound_pallet_customer_plan(customer_id,plan_id)')); PREPARE inbound_customer_scope_stmt FROM @ddl; EXECUTE inbound_customer_scope_stmt; DEALLOCATE PREPARE inbound_customer_scope_stmt; SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='ll_inbound_plan_pallet_allocation' AND column_name='customer_id'), 'SELECT 1', 'ALTER TABLE ll_inbound_plan_pallet_allocation ADD COLUMN customer_id VARCHAR(128) NOT NULL DEFAULT '''' COMMENT ''客户账号标识'' AFTER id')); PREPARE inbound_customer_scope_stmt FROM @ddl; EXECUTE inbound_customer_scope_stmt; DEALLOCATE PREPARE inbound_customer_scope_stmt; -UPDATE ll_inbound_plan_pallet_allocation a JOIN ll_inbound_plan p ON p.id=a.plan_id SET a.customer_id=p.customer_id WHERE a.customer_id=''; +UPDATE ll_inbound_plan_pallet_allocation a JOIN ll_inbound_plan p ON p.id=a.plan_id SET a.customer_id=p.customer_id WHERE a.customer_id IS NULL OR a.customer_id=''; ALTER TABLE ll_inbound_plan_pallet_allocation MODIFY COLUMN customer_id VARCHAR(128) NOT NULL COMMENT '客户账号标识'; SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name='ll_inbound_plan_pallet_allocation' AND index_name='idx_ll_inbound_allocation_customer_plan'), 'SELECT 1', 'ALTER TABLE ll_inbound_plan_pallet_allocation ADD KEY idx_ll_inbound_allocation_customer_plan(customer_id,plan_id)')); PREPARE inbound_customer_scope_stmt FROM @ddl; EXECUTE inbound_customer_scope_stmt; DEALLOCATE PREPARE inbound_customer_scope_stmt; +DROP PROCEDURE IF EXISTS assert_inbound_plan_customer_scope_ready; +DELIMITER $$ +CREATE PROCEDURE assert_inbound_plan_customer_scope_ready() +BEGIN + IF EXISTS (SELECT 1 FROM ll_inbound_plan_line l LEFT JOIN ll_inbound_plan p ON p.id=l.plan_id WHERE l.customer_id IS NULL OR l.customer_id='' OR p.id IS NULL OR l.customer_id<>p.customer_id) + OR EXISTS (SELECT 1 FROM ll_inbound_plan_pallet v LEFT JOIN ll_inbound_plan p ON p.id=v.plan_id WHERE v.customer_id IS NULL OR v.customer_id='' OR p.id IS NULL OR v.customer_id<>p.customer_id) + OR EXISTS (SELECT 1 FROM ll_inbound_plan_pallet_allocation a LEFT JOIN ll_inbound_plan p ON p.id=a.plan_id WHERE a.customer_id IS NULL OR a.customer_id='' OR p.id IS NULL OR a.customer_id<>p.customer_id) THEN + SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='Inbound plan child customer scope migration is incomplete'; + END IF; +END$$ +DELIMITER ; +CALL assert_inbound_plan_customer_scope_ready(); +DROP PROCEDURE assert_inbound_plan_customer_scope_ready; SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='ll_inbound_plan_line' AND column_name='destination_type'), 'ALTER TABLE ll_inbound_plan_line MODIFY COLUMN destination_type VARCHAR(32) NOT NULL DEFAULT ''OVERSEAS'' COMMENT ''目的类型：FBA按箱数接收、OVERSEAS逐箱扫码''', 'ALTER TABLE ll_inbound_plan_line ADD COLUMN destination_type VARCHAR(32) NOT NULL DEFAULT ''OVERSEAS'' COMMENT ''目的类型：FBA按箱数接收、OVERSEAS逐箱扫码'' AFTER inbound_code')); PREPARE receiving_container_line_stmt FROM @ddl; EXECUTE receiving_container_line_stmt; DEALLOCATE PREPARE receiving_container_line_stmt; UPDATE ll_inbound_plan_line l JOIN ll_order o ON o.id=l.order_id diff --git a/lastleg-service/src/main/resources/openapi/overseas-business-v1.yaml b/lastleg-service/src/main/resources/openapi/overseas-business-v1.yaml index a485c55f7..f3de2ced2 100644 --- a/lastleg-service/src/main/resources/openapi/overseas-business-v1.yaml +++ b/lastleg-service/src/main/resources/openapi/overseas-business-v1.yaml @@ -8,14 +8,14 @@ paths: summary: 查询当前客户待签收柜号 description: 权限 lastleg:pda:operate；客户与操作账号均从 PDA Token 获取，不接收 customerId。 security: [{ bearerAuth: [] }] - responses: { '200': { description: 柜号候选及箱数进度 }, '403': { description: 缺少 PDA 操作权限 } } + responses: { '200': { description: 柜号候选及箱数进度, content: { application/json: { schema: { $ref: '#/components/schemas/ReceivingCandidateListApiResponse' } } } }, '403': { description: 缺少 PDA 操作权限 } } /api/pda/v1/last-leg/receiving-containers/{containerNo}: get: summary: 查询待签收柜号详情 description: 权限 lastleg:pda:operate；返回柜内全部入仓码、目的类型、目的仓、托数和实收进度。 security: [{ bearerAuth: [] }] parameters: [{ name: containerNo, in: path, required: true, schema: { type: string } }] - responses: { '200': { description: 柜号详情 }, '409': { description: 当前账号已锁定其他柜号 } } + responses: { '200': { description: 柜号详情, content: { application/json: { schema: { $ref: '#/components/schemas/ReceivingSessionApiResponse' } } } }, '409': { description: 当前账号已锁定其他柜号 } } /api/pda/v1/last-leg/receiving-containers/{containerNo}/prepare: post: summary: 选择当前待签收柜号 @@ -23,62 +23,62 @@ paths: security: [{ bearerAuth: [] }] parameters: [{ name: containerNo, in: path, required: true, schema: { type: string } }] requestBody: { required: true, content: { application/json: { schema: { $ref: '#/components/schemas/ReceivingAction' } } } } - responses: { '200': { description: 当前柜会话 }, '409': { description: 账号或柜号已有未完成会话 } } + responses: { '200': { description: 当前柜会话, content: { application/json: { schema: { $ref: '#/components/schemas/ReceivingSessionApiResponse' } } } }, '409': { description: 账号或柜号已有未完成会话 } } /api/pda/v1/last-leg/receiving-containers/current/unload-photos/files: post: summary: 上传当前柜签收照片 description: 权限 lastleg:pda:operate；multipart 文件归属当前 WAIT_SIGN 会话。 security: [{ bearerAuth: [] }] requestBody: { required: true, content: { multipart/form-data: { schema: { type: object, required: [file], properties: { file: { type: string, format: binary } } } } } } - responses: { '200': { description: 文件标识 }, '409': { description: 当前柜已签收 } } + responses: { '200': { description: 文件标识, content: { application/json: { schema: { $ref: '#/components/schemas/ReceivingPhotoApiResponse' } } } }, '409': { description: 当前柜已签收 } } /api/pda/v1/last-leg/receiving-containers/current/sign: post: summary: 签收当前柜 description: 权限 lastleg:pda:operate；校验非空照片归属后由 WAIT_SIGN 推进 SIGNED。 security: [{ bearerAuth: [] }] requestBody: { required: true, content: { application/json: { schema: { $ref: '#/components/schemas/ReceivingSign' } } } } - responses: { '200': { description: 已签收会话 }, '409': { description: 状态或照片校验失败 } } + responses: { '200': { description: 已签收会话, content: { application/json: { schema: { $ref: '#/components/schemas/ReceivingSessionApiResponse' } } } }, '409': { description: 状态或照片校验失败 } } /api/pda/v1/last-leg/receiving-containers/current: get: summary: 恢复当前账号已签柜 - description: 权限 lastleg:pda:operate；返回 SIGNED 或 RECEIVING 当前柜，不依赖 planNo。 + description: 权限 lastleg:pda:operate；返回 WAIT_SIGN、SIGNED 或 RECEIVING 当前柜，不依赖 planNo。 security: [{ bearerAuth: [] }] - responses: { '200': { description: 当前柜完整进度 }, '404': { description: 无当前柜 } } + responses: { '200': { description: 当前柜完整进度, content: { application/json: { schema: { $ref: '#/components/schemas/ReceivingSessionApiResponse' } } } }, '404': { description: 无当前柜 } } /api/pda/v1/last-leg/receiving-containers/current/cartons/resolve: post: summary: 解析当前柜箱唛 description: 权限 lastleg:pda:operate；仅 OVERSEAS 目的类型可逐箱收货，返回该箱所属入仓码完整信息。 security: [{ bearerAuth: [] }] requestBody: { required: true, content: { application/json: { schema: { type: object, required: [cartonMark], properties: { cartonMark: { type: string } } } } } } - responses: { '200': { description: 箱唛与入仓码信息 }, '409': { description: 跨柜、重复箱或目的类型不符 } } + responses: { '200': { description: 箱唛与入仓码信息, content: { application/json: { schema: { $ref: '#/components/schemas/ReceivingCartonResolvedApiResponse' } } } }, '409': { description: 跨柜、重复箱或目的类型不符 } } /api/pda/v1/last-leg/receiving-containers/current/pallets/resolve: post: summary: 解析当前柜实物托盘码 description: 权限 lastleg:pda:operate；托盘固定从 Warehouse Overseas 公共资源作用域解析并校验实际仓与目的仓。 security: [{ bearerAuth: [] }] requestBody: { required: true, content: { application/json: { schema: { type: object, required: [palletCode, inboundCode], properties: { palletCode: { type: string }, inboundCode: { type: string } } } } } } - responses: { '200': { description: 可用托盘与目的匹配结果 }, '409': { description: 仓库、目的仓或托盘状态不符 } } + responses: { '200': { description: 可用托盘与目的匹配结果, content: { application/json: { schema: { $ref: '#/components/schemas/ReceivingPalletResolvedApiResponse' } } } }, '409': { description: 仓库、目的仓或托盘状态不符 } } /api/pda/v1/last-leg/receiving-containers/current/receipts/bulk: post: summary: FBA 入仓码按箱数收货 description: 权限 lastleg:pda:operate；只允许 destinationType=FBA，OVERSEAS 在服务端硬拒绝；不接收 customerId、planNo、allocationId、warehouseId、locationId。 security: [{ bearerAuth: [] }] requestBody: { required: true, content: { application/json: { schema: { $ref: '#/components/schemas/ReceivingBulk' } } } } - responses: { '200': { description: 本次与累计实收进度 }, '409': { description: 海外仓批量、超收或托盘不匹配 } } + responses: { '200': { description: 本次与累计实收进度, content: { application/json: { schema: { $ref: '#/components/schemas/ReceivingReceiptApiResponse' } } } }, '409': { description: 海外仓批量、超收或托盘不匹配 } } /api/pda/v1/last-leg/receiving-containers/current/receipts/carton: post: summary: 海外仓入仓码逐箱收货 description: 权限 lastleg:pda:operate；只允许 destinationType=OVERSEAS，拒绝跨柜、重复箱和不匹配托盘。 security: [{ bearerAuth: [] }] requestBody: { required: true, content: { application/json: { schema: { $ref: '#/components/schemas/ReceivingCarton' } } } } - responses: { '200': { description: 本次与累计实收进度 }, '409': { description: FBA逐箱、跨柜、重复箱或托盘不匹配 } } + responses: { '200': { description: 本次与累计实收进度, content: { application/json: { schema: { $ref: '#/components/schemas/ReceivingReceiptApiResponse' } } } }, '409': { description: FBA逐箱、跨柜、重复箱或托盘不匹配 } } /api/pda/v1/last-leg/receiving-containers/current/complete: post: summary: 完成当前柜收货并入库存 description: 权限 lastleg:pda:operate；仅柜内每个入仓码实收总数精确相等后完成，库存、计划、尾程状态与卸柜 Outbox 同事务落地。 security: [{ bearerAuth: [] }] requestBody: { required: true, content: { application/json: { schema: { $ref: '#/components/schemas/ReceivingAction' } } } } - responses: { '200': { description: COMPLETED 柜号会话 }, '409': { description: 数量未核平或状态冲突 } } + responses: { '200': { description: COMPLETED 柜号会话, content: { application/json: { schema: { $ref: '#/components/schemas/ReceivingSessionApiResponse' } } } }, '409': { description: 数量未核平或状态冲突 } } /api/platform/v1/last-leg/orders: get: summary: 分页查询尾程订单 diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxServiceTest.java new file mode 100644 index 000000000..4712eba02 --- /dev/null +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxServiceTest.java @@ -0,0 +1,136 @@ +package com.baiyue.lastleg.inboundplan.pda.service; + +import com.baiyue.common.event.EventEnvelope; +import com.baiyue.common.event.EventTypes; +import com.baiyue.common.event.OverseasContainerUnloadedV1; +import com.baiyue.lastleg.entity.LastlegEventOutbox; +import com.baiyue.lastleg.entity.LastlegOrder; +import com.baiyue.lastleg.inboundplan.entity.InboundPlan; +import com.baiyue.lastleg.inboundplan.pda.entity.InboundPdaTask; +import com.baiyue.lastleg.mapper.LastlegEventOutboxMapper; +import com.baiyue.lastleg.mapper.LastlegOrderMapper; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.springframework.test.util.ReflectionTestUtils; + +import java.time.Instant; +import java.time.LocalDateTime; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class InboundUnloadCompletedOutboxServiceTest { + private final LastlegEventOutboxMapper mapper = mock(LastlegEventOutboxMapper.class); + private final LastlegOrderMapper orders = mock(LastlegOrderMapper.class); + private final ObjectMapper json = new ObjectMapper().findAndRegisterModules(); + private final InboundUnloadCompletedOutboxService service = new InboundUnloadCompletedOutboxService(); + + @BeforeEach + void setup() { + ReflectionTestUtils.setField(service, "mapper", mapper); + ReflectionTestUtils.setField(service, "orders", orders); + ReflectionTestUtils.setField(service, "objectMapper", json); + } + + @Test + void freezesContainerCompletionSeparatelyFromPhotoUpload() throws Exception { + Instant photoUploadedAt = Instant.parse("2026-08-24T08:00:00Z"); + Instant containerCompletedAt = Instant.parse("2026-08-24T09:30:00Z"); + + service.publishContainerCompleted("100", "booking-1", "MSCU1234567", + containerCompletedAt, photoUploadedAt, 7); + + ArgumentCaptor captured = ArgumentCaptor.forClass(LastlegEventOutbox.class); + verify(mapper).insert(captured.capture()); + LastlegEventOutbox row = captured.getValue(); + EventEnvelope envelope = json.readValue(row.getPayloadJson(), EventEnvelope.class); + OverseasContainerUnloadedV1 payload = json.treeToValue( + envelope.payload(), OverseasContainerUnloadedV1.class); + assertThat(payload.unloadingCompletedAt()).isEqualTo(containerCompletedAt); + assertThat(payload.unloadingPhotoUploadedAt()).isEqualTo(photoUploadedAt); + assertThat(row.getEventKey()).isEqualTo(EventTypes.OVERSEAS_CONTAINER_UNLOADED + ":booking-1"); + assertThat(envelope.aggregateId()).isEqualTo("booking-1"); + } + + @Test + void legacySinglePlanEntryUsesTaskCompletionTime() throws Exception { + InboundPlan plan = new InboundPlan(); + plan.setSourceOrderId(11L); + plan.setContainerNo("MSCU1234567"); + InboundPdaTask task = new InboundPdaTask(); + task.setCustomerId("100"); + task.setCompletedAt(LocalDateTime.parse("2026-08-24T09:30:00")); + task.setUnloadPhotoUploadedAt(LocalDateTime.parse("2026-08-24T08:00:00")); + task.setVersion(3); + LastlegOrder order = new LastlegOrder(); + order.setSourceContainerBookingId("booking-1"); + when(orders.selectById(11L)).thenReturn(order); + + service.publish(plan, task, List.of(1L)); + + ArgumentCaptor captured = ArgumentCaptor.forClass(LastlegEventOutbox.class); + verify(mapper).insert(captured.capture()); + EventEnvelope envelope = json.readValue(captured.getValue().getPayloadJson(), EventEnvelope.class); + OverseasContainerUnloadedV1 payload = json.treeToValue( + envelope.payload(), OverseasContainerUnloadedV1.class); + assertThat(payload.unloadingCompletedAt()).isEqualTo(Instant.parse("2026-08-24T09:30:00Z")); + assertThat(payload.unloadingPhotoUploadedAt()).isEqualTo(Instant.parse("2026-08-24T08:00:00Z")); + } + + @Test + void repeatedContainerCompletionDoesNotPublishAnotherEvent() { + LastlegEventOutbox existing = existing("100", "booking-1", "MSCU1234567"); + when(mapper.findByEventKey(EventTypes.OVERSEAS_CONTAINER_UNLOADED + ":booking-1")) + .thenReturn(existing); + + service.publishContainerCompleted("100", "booking-1", "MSCU1234567", + Instant.parse("2026-08-24T10:00:00Z"), Instant.parse("2026-08-24T08:00:00Z"), 8); + + verify(mapper, never()).insert(any(LastlegEventOutbox.class)); + } + + @Test + void sameIdempotencyKeyCannotRepresentAnotherPhysicalContainer() { + LastlegEventOutbox existing = existing("100", "booking-1", "MSCU1234567"); + when(mapper.findByEventKey(EventTypes.OVERSEAS_CONTAINER_UNLOADED + ":booking-1")) + .thenReturn(existing); + + assertThatThrownBy(() -> service.publishContainerCompleted("100", "booking-1", "TGHU7654321", + Instant.parse("2026-08-24T10:00:00Z"), Instant.parse("2026-08-24T08:00:00Z"), 8)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("conflicts with another container"); + } + + private LastlegEventOutbox existing(String customerId, String bookingId, String containerNo) { + OverseasContainerUnloadedV1 payload = new OverseasContainerUnloadedV1(bookingId, containerNo, + Instant.parse("2026-08-24T09:30:00Z"), Instant.parse("2026-08-24T08:00:00Z"), null); + EventEnvelope envelope = new EventEnvelope("123e4567-e89b-12d3-a456-426614174000", + EventTypes.OVERSEAS_CONTAINER_UNLOADED, 1, Instant.parse("2026-08-24T09:30:01Z"), + com.baiyue.common.event.RegionCode.OVERSEAS, com.baiyue.common.event.RegionCode.CN, + customerId, "CONTAINER_BOOKING", bookingId, 7, 1L, + "123e4567-e89b-12d3-a456-426614174001", + EventTypes.OVERSEAS_CONTAINER_UNLOADED + ':' + bookingId, null, + json.valueToTree(payload), "sha256:" + "0".repeat(64)); + LastlegEventOutbox row = new LastlegEventOutbox(); + row.setCustomerId(customerId); + row.setAggregateId(bookingId); + row.setPayloadJson(write(envelope)); + return row; + } + + private String write(Object value) { + try { + return json.writeValueAsString(value); + } catch (Exception exception) { + throw new IllegalStateException(exception); + } + } +}