From c194dd286289fcb603b9fb181326e020221fcffd Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Mon, 5 Oct 2026 03:37:13 +0800 Subject: [PATCH] fix(transfer): reserve receipt resources with durable recovery intents --- .../gateway/HttpWarehouseResourceGateway.java | 13 +- .../WarehouseTransferReceiptCommand.java | 29 +++ ...WarehouseTransferReceiptCommandMapper.java | 24 ++ .../WarehouseTransferReceiptCoordinator.java | 208 +++++++++++++++ .../service/WarehouseTransferService.java | 21 +- .../openapi/overseas-business-v1.yaml | 34 +++ .../HttpWarehouseResourceGatewayTest.java | 47 ++++ ...useTransferReceiptOpenApiContractTest.java | 36 +++ ...rehouseTransferReceiptCoordinatorTest.java | 245 ++++++++++++++++++ .../service/WarehouseTransferServiceTest.java | 13 + 10 files changed, 661 insertions(+), 9 deletions(-) create mode 100644 lastleg-service/src/main/java/com/baiyue/lastleg/transfer/entity/WarehouseTransferReceiptCommand.java create mode 100644 lastleg-service/src/main/java/com/baiyue/lastleg/transfer/mapper/WarehouseTransferReceiptCommandMapper.java create mode 100644 lastleg-service/src/main/java/com/baiyue/lastleg/transfer/service/WarehouseTransferReceiptCoordinator.java create mode 100644 lastleg-service/src/test/java/com/baiyue/lastleg/transfer/WarehouseTransferReceiptOpenApiContractTest.java create mode 100644 lastleg-service/src/test/java/com/baiyue/lastleg/transfer/service/WarehouseTransferReceiptCoordinatorTest.java 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 4983eda0d..2a56165fe 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 @@ -75,7 +75,15 @@ public class HttpWarehouseResourceGateway implements WarehouseResourceGateway { response.message() == null || response.message().isBlank() ? "Warehouse resource reservation was rejected" : response.message()); } - return response.data().lockId(); + ReserveData lock = response.data(); + if (!"200".equals(response.code()) || lock.lockId() <= 0 || !"ACTIVE".equals(lock.status()) + || !Long.valueOf(Long.parseLong(customerId)).equals(lock.customerId()) + || !Long.valueOf(warehouseId).equals(lock.warehouseId()) || !Long.valueOf(palletId).equals(lock.palletId()) + || !Long.valueOf(locationId).equals(lock.locationId()) || !java.util.Objects.equals(businessType, lock.businessType()) + || !java.util.Objects.equals(businessId, lock.businessId()) || !java.util.Objects.equals(reservationKey, lock.reservationKey())) { + throw new ResponseStatusException(HttpStatus.CONFLICT, "Warehouse reservation is not the requested ACTIVE lock"); + } + return lock.lockId(); } @Override @@ -340,7 +348,8 @@ public class HttpWarehouseResourceGateway implements WarehouseResourceGateway { String locationType, String locationStatus, boolean reservable, String reservationBlockReason) {} private record ReserveEnvelope(String code, String message, ReserveData data) {} - private record ReserveData(long lockId) {} + private record ReserveData(long lockId, Long customerId, Long warehouseId, Long palletId, Long locationId, + String businessType, String businessId, String reservationKey, String status) {} private record WarehouseReferenceResponse(String code, String message, WarehouseReference data) {} private record WarehouseLocationResponse(String code, String message, WarehouseLocationReference data) {} private record PalletBatchEnvelope(String code, String message, PalletBatchResponse data) {} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/transfer/entity/WarehouseTransferReceiptCommand.java b/lastleg-service/src/main/java/com/baiyue/lastleg/transfer/entity/WarehouseTransferReceiptCommand.java new file mode 100644 index 000000000..fc941b84e --- /dev/null +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/transfer/entity/WarehouseTransferReceiptCommand.java @@ -0,0 +1,29 @@ +package com.baiyue.lastleg.transfer.entity; + +import com.baomidou.mybatisplus.annotation.IdType; +import com.baomidou.mybatisplus.annotation.TableId; +import com.baomidou.mybatisplus.annotation.TableName; +import lombok.Data; +import java.time.LocalDateTime; + +@Data +@TableName("ll_warehouse_transfer_receipt_command") +public class WarehouseTransferReceiptCommand { + @TableId(type = IdType.INPUT) private Long id; + private String customerId; + private Long transferOrderId; + private Long sourceInventoryPositionId; + private String idempotencyKey; + private String requestHash; + private String requestJson; + private Long targetWarehouseId; + private Long targetPalletId; + private Long targetLocationId; + private String targetHash; + private String targetJson; + private String reservationKey; + private Long resourceLockId; + private String status; + private LocalDateTime createdAt; + private LocalDateTime updatedAt; +} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/transfer/mapper/WarehouseTransferReceiptCommandMapper.java b/lastleg-service/src/main/java/com/baiyue/lastleg/transfer/mapper/WarehouseTransferReceiptCommandMapper.java new file mode 100644 index 000000000..f0f40d2dd --- /dev/null +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/transfer/mapper/WarehouseTransferReceiptCommandMapper.java @@ -0,0 +1,24 @@ +package com.baiyue.lastleg.transfer.mapper; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import com.baiyue.lastleg.transfer.entity.WarehouseTransferReceiptCommand; +import org.apache.ibatis.annotations.Mapper; +import org.apache.ibatis.annotations.Param; +import org.apache.ibatis.annotations.Select; + +@Mapper +public interface WarehouseTransferReceiptCommandMapper extends BaseMapper { + @Select("SELECT * FROM ll_warehouse_transfer_receipt_command WHERE customer_id=#{customerId} " + + "AND transfer_order_id=#{transferId} AND source_inventory_position_id=#{sourceId} LIMIT 1 FOR UPDATE") + WarehouseTransferReceiptCommand lockSource(@Param("customerId") String customerId, + @Param("transferId") Long transferId, @Param("sourceId") Long sourceId); + + @Select("SELECT * FROM ll_warehouse_transfer_receipt_command WHERE customer_id=#{customerId} " + + "AND idempotency_key=#{key} LIMIT 1") + WarehouseTransferReceiptCommand findKey(@Param("customerId") String customerId, @Param("key") String key); + + @Select("SELECT * FROM ll_warehouse_transfer_receipt_command WHERE customer_id=#{customerId} " + + "AND transfer_order_id=#{transferId} AND target_pallet_id=#{palletId} ORDER BY id LIMIT 1") + WarehouseTransferReceiptCommand findTarget(@Param("customerId") String customerId, + @Param("transferId") Long transferId, @Param("palletId") Long palletId); +} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/transfer/service/WarehouseTransferReceiptCoordinator.java b/lastleg-service/src/main/java/com/baiyue/lastleg/transfer/service/WarehouseTransferReceiptCoordinator.java new file mode 100644 index 000000000..c31bda765 --- /dev/null +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/transfer/service/WarehouseTransferReceiptCoordinator.java @@ -0,0 +1,208 @@ +package com.baiyue.lastleg.transfer.service; + +import com.baomidou.mybatisplus.core.toolkit.IdWorker; +import com.baiyue.lastleg.client.DomesticWarehouseDirectoryClient; +import com.baiyue.lastleg.fleet.driver.FleetWarehouseClient; +import com.baiyue.lastleg.inboundplan.pda.gateway.WarehouseResourceGateway; +import com.baiyue.lastleg.transfer.dto.WarehouseTransferRequests; +import com.baiyue.lastleg.transfer.dto.WarehouseTransferResponses; +import com.baiyue.lastleg.transfer.entity.WarehouseTransferOrder; +import com.baiyue.lastleg.transfer.entity.WarehouseTransferReceiptCommand; +import com.baiyue.lastleg.transfer.mapper.WarehouseTransferInventoryAllocationMapper; +import com.baiyue.lastleg.transfer.mapper.WarehouseTransferOrderMapper; +import com.baiyue.lastleg.transfer.mapper.WarehouseTransferReceiptCommandMapper; +import com.baiyue.lastleg.transfer.mapper.WarehouseTransferReceiptMapper; +import com.fasterxml.jackson.databind.ObjectMapper; +import jakarta.annotation.Resource; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.http.HttpStatus; +import org.springframework.stereotype.Service; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.transaction.support.TransactionTemplate; +import org.springframework.web.server.ResponseStatusException; + +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.time.LocalDateTime; +import java.util.HexFormat; +import java.util.Objects; +import java.util.Set; +import java.util.function.Function; +import java.util.function.Supplier; + +/** Durable per-source receipt intents share one physical reservation per transfer target pallet. */ +@Service +public class WarehouseTransferReceiptCoordinator { + @Resource private WarehouseTransferOrderMapper transfers; + @Resource private WarehouseTransferInventoryAllocationMapper allocations; + @Resource private WarehouseTransferReceiptMapper receipts; + @Resource private WarehouseTransferReceiptCommandMapper commands; + @Resource private WarehouseResourceGateway warehouse; + @Resource private DomesticWarehouseDirectoryClient directory; + @Resource private FleetWarehouseClient catalog; + @Resource private PlatformTransactionManager transactionManager; + @Resource private ObjectMapper objectMapper; + @Value("${baiyue.security.internal-service-token:}") private String internalToken; + + public record Target(String warehouseId, String warehouseCode, String warehouseName, + Long palletId, String palletNo, Long locationId, String locationCode, + String fbaId, String fbaCode, String fbaName) { } + public record Context(Target target, long resourceLockId, long commandId) { } + private record Binding(String warehouseId, Long palletId, Long locationId, String fbaCode) { } + private record ResolvedTarget(Target target, String palletStatus) { } + private record Admission(WarehouseTransferOrder transfer, boolean replay) { } + + public WarehouseTransferResponses.Detail receive(String customerId, String transferNo, String operatorId, + WarehouseTransferRequests.Receipt request, Supplier replay, + Function stockWork) { + if (TransactionSynchronizationManager.isActualTransactionActive()) { + throw new IllegalStateException("Receipt resource coordination must start outside the stock transaction"); + } + String requestHash = hash(request); + Admission first = transaction().execute(status -> admit(customerId, transferNo, request, requestHash)); + if (first.replay()) return replay.get(); + ResolvedTarget resolved = resolveTarget(first.transfer(), request); + Target target = resolved.target(); + WarehouseTransferReceiptCommand command = transaction().execute(status -> { + Admission current = admit(customerId, transferNo, request, requestHash); + if (current.replay()) return null; + return prepare(customerId, current.transfer(), request, requestHash, target, resolved.palletStatus()); + }); + if (command == null) return replay.get(); + + // Idempotent remote ownership survives a later local rollback. Never release it blindly. + long lockId = warehouse.reserve(customerId, command.getTargetWarehouseId(), command.getTargetPalletId(), + command.getTargetLocationId(), "WAREHOUSE_TRANSFER_RECEIPT", command.getTargetHash(), command.getReservationKey()); + var activeWarehouse = warehouse.requireActiveSharedWarehouse(command.getTargetWarehouseId()); + require(activeWarehouse != null && activeWarehouse.warehouseId() == command.getTargetWarehouseId() + && "ACTIVE".equals(activeWarehouse.status()), "Receiving warehouse changed during reservation"); + var occupied = warehouse.resolvePallet(command.getTargetPalletId()); + require(lockId > 0 && occupied != null && occupied.palletId() == target.palletId() + && occupied.warehouseId() == command.getTargetWarehouseId() + && occupied.locationId() == target.locationId() && "OCCUPIED".equals(occupied.status()) + && target.palletNo().equals(occupied.palletNo()) && target.fbaCode().equals(occupied.fbaWarehouseCode()), + "Reserved target pallet no longer matches the receipt"); + transaction().executeWithoutResult(status -> { + WarehouseTransferReceiptCommand current = lockCommand(command); + require(current.getResourceLockId() == null || current.getResourceLockId() == lockId, + "Receipt resource lock changed"); + if (!"COMPLETED".equals(current.getStatus())) { + current.setResourceLockId(lockId); current.setStatus("RESERVED"); + current.setUpdatedAt(LocalDateTime.now()); commands.updateById(current); + } + }); + return transaction().execute(status -> { + Admission current = admit(customerId, transferNo, request, requestHash); + WarehouseTransferReceiptCommand owned = lockCommand(command); + if (current.replay()) return replay.get(); + require("RESERVED".equals(owned.getStatus()) && Objects.equals(owned.getResourceLockId(), lockId), + "Receipt has no active reserved resource"); + WarehouseTransferResponses.Detail result = stockWork.apply(new Context(target, lockId, owned.getId())); + owned.setStatus("COMPLETED"); owned.setUpdatedAt(LocalDateTime.now()); commands.updateById(owned); + return result; + }); + } + + private Admission admit(String customerId, String transferNo, WarehouseTransferRequests.Receipt request, + String requestHash) { + WarehouseTransferOrder transfer = transfers.lockNo(customerId, transferNo); + if (transfer == null) throw new ResponseStatusException(HttpStatus.NOT_FOUND, "Warehouse transfer not found"); + var received = receipts.findSource(customerId, transfer.getId(), request.sourceInventoryPositionId()); + if (received != null) { + require(requestHash.equals(received.getRequestHash()), "Receipt original request changed"); + return new Admission(transfer, true); + } + require(Set.of("OUTBOUNDED", "PARTIALLY_RECEIVED").contains(transfer.getStatus()), + "Only outbound transfer stock can be received"); + require(allocations.lockTransfer(customerId, transfer.getId()).stream().anyMatch(a -> + Objects.equals(a.getInventoryPositionId(), request.sourceInventoryPositionId()) && "SHIPPED".equals(a.getStatus())), + "Source position is not awaiting this transfer receipt"); + WarehouseTransferReceiptCommand existing = commands.lockSource(customerId, transfer.getId(), request.sourceInventoryPositionId()); + if (existing != null) require(existing.getRequestHash().equals(requestHash) + && existing.getIdempotencyKey().equals(request.idempotencyKey().trim()), "Receipt original command changed"); + return new Admission(transfer, false); + } + + private WarehouseTransferReceiptCommand prepare(String customerId, WarehouseTransferOrder transfer, + WarehouseTransferRequests.Receipt request, String requestHash, Target target, String palletStatus) { + require(Objects.equals(transfer.getFbaWarehouseCode(), target.fbaCode()), "Frozen transfer destination changed"); + String targetHash = hash(new Binding(target.warehouseId(), target.palletId(), target.locationId(), target.fbaCode())); + WarehouseTransferReceiptCommand shared = commands.findTarget(customerId, transfer.getId(), target.palletId()); + require(!"OCCUPIED".equals(palletStatus) || shared != null, "Occupied target pallet is not reserved by this transfer"); + if (shared != null) require(shared.getTargetHash().equals(targetHash), "Transfer target pallet binding changed"); + WarehouseTransferReceiptCommand existing = commands.lockSource(customerId, transfer.getId(), request.sourceInventoryPositionId()); + if (existing != null) { + require(existing.getTargetHash().equals(targetHash), "Receipt frozen target changed"); + return existing; + } + require(commands.findKey(customerId, request.idempotencyKey().trim()) == null, "Receipt key belongs to another source"); + var value = new WarehouseTransferReceiptCommand(); value.setId(IdWorker.getId()); value.setCustomerId(customerId); + value.setTransferOrderId(transfer.getId()); value.setSourceInventoryPositionId(request.sourceInventoryPositionId()); + value.setIdempotencyKey(request.idempotencyKey().trim()); value.setRequestHash(requestHash); value.setRequestJson(json(request)); + value.setTargetWarehouseId(Long.parseLong(target.warehouseId())); value.setTargetPalletId(target.palletId()); + value.setTargetLocationId(target.locationId()); value.setTargetHash(targetHash); value.setTargetJson(json(target)); + value.setReservationKey("WTR_RECEIPT_RES:" + transfer.getId() + ":" + target.palletId()); + value.setStatus("PREPARED"); value.setCreatedAt(LocalDateTime.now()); value.setUpdatedAt(value.getCreatedAt()); + commands.insert(value); return value; + } + + private WarehouseTransferReceiptCommand lockCommand(WarehouseTransferReceiptCommand expected) { + var value = commands.lockSource(expected.getCustomerId(), expected.getTransferOrderId(), expected.getSourceInventoryPositionId()); + require(value != null && Objects.equals(value.getId(), expected.getId()) + && value.getRequestHash().equals(expected.getRequestHash()) && value.getTargetHash().equals(expected.getTargetHash()), + "Receipt intent changed"); + return value; + } + + private ResolvedTarget resolveTarget(WarehouseTransferOrder transfer, WarehouseTransferRequests.Receipt request) { + long warehouseId = positive(request.destinationOverseasWarehouseId(), "Receiving warehouse ID is invalid"); + var master = warehouse.requireActiveSharedWarehouse(warehouseId); + require(master != null && master.warehouseId() == warehouseId && "ACTIVE".equals(master.status()), + "Receiving warehouse is not active shared overseas receiving"); + snapshot(request.destinationOverseasWarehouseCode(), master.warehouseCode(), "Receiving warehouse code changed"); + snapshot(request.destinationOverseasWarehouseName(), master.warehouseName(), "Receiving warehouse name changed"); + var pallet = warehouse.resolvePallet(request.destinationPalletId()); + require(pallet != null && pallet.palletId() == request.destinationPalletId() && pallet.warehouseId() == warehouseId + && Set.of("IDLE", "OCCUPIED").contains(pallet.status()) && request.destinationPalletNo().trim().equals(pallet.palletNo()), + "Target pallet does not belong to receiving warehouse"); + long locationId = request.warehouseLocationId() == null ? pallet.locationId() : request.warehouseLocationId(); + require(locationId > 0 && pallet.locationId() == locationId, "Target pallet is not bound to requested location"); + var location = warehouse.resolveLocation(warehouseId, locationId); + require(location != null && location.locationId() == locationId && Set.of("ACTIVE", "IDLE", "OCCUPIED").contains(location.status()), + "Target location does not belong to receiving warehouse"); + snapshot(request.warehouseLocationCode(), location.locationCode(), "Target location code changed"); + String frozenFba = transfer.getFbaWarehouseCode(); + require(frozenFba != null && !frozenFba.isBlank() && frozenFba.equals(request.destinationWarehouseCode().trim()) + && frozenFba.equals(pallet.fbaWarehouseCode()), "Target pallet FBA differs from frozen transfer destination"); + var activeCatalog = catalog.activeWarehouses(internalToken); + require(activeCatalog != null && activeCatalog.successful() && activeCatalog.data() != null, + "Authoritative FBA warehouse catalog is unavailable"); + var candidates = activeCatalog.data().stream().filter(w -> "FBA".equals(w.warehouseType()) + && frozenFba.equals(w.warehouseCode()) && w.id() != null && w.id() > 0).toList(); + require(candidates.size() == 1, "Frozen FBA warehouse is absent or ambiguous"); + long fbaId = candidates.get(0).id(); + snapshot(transfer.getDestinationWarehouseId(), Long.toString(fbaId), "Frozen FBA ID is not authoritative"); + snapshot(request.destinationWarehouseId(), Long.toString(fbaId), "Receipt FBA ID is not authoritative"); + var response = directory.activeWarehouse(fbaId, internalToken); + var fba = response == null || response.data() == null ? null : response.data().warehouse(); + require(response != null && response.successful() && fba != null && Long.valueOf(fbaId).equals(fba.id()) + && Long.valueOf(0).equals(fba.tenantId()) && "ACTIVE".equals(fba.status()) && "FBA".equals(fba.warehouseType()) + && frozenFba.equals(fba.warehouseCode()), "FBA warehouse does not match frozen destination"); + snapshot(request.destinationWarehouseName(), fba.warehouseName(), "FBA warehouse name changed"); + return new ResolvedTarget(new Target(Long.toString(warehouseId), master.warehouseCode(), master.warehouseName(), pallet.palletId(), + pallet.palletNo(), locationId, location.locationCode(), Long.toString(fbaId), frozenFba, fba.warehouseName()), pallet.status()); + } + + private TransactionTemplate transaction() { + TransactionTemplate template = new TransactionTemplate(transactionManager); + template.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); return template; + } + private String json(Object value) { try { return objectMapper.writeValueAsString(value); } catch (Exception e) { throw new IllegalStateException("Receipt snapshot serialization failed", e); } } + private String hash(Object value) { try { return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(json(value).getBytes(StandardCharsets.UTF_8))); } catch (Exception e) { throw new IllegalStateException("Receipt hash failed", e); } } + private static long positive(String value, String message) { try { long id = Long.parseLong(value.trim()); require(id > 0, message); return id; } catch (NumberFormatException | NullPointerException e) { throw new ResponseStatusException(HttpStatus.CONFLICT, message); } } + private static String text(String value) { return value == null || value.isBlank() ? null : value.trim(); } + private static void snapshot(String supplied, String expected, String message) { if (text(supplied) != null) require(Objects.equals(text(supplied), expected), message); } + private static void require(boolean value, String message) { if (!value) throw new ResponseStatusException(HttpStatus.CONFLICT, message); } +} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/transfer/service/WarehouseTransferService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/transfer/service/WarehouseTransferService.java index eb3f5a7e8..fef0d4215 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/transfer/service/WarehouseTransferService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/transfer/service/WarehouseTransferService.java @@ -126,6 +126,7 @@ public class WarehouseTransferService { @Resource private RelabelApplicationMapper relabelApplications; @Resource private RelabelApplicationService relabelService; @Resource private WarehouseResourceGateway warehouseResources; + @Resource private WarehouseTransferReceiptCoordinator receiptCoordinator; @Resource private InboundPlanMapper inboundPlans; @Resource private InboundPlanLineMapper inboundPlanLines; @Resource private InboundPlanPalletMapper inboundPlanPallets; @@ -1214,9 +1215,15 @@ public class WarehouseTransferService { return detail(customerId, transferNo); } - @Transactional public WarehouseTransferResponses.Detail receive(String customerId, String transferNo, String operatorId, WarehouseTransferRequests.Receipt request) { + return receiptCoordinator.receive(customerId, transferNo, operatorId, request, + () -> detail(customerId, transferNo), + context -> receiveReserved(customerId, transferNo, operatorId, request, context)); + } + + private WarehouseTransferResponses.Detail receiveReserved(String customerId, String transferNo, String operatorId, + WarehouseTransferRequests.Receipt request, WarehouseTransferReceiptCoordinator.Context context) { WarehouseTransferOrder transfer = transfers.lockNo(customerId, transferNo); if (transfer == null) throw notFound("Warehouse transfer order not found"); if (!Set.of(OUTBOUNDED, PARTIALLY_RECEIVED).contains(transfer.getStatus())) { @@ -1247,17 +1254,17 @@ public class WarehouseTransferService { if (sourceBalance == null || !customerId.equals(sourceBalance.getCustomerId())) { throw conflict("Source inventory balance is missing for warehouse transfer receipt"); } + WarehouseTransferReceiptCoordinator.Target target = context.target(); InventoryCommandResponse result = inventory.stockIn(customerId, operatorId, new InventoryCommandRequests.StockIn(customerId, sourceBalance.getCustomerName(), - request.destinationOverseasWarehouseId().trim(), trim(request.destinationOverseasWarehouseCode()), - trim(request.destinationOverseasWarehouseName()), trim(request.destinationWarehouseId()), - request.destinationWarehouseCode().trim(), trim(request.destinationWarehouseName()), + target.warehouseId(), target.warehouseCode(), target.warehouseName(), target.fbaId(), + target.fbaCode(), target.fbaName(), transfer.getLastlegOrderId(), transfer.getLastlegOrderNo(), transferNo, Boolean.TRUE.equals(transfer.getNeedRelabel()) ? allocation.getSkuCode() : sourceBalance.getSkuCode(), sourceBalance.getProductNameZh(), sourceBalance.getProductNameEn(), - request.destinationPalletId(), request.destinationPalletNo().trim(), - trim(request.batchNo()) == null ? transferNo : request.batchNo().trim(), request.warehouseLocationId(), - trim(request.warehouseLocationCode()), allocation.getPackageCount(), transferNo, + target.palletId(), target.palletNo(), + trim(request.batchNo()) == null ? transferNo : request.batchNo().trim(), target.locationId(), + target.locationCode(), allocation.getPackageCount(), transferNo, "WAREHOUSE_TRANSFER_RECEIPT:" + transferNo, operationKey, trim(request.remark()))); LocalDateTime now = LocalDateTime.now(); WarehouseTransferReceipt receipt = new WarehouseTransferReceipt(); 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 8d6264a1a..62b0bcf3f 100644 --- a/lastleg-service/src/main/resources/openapi/overseas-business-v1.yaml +++ b/lastleg-service/src/main/resources/openapi/overseas-business-v1.yaml @@ -1075,6 +1075,40 @@ paths: responses: '200': { description: 原转运详情；allocations内skuCode为锁定的实际发货SKU，换标后返回新SKU；不包括已被换标替换的历史分配 } '409': { description: 来源不可用或幂等载荷冲突 } + /api/platform/v1/last-leg/warehouse-transfers/{transferNo}/receipts: + post: + summary: 已有库存转运逐来源在目标仓实收 + description: 权限lastleg:warehouse-transfer:manage；必须指定customerId。只接收本票SHIPPED来源，服务端读取实际OVERSEAS/ACTIVE/RECEIVING仓、托盘及绑定库位、权威FBA主档并核对冻结目的仓。先持久化完整原请求，再占用资源；同转运同目标托盘的多来源共享一次ACTIVE资源锁。远端成功但本地事务失败时保留原命令，原键原载荷重试继续；新键或变更载荷不能绕过。库存、收货记录和最终换标费用结算命令同本地事务提交，成功重放不重复入账或占用，未知结果须先核对原命令。 + security: [{ bearerAuth: [] }] + parameters: + - { name: transferNo, in: path, required: true, schema: { type: string } } + - { name: customerId, in: query, required: true, schema: { type: string } } + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [sourceInventoryPositionId, destinationOverseasWarehouseId, destinationWarehouseCode, destinationPalletId, destinationPalletNo, idempotencyKey] + properties: + sourceInventoryPositionId: { type: integer, format: int64 } + destinationOverseasWarehouseId: { type: string, maxLength: 128 } + destinationOverseasWarehouseCode: { type: string, nullable: true, maxLength: 128 } + destinationOverseasWarehouseName: { type: string, nullable: true, maxLength: 255 } + destinationWarehouseId: { type: string, nullable: true, maxLength: 128 } + destinationWarehouseCode: { type: string, maxLength: 128 } + destinationWarehouseName: { type: string, nullable: true, maxLength: 255 } + destinationPalletId: { type: integer, format: int64 } + destinationPalletNo: { type: string, maxLength: 128 } + batchNo: { type: string, nullable: true, maxLength: 128 } + warehouseLocationId: { type: integer, format: int64, nullable: true, description: 未传时使用该实物托盘绑定库位 } + warehouseLocationCode: { type: string, nullable: true, maxLength: 128 } + idempotencyKey: { type: string, maxLength: 255 } + remark: { type: string, nullable: true, maxLength: 1000 } + responses: + '200': { description: 原转运详情，部分来源PARTIALLY_RECEIVED、全部来源RECEIVED；原键原载荷返回原收货结果 } + '409': { description: 原请求冲突、来源状态或目标资源归属不匹配；不得换键绕过 } + '503': { description: 资源服务不可用，保留原收货命令待核对恢复 } /api/platform/v1/last-leg/warehouse-transfers/{transferNo}/outbound: post: summary: 已有库存转运实际出库 diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/pda/gateway/HttpWarehouseResourceGatewayTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/pda/gateway/HttpWarehouseResourceGatewayTest.java index 354e31c66..6b3c47e9a 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/pda/gateway/HttpWarehouseResourceGatewayTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/pda/gateway/HttpWarehouseResourceGatewayTest.java @@ -21,6 +21,53 @@ class HttpWarehouseResourceGatewayTest { private HttpWarehouseResourceGateway gateway; private MockRestServiceServer server; + @Test + void reserveAcceptsOnlyCompleteMatchingActiveLockWithStringEncodedLongs() { + server.expect(requestTo("http://warehouse/internal/v1/warehouse-resource-locks")) + .andRespond(withSuccess(""" + {"code":"200","data":{"lockId":"51","customerId":"1","warehouseId":"2", + "palletId":"3","locationId":"4","businessType":"WAREHOUSE_TRANSFER_RECEIPT", + "businessId":"target-hash","reservationKey":"resource-1","status":"ACTIVE"}} + """, MediaType.APPLICATION_JSON)); + assertThat(gateway.reserve("1", 2, 3, 4, "WAREHOUSE_TRANSFER_RECEIPT", "target-hash", "resource-1")).isEqualTo(51); + server.verify(); + } + + @Test + void reserveRejectsReleasedHistoricalLockRatherThanTreatingOldKeyAsNewOccupancy() { + server.expect(requestTo("http://warehouse/internal/v1/warehouse-resource-locks")) + .andRespond(withSuccess(""" + {"code":"200","data":{"lockId":51,"customerId":1,"warehouseId":2,"palletId":3, + "locationId":4,"businessType":"WAREHOUSE_TRANSFER_RECEIPT","businessId":"target-hash", + "reservationKey":"resource-1","status":"RELEASED"}} + """, MediaType.APPLICATION_JSON)); + assertThatThrownBy(() -> gateway.reserve("1", 2, 3, 4, "WAREHOUSE_TRANSFER_RECEIPT", "target-hash", "resource-1")) + .isInstanceOf(ResponseStatusException.class).hasMessageContaining("requested ACTIVE lock"); + server.verify(); + } + + @Test + void reserveRejectsWrongOwnershipEvenWhenReturnedLockIsActive() { + server.expect(requestTo("http://warehouse/internal/v1/warehouse-resource-locks")) + .andRespond(withSuccess(""" + {"code":"200","data":{"lockId":51,"customerId":99,"warehouseId":2,"palletId":3, + "locationId":4,"businessType":"WAREHOUSE_TRANSFER_RECEIPT","businessId":"target-hash", + "reservationKey":"resource-1","status":"ACTIVE"}} + """, MediaType.APPLICATION_JSON)); + assertThatThrownBy(() -> gateway.reserve("1", 2, 3, 4, "WAREHOUSE_TRANSFER_RECEIPT", "target-hash", "resource-1")) + .isInstanceOf(ResponseStatusException.class).hasMessageContaining("requested ACTIVE lock"); + server.verify(); + } + + @Test + void reserveRejectsMissingActiveStateInsteadOfAcceptingBareLockId() { + server.expect(requestTo("http://warehouse/internal/v1/warehouse-resource-locks")) + .andRespond(withSuccess("{\"code\":\"200\",\"data\":{\"lockId\":51}}", MediaType.APPLICATION_JSON)); + assertThatThrownBy(() -> gateway.reserve("1", 2, 3, 4, "WAREHOUSE_TRANSFER_RECEIPT", "target-hash", "resource-1")) + .isInstanceOf(ResponseStatusException.class).hasMessageContaining("requested ACTIVE lock"); + server.verify(); + } + @BeforeEach void setUp() { RestClient.Builder builder = RestClient.builder(); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/transfer/WarehouseTransferReceiptOpenApiContractTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/transfer/WarehouseTransferReceiptOpenApiContractTest.java new file mode 100644 index 000000000..a37b6e306 --- /dev/null +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/transfer/WarehouseTransferReceiptOpenApiContractTest.java @@ -0,0 +1,36 @@ +package com.baiyue.lastleg.transfer; + +import com.baiyue.lastleg.transfer.dto.WarehouseTransferRequests; +import org.junit.jupiter.api.Test; +import org.yaml.snakeyaml.Yaml; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; +import static org.assertj.core.api.Assertions.assertThat; + +class WarehouseTransferReceiptOpenApiContractTest { + @Test + @SuppressWarnings("unchecked") + void actualReceiptEndpointDocumentsExactDtoAndDurableResourceBoundary() { + Map document = new Yaml().load(getClass().getResourceAsStream("/openapi/overseas-business-v1.yaml")); + var paths=(Map)document.get("paths"); + var path=(Map)paths.get("/api/platform/v1/last-leg/warehouse-transfers/{transferNo}/receipts"); + assertThat(path).isNotNull(); + var post=(Map)path.get("post"); + assertThat((String)post.get("description")).contains("lastleg:warehouse-transfer:manage", "原键原载荷", "共享一次ACTIVE资源锁"); + var body=(Map)post.get("requestBody"); + var content=(Map)body.get("content"); + var media=(Map)content.get("application/json"); + var schema=(Map)media.get("schema"); + var properties=(Map)schema.get("properties"); + Set fields=Arrays.stream(WarehouseTransferRequests.Receipt.class.getRecordComponents()) + .map(java.lang.reflect.RecordComponent::getName).collect(Collectors.toSet()); + assertThat(properties.keySet()).isEqualTo(fields); + assertThat((List)schema.get("required")).containsExactlyInAnyOrder( + "sourceInventoryPositionId","destinationOverseasWarehouseId","destinationWarehouseCode", + "destinationPalletId","destinationPalletNo","idempotencyKey"); + assertThat((Map)post.get("responses")).containsKeys("200","409","503"); + } +} diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/transfer/service/WarehouseTransferReceiptCoordinatorTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/transfer/service/WarehouseTransferReceiptCoordinatorTest.java new file mode 100644 index 000000000..8030daa98 --- /dev/null +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/transfer/service/WarehouseTransferReceiptCoordinatorTest.java @@ -0,0 +1,245 @@ +package com.baiyue.lastleg.transfer.service; + +import com.baomidou.mybatisplus.annotation.TableName; +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.extension.spring.MybatisSqlSessionFactoryBean; +import com.baiyue.common.core.api.ApiResponse; +import com.baiyue.lastleg.client.DomesticWarehouseDirectoryClient; +import com.baiyue.lastleg.fleet.driver.FleetWarehouseClient; +import com.baiyue.lastleg.inboundplan.pda.gateway.WarehouseResourceGateway; +import com.baiyue.lastleg.transfer.dto.WarehouseTransferRequests; +import com.baiyue.lastleg.transfer.entity.*; +import com.baiyue.lastleg.transfer.mapper.*; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mybatis.spring.SqlSessionTemplate; +import org.springframework.http.HttpStatus; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.jdbc.datasource.DriverManagerDataSource; +import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.web.server.ResponseStatusException; + +import java.lang.reflect.Modifier; +import java.math.BigDecimal; +import java.time.LocalDateTime; +import java.util.*; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +class WarehouseTransferReceiptCoordinatorTest { + JdbcTemplate jdbc; SqlSessionTemplate session; WarehouseTransferReceiptCoordinator coordinator; + WarehouseResourceGateway warehouse; DomesticWarehouseDirectoryClient directory; FleetWarehouseClient catalog; + ObjectMapper json = new ObjectMapper().findAndRegisterModules(); + AtomicInteger occupancies = new AtomicInteger(), reserveCalls = new AtomicInteger(), stockCalls = new AtomicInteger(); + AtomicBoolean loseReserveResponse = new AtomicBoolean(), failLocal = new AtomicBoolean(); + Map remote = new HashMap<>(); + record RemoteLock(String customer, String key, String hash, long warehouse, long location, long lockId) { } + + @BeforeEach void setup() throws Exception { + var ds = new DriverManagerDataSource("jdbc:h2:mem:" + UUID.randomUUID() + ";MODE=MYSQL;DB_CLOSE_DELAY=-1", "sa", ""); + jdbc = new JdbcTemplate(ds); + for (Class type : List.of(WarehouseTransferOrder.class, WarehouseTransferInventoryAllocation.class, + WarehouseTransferReceipt.class, WarehouseTransferReceiptCommand.class)) { + List fields = new ArrayList<>(); + for (var f : type.getDeclaredFields()) { + if (Modifier.isStatic(f.getModifiers())) continue; + String name = f.getName().replaceAll("([a-z])([A-Z])", "$1_$2").toLowerCase(Locale.ROOT); + String sqlType = f.getType() == String.class ? "VARCHAR(16000)" : f.getType() == LocalDateTime.class + ? "TIMESTAMP" : f.getType() == BigDecimal.class ? "DECIMAL(24,6)" : f.getType() == Boolean.class ? "BOOLEAN" : "BIGINT"; + fields.add(name + " " + sqlType + (name.equals("id") ? " PRIMARY KEY" : "")); + } + jdbc.execute("CREATE TABLE " + type.getAnnotation(TableName.class).value() + " (" + String.join(",", fields) + ")"); + } + jdbc.execute("CREATE UNIQUE INDEX receipt_command_source ON ll_warehouse_transfer_receipt_command(transfer_order_id,source_inventory_position_id)"); + jdbc.execute("CREATE UNIQUE INDEX receipt_command_key ON ll_warehouse_transfer_receipt_command(customer_id,idempotency_key)"); + jdbc.execute("CREATE UNIQUE INDEX receipt_source ON ll_warehouse_transfer_receipt(transfer_order_id,source_inventory_position_id)"); + jdbc.execute("CREATE TABLE local_effect(type VARCHAR(30), source_id BIGINT)"); + var configuration = new MybatisConfiguration(); configuration.setMapUnderscoreToCamelCase(true); + for (Class mapper : List.of(WarehouseTransferOrderMapper.class, WarehouseTransferInventoryAllocationMapper.class, + WarehouseTransferReceiptMapper.class, WarehouseTransferReceiptCommandMapper.class)) configuration.addMapper(mapper); + var factory = new MybatisSqlSessionFactoryBean(); factory.setDataSource(ds); factory.setConfiguration(configuration); + session = new SqlSessionTemplate(factory.getObject()); + coordinator = new WarehouseTransferReceiptCoordinator(); + ReflectionTestUtils.setField(coordinator, "transfers", session.getMapper(WarehouseTransferOrderMapper.class)); + ReflectionTestUtils.setField(coordinator, "allocations", session.getMapper(WarehouseTransferInventoryAllocationMapper.class)); + ReflectionTestUtils.setField(coordinator, "receipts", session.getMapper(WarehouseTransferReceiptMapper.class)); + ReflectionTestUtils.setField(coordinator, "commands", session.getMapper(WarehouseTransferReceiptCommandMapper.class)); + ReflectionTestUtils.setField(coordinator, "transactionManager", new DataSourceTransactionManager(ds)); + ReflectionTestUtils.setField(coordinator, "objectMapper", json); + warehouse = mock(WarehouseResourceGateway.class); directory = mock(DomesticWarehouseDirectoryClient.class); catalog = mock(FleetWarehouseClient.class); + ReflectionTestUtils.setField(coordinator, "warehouse", warehouse); ReflectionTestUtils.setField(coordinator, "directory", directory); + ReflectionTestUtils.setField(coordinator, "catalog", catalog); ReflectionTestUtils.setField(coordinator, "internalToken", "test-token"); + when(warehouse.requireActiveSharedWarehouse(9L)).thenReturn(new WarehouseResourceGateway.WarehouseReference(9, "NY", "NY warehouse", "ACTIVE")); + when(warehouse.resolvePallet(anyLong())).thenAnswer(call -> pallet(call.getArgument(0), remote.containsKey(call.getArgument(0)) ? "OCCUPIED" : "IDLE")); + when(warehouse.resolveLocation(eq(9L), anyLong())).thenAnswer(call -> new WarehouseResourceGateway.WarehouseLocationReference( + call.getArgument(1), "LOC-" + call.getArgument(1), "STORAGE", new BigDecimal("1000"), "ACTIVE")); + when(catalog.activeWarehouses("test-token")).thenReturn(ApiResponse.success(List.of( + new FleetWarehouseClient.Warehouse(30L, "ABE4", "ABE4 warehouse", "FBA", "US")))); + when(directory.activeWarehouse(30L, "test-token")).thenReturn(ApiResponse.success(new DomesticWarehouseDirectoryClient.Detail( + new DomesticWarehouseDirectoryClient.Warehouse(30L, 0L, "ABE4", "ABE4 warehouse", "FBA", "US", "ACTIVE"), List.of()))); + when(warehouse.reserve(anyString(), anyLong(), anyLong(), anyLong(), anyString(), anyString(), anyString())).thenAnswer(call -> { + assertFalse(TransactionSynchronizationManager.isActualTransactionActive(), "Remote reserve must not hold local transaction locks"); + reserveCalls.incrementAndGet(); long pallet = call.getArgument(2), wh = call.getArgument(1), location = call.getArgument(3); + String customer = call.getArgument(0), key = call.getArgument(6), hash = call.getArgument(5); + RemoteLock existing = remote.get(pallet); + if (existing != null && (!existing.customer.equals(customer) || !existing.key.equals(key) || !existing.hash.equals(hash) + || existing.warehouse != wh || existing.location != location)) throw conflict("Other reservation owns pallet"); + if (existing == null) { existing = new RemoteLock(customer, key, hash, wh, location, 500L + pallet); remote.put(pallet, existing); occupancies.incrementAndGet(); } + if (loseReserveResponse.getAndSet(false)) throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE, "response lost after reserve commit"); + return existing.lockId; + }); + addTransfer(1, "WT-1", 101); + } + + private WarehouseResourceGateway.WarehousePalletReference pallet(long id, String state) { + return new WarehouseResourceGateway.WarehousePalletReference(id, "PAL-" + id, 9, id + 20, state, "ABE4", remote.containsKey(id) ? 1 : 0); + } + private void addTransfer(long id, String no, long sourceId) { + var head = new WarehouseTransferOrder(); head.setId(id); head.setCustomerId("7"); head.setTransferNo(no); + head.setStatus("OUTBOUNDED"); head.setFbaWarehouseCode("ABE4"); head.setDestinationWarehouseId("30"); head.setDestinationWarehouseName("ABE4 warehouse"); + session.getMapper(WarehouseTransferOrderMapper.class).insert(head); addSource(id, sourceId); + } + private void addSource(long transfer, long source) { + var allocation = new WarehouseTransferInventoryAllocation(); allocation.setId(source + 1000); allocation.setCustomerId("7"); + allocation.setTransferOrderId(transfer); allocation.setInventoryPositionId(source); allocation.setPackageCount(1); allocation.setStatus("SHIPPED"); + session.getMapper(WarehouseTransferInventoryAllocationMapper.class).insert(allocation); + } + private WarehouseTransferRequests.Receipt request(long source, long pallet, String key) { + return new WarehouseTransferRequests.Receipt(source, "9", "NY", "NY warehouse", "30", "ABE4", "ABE4 warehouse", + pallet, "PAL-" + pallet, "BATCH", pallet + 20, "LOC-" + (pallet + 20), key, "TEST"); + } + private void receive(String no, WarehouseTransferRequests.Receipt request) { + coordinator.receive("7", no, "operator", request, () -> null, context -> { + assertTrue(TransactionSynchronizationManager.isActualTransactionActive()); stockCalls.incrementAndGet(); + assertEquals(30L, Long.parseLong(context.target().fbaId())); assertTrue(context.resourceLockId() > 0); + jdbc.update("INSERT INTO local_effect(type,source_id) VALUES('STOCK_IN',?)", request.sourceInventoryPositionId()); + jdbc.update("UPDATE ll_warehouse_transfer_inventory_allocation SET status='RECEIVED' WHERE customer_id='7' AND inventory_position_id=?", request.sourceInventoryPositionId()); + var head = session.getMapper(WarehouseTransferOrderMapper.class).findNo("7", no); + var receipt = new WarehouseTransferReceipt(); receipt.setId(request.sourceInventoryPositionId() + 2000); + receipt.setCustomerId("7"); receipt.setTransferOrderId(head.getId()); receipt.setSourceInventoryPositionId(request.sourceInventoryPositionId()); + receipt.setRequestHash(hash(request)); receipt.setIdempotencyKey(request.idempotencyKey()); receipt.setPackageCount(1); + session.getMapper(WarehouseTransferReceiptMapper.class).insert(receipt); + long left = jdbc.queryForObject("SELECT COUNT(*) FROM ll_warehouse_transfer_inventory_allocation WHERE transfer_order_id=? AND status='SHIPPED'", Long.class, head.getId()); + jdbc.update("UPDATE ll_warehouse_transfer_order SET status=? WHERE id=?", left == 0 ? "RECEIVED" : "PARTIALLY_RECEIVED", head.getId()); + if (left == 0) { jdbc.update("INSERT INTO local_effect(type,source_id) VALUES('FEE_APPROVED',?)", request.sourceInventoryPositionId()); jdbc.update("INSERT INTO local_effect(type,source_id) VALUES('SETTLE',?)", request.sourceInventoryPositionId()); } + if (failLocal.getAndSet(false)) throw new IllegalStateException("local stock/fee transaction failed"); + return null; + }); + } + private String hash(Object value) { try { return HexFormat.of().formatHex(java.security.MessageDigest.getInstance("SHA-256").digest(json.writeValueAsBytes(value))); } catch (Exception e) { throw new AssertionError(e); } } + private long count(String table) { return jdbc.queryForObject("SELECT COUNT(*) FROM " + table, Long.class); } + private String state(long source) { return jdbc.queryForObject("SELECT status FROM ll_warehouse_transfer_receipt_command WHERE source_inventory_position_id=?", String.class, source); } + private static ResponseStatusException conflict(String message) { return new ResponseStatusException(HttpStatus.CONFLICT, message); } + + @Test void originalReceiptReplayDoesNotReserveOrStockAgainEvenAfterPhysicalResourceChanges() { + var request = request(101, 20, "KEY-1"); receive("WT-1", request); remote.clear(); receive("WT-1", request); + assertEquals(1, reserveCalls.get()); assertEquals(1, occupancies.get()); assertEquals(1, stockCalls.get()); + assertEquals(1, count("ll_warehouse_transfer_receipt")); assertEquals("COMPLETED", state(101)); + } + @Test void multipleSourcesShareOnePhysicalReservationAndCompleteDistinctReceipts() { + addSource(1, 102); receive("WT-1", request(101, 20, "KEY-1")); receive("WT-1", request(102, 20, "KEY-2")); + assertEquals(1, occupancies.get()); assertEquals(2, stockCalls.get()); assertEquals(2, count("ll_warehouse_transfer_receipt")); + assertEquals(1L, jdbc.queryForObject("SELECT COUNT(DISTINCT resource_lock_id) FROM ll_warehouse_transfer_receipt_command", Long.class)); + assertEquals(1L, jdbc.queryForObject("SELECT COUNT(DISTINCT target_hash) FROM ll_warehouse_transfer_receipt_command", Long.class)); + assertEquals(2L, jdbc.queryForObject("SELECT COUNT(DISTINCT request_hash) FROM ll_warehouse_transfer_receipt_command", Long.class)); + assertEquals(1L, jdbc.queryForObject("SELECT COUNT(*) FROM local_effect WHERE type='SETTLE'", Long.class)); + } + @Test void remoteCommittedButLocalStockAndFeeFailureRollsBackThenSameRequestRecovers() { + var request = request(101, 20, "KEY-1"); failLocal.set(true); + assertThrows(IllegalStateException.class, () -> receive("WT-1", request)); + assertEquals("RESERVED", state(101)); assertEquals(0, count("local_effect")); assertEquals(0, count("ll_warehouse_transfer_receipt")); + assertEquals("SHIPPED", jdbc.queryForObject("SELECT status FROM ll_warehouse_transfer_inventory_allocation WHERE inventory_position_id=101", String.class)); + receive("WT-1", request); assertEquals(1, occupancies.get()); assertEquals("COMPLETED", state(101)); + assertEquals(1L, jdbc.queryForObject("SELECT COUNT(*) FROM local_effect WHERE type='STOCK_IN'", Long.class)); + verify(warehouse, never()).release(anyString(), anyLong()); + } + @Test void responseLostAfterRemoteCommitLeavesPreparedIntentRecoverableWithoutNewOccupancy() { + loseReserveResponse.set(true); var request = request(101, 20, "KEY-1"); + assertThrows(ResponseStatusException.class, () -> receive("WT-1", request)); + assertEquals("PREPARED", state(101)); assertEquals(0, stockCalls.get()); assertEquals(1, occupancies.get()); + receive("WT-1", request); assertEquals("COMPLETED", state(101)); assertEquals(1, occupancies.get()); + } + @Test void failedFirstSourceDoesNotPreventSecondSourceThenOriginalRecoveryOnSamePallet() { + addSource(1, 102); failLocal.set(true); var first = request(101, 20, "KEY-1"); + assertThrows(IllegalStateException.class, () -> receive("WT-1", first)); + receive("WT-1", request(102, 20, "KEY-2")); receive("WT-1", first); + assertEquals(1, occupancies.get()); assertEquals(2, count("ll_warehouse_transfer_receipt")); + assertEquals(1L, jdbc.queryForObject("SELECT COUNT(*) FROM local_effect WHERE type='SETTLE'", Long.class)); + } + @Test void changedOriginalKeyPayloadOrTargetCannotBypassPreparedIntent() { + failLocal.set(true); assertThrows(IllegalStateException.class, () -> receive("WT-1", request(101, 20, "KEY-1"))); + assertThrows(ResponseStatusException.class, () -> receive("WT-1", request(101, 20, "NEW-KEY"))); + assertThrows(ResponseStatusException.class, () -> receive("WT-1", request(101, 21, "KEY-1"))); + var original = request(101, 20, "KEY-1"); var changed = new WarehouseTransferRequests.Receipt(101L, "9", "NY", "NY warehouse", "30", "ABE4", "ABE4 warehouse", 20L, "PAL-20", "CHANGED-BATCH", 40L, "LOC-40", "KEY-1", "TEST"); + assertThrows(ResponseStatusException.class, () -> receive("WT-1", changed)); + assertEquals(1, reserveCalls.get()); assertEquals(1, count("ll_warehouse_transfer_receipt_command")); assertEquals(0, count("ll_warehouse_transfer_receipt")); + } + @Test void sameKeyCannotBeReusedForAnotherSource() { + addSource(1, 102); receive("WT-1", request(101, 20, "KEY-1")); + assertThrows(ResponseStatusException.class, () -> receive("WT-1", request(102, 20, "KEY-1"))); + assertEquals(1, occupancies.get()); assertEquals(1, count("ll_warehouse_transfer_receipt_command")); + } + @Test void occupiedPalletOfAnotherTransferIsNotReused() { + receive("WT-1", request(101, 20, "KEY-1")); addTransfer(2, "WT-2", 201); + assertThrows(ResponseStatusException.class, () -> receive("WT-2", request(201, 20, "KEY-3"))); + assertEquals(1, occupancies.get()); assertEquals(1, count("ll_warehouse_transfer_receipt")); + assertEquals(1, count("ll_warehouse_transfer_receipt_command")); + receive("WT-2", request(201, 21, "KEY-3")); + assertEquals(2, occupancies.get()); assertEquals(2, count("ll_warehouse_transfer_receipt")); + } + @Test void twoRealResourcesInSameWarehouseStillRequireExactPalletLocationAssociation() { + var wrong = new WarehouseTransferRequests.Receipt(101L, "9", "NY", "NY warehouse", "30", "ABE4", "ABE4 warehouse", 20L, "PAL-20", "BATCH", 41L, "LOC-41", "KEY-1", "TEST"); + assertThrows(ResponseStatusException.class, () -> receive("WT-1", wrong)); assertEquals(0, reserveCalls.get()); assertEquals(0, count("ll_warehouse_transfer_receipt_command")); + } + @Test void omittedFbaIdsStillResolveAuthoritativeCatalogAndActivePublicMaster() { + jdbc.update("UPDATE ll_warehouse_transfer_order SET destination_warehouse_id=NULL WHERE id=1"); + var request = new WarehouseTransferRequests.Receipt(101L, "9", null, null, null, "ABE4", null, 20L, "PAL-20", "BATCH", null, null, "KEY-1", "TEST"); + receive("WT-1", request); verify(catalog).activeWarehouses("test-token"); verify(directory).activeWarehouse(30L, "test-token"); + assertEquals(1, occupancies.get()); + } + @Test void omittedIdsDoNotAllowStringOnlyFbaWhenAuthorityHasNoMatchingWarehouse() { + jdbc.update("UPDATE ll_warehouse_transfer_order SET destination_warehouse_id=NULL WHERE id=1"); + when(catalog.activeWarehouses("test-token")).thenReturn(ApiResponse.success(List.of())); + var request = new WarehouseTransferRequests.Receipt(101L, "9", null, null, null, "ABE4", null, 20L, "PAL-20", "BATCH", null, null, "KEY-1", "TEST"); + assertThrows(ResponseStatusException.class, () -> receive("WT-1", request)); assertEquals(0, count("ll_warehouse_transfer_receipt_command")); assertEquals(0, occupancies.get()); + } + @Test void inactiveOrForeignTypeFbaMasterStopsBeforeResourceOccupation() { + when(directory.activeWarehouse(30L, "test-token")).thenReturn(ApiResponse.success(new DomesticWarehouseDirectoryClient.Detail( + new DomesticWarehouseDirectoryClient.Warehouse(30L, 0L, "ABE4", "ABE4 warehouse", "OVERSEAS", "US", "ACTIVE"), List.of()))); + assertThrows(ResponseStatusException.class, () -> receive("WT-1", request(101, 20, "KEY-1"))); assertEquals(0, occupancies.get()); + when(directory.activeWarehouse(30L, "test-token")).thenReturn(ApiResponse.success(new DomesticWarehouseDirectoryClient.Detail( + new DomesticWarehouseDirectoryClient.Warehouse(30L, 0L, "ABE4", "ABE4 warehouse", "FBA", "US", "DISABLED"), List.of()))); + assertThrows(ResponseStatusException.class, () -> receive("WT-1", request(101, 20, "KEY-1"))); assertEquals(0, count("ll_warehouse_transfer_receipt_command")); + } + @Test void inactiveReceivingWarehouseOrForeignPalletCannotCreateReceiptCommand() { + when(warehouse.requireActiveSharedWarehouse(9L)).thenReturn(new WarehouseResourceGateway.WarehouseReference(9, "NY", "NY warehouse", "DISABLED")); + assertThrows(ResponseStatusException.class, () -> receive("WT-1", request(101, 20, "KEY-1"))); + when(warehouse.requireActiveSharedWarehouse(9L)).thenReturn(new WarehouseResourceGateway.WarehouseReference(9, "NY", "NY warehouse", "ACTIVE")); + when(warehouse.resolvePallet(20L)).thenReturn(new WarehouseResourceGateway.WarehousePalletReference(20, "PAL-20", 99, 40, "IDLE", "ABE4", 0)); + assertThrows(ResponseStatusException.class, () -> receive("WT-1", request(101, 20, "KEY-1"))); assertEquals(0, count("ll_warehouse_transfer_receipt_command")); + } + @Test void wrongFrozenFbaAndWrongMasterIdAreRejected() { + when(warehouse.resolvePallet(20L)).thenReturn(new WarehouseResourceGateway.WarehousePalletReference(20, "PAL-20", 9, 40, "IDLE", "ABE3", 0)); + assertThrows(ResponseStatusException.class, () -> receive("WT-1", request(101, 20, "KEY-1"))); assertEquals(0, occupancies.get()); + when(warehouse.resolvePallet(20L)).thenReturn(pallet(20, "IDLE")); + when(directory.activeWarehouse(30L, "test-token")).thenReturn(ApiResponse.success(new DomesticWarehouseDirectoryClient.Detail( + new DomesticWarehouseDirectoryClient.Warehouse(31L, 0L, "ABE4", "ABE4 warehouse", "FBA", "US", "ACTIVE"), List.of()))); + assertThrows(ResponseStatusException.class, () -> receive("WT-1", request(101, 20, "KEY-1"))); assertEquals(0, count("ll_warehouse_transfer_receipt_command")); + } + @Test void sourceWithoutShippedOwnershipIsRejectedBeforeAnyRemoteCall() { + assertThrows(ResponseStatusException.class, () -> receive("WT-1", request(999, 20, "KEY-1"))); + assertEquals(0, reserveCalls.get()); verify(warehouse, never()).requireActiveSharedWarehouse(anyLong()); + } + @Test void legitimateDifferentTargetPalletsReceiveIndependentSourcesWithoutReducingCapability() { + addSource(1, 102); receive("WT-1", request(101, 20, "KEY-1")); receive("WT-1", request(102, 21, "KEY-2")); + assertEquals(2, occupancies.get()); assertEquals(2, count("ll_warehouse_transfer_receipt")); + assertEquals(2L, jdbc.queryForObject("SELECT COUNT(DISTINCT resource_lock_id) FROM ll_warehouse_transfer_receipt_command", Long.class)); + } +} diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/transfer/service/WarehouseTransferServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/transfer/service/WarehouseTransferServiceTest.java index ee4ba4044..7db6c3cf5 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/transfer/service/WarehouseTransferServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/transfer/service/WarehouseTransferServiceTest.java @@ -173,10 +173,23 @@ class WarehouseTransferServiceTest { @Mock InventoryCommandMapper inventoryCommands; @Mock com.baiyue.lastleg.relabel.service.RelabelApplicationService relabelService; private WarehouseTransferService service; + @Mock WarehouseTransferReceiptCoordinator receiptCoordinator; @BeforeEach void setUp() { service = new WarehouseTransferService(); + ReflectionTestUtils.setField(service, "receiptCoordinator", receiptCoordinator); + lenient().when(receiptCoordinator.receive(anyString(), anyString(), anyString(), any(), any(), any())) + .thenAnswer(call -> { + WarehouseTransferRequests.Receipt request = call.getArgument(3); + var target = new WarehouseTransferReceiptCoordinator.Target(request.destinationOverseasWarehouseId(), + request.destinationOverseasWarehouseCode(), request.destinationOverseasWarehouseName(), + request.destinationPalletId(), request.destinationPalletNo(), request.warehouseLocationId(), + request.warehouseLocationCode(), request.destinationWarehouseId(), request.destinationWarehouseCode(), + request.destinationWarehouseName()); + java.util.function.Function work = call.getArgument(5); + return work.apply(new WarehouseTransferReceiptCoordinator.Context(target, 51L, 61L)); + }); ReflectionTestUtils.setField(service, "transfers", transfers); ReflectionTestUtils.setField(service, "transferItems", transferItems); ReflectionTestUtils.setField(service, "transferCartons", transferCartons);