1

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) { }
 }

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()

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);

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<PdaResponses.Task> completeContainer(String customerId, List<InboundPlan> containerPlans) {
+        if (containerPlans == null || containerPlans.isEmpty()) throw conflict("Receiving container has no inbound plans");
+        List<InboundPlan> 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<CompletionContext> 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<Long> 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<Long, Integer> pickedByLine = new HashMap<>();
         List<InboundPlanPalletAllocation> 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<Long> photoFileIds,
+                                     List<InboundPlanLine> lines,
+                                     List<InboundPlanPalletAllocation> allocations,
+                                     List<InboundPlanCarton> 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); }

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<Long> 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) {

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<InboundPlan> 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<InboundPlan> 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<InboundPlan> 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) { }
 }

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<LastlegEventOutbox> {
+    @Select("SELECT * FROM ll_event_outbox WHERE event_key=#{eventKey} LIMIT 1")
+    LastlegEventOutbox findByEventKey(@Param("eventKey") String eventKey);
 }