fix(transfer): reserve receipt resources with durable recovery intents

This commit is contained in:
2026-10-05 03:37:13 +08:00
parent 3f8ac23aba
commit c194dd2862
10 changed files with 661 additions and 9 deletions
@@ -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) {}
@@ -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;
}
@@ -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<WarehouseTransferReceiptCommand> {
@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);
}
@@ -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<WarehouseTransferResponses.Detail> replay,
Function<Context, WarehouseTransferResponses.Detail> 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); }
}
@@ -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();