Merge remote-tracking branch 'origin/test' into dev
# Conflicts: # deploy/sql/海外区域统一增量-MySQL8.sql # lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/gateway/HttpWarehouseResourceGateway.java # lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/gateway/WarehouseResourceGateway.java # lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/ReceivingContainerService.java # lastleg-service/src/main/resources/db/lastleg.sql # lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/pda/gateway/HttpWarehouseResourceGatewayTest.java
This commit is contained in:
+1
@@ -19,6 +19,7 @@ public class ContainerPlanConversionCommand {
|
||||
private Long warehouseId;
|
||||
private String batchKey;
|
||||
private String status;
|
||||
private String leaseToken;
|
||||
private String preparationJson;
|
||||
private String resultJson;
|
||||
private String lastError;
|
||||
|
||||
+21
-5
@@ -17,17 +17,33 @@ public interface ContainerPlanConversionCommandMapper extends BaseMapper<Contain
|
||||
@Select("SELECT * FROM ll_container_plan_conversion_command WHERE source_container_booking_id=#{bookingId} LIMIT 1")
|
||||
ContainerPlanConversionCommand findByBooking(@Param("bookingId") String bookingId);
|
||||
|
||||
@Update("UPDATE ll_container_plan_conversion_command SET status=#{status},result_json=COALESCE(#{resultJson},result_json),last_error=#{lastError},updated_at=#{updatedAt} WHERE id=#{id} AND status=#{expectedStatus}")
|
||||
@Update("UPDATE ll_container_plan_conversion_command SET status=#{status},result_json=COALESCE(#{resultJson},result_json),last_error=#{lastError},updated_at=#{updatedAt} WHERE id=#{id} AND status=#{expectedStatus} AND lease_token=#{leaseToken}")
|
||||
int updateState(@Param("id") long id, @Param("expectedStatus") String expectedStatus,
|
||||
@Param("leaseToken") String leaseToken,
|
||||
@Param("status") String status,
|
||||
@Param("resultJson") String resultJson, @Param("lastError") String lastError,
|
||||
@Param("updatedAt") LocalDateTime updatedAt);
|
||||
|
||||
@Update("UPDATE ll_container_plan_conversion_command SET status=#{nextStatus},last_error=NULL,updated_at=#{updatedAt} WHERE id=#{id} AND status=#{expectedStatus}")
|
||||
@Update("UPDATE ll_container_plan_conversion_command SET status=#{nextStatus},lease_token=#{leaseToken},last_error=NULL,updated_at=#{updatedAt} WHERE id=#{id} AND status=#{expectedStatus}")
|
||||
int claim(@Param("id") long id, @Param("expectedStatus") String expectedStatus,
|
||||
@Param("nextStatus") String nextStatus, @Param("updatedAt") LocalDateTime updatedAt);
|
||||
@Param("nextStatus") String nextStatus, @Param("leaseToken") String leaseToken,
|
||||
@Param("updatedAt") LocalDateTime updatedAt);
|
||||
|
||||
@Update("UPDATE ll_container_plan_conversion_command SET last_error=NULL,updated_at=#{updatedAt} WHERE id=#{id} AND status='APPLYING' AND updated_at=#{expectedUpdatedAt}")
|
||||
@Update("UPDATE ll_container_plan_conversion_command SET lease_token=#{leaseToken},last_error=NULL,updated_at=#{updatedAt} WHERE id=#{id} AND status='APPLYING' AND updated_at=#{expectedUpdatedAt}")
|
||||
int reclaimStale(@Param("id") long id, @Param("expectedUpdatedAt") LocalDateTime expectedUpdatedAt,
|
||||
@Param("updatedAt") LocalDateTime updatedAt);
|
||||
@Param("leaseToken") String leaseToken, @Param("updatedAt") LocalDateTime updatedAt);
|
||||
|
||||
@Update("""
|
||||
UPDATE ll_container_plan_conversion_command
|
||||
SET idempotency_key=#{key},request_hash=#{requestHash},snapshot_version=#{snapshotVersion},
|
||||
snapshot_hash=#{snapshotHash},warehouse_id=#{warehouseId},batch_key=#{batchKey},
|
||||
status='PREPARED',lease_token=NULL,preparation_json=#{preparationJson},result_json=NULL,last_error=NULL,
|
||||
updated_at=#{updatedAt}
|
||||
WHERE id=#{id} AND status='FAILED'
|
||||
""")
|
||||
int replaceFailed(@Param("id") long id, @Param("key") String key,
|
||||
@Param("requestHash") String requestHash, @Param("snapshotVersion") long snapshotVersion,
|
||||
@Param("snapshotHash") String snapshotHash, @Param("warehouseId") long warehouseId,
|
||||
@Param("batchKey") String batchKey, @Param("preparationJson") String preparationJson,
|
||||
@Param("updatedAt") LocalDateTime updatedAt);
|
||||
}
|
||||
|
||||
+45
-22
@@ -117,36 +117,37 @@ public class HttpWarehouseResourceGateway implements WarehouseResourceGateway {
|
||||
|
||||
@Override
|
||||
public WarehousePalletReference resolvePallet(String palletCode) {
|
||||
WarehousePalletData data = palletData(palletCode);
|
||||
requireReservable(data);
|
||||
return pallet(data);
|
||||
WarehousePalletResponse response = lookupPallet(palletCode);
|
||||
requireReservable(response.data());
|
||||
return reference(response.data());
|
||||
}
|
||||
|
||||
@Override
|
||||
public WarehousePalletReference inspectPallet(String palletCode) {
|
||||
return pallet(palletData(palletCode));
|
||||
return reference(lookupPallet(palletCode).data());
|
||||
}
|
||||
|
||||
private WarehousePalletData palletData(String palletCode) {
|
||||
WarehousePalletResponse response = client.get()
|
||||
.uri("/internal/v1/warehouse-pallets/resolve/{palletCode}", palletCode)
|
||||
.header("X-Internal-Service-Token", internalToken)
|
||||
.retrieve().onStatus(HttpStatusCode::isError, (request, result) -> {
|
||||
throw palletLookupFailure(result.getStatusCode());
|
||||
}).body(WarehousePalletResponse.class);
|
||||
if (response != null && "A0404".equals(response.code())) {
|
||||
throw new ResponseStatusException(HttpStatus.NOT_FOUND, "Warehouse pallet was not found");
|
||||
@Override
|
||||
public WarehousePalletReference resolveReservedPallet(String palletCode, String customerId, long warehouseId,
|
||||
long palletId, long locationId, long resourceLockId) {
|
||||
if (customerId == null || customerId.isBlank() || warehouseId <= 0 || palletId <= 0
|
||||
|| locationId <= 0 || resourceLockId <= 0) {
|
||||
throw new ResponseStatusException(HttpStatus.CONFLICT,
|
||||
"Reserved pallet binding is incomplete");
|
||||
}
|
||||
if (response == null || response.data() == null) {
|
||||
throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
|
||||
"Warehouse pallet service returned no usable response");
|
||||
try {
|
||||
Long.parseLong(customerId);
|
||||
} catch (NumberFormatException exception) {
|
||||
throw new ResponseStatusException(HttpStatus.CONFLICT,
|
||||
"Reserved pallet customer binding is invalid");
|
||||
}
|
||||
return response.data();
|
||||
}
|
||||
|
||||
private WarehousePalletReference pallet(WarehousePalletData data) {
|
||||
return new WarehousePalletReference(data.id(), data.palletCode(), data.warehouseId(), data.locationId(),
|
||||
data.status(), data.fbaWarehouseCode(), data.unitPrice(), data.version());
|
||||
WarehousePalletData pallet = lookupPallet(palletCode).data();
|
||||
if (pallet.id() != palletId || pallet.warehouseId() != warehouseId || pallet.locationId() != locationId
|
||||
|| !"OCCUPIED".equals(pallet.status())) {
|
||||
throw new ResponseStatusException(HttpStatus.CONFLICT,
|
||||
"Scanned pallet does not match the reserved inbound plan resource");
|
||||
}
|
||||
return reference(pallet);
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -169,6 +170,28 @@ public class HttpWarehouseResourceGateway implements WarehouseResourceGateway {
|
||||
status.value() == 404 ? "Warehouse pallet was not found" : "Warehouse pallet lookup failed");
|
||||
}
|
||||
|
||||
private WarehousePalletResponse lookupPallet(String palletCode) {
|
||||
WarehousePalletResponse response = client.get()
|
||||
.uri("/internal/v1/warehouse-pallets/resolve/{palletCode}", palletCode)
|
||||
.header("X-Internal-Service-Token", internalToken)
|
||||
.retrieve().onStatus(HttpStatusCode::isError, (request, result) -> {
|
||||
throw palletLookupFailure(result.getStatusCode());
|
||||
}).body(WarehousePalletResponse.class);
|
||||
if (response != null && "A0404".equals(response.code())) {
|
||||
throw new ResponseStatusException(HttpStatus.NOT_FOUND, "Warehouse pallet was not found");
|
||||
}
|
||||
if (response == null || response.data() == null) {
|
||||
throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
|
||||
"Warehouse pallet service returned no usable response");
|
||||
}
|
||||
return response;
|
||||
}
|
||||
|
||||
private WarehousePalletReference reference(WarehousePalletData pallet) {
|
||||
return new WarehousePalletReference(pallet.id(), pallet.palletCode(), pallet.warehouseId(),
|
||||
pallet.locationId(), pallet.status(), pallet.fbaWarehouseCode(), pallet.unitPrice(), pallet.version());
|
||||
}
|
||||
|
||||
private ResponseStatusException warehouseReservationFailure(HttpStatusCode status) {
|
||||
HttpStatus mapped = status.value() == 409 ? HttpStatus.CONFLICT
|
||||
: status.value() == 404 ? HttpStatus.NOT_FOUND
|
||||
|
||||
+4
@@ -14,6 +14,10 @@ public interface WarehouseResourceGateway {
|
||||
default WarehousePalletReference inspectPallet(String palletCode) {
|
||||
return resolvePallet(palletCode);
|
||||
}
|
||||
default WarehousePalletReference resolveReservedPallet(String palletCode, String customerId, long warehouseId,
|
||||
long palletId, long locationId, long resourceLockId) {
|
||||
throw new UnsupportedOperationException("Reserved warehouse pallet resolution is not configured");
|
||||
}
|
||||
default WarehouseLocationReference resolveLocation(long warehouseId, String locationCode) {
|
||||
throw new UnsupportedOperationException("Warehouse location resolution is not configured");
|
||||
}
|
||||
|
||||
+37
-13
@@ -113,9 +113,8 @@ public class InboundPdaWorkflowService {
|
||||
InboundPlanPallet pallet = requirePallet(plan.getId(), planPalletId);
|
||||
validateDestinationWarehouse(pallet, request.warehouseId());
|
||||
if (pallet.getWarehouseResourceLockId() != null) {
|
||||
if (Objects.equals(pallet.getWarehousePalletId(), request.palletId())
|
||||
&& Objects.equals(pallet.getWarehouseLocationId(), request.locationId())) return;
|
||||
throw conflict("Planned pallet is already assigned to another warehouse resource");
|
||||
confirmReservedAssignment(plan, pallet, request.palletId(), request.locationId(), null);
|
||||
return;
|
||||
}
|
||||
String destinationKey = destinationKey(pallet);
|
||||
String businessId = plan.getContainerNo() + ":" + destinationKey;
|
||||
@@ -130,12 +129,7 @@ public class InboundPdaWorkflowService {
|
||||
pallet.setAssignmentStatus("ASSIGNED");
|
||||
pallet.setLockedAt(LocalDateTime.now());
|
||||
repository.update(pallet);
|
||||
List<InboundPlanPallet> pallets = repository.pallets(plan.getId());
|
||||
if (!pallets.isEmpty() && pallets.stream().allMatch(value -> !isNotAssigned(value))) {
|
||||
plan.setStatus("ASSIGNED");
|
||||
plan.setUpdatedAt(LocalDateTime.now());
|
||||
repository.update(plan);
|
||||
}
|
||||
promotePlanWhenAllPalletsAssigned(plan);
|
||||
}
|
||||
|
||||
@Transactional
|
||||
@@ -167,13 +161,13 @@ public class InboundPdaWorkflowService {
|
||||
@Transactional
|
||||
public Long assignResolvedPallet(String customerId, InboundPlan plan, InboundPlanPallet pallet,
|
||||
WarehouseResourceGateway.WarehousePalletReference resource) {
|
||||
if (pallet.getWarehouseResourceLockId() != null) {
|
||||
if (Objects.equals(pallet.getWarehousePalletId(), resource.palletId())) return null;
|
||||
throw conflict("Destination pallet is already bound to another physical pallet");
|
||||
}
|
||||
if (resource.palletNo() == null || resource.palletNo().isBlank()) {
|
||||
throw conflict("Resolved physical pallet has no pallet code");
|
||||
}
|
||||
if (pallet.getWarehouseResourceLockId() != null) {
|
||||
confirmReservedAssignment(plan, pallet, resource.palletId(), resource.locationId(), resource.palletNo());
|
||||
return null;
|
||||
}
|
||||
String destinationKey = destinationKey(pallet);
|
||||
String businessId = plan.getContainerNo() + ":" + destinationKey;
|
||||
String reservationKey = "INBOUND_CONTAINER_DESTINATION_PALLET:" + businessId + ":" + resource.palletId();
|
||||
@@ -184,6 +178,7 @@ public class InboundPdaWorkflowService {
|
||||
pallet.setWarehouseLocationId(resource.locationId());
|
||||
pallet.setWarehouseResourceLockId(lockId); pallet.setAssignmentStatus("ASSIGNED");
|
||||
pallet.setLockedAt(LocalDateTime.now()); repository.update(pallet);
|
||||
promotePlanWhenAllPalletsAssigned(plan);
|
||||
return lockId;
|
||||
} catch (RuntimeException failure) {
|
||||
compensateReservation(customerId, lockId, failure);
|
||||
@@ -878,6 +873,35 @@ public class InboundPdaWorkflowService {
|
||||
|| !("ASSIGNED".equals(pallet.getAssignmentStatus()) || "LOCKED".equals(pallet.getAssignmentStatus()));
|
||||
}
|
||||
|
||||
private void confirmReservedAssignment(InboundPlan plan, InboundPlanPallet pallet,
|
||||
long palletId, long locationId, String palletNo) {
|
||||
boolean sameReservedResource = pallet.getWarehouseResourceLockId() != null
|
||||
&& pallet.getWarehouseResourceLockId() > 0
|
||||
&& Objects.equals(pallet.getWarehousePalletId(), palletId)
|
||||
&& Objects.equals(pallet.getWarehouseLocationId(), locationId);
|
||||
if (!sameReservedResource) {
|
||||
throw conflict("Planned pallet is already assigned to another physical pallet or location");
|
||||
}
|
||||
if ("ASSIGNED".equals(pallet.getAssignmentStatus()) || "LOCKED".equals(pallet.getAssignmentStatus())) return;
|
||||
if (!"RESERVED".equals(pallet.getAssignmentStatus())) {
|
||||
throw conflict("Reserved physical pallet has an invalid assignment status");
|
||||
}
|
||||
if (palletNo != null && !palletNo.isBlank()) pallet.setPalletNo(palletNo.trim());
|
||||
pallet.setAssignmentStatus("ASSIGNED");
|
||||
if (pallet.getLockedAt() == null) pallet.setLockedAt(LocalDateTime.now());
|
||||
repository.update(pallet);
|
||||
promotePlanWhenAllPalletsAssigned(plan);
|
||||
}
|
||||
|
||||
private void promotePlanWhenAllPalletsAssigned(InboundPlan plan) {
|
||||
List<InboundPlanPallet> plannedPallets = repository.pallets(plan.getId());
|
||||
if (plannedPallets.isEmpty() || plannedPallets.stream().anyMatch(this::isNotAssigned)) return;
|
||||
if (!java.util.Set.of("PENDING_ASSIGNMENT", "RESERVED").contains(plan.getStatus())) return;
|
||||
plan.setStatus("ASSIGNED");
|
||||
plan.setUpdatedAt(LocalDateTime.now());
|
||||
repository.update(plan);
|
||||
}
|
||||
|
||||
private InboundPlan requirePlan(String customerId, String planNo) {
|
||||
InboundPlan plan = repository.lockPlan(customerId, planNo);
|
||||
if (plan == null) throw notFound("Inbound plan not found");
|
||||
|
||||
+52
-14
@@ -274,14 +274,8 @@ public class ReceivingContainerService {
|
||||
ReceivingContainerSession session = requireSigned(operatorKey);
|
||||
String customerId = session.getCustomerId();
|
||||
InboundRef ref = requireInbound(customerId, session.getContainerNo(), request.inboundCode());
|
||||
WarehouseResourceGateway.WarehousePalletReference pallet;
|
||||
try {
|
||||
pallet = warehouse.resolvePallet(request.palletCode());
|
||||
} catch (IllegalArgumentException exception) {
|
||||
throw notFound("Pallet was not found");
|
||||
}
|
||||
if (pallet == null) throw notFound("Pallet was not found");
|
||||
validatePallet(ref, pallet);
|
||||
PalletSelection selection = resolveReceivingPallet(ref, request.palletCode(), false);
|
||||
WarehouseResourceGateway.WarehousePalletReference pallet = selection.resource();
|
||||
return new ReceivingContainerResponses.PalletResolved(pallet.palletNo(), pallet.warehouseId(), pallet.status(),
|
||||
pallet.fbaWarehouseCode(), ref.line().getInboundCode(), destinationType(ref.line(), ref.order()), true);
|
||||
}
|
||||
@@ -301,9 +295,9 @@ public class ReceivingContainerService {
|
||||
List<FbaSkuAllocation> skuAllocations = allocateFbaSkuReceipts(customerId, fbaPlanRefs, request.packageCount());
|
||||
skuAllocations.forEach(allocation -> workflow.validateBulkResolved(customerId, allocation.ref().plan(),
|
||||
allocation.ref().line(), allocation.packages()));
|
||||
WarehouseResourceGateway.WarehousePalletReference resource = warehouse.resolvePallet(request.palletCode());
|
||||
validatePallet(ref, resource);
|
||||
InboundPlanPallet pallet = choosePallet(ref, resource);
|
||||
PalletSelection selection = resolveReceivingPallet(ref, request.palletCode(), true);
|
||||
WarehouseResourceGateway.WarehousePalletReference resource = selection.resource();
|
||||
InboundPlanPallet pallet = selection.pallet();
|
||||
Long resourceLockId = workflow.assignResolvedPallet(customerId, ref.plan(), pallet, resource);
|
||||
try {
|
||||
int accepted = 0;
|
||||
@@ -344,9 +338,9 @@ public class ReceivingContainerService {
|
||||
if (!"OVERSEAS".equals(destinationType(ref.line(), ref.order()))) {
|
||||
throw conflict("FBA destination must use bulk count");
|
||||
}
|
||||
WarehouseResourceGateway.WarehousePalletReference resource = warehouse.inspectPallet(request.palletCode());
|
||||
validatePallet(ref, resource);
|
||||
InboundPlanPallet pallet = choosePallet(ref, resource);
|
||||
PalletSelection selection = resolveReceivingPallet(ref, request.palletCode(), true);
|
||||
WarehouseResourceGateway.WarehousePalletReference resource = selection.resource();
|
||||
InboundPlanPallet pallet = selection.pallet();
|
||||
workflow.validateCartonResolved(customerId, ref.plan(), ref.line(), pallet, request.cartonMark());
|
||||
Long resourceLockId = workflow.assignResolvedPallet(customerId, ref.plan(), pallet, resource);
|
||||
try {
|
||||
@@ -639,6 +633,47 @@ public class ReceivingContainerService {
|
||||
return candidates.get(0);
|
||||
}
|
||||
|
||||
private PalletSelection resolveReceivingPallet(InboundRef ref, String palletCode, boolean requirePlanPallet) {
|
||||
List<InboundPlanPallet> reserved = pallets.selectList(new LambdaQueryWrapper<InboundPlanPallet>()
|
||||
.eq(InboundPlanPallet::getCustomerId, ref.order().getCustomerId())
|
||||
.eq(InboundPlanPallet::getPlanId, ref.plan().getId())).stream()
|
||||
.filter(value -> sameDestination(ref.line(), value))
|
||||
.filter(value -> palletCode.trim().equals(value.getPalletNo()))
|
||||
.filter(this::hasConfirmedReservation)
|
||||
.toList();
|
||||
if (reserved.size() > 1) {
|
||||
throw conflict("Pallet code matches multiple reserved resources in this inbound plan");
|
||||
}
|
||||
WarehouseResourceGateway.WarehousePalletReference resource;
|
||||
InboundPlanPallet pallet;
|
||||
if (reserved.size() == 1) {
|
||||
pallet = reserved.get(0);
|
||||
long warehouseId = ref.plan().getOverseasWarehouseId() == null
|
||||
? Long.parseLong(ref.order().getOverseasWarehouseId()) : ref.plan().getOverseasWarehouseId();
|
||||
resource = warehouse.resolveReservedPallet(palletCode.trim(), ref.order().getCustomerId(), warehouseId,
|
||||
pallet.getWarehousePalletId(), pallet.getWarehouseLocationId(), pallet.getWarehouseResourceLockId());
|
||||
} else {
|
||||
try {
|
||||
resource = warehouse.resolvePallet(palletCode.trim());
|
||||
} catch (IllegalArgumentException exception) {
|
||||
throw notFound("Pallet was not found");
|
||||
}
|
||||
if (resource == null) throw notFound("Pallet was not found");
|
||||
pallet = requirePlanPallet ? choosePallet(ref, resource) : null;
|
||||
}
|
||||
validatePallet(ref, resource);
|
||||
return new PalletSelection(pallet, resource);
|
||||
}
|
||||
|
||||
private boolean hasConfirmedReservation(InboundPlanPallet pallet) {
|
||||
return pallet.getWarehousePalletId() != null && pallet.getWarehousePalletId() > 0
|
||||
&& pallet.getWarehouseLocationId() != null && pallet.getWarehouseLocationId() > 0
|
||||
&& pallet.getWarehouseResourceLockId() != null && pallet.getWarehouseResourceLockId() > 0
|
||||
&& ("RESERVED".equals(pallet.getAssignmentStatus())
|
||||
|| "ASSIGNED".equals(pallet.getAssignmentStatus())
|
||||
|| "LOCKED".equals(pallet.getAssignmentStatus()));
|
||||
}
|
||||
|
||||
private void validatePallet(InboundRef ref, WarehouseResourceGateway.WarehousePalletReference pallet) {
|
||||
if ("SCRAPPED".equals(pallet.status())) throw conflict("Pallet is scrapped");
|
||||
long actualWarehouse;
|
||||
@@ -666,6 +701,9 @@ public class ReceivingContainerService {
|
||||
return Objects.equals(line.getDestinationWarehouseCode(), pallet.getDestinationWarehouseCode());
|
||||
}
|
||||
|
||||
private record PalletSelection(InboundPlanPallet pallet,
|
||||
WarehouseResourceGateway.WarehousePalletReference resource) { }
|
||||
|
||||
private ReceivingContainerResponses.Session snapshot(ReceivingContainerSession session, List<InboundPlan> containerPlans) {
|
||||
List<ReceivingContainerResponses.Inbound> inbounds = refs(session.getCustomerId(), session.getContainerNo()).stream()
|
||||
.map(this::inbound).toList();
|
||||
|
||||
+136
-48
@@ -46,82 +46,108 @@ public class ContainerInboundPlanConversionCoordinator {
|
||||
String bookingId = required(request.sourceContainerBookingId(), 128, "来源柜预订 ID 不能为空");
|
||||
String key = required(idempotencyKey, 128, "Idempotency-Key 必须为1至128个字符");
|
||||
String requestHash = hash(List.of(bookingId, request.snapshotVersion(), request.snapshotHash()));
|
||||
ContainerPlanConversionCommand resolved = commands.find(bookingId, key);
|
||||
if (resolved == null) resolved = commands.findByBooking(bookingId);
|
||||
if (resolved == null) resolved = createCommand(bookingId, key, request, requestHash);
|
||||
ContainerPlanConversionCommand direct = commands.find(bookingId, key);
|
||||
ContainerPlanConversionCommand resolved = direct == null ? commands.findByBooking(bookingId) : direct;
|
||||
if (resolved == null) {
|
||||
resolved = createCommand(bookingId, key, request, requestHash);
|
||||
} else if ("COMPLETED".equals(resolved.getStatus())) {
|
||||
if (direct != null) requireSameRequest(resolved, requestHash);
|
||||
return result(resolved);
|
||||
} else if ("FAILED".equals(resolved.getStatus())) {
|
||||
resolved = replaceFailedCommand(resolved, bookingId, key, request, requestHash);
|
||||
} else {
|
||||
requireSameRequest(resolved, requestHash);
|
||||
}
|
||||
if ("COMPLETED".equals(resolved.getStatus())) {
|
||||
requireSameRequest(resolved, requestHash);
|
||||
return result(resolved);
|
||||
}
|
||||
final ContainerPlanConversionCommand command = resolved;
|
||||
requireSameRequest(command, requestHash);
|
||||
Preparation preparation = read(command.getPreparationJson(), Preparation.class);
|
||||
|
||||
if ("COMPLETED".equals(command.getStatus())) return result(command);
|
||||
boolean applyingLeaseOwned = false;
|
||||
if ("APPLYING".equals(command.getStatus())) {
|
||||
LocalDateTime now = LocalDateTime.now();
|
||||
String leaseToken = IdWorker.getIdStr();
|
||||
if (command.getUpdatedAt() == null || command.getUpdatedAt().plus(APPLYING_LEASE).isAfter(now)
|
||||
|| commands.reclaimStale(command.getId(), command.getUpdatedAt(), now) != 1) {
|
||||
|| commands.reclaimStale(command.getId(), command.getUpdatedAt(), leaseToken, now) != 1) {
|
||||
throw new ResponseStatusException(HttpStatus.CONFLICT, "该柜正在生成入库计划，请勿重复提交");
|
||||
}
|
||||
command.setLeaseToken(leaseToken);
|
||||
command.setUpdatedAt(now);
|
||||
applyingLeaseOwned = true;
|
||||
}
|
||||
if (!"LOCAL_COMMITTED".equals(command.getStatus())) {
|
||||
if (commands.claim(command.getId(), "PREPARED", "APPLYING", LocalDateTime.now()) != 1) {
|
||||
if ("PREPARED".equals(command.getStatus())) {
|
||||
String leaseToken = IdWorker.getIdStr();
|
||||
if (commands.claim(command.getId(), "PREPARED", "APPLYING", leaseToken, LocalDateTime.now()) != 1) {
|
||||
ContainerPlanConversionCommand current = commands.findByBooking(bookingId);
|
||||
if (current != null && "COMPLETED".equals(current.getStatus())) return result(current);
|
||||
throw new ResponseStatusException(HttpStatus.CONFLICT, "该柜正在生成入库计划，请勿重复提交");
|
||||
}
|
||||
command.setStatus("APPLYING");
|
||||
command.setLeaseToken(leaseToken);
|
||||
applyingLeaseOwned = true;
|
||||
}
|
||||
if (!applyingLeaseOwned && !"LOCAL_COMMITTED".equals(command.getStatus())) {
|
||||
throw new ResponseStatusException(HttpStatus.CONFLICT, "柜级转换命令状态无效");
|
||||
}
|
||||
if (preparation.items().isEmpty()) {
|
||||
try {
|
||||
List<InboundPlanResponse> existing = inTransaction(() -> {
|
||||
List<InboundPlanResponse> converted = plans.convertContainerReserved(bookingId,
|
||||
request.snapshotVersion(), request.snapshotHash(), List.of());
|
||||
state(command.getId(), "APPLYING", "COMPLETED", write(converted), null);
|
||||
state(command, "APPLYING", "COMPLETED", write(converted), null);
|
||||
return converted;
|
||||
});
|
||||
command.setStatus("COMPLETED");
|
||||
return existing;
|
||||
} catch (RuntimeException failure) {
|
||||
resetForRetry(command, failure);
|
||||
ContainerPlanConversionCommand current = commands.findByBooking(bookingId);
|
||||
if (current != null && "COMPLETED".equals(current.getStatus())) return result(current);
|
||||
failWithoutResources(command, failure);
|
||||
throw failure;
|
||||
}
|
||||
}
|
||||
|
||||
WarehouseResourceGateway.PalletBatchEnsure ensure = new WarehouseResourceGateway.PalletBatchEnsure(
|
||||
preparation.warehouseId(), command.getBatchKey(), preparation.items().stream().map(item ->
|
||||
new WarehouseResourceGateway.PalletBatchItemRequest(Long.parseLong(item.customerId()), item.businessKey(),
|
||||
item.planNo(), item.globalSequenceNo(), "STANDARD", item.fbaWarehouseCode())).toList());
|
||||
WarehouseResourceGateway.PalletBatchResponse reservation;
|
||||
try {
|
||||
reservation = warehouse.ensurePalletBatch(ensure);
|
||||
requireBatch(reservation, command.getBatchKey(), preparation.items().size(), "RESERVED", "CONFIRMED");
|
||||
} catch (RuntimeException failure) {
|
||||
resetForRetry(command, failure);
|
||||
throw failure;
|
||||
}
|
||||
|
||||
List<InboundPlanResponse> localResult;
|
||||
if ("LOCAL_COMMITTED".equals(command.getStatus())) {
|
||||
localResult = result(command);
|
||||
} else {
|
||||
WarehouseResourceGateway.PalletBatchEnsure ensure = new WarehouseResourceGateway.PalletBatchEnsure(
|
||||
preparation.warehouseId(), command.getBatchKey(), preparation.items().stream().map(item ->
|
||||
new WarehouseResourceGateway.PalletBatchItemRequest(Long.parseLong(item.customerId()),
|
||||
item.businessKey(), item.planNo(), item.globalSequenceNo(), "STANDARD",
|
||||
item.fbaWarehouseCode())).toList());
|
||||
WarehouseResourceGateway.PalletBatchResponse reservation;
|
||||
try {
|
||||
reservation = warehouse.ensurePalletBatch(ensure);
|
||||
requireBatch(reservation, command.getBatchKey(), preparation.items().size(), "RESERVED", "CONFIRMED");
|
||||
} catch (RuntimeException failure) {
|
||||
compensateAndFail(command, preparation, failure);
|
||||
throw failure;
|
||||
}
|
||||
try {
|
||||
List<InboundPlanService.PreparedPlan> preparedPlans = bind(preparation, reservation);
|
||||
localResult = inTransaction(() -> {
|
||||
List<InboundPlanResponse> created = plans.convertContainerReserved(bookingId,
|
||||
request.snapshotVersion(), request.snapshotHash(), preparedPlans);
|
||||
state(command.getId(), "APPLYING", "LOCAL_COMMITTED", write(created), null);
|
||||
state(command, "APPLYING", "LOCAL_COMMITTED", write(created), null);
|
||||
return created;
|
||||
});
|
||||
command.setStatus("LOCAL_COMMITTED");
|
||||
} catch (RuntimeException failure) {
|
||||
try {
|
||||
warehouse.compensatePalletBatch(new WarehouseResourceGateway.PalletBatchCompensation(
|
||||
preparation.warehouseId(), command.getBatchKey(), "COMPENSATE:" + command.getId(),
|
||||
"Lastleg local transaction did not commit"));
|
||||
} catch (RuntimeException compensationFailure) {
|
||||
failure.addSuppressed(compensationFailure);
|
||||
ContainerPlanConversionCommand current = commands.findByBooking(bookingId);
|
||||
if (current != null && "COMPLETED".equals(current.getStatus())) {
|
||||
return result(current);
|
||||
}
|
||||
if (current != null && "LOCAL_COMMITTED".equals(current.getStatus())) {
|
||||
command.setStatus(current.getStatus());
|
||||
command.setResultJson(current.getResultJson());
|
||||
localResult = result(current);
|
||||
} else {
|
||||
compensateAndFail(command, preparation, failure);
|
||||
throw failure;
|
||||
}
|
||||
resetForRetry(command, failure);
|
||||
throw failure;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -136,6 +162,35 @@ public class ContainerInboundPlanConversionCoordinator {
|
||||
private ContainerPlanConversionCommand createCommand(String bookingId, String key,
|
||||
ConvertContainerInboundPlanRequest request,
|
||||
String requestHash) {
|
||||
long commandId = IdWorker.getId();
|
||||
ContainerPlanConversionCommand command = prepareCommand(commandId, commandId, bookingId, key, request, requestHash);
|
||||
try { inTransaction(() -> { commands.insert(command); return null; }); }
|
||||
catch (DuplicateKeyException concurrent) {
|
||||
ContainerPlanConversionCommand existing = commands.findByBooking(bookingId);
|
||||
if (existing == null) throw concurrent;
|
||||
return existing;
|
||||
}
|
||||
return command;
|
||||
}
|
||||
|
||||
private ContainerPlanConversionCommand replaceFailedCommand(ContainerPlanConversionCommand previous,
|
||||
String bookingId, String key,
|
||||
ConvertContainerInboundPlanRequest request,
|
||||
String requestHash) {
|
||||
ContainerPlanConversionCommand replacement = prepareCommand(previous.getId(), IdWorker.getId(),
|
||||
bookingId, key, request, requestHash);
|
||||
int updated = commands.replaceFailed(previous.getId(), key, requestHash, request.snapshotVersion(),
|
||||
request.snapshotHash(), replacement.getWarehouseId(), replacement.getBatchKey(),
|
||||
replacement.getPreparationJson(), replacement.getUpdatedAt());
|
||||
if (updated != 1) {
|
||||
throw new ResponseStatusException(HttpStatus.CONFLICT, "该柜正在刷新入库计划，请勿重复提交");
|
||||
}
|
||||
return replacement;
|
||||
}
|
||||
|
||||
private ContainerPlanConversionCommand prepareCommand(long commandId, long generationId, String bookingId,
|
||||
String key, ConvertContainerInboundPlanRequest request,
|
||||
String requestHash) {
|
||||
ContainerInboundPlanPreviewResponse preview = plans.previewContainer(bookingId);
|
||||
if (preview.snapshotVersion() != request.snapshotVersion()
|
||||
|| !Objects.equals(preview.snapshotHash(), request.snapshotHash()) || !preview.ready()) {
|
||||
@@ -146,7 +201,6 @@ public class ContainerInboundPlanConversionCoordinator {
|
||||
catch (RuntimeException exception) {
|
||||
throw new ResponseStatusException(HttpStatus.CONFLICT, "柜级接柜仓 ID 无效", exception);
|
||||
}
|
||||
long commandId = IdWorker.getId();
|
||||
List<PreparedItem> items = new ArrayList<>();
|
||||
int globalSequence = 0;
|
||||
for (ContainerInboundPlanPreviewResponse.Order order : preview.groups().stream()
|
||||
@@ -155,7 +209,7 @@ public class ContainerInboundPlanConversionCoordinator {
|
||||
String planNo = numbers.next("INBOUND_PLAN", order.destinationWarehouseCode(), planId);
|
||||
for (int localSequence = 1; localSequence <= order.suggestedPalletCount(); localSequence++) {
|
||||
items.add(new PreparedItem(order.customerId(), Long.parseLong(order.orderId()), planId, planNo,
|
||||
localSequence, ++globalSequence, "CIP:" + commandId + ":" + globalSequence,
|
||||
localSequence, ++globalSequence, "CIP:" + generationId + ":" + globalSequence,
|
||||
"FBA".equals(order.destinationType()) ? order.destinationWarehouseCode() : null));
|
||||
}
|
||||
}
|
||||
@@ -164,15 +218,9 @@ public class ContainerInboundPlanConversionCoordinator {
|
||||
command.setId(commandId); command.setSourceContainerBookingId(bookingId); command.setIdempotencyKey(key);
|
||||
command.setRequestHash(requestHash); command.setSnapshotVersion(request.snapshotVersion());
|
||||
command.setSnapshotHash(request.snapshotHash()); command.setWarehouseId(warehouseId);
|
||||
command.setBatchKey("CIP:" + hash(bookingId) + ":" + commandId); command.setStatus("PREPARED");
|
||||
command.setBatchKey("CIP:" + hash(bookingId) + ":" + generationId); command.setStatus("PREPARED");
|
||||
command.setPreparationJson(write(preparation)); command.setCreatedAt(LocalDateTime.now());
|
||||
command.setUpdatedAt(command.getCreatedAt());
|
||||
try { inTransaction(() -> { commands.insert(command); return null; }); }
|
||||
catch (DuplicateKeyException concurrent) {
|
||||
ContainerPlanConversionCommand existing = commands.findByBooking(bookingId);
|
||||
if (existing == null) throw concurrent;
|
||||
return existing;
|
||||
}
|
||||
return command;
|
||||
}
|
||||
|
||||
@@ -213,11 +261,26 @@ public class ContainerInboundPlanConversionCoordinator {
|
||||
|
||||
private void requireBatch(WarehouseResourceGateway.PalletBatchResponse response, String batchKey, int size,
|
||||
String... acceptedStatuses) {
|
||||
if (response == null || !Objects.equals(batchKey, response.batchKey()) || response.items() == null
|
||||
if (response == null || response.batchId() <= 0 || response.attemptNo() <= 0
|
||||
|| !Objects.equals(batchKey, response.batchKey()) || response.items() == null
|
||||
|| response.items().size() != size || java.util.Arrays.stream(acceptedStatuses)
|
||||
.noneMatch(status -> status.equals(response.status()))) {
|
||||
throw new ResponseStatusException(HttpStatus.CONFLICT, "海外仓托盘批次响应不完整或状态不一致");
|
||||
}
|
||||
String expectedItemStatus = switch (response.status()) {
|
||||
case "RESERVED" -> "RESERVED";
|
||||
case "CONFIRMED" -> "CONFIRMED";
|
||||
case "COMPENSATED" -> "RELEASED";
|
||||
default -> null;
|
||||
};
|
||||
boolean invalidItem = expectedItemStatus == null || response.items().stream().anyMatch(item ->
|
||||
item.customerId() <= 0 || item.sequenceNo() <= 0 || item.palletId() <= 0 || item.locationId() <= 0
|
||||
|| item.resourceLockId() <= 0 || trimToNull(item.businessKey()) == null
|
||||
|| trimToNull(item.inboundPlanNo()) == null || trimToNull(item.palletCode()) == null
|
||||
|| item.palletCode().length() > 128 || !expectedItemStatus.equals(item.status()));
|
||||
if (invalidItem) {
|
||||
throw new ResponseStatusException(HttpStatus.CONFLICT, "海外仓托盘批次明细不完整或状态不一致");
|
||||
}
|
||||
}
|
||||
|
||||
private void requireSameRequest(ContainerPlanConversionCommand command, String requestHash) {
|
||||
@@ -235,19 +298,41 @@ public class ContainerInboundPlanConversionCoordinator {
|
||||
}
|
||||
|
||||
private void complete(ContainerPlanConversionCommand command, List<InboundPlanResponse> result) {
|
||||
inTransaction(() -> { state(command.getId(), "LOCAL_COMMITTED", "COMPLETED", write(result), null); return null; });
|
||||
inTransaction(() -> { state(command, "LOCAL_COMMITTED", "COMPLETED", write(result), null); return null; });
|
||||
command.setStatus("COMPLETED");
|
||||
}
|
||||
|
||||
private void resetForRetry(ContainerPlanConversionCommand command, RuntimeException failure) {
|
||||
private void failWithoutResources(ContainerPlanConversionCommand command, RuntimeException failure) {
|
||||
if ("APPLYING".equals(command.getStatus())) {
|
||||
state(command.getId(), "APPLYING", "PREPARED", null, safeMessage(failure));
|
||||
command.setStatus("PREPARED");
|
||||
state(command, "APPLYING", "FAILED", null, safeMessage(failure));
|
||||
command.setStatus("FAILED");
|
||||
}
|
||||
}
|
||||
|
||||
private void state(long id, String expectedStatus, String status, String resultJson, String lastError) {
|
||||
if (commands.updateState(id, expectedStatus, status, resultJson, lastError, LocalDateTime.now()) != 1) {
|
||||
private void compensateAndFail(ContainerPlanConversionCommand command, Preparation preparation,
|
||||
RuntimeException failure) {
|
||||
try {
|
||||
WarehouseResourceGateway.PalletBatchResponse compensated = warehouse.compensatePalletBatch(
|
||||
new WarehouseResourceGateway.PalletBatchCompensation(preparation.warehouseId(),
|
||||
command.getBatchKey(), "COMPENSATE:" + command.getBatchKey(),
|
||||
"Lastleg local transaction did not commit"));
|
||||
requireBatch(compensated, command.getBatchKey(), preparation.items().size(), "COMPENSATED");
|
||||
failWithoutResources(command, failure);
|
||||
} catch (ResponseStatusException compensationFailure) {
|
||||
if (compensationFailure.getStatusCode().value() == 404) {
|
||||
failWithoutResources(command, failure);
|
||||
return;
|
||||
}
|
||||
failure.addSuppressed(compensationFailure);
|
||||
} catch (RuntimeException compensationFailure) {
|
||||
failure.addSuppressed(compensationFailure);
|
||||
}
|
||||
}
|
||||
|
||||
private void state(ContainerPlanConversionCommand command, String expectedStatus, String status,
|
||||
String resultJson, String lastError) {
|
||||
if (commands.updateState(command.getId(), expectedStatus, command.getLeaseToken(), status,
|
||||
resultJson, lastError, LocalDateTime.now()) != 1) {
|
||||
throw new IllegalStateException("无法更新柜级转换命令状态");
|
||||
}
|
||||
}
|
||||
@@ -281,6 +366,9 @@ public class ContainerInboundPlanConversionCoordinator {
|
||||
String message = value.getMessage();
|
||||
return message == null ? value.getClass().getSimpleName() : message.substring(0, Math.min(1000, message.length()));
|
||||
}
|
||||
private static String trimToNull(String value) {
|
||||
return value == null || value.isBlank() ? null : value.trim();
|
||||
}
|
||||
|
||||
public record Preparation(long warehouseId, List<PreparedItem> items) { }
|
||||
public record PreparedItem(String customerId, long orderId, long planId, String planNo,
|
||||
|
||||
+2
-1
@@ -198,7 +198,8 @@ public class InboundPlanService {
|
||||
.map(ContainerInboundPlanPreviewResponse.Order::plannedPalletCount)
|
||||
.filter(Objects::nonNull).mapToInt(Integer::intValue).sum();
|
||||
long snapshotVersion = rows.stream().mapToLong(InboundPlanOrderMapper.ContainerPreviewOrderRow::version).max().orElse(0);
|
||||
String snapshotHash = sha256(List.of(bookingId, containerNo, warehouseId, snapshotVersion, orders));
|
||||
String snapshotHash = sha256(java.util.Arrays.asList(
|
||||
bookingId, containerNo, warehouseId, snapshotVersion, orders));
|
||||
return new ContainerInboundPlanPreviewResponse(bookingId, containerNo, warehouseId, warehouseCode,
|
||||
warehouseName, snapshotVersion, snapshotHash, orders.size(), orders.size() - blocked, blocked,
|
||||
pendingOrders, pendingPallets, existingPallets,
|
||||
|
||||
Reference in New Issue
Block a user