merge: sync latest test environment changes

This commit is contained in:
2026-09-09 22:45:42 +08:00
26 changed files with 217 additions and 184 deletions
@@ -13,7 +13,6 @@ public class ReceivingContainerOperation {
@TableId(type = IdType.INPUT)
private Long id;
private Long sessionId;
private String customerId;
private String operatorKey;
private String action;
private String idempotencyKey;
@@ -15,7 +15,6 @@ import java.time.LocalDateTime;
public class ReceivingContainerSession {
@TableId(type = IdType.INPUT)
private Long id;
private String customerId;
private String sourceContainerBookingId;
private String containerFactHash;
private String containerNo;
@@ -546,33 +546,34 @@ public class InboundPdaWorkflowService {
@Transactional
public PdaResponses.Task complete(String customerId, String planNo) {
InboundPlan plan = requirePlan(customerId, planNo);
return completeContainer(customerId, List.of(plan)).get(0);
return completeContainer(List.of(plan)).get(0);
}
/** Completes every plan of one physical container and emits exactly one container-level outbox event. */
@Transactional
public List<PdaResponses.Task> completeContainer(String customerId, List<InboundPlan> containerPlans) {
public List<PdaResponses.Task> completeContainer(List<InboundPlan> containerPlans) {
if (containerPlans == null || containerPlans.isEmpty()) throw conflict("Receiving container has no inbound plans");
List<InboundPlan> orderedPlans = containerPlans.stream().sorted(Comparator.comparing(InboundPlan::getId)).toList();
String containerNo = orderedPlans.get(0).getContainerNo();
if (containerNo == null || containerNo.isBlank() || orderedPlans.stream().anyMatch(plan ->
!customerId.equals(plan.getCustomerId()) || !containerNo.equals(plan.getContainerNo()))) {
throw conflict("Inbound plans must belong to one customer and one physical container");
plan.getCustomerId() == null || plan.getCustomerId().isBlank() || !containerNo.equals(plan.getContainerNo()))) {
throw conflict("Inbound plans must have an owning customer and belong to one physical container");
}
List<CompletionContext> contexts = orderedPlans.stream().map(plan -> completionContext(customerId, plan)).toList();
List<CompletionContext> contexts = orderedPlans.stream()
.map(plan -> completionContext(plan.getCustomerId(), plan)).toList();
boolean anyCompleted = contexts.stream().anyMatch(context -> "COMPLETED".equals(context.task().getStatus()));
boolean allCompleted = contexts.stream().allMatch(context -> "COMPLETED".equals(context.task().getStatus()));
if (allCompleted) return contexts.stream().map(context -> task(context.task(), context.plan().getPlanNo())).toList();
if (anyCompleted) throw conflict("Container plans are only partially completed; manual reconciliation is required");
LocalDateTime now = LocalDateTime.now();
for (CompletionContext context : contexts) completePlan(customerId, context, now);
for (CompletionContext context : contexts) completePlan(context.plan().getCustomerId(), context, now);
String sourceContainerBookingId = sourceContainerBookingId(contexts);
LocalDateTime photoUploadedAt = contexts.stream().map(context -> context.task().getUnloadPhotoUploadedAt())
.max(LocalDateTime::compareTo).orElseThrow();
long aggregateVersion = contexts.stream().map(CompletionContext::task)
.map(InboundPdaTask::getVersion).filter(Objects::nonNull).mapToLong(Integer::longValue).max().orElse(1L);
unloadCompletedOutbox.publishContainerCompleted(customerId, sourceContainerBookingId, containerNo,
unloadCompletedOutbox.publishContainerCompleted(sourceContainerBookingId, containerNo,
now.toInstant(ZoneOffset.UTC), photoUploadedAt.toInstant(ZoneOffset.UTC), aggregateVersion);
return contexts.stream().map(context -> task(context.task(), context.plan().getPlanNo())).toList();
}
@@ -47,7 +47,7 @@ public class InboundUnloadCompletedOutboxService {
if (task.getUnloadPhotoUploadedAt() == null) {
throw new IllegalStateException("Inbound task has no unload photo upload time");
}
publishContainerCompleted(task.getCustomerId(), sourceOrder.getSourceContainerBookingId(),
publishContainerCompleted(sourceOrder.getSourceContainerBookingId(),
plan.getContainerNo(), task.getCompletedAt().toInstant(ZoneOffset.UTC),
task.getUnloadPhotoUploadedAt().toInstant(ZoneOffset.UTC),
Math.max(1, task.getVersion() == null ? 1 : task.getVersion()));
@@ -56,10 +56,9 @@ public class InboundUnloadCompletedOutboxService {
/**
* 柜级唯一发布入口。调用方只能在同一物理柜的全部计划完成后调用一次;重复调用保留首次冻结事实。
*/
public void publishContainerCompleted(String customerId, String sourceContainerBookingId,
String containerNo, Instant unloadingCompletedAt,
Instant unloadingPhotoUploadedAt, long aggregateVersion) {
requireText(customerId, "customerId");
public void publishContainerCompleted(String sourceContainerBookingId,
String containerNo, Instant unloadingCompletedAt,
Instant unloadingPhotoUploadedAt, long aggregateVersion) {
requireText(sourceContainerBookingId, "sourceContainerBookingId");
requireText(containerNo, "containerNo");
Objects.requireNonNull(unloadingCompletedAt, "unloadingCompletedAt");
@@ -71,7 +70,7 @@ public class InboundUnloadCompletedOutboxService {
String eventKey = EventTypes.OVERSEAS_CONTAINER_UNLOADED + ':' + sourceContainerBookingId;
LastlegEventOutbox existing = mapper.findByEventKey(eventKey);
if (existing != null) {
requireSameContainer(existing, customerId, sourceContainerBookingId, containerNo);
requireSameContainer(existing, sourceContainerBookingId, containerNo);
return;
}
var payload = new OverseasContainerUnloadedV1(sourceContainerBookingId,
@@ -79,12 +78,12 @@ public class InboundUnloadCompletedOutboxService {
JsonNode payloadNode = objectMapper.valueToTree(payload);
String traceId = EventIdentifiers.newUuid();
EventEnvelope envelope = new EventEnvelope(EventIdentifiers.newUuid(), EventTypes.OVERSEAS_CONTAINER_UNLOADED,
1, Instant.now(), RegionCode.OVERSEAS, RegionCode.CN, customerId, "CONTAINER_BOOKING",
1, Instant.now(), RegionCode.OVERSEAS, RegionCode.CN, "0", "CONTAINER_BOOKING",
sourceContainerBookingId, Math.max(1, aggregateVersion), 1L, traceId, eventKey, null, payloadNode,
EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(payloadNode))).requireValid();
LastlegEventOutbox event = new LastlegEventOutbox();
event.setId(IdWorker.getId()); event.setCustomerId(customerId);
event.setId(IdWorker.getId()); event.setCustomerId("0");
event.setAggregateType(envelope.aggregateType()); event.setAggregateId(envelope.aggregateId());
event.setEventType(envelope.eventType()); event.setEventVersion(envelope.eventVersion());
event.setEventKey(eventKey); event.setPayloadJson(json(envelope)); event.setStatus("PENDING");
@@ -94,17 +93,17 @@ public class InboundUnloadCompletedOutboxService {
} catch (DuplicateKeyException exception) {
LastlegEventOutbox raced = mapper.findByEventKey(eventKey);
if (raced == null) throw exception;
requireSameContainer(raced, customerId, sourceContainerBookingId, containerNo);
requireSameContainer(raced, sourceContainerBookingId, containerNo);
}
}
private void requireSameContainer(LastlegEventOutbox existing, String customerId,
private void requireSameContainer(LastlegEventOutbox existing,
String sourceContainerBookingId, String containerNo) {
try {
EventEnvelope envelope = objectMapper.readValue(existing.getPayloadJson(), EventEnvelope.class);
OverseasContainerUnloadedV1 frozen = objectMapper.treeToValue(
envelope.payload(), OverseasContainerUnloadedV1.class);
if (!Objects.equals(existing.getCustomerId(), customerId)
if (!"0".equals(existing.getCustomerId())
|| !Objects.equals(existing.getAggregateId(), sourceContainerBookingId)
|| !Objects.equals(frozen.containerNo(), containerNo)) {
throw new IllegalStateException("Unload-completed event key conflicts with another container");
@@ -59,6 +59,7 @@ import java.util.stream.Collectors;
@Service
public class ReceivingContainerService {
private static final String PLATFORM_SCOPE = "0";
private static final String PHOTO_DOWNLOAD_PATH =
"/api/pda/v1/last-leg/receiving-containers/current/unload-photos/files/{fileId}/download";
private static final String PALLET_MERGE_HINT =
@@ -125,10 +126,9 @@ public class ReceivingContainerService {
}
public ReceivingContainerResponses.Session detail(String operatorKey, long overseasWarehouseId, String containerNo) {
requireReadyContainer(containerNo, overseasWarehouseId, false);
List<InboundPlan> containerPlans = requirePlans(containerNo);
String customerId = requireSingleCustomer(containerPlans, containerNo);
if (!warehouseEligible(containerPlans, overseasWarehouseId)) throw conflict("当前柜已分配至其他海外接柜仓,无法在本仓操作");
requireReceivingAvailable(customerId, containerNo);
ReceivingContainerSession current = sessions.findActiveByOperator(activeOperatorKey(operatorKey));
requireSessionWarehouse(current, overseasWarehouseId);
if (current != null && !containerNo.equals(current.getContainerNo())) {
@@ -141,7 +141,7 @@ public class ReceivingContainerService {
WarehouseResourceGateway.WarehouseReference receivingWarehouse =
warehouse.requireActiveSharedWarehouse(overseasWarehouseId);
return snapshot(current != null && containerNo.equals(current.getContainerNo()) ? current : waitingView(
customerId, operatorKey, receivingWarehouse, containerNo), containerPlans);
operatorKey, receivingWarehouse, containerNo), containerPlans);
}
@Transactional
@@ -158,9 +158,7 @@ public class ReceivingContainerService {
return result;
}
List<InboundPlan> containerPlans = requirePlans(containerNo);
String customerId = requireSingleCustomer(containerPlans, containerNo);
if (!warehouseEligible(containerPlans, overseasWarehouseId)) throw conflict("当前柜已分配至其他海外接柜仓,无法在本仓操作");
requireReceivingAvailable(customerId, containerNo);
String activeOperatorKey = activeOperatorKey(operatorKey);
ReceivingContainerSession active = sessions.lockActiveByOperator(activeOperatorKey);
requireSessionWarehouse(active, overseasWarehouseId);
@@ -179,7 +177,7 @@ public class ReceivingContainerService {
}
LocalDateTime now = LocalDateTime.now();
ReceivingContainerSession session = new ReceivingContainerSession();
session.setId(IdWorker.getId()); session.setCustomerId(customerId); session.setContainerNo(containerNo);
session.setId(IdWorker.getId()); session.setContainerNo(containerNo);
session.setSourceContainerBookingId(readiness.sourceContainerBookingId());
session.setContainerFactHash(readiness.factHash());
session.setOverseasWarehouseId(overseasWarehouseId);
@@ -197,9 +195,8 @@ public class ReceivingContainerService {
public ReceivingContainerResponses.Photo uploadPhoto(String operatorKey, long operatorId, MultipartFile file) {
ReceivingContainerSession session = requireCurrent(operatorKey, true);
String customerId = session.getCustomerId();
if (!"WAIT_SIGN".equals(session.getStatus())) throw conflict("Unload photos can only be uploaded before signing");
UnloadPhotoFileGateway.UploadedFile uploaded = unloadPhotos.upload(customerId, session.getId(), operatorId, file);
UnloadPhotoFileGateway.UploadedFile uploaded = unloadPhotos.upload(PLATFORM_SCOPE, session.getId(), operatorId, file);
return new ReceivingContainerResponses.Photo(Long.toString(uploaded.fileId()), uploaded.fileName(), uploaded.contentType(), uploaded.sizeBytes());
}
@@ -209,24 +206,24 @@ public class ReceivingContainerService {
ReceivingContainerOperation replay = replay(operatorKey, "SIGN", request.idempotencyKey(), request);
if (replay != null) return read(replay, ReceivingContainerResponses.Session.class);
ReceivingContainerSession session = requireCurrent(operatorKey, true);
String customerId = session.getCustomerId();
if (!"WAIT_SIGN".equals(session.getStatus())) throw conflict("Container has already been signed");
List<ReceivingContainerResponses.SignedPhoto> photos = validateSignPhotos(request);
List<Long> photoFileIds = photos.stream().map(photo -> positiveFileId(photo.fileId())).toList();
photoFileIds.forEach(fileId -> unloadPhotos.validate(customerId, session.getId(), fileId));
photoFileIds.forEach(fileId -> unloadPhotos.validate(PLATFORM_SCOPE, session.getId(), fileId));
LocalDateTime now = LocalDateTime.now();
session.setSignPhotos(write(photos)); session.setStatus("SIGNED"); session.setSignedAt(now);
session.setOperatorName(operatorName); session.setUpdatedAt(now); update(session);
for (InboundPlan plan : requirePlans(customerId, session.getContainerNo())) {
workflow.ensureSignedTask(customerId, plan, operatorKey, operatorName, photoFileIds, now);
List<InboundPlan> containerPlans = requirePlans(session.getContainerNo());
for (InboundPlan plan : containerPlans) {
workflow.ensureSignedTask(plan.getCustomerId(), plan, operatorKey, operatorName, photoFileIds, now);
}
ReceivingContainerResponses.Session result = snapshot(session, requirePlans(customerId, session.getContainerNo()));
ReceivingContainerResponses.Session result = snapshot(session, containerPlans);
save(session, "SIGN", request.idempotencyKey(), request, result); return result;
}
public ReceivingContainerResponses.Session current(String operatorKey) {
ReceivingContainerSession session = requireCurrent(operatorKey, false);
return snapshot(session, requirePlans(session.getCustomerId(), session.getContainerNo()));
return snapshot(session, requirePlans(session.getContainerNo()));
}
public UnloadPhotoFileGateway.DownloadedFile downloadCurrentPhoto(String operatorKey, String fileId) {
@@ -236,7 +233,7 @@ public class ReceivingContainerService {
throw new ResponseStatusException(HttpStatus.FORBIDDEN,
"Unload photo is not part of the current receiving session");
}
return unloadPhotos.download(session.getCustomerId(), session.getId(), parsedFileId);
return unloadPhotos.download(PLATFORM_SCOPE, session.getId(), parsedFileId);
}
public void requireCurrentWarehouse(String operatorKey, long overseasWarehouseId) {
@@ -246,9 +243,8 @@ public class ReceivingContainerService {
public ReceivingContainerResponses.CartonResolved resolveCarton(String operatorKey,
ReceivingContainerRequests.CartonResolve request) {
ReceivingContainerSession session = requireSigned(operatorKey);
String customerId = session.getCustomerId();
String inboundCode = cartonInboundCode(request.cartonMark());
List<InboundRef> matches = refs(customerId, session.getContainerNo()).stream()
List<InboundRef> matches = refs(session.getContainerNo()).stream()
.filter(value -> inboundCode.equals(value.line().getInboundCode())).toList();
CartonRef carton;
if (matches.size() > 1) {
@@ -256,9 +252,9 @@ public class ReceivingContainerService {
} else if (matches.size() == 1 && "FBA".equals(destinationType(matches.get(0).line(), matches.get(0).order()))) {
carton = new CartonRef(matches.get(0));
} else if (matches.size() == 1) {
carton = requireCartonRef(customerId, session.getContainerNo(), request.cartonMark());
carton = requireCartonRef(matches.get(0).order().getCustomerId(), session.getContainerNo(), request.cartonMark());
} else {
carton = requireUnmatchedFbaCartonPrefix(customerId, inboundCode);
carton = requireUnmatchedFbaCartonPrefix(inboundCode);
}
InboundRef ref = carton.ref();
ReceivingContainerResponses.Inbound inbound = inbound(ref);
@@ -273,8 +269,7 @@ public class ReceivingContainerService {
public ReceivingContainerResponses.PalletResolved resolvePallet(String operatorKey,
ReceivingContainerRequests.PalletResolve request) {
ReceivingContainerSession session = requireSigned(operatorKey);
String customerId = session.getCustomerId();
InboundRef ref = requireInbound(customerId, session.getContainerNo(), request.inboundCode());
InboundRef ref = requireInbound(session.getContainerNo(), request.inboundCode());
PalletSelection selection = resolveReceivingPallet(ref, request.palletCode(), false);
WarehouseResourceGateway.WarehousePalletReference pallet = selection.resource();
return new ReceivingContainerResponses.PalletResolved(pallet.palletNo(), pallet.warehouseId(), pallet.status(),
@@ -287,15 +282,15 @@ public class ReceivingContainerService {
ReceivingContainerOperation replay = replay(operatorKey, "BULK_RECEIVE", request.idempotencyKey(), request);
if (replay != null) return read(replay, ReceivingContainerResponses.Receipt.class);
ReceivingContainerSession session = requireSigned(operatorKey);
String customerId = session.getCustomerId();
InboundRef ref = requireInbound(customerId, session.getContainerNo(), request.inboundCode());
InboundRef ref = requireInbound(session.getContainerNo(), request.inboundCode());
String customerId = ref.order().getCustomerId();
if ("ONE_CLICK_DROPSHIP".equals(ref.order().getOrderType())) {
throw conflict("One-click dropship inbound must be operated by complete source pallet");
}
if (!"FBA".equals(destinationType(ref.line(), ref.order()))) {
throw conflict("OVERSEAS destination must use carton-by-carton scanning; bulk count is forbidden");
}
List<InboundRef> fbaPlanRefs = requireFbaPlanRefs(customerId, session.getContainerNo(), ref);
List<InboundRef> fbaPlanRefs = requireFbaPlanRefs(session.getContainerNo(), ref);
List<FbaSkuAllocation> skuAllocations = allocateFbaSkuReceipts(customerId, fbaPlanRefs, request.packageCount());
skuAllocations.forEach(allocation -> workflow.validateBulkResolved(customerId, allocation.ref().plan(),
allocation.ref().line(), allocation.packages()));
@@ -331,12 +326,14 @@ public class ReceivingContainerService {
ReceivingContainerOperation replay = replay(operatorKey, "CARTON_RECEIVE", request.idempotencyKey(), request);
if (replay != null) return read(replay, ReceivingContainerResponses.Receipt.class);
ReceivingContainerSession session = requireSigned(operatorKey);
String customerId = session.getCustomerId();
List<InboundRef> currentRefs = refs(customerId, session.getContainerNo());
List<InboundRef> currentRefs = refs(session.getContainerNo());
if (currentRefs.stream().noneMatch(value ->
"OVERSEAS".equals(destinationType(value.line(), value.order())))) {
throw conflict("FBA destination must use bulk count");
}
String inboundCode = cartonInboundCode(request.cartonMark());
InboundRef inboundRef = requireInbound(session.getContainerNo(), inboundCode);
String customerId = inboundRef.order().getCustomerId();
CartonRef carton = requireCartonRef(customerId, session.getContainerNo(), request.cartonMark());
InboundRef ref = carton.ref();
if (!"OVERSEAS".equals(destinationType(ref.line(), ref.order()))) {
@@ -370,10 +367,10 @@ public class ReceivingContainerService {
ReceivingContainerOperation replay = replay(operatorKey, "PALLET_RECEIVE", request.idempotencyKey(), request);
if (replay != null) return read(replay, ReceivingContainerResponses.Receipt.class);
ReceivingContainerSession session = requireSigned(operatorKey);
String customerId = session.getCustomerId();
List<DropshipPalletRef> matches = new ArrayList<>();
for (InboundRef ref : refs(customerId, session.getContainerNo())) {
for (InboundRef ref : refs(session.getContainerNo())) {
if (!"ONE_CLICK_DROPSHIP".equals(ref.order().getOrderType())) continue;
String customerId = ref.order().getCustomerId();
List<InboundPlanPallet> sourcePallets = pallets.selectList(new LambdaQueryWrapper<InboundPlanPallet>()
.eq(InboundPlanPallet::getCustomerId, customerId)
.eq(InboundPlanPallet::getPlanId, ref.plan().getId())
@@ -391,6 +388,7 @@ public class ReceivingContainerService {
if (matches.isEmpty()) throw notFound("Source pallet is not part of the current container");
if (matches.size() != 1) throw conflict("Source pallet is duplicated inside the current container");
DropshipPalletRef match = matches.get(0);
String customerId = match.ref().order().getCustomerId();
int accepted = value(match.allocation().getPlannedPackageCount());
workflow.pickWholePallet(customerId, match.ref().plan().getPlanNo(), match.allocation().getId(),
new PdaRequests.WholePalletOperation(request.idempotencyKey() + ":PICK", operatorKey, operatorName));
@@ -410,12 +408,11 @@ public class ReceivingContainerService {
ReceivingContainerOperation replay = replay(operatorKey, "COMPLETE", request.idempotencyKey(), request);
if (replay != null) return read(replay, ReceivingContainerResponses.Session.class);
ReceivingContainerSession session = requireSigned(operatorKey);
String customerId = session.getCustomerId();
List<InboundPlan> containerPlans = requirePlans(customerId, session.getContainerNo());
List<InboundPlan> containerPlans = requirePlans(session.getContainerNo());
Totals totals = totals(containerPlans);
if (totals.total() != totals.received()) throw conflict("Every inbound code must be fully received before completing the container");
validateFbaSkuBalance(customerId, session.getContainerNo());
workflow.completeContainer(customerId, containerPlans);
validateFbaSkuBalance(session.getContainerNo());
workflow.completeContainer(containerPlans);
LocalDateTime now = LocalDateTime.now();
session.setStatus("COMPLETED"); session.setCompletedAt(now); session.setUpdatedAt(now);
session.setActiveOperatorKey(null); session.setActiveContainerKey(null); update(session);
@@ -489,7 +486,6 @@ public class ReceivingContainerService {
Set<String> warehouseNames = orderFacts.stream().map(LastlegOrder::getOverseasWarehouseName)
.map(ReceivingContainerService::trim).filter(Objects::nonNull).collect(Collectors.toSet());
boolean warehouseSnapshotConsistent = warehouseCodes.size() == 1 && warehouseNames.size() == 1;
Set<String> customers = orderFacts.stream().map(LastlegOrder::getCustomerId).collect(Collectors.toSet());
String containerNo = orderFacts.stream().map(LastlegOrder::getContainerNo).filter(Objects::nonNull).findFirst().orElse(null);
int totalOrders = orderFacts.size();
int inboundCodes = (int) orderFacts.stream().map(LastlegOrder::getInboundCode).map(ReceivingContainerService::trim)
@@ -499,14 +495,13 @@ public class ReceivingContainerService {
return values.size() == 1 && plansById.containsKey(values.get(0).getPlanId());
}).count();
int planCount = matchedPlans.size();
boolean allPlansReady = totalOrders > 0 && customers.size() == 1 && inboundCodes == totalOrders
boolean allPlansReady = totalOrders > 0 && inboundCodes == totalOrders
&& plannedOrders == totalOrders && planCount == totalOrders;
List<InboundPlan> containerPlans = planFacts.stream()
.filter(plan -> containerNo.equals(plan.getContainerNo())).toList();
boolean allPlansCompleted = !containerPlans.isEmpty()
&& containerPlans.stream().allMatch(plan -> "COMPLETED".equals(plan.getStatus()));
String customerId = customers.size() == 1 ? customers.iterator().next() : null;
boolean shippingReady = customerId != null && shippingGates.receivingAvailable(customerId, containerNo) == 1;
boolean shippingReady = shippingGates.receivingAvailable(sourceContainerBookingId, containerNo) == 1;
boolean operable = allPlansReady && shippingReady && warehouseSnapshotConsistent;
int pending = Math.max(0, totalOrders - plannedOrders);
String blockReason = operable ? null : (!warehouseSnapshotConsistent
@@ -564,42 +559,28 @@ public class ReceivingContainerService {
List<InboundPlan> values = plans.selectList(new LambdaQueryWrapper<InboundPlan>()
.eq(InboundPlan::getContainerNo, containerNo).orderByAsc(InboundPlan::getId));
if (values.isEmpty()) throw notFound("Receiving container was not found");
requireSingleCustomer(values, containerNo);
if (values.stream().anyMatch(value -> trim(value.getCustomerId()) == null)) {
throw conflict("Every inbound plan in the receiving container must have an owning customer");
}
if (values.stream().allMatch(value -> "COMPLETED".equals(value.getStatus()))) {
throw conflict("Receiving container is already completed");
}
return values;
}
private List<InboundPlan> requirePlans(String customerId, String containerNo) {
List<InboundPlan> values = plansForContainer(customerId, containerNo);
if (values.isEmpty()) throw notFound("Receiving container was not found");
if (values.stream().allMatch(value -> "COMPLETED".equals(value.getStatus()))) {
throw conflict("Receiving container is already completed");
}
return values;
}
private List<InboundPlan> plansForContainer(String customerId, String containerNo) {
private List<InboundPlan> plansForContainer(String containerNo) {
List<InboundPlan> values = plans.selectList(new LambdaQueryWrapper<InboundPlan>()
.eq(InboundPlan::getCustomerId, customerId).eq(InboundPlan::getContainerNo, containerNo)
.eq(InboundPlan::getContainerNo, containerNo)
.orderByAsc(InboundPlan::getId));
if (values.isEmpty()) throw notFound("Receiving container was not found");
return values;
}
private String requireSingleCustomer(List<InboundPlan> values, String containerNo) {
Set<String> customers = values.stream().map(InboundPlan::getCustomerId).map(ReceivingContainerService::trim)
.filter(Objects::nonNull).collect(Collectors.toCollection(java.util.LinkedHashSet::new));
if (customers.size() != 1 || values.stream().anyMatch(value -> trim(value.getCustomerId()) == null)) {
throw conflict("Receiving container must resolve to exactly one customer: " + containerNo);
}
return customers.iterator().next();
}
private List<InboundRef> refs(String customerId, String containerNo) {
private List<InboundRef> refs(String containerNo) {
List<InboundRef> result = new ArrayList<>();
for (InboundPlan plan : plansForContainer(customerId, containerNo)) {
for (InboundPlan plan : plansForContainer(containerNo)) {
String customerId = trim(plan.getCustomerId());
if (customerId == null) throw conflict("Inbound plan has no owning customer");
for (InboundPlanLine line : lines.selectList(new LambdaQueryWrapper<InboundPlanLine>()
.eq(InboundPlanLine::getCustomerId, customerId).eq(InboundPlanLine::getPlanId, plan.getId())
.orderByAsc(InboundPlanLine::getId))) {
@@ -611,8 +592,8 @@ public class ReceivingContainerService {
return result;
}
private InboundRef requireInbound(String customerId, String containerNo, String inboundCode) {
List<InboundRef> matches = refs(customerId, containerNo).stream()
private InboundRef requireInbound(String containerNo, String inboundCode) {
List<InboundRef> matches = refs(containerNo).stream()
.filter(value -> inboundCode.equals(value.line().getInboundCode())).toList();
if (matches.isEmpty()) throw notFound("Inbound code is not part of the current container");
Set<Long> planIds = matches.stream().map(value -> value.plan().getId()).collect(Collectors.toSet());
@@ -624,7 +605,8 @@ public class ReceivingContainerService {
InboundPlanCarton carton = cartons.selectOne(new LambdaQueryWrapper<InboundPlanCarton>()
.eq(InboundPlanCarton::getCustomerId, customerId).eq(InboundPlanCarton::getCartonMark, cartonMark));
if (carton == null) throw notFound("Carton mark was not found");
List<InboundRef> matches = refs(customerId, containerNo).stream()
List<InboundRef> matches = refs(containerNo).stream()
.filter(value -> customerId.equals(value.order().getCustomerId()))
.filter(value -> carton.getLineId() == null || Objects.equals(value.line().getId(), carton.getLineId()))
.filter(value -> carton.getPlanId() == null || Objects.equals(value.plan().getId(), carton.getPlanId()))
.filter(value -> Objects.equals(value.line().getOrderId(), carton.getOrderId())).toList();
@@ -640,16 +622,15 @@ public class ReceivingContainerService {
return inboundCode;
}
private CartonRef requireUnmatchedFbaCartonPrefix(String customerId, String inboundCode) {
private CartonRef requireUnmatchedFbaCartonPrefix(String inboundCode) {
boolean exists = lines.selectList(new LambdaQueryWrapper<InboundPlanLine>()
.eq(InboundPlanLine::getCustomerId, customerId)
.eq(InboundPlanLine::getInboundCode, inboundCode)).stream().findAny().isPresent();
if (exists) throw conflict("Carton mark belongs to another container");
throw notFound("Carton mark was not found");
}
private List<InboundRef> requireFbaPlanRefs(String customerId, String containerNo, InboundRef selected) {
List<InboundRef> values = refs(customerId, containerNo).stream()
private List<InboundRef> requireFbaPlanRefs(String containerNo, InboundRef selected) {
List<InboundRef> values = refs(containerNo).stream()
.filter(value -> Objects.equals(value.plan().getId(), selected.plan().getId()))
.sorted(Comparator.comparing(value -> value.line().getId())).toList();
if (values.isEmpty() || values.stream().anyMatch(value ->
@@ -678,11 +659,12 @@ public class ReceivingContainerService {
}
private PalletSelection resolveReceivingPallet(InboundRef ref, String palletCode, boolean requirePlanPallet) {
String normalizedPalletCode = palletCode.trim();
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(value -> matchesPlannedPalletCode(value, normalizedPalletCode))
.filter(this::hasConfirmedReservation)
.toList();
if (reserved.size() > 1) {
@@ -694,11 +676,11 @@ public class ReceivingContainerService {
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,
resource = warehouse.resolveReservedPallet(pallet.getPalletNo(), ref.order().getCustomerId(), warehouseId,
pallet.getWarehousePalletId(), pallet.getWarehouseLocationId(), pallet.getWarehouseResourceLockId());
} else {
try {
resource = warehouse.resolveAvailablePallet(palletCode.trim());
resource = warehouse.resolveAvailablePallet(normalizedPalletCode);
} catch (IllegalArgumentException exception) {
throw notFound("Pallet was not found");
}
@@ -709,6 +691,11 @@ public class ReceivingContainerService {
return new PalletSelection(pallet, resource);
}
private boolean matchesPlannedPalletCode(InboundPlanPallet pallet, String scannedCode) {
return scannedCode.equals(trim(pallet.getPalletNo()))
|| scannedCode.equals(trim(pallet.getSourcePalletNo()));
}
private boolean hasConfirmedReservation(InboundPlanPallet pallet) {
return pallet.getWarehousePalletId() != null && pallet.getWarehousePalletId() > 0
&& pallet.getWarehouseLocationId() != null && pallet.getWarehouseLocationId() > 0
@@ -752,7 +739,7 @@ public class ReceivingContainerService {
InboundPlanPalletAllocation allocation) { }
private ReceivingContainerResponses.Session snapshot(ReceivingContainerSession session, List<InboundPlan> containerPlans) {
List<ReceivingContainerResponses.Inbound> inbounds = refs(session.getCustomerId(), session.getContainerNo()).stream()
List<ReceivingContainerResponses.Inbound> inbounds = refs(session.getContainerNo()).stream()
.map(this::inbound).toList();
int total = inbounds.stream().mapToInt(ReceivingContainerResponses.Inbound::totalPackages).sum();
int received = inbounds.stream().mapToInt(ReceivingContainerResponses.Inbound::receivedPackages).sum();
@@ -796,8 +783,8 @@ public class ReceivingContainerService {
}
private Totals totals(List<InboundPlan> containerPlans) {
List<ReceivingContainerResponses.Inbound> values = refs(containerPlans.get(0).getCustomerId(),
containerPlans.get(0).getContainerNo()).stream().map(this::inbound).toList();
List<ReceivingContainerResponses.Inbound> values = refs(containerPlans.get(0).getContainerNo())
.stream().map(this::inbound).toList();
List<String> destinationCodes = values.stream().map(this::destinationCode)
.filter(Objects::nonNull).distinct().sorted().toList();
return new Totals(values.size(), destinationCodes,
@@ -903,14 +890,15 @@ public class ReceivingContainerService {
}
private void recordSkuReceipt(ReceivingContainerSession session, FbaSkuAllocation sku, InboundPlanPallet pallet) {
List<ReceivingSkuReceipt> existing = skuReceipts.lockLineSku(session.getCustomerId(),
String customerId = sku.ref().order().getCustomerId();
List<ReceivingSkuReceipt> existing = skuReceipts.lockLineSku(customerId,
sku.ref().line().getId(), sku.skuCode());
ReceivingSkuReceipt receipt = existing.stream()
.filter(value -> Objects.equals(value.getPalletId(), pallet.getId())).findFirst().orElse(null);
LocalDateTime now = LocalDateTime.now();
if (receipt == null) {
receipt = new ReceivingSkuReceipt(); receipt.setId(IdWorker.getId());
receipt.setCustomerId(session.getCustomerId()); receipt.setSessionId(session.getId());
receipt.setCustomerId(customerId); receipt.setSessionId(session.getId());
receipt.setPlanId(sku.ref().plan().getId()); receipt.setLineId(sku.ref().line().getId());
receipt.setPalletId(pallet.getId()); receipt.setOrderId(sku.ref().order().getId());
receipt.setInboundCode(sku.ref().line().getInboundCode()); receipt.setSkuCode(sku.skuCode());
@@ -924,10 +912,11 @@ public class ReceivingContainerService {
}
}
private void validateFbaSkuBalance(String customerId, String containerNo) {
for (InboundRef ref : refs(customerId, containerNo)) {
private void validateFbaSkuBalance(String containerNo) {
for (InboundRef ref : refs(containerNo)) {
if (!"FBA".equals(destinationType(ref.line(), ref.order()))) continue;
if ("ONE_CLICK_DROPSHIP".equals(ref.order().getOrderType())) continue;
String customerId = ref.order().getCustomerId();
List<SkuExpectation> skuCodes = skuExpectations(ref);
int received = 0;
for (SkuExpectation sku : skuCodes) {
@@ -962,10 +951,10 @@ public class ReceivingContainerService {
if (sessions.updateById(session) != 1) throw conflict("Receiving container session changed concurrently");
}
private ReceivingContainerSession waitingView(String customerId, String operatorKey,
private ReceivingContainerSession waitingView(String operatorKey,
WarehouseResourceGateway.WarehouseReference warehouse,
String containerNo) {
ReceivingContainerSession value = new ReceivingContainerSession(); value.setCustomerId(customerId);
ReceivingContainerSession value = new ReceivingContainerSession();
value.setOverseasWarehouseId(warehouse.warehouseId());
value.setOverseasWarehouseCode(warehouse.warehouseCode());
value.setOverseasWarehouseName(warehouse.warehouseName());
@@ -973,12 +962,6 @@ public class ReceivingContainerService {
return value;
}
private boolean eligible(List<InboundPlan> containerPlans, long overseasWarehouseId) {
String customerId = requireSingleCustomer(containerPlans, containerPlans.get(0).getContainerNo());
return warehouseEligible(containerPlans, overseasWarehouseId)
&& shippingGates.receivingAvailable(customerId, containerPlans.get(0).getContainerNo()) == 1;
}
private boolean warehouseEligible(List<InboundPlan> containerPlans, long overseasWarehouseId) {
Set<Long> assigned = containerPlans.stream().map(InboundPlan::getOverseasWarehouseId)
.filter(Objects::nonNull).collect(Collectors.toSet());
@@ -1004,12 +987,6 @@ public class ReceivingContainerService {
}
}
private void requireReceivingAvailable(String customerId, String containerNo) {
if (shippingGates.receivingAvailable(customerId, containerNo) != 1) {
throw conflict("Container has not reached PICKUP_READY and is not available for overseas receiving");
}
}
private void requireSessionWarehouse(ReceivingContainerSession session, long overseasWarehouseId) {
if (session != null && !Objects.equals(session.getOverseasWarehouseId(), overseasWarehouseId)) {
throw conflict("Current container belongs to another overseas warehouse");
@@ -1029,7 +1006,7 @@ public class ReceivingContainerService {
private void save(ReceivingContainerSession session, String action, String key, Object request, Object result) {
key = idempotencyKey(key);
ReceivingContainerOperation value = new ReceivingContainerOperation(); value.setId(IdWorker.getId());
value.setSessionId(session.getId()); value.setCustomerId(session.getCustomerId());
value.setSessionId(session.getId());
value.setOperatorKey(session.getOperatorKey()); value.setAction(action); value.setIdempotencyKey(key);
value.setRequestHash(hash(request)); value.setRequestJson(write(request)); value.setResultJson(write(result));
value.setCreatedAt(LocalDateTime.now()); operations.insert(value);
@@ -117,7 +117,7 @@ public class LoadingCompletedConsumer {
private void freezeContainerIdentity(EventEnvelope envelope, String sourceContainerBookingId,
String containerNo) {
shippingGates.initialize(sourceContainerBookingId, envelope.tenantId(), "",
shippingGates.initialize(sourceContainerBookingId, "",
containerNo, 0, envelope.handoverVersion());
ContainerShippingGate frozen = shippingGates.lock(sourceContainerBookingId);
if (frozen == null) {
@@ -12,7 +12,6 @@ import java.time.LocalDateTime;
public class ContainerShippingGate {
@TableId(type = IdType.INPUT)
private String containerBookingId;
private String customerId;
private String oceanBookingId;
private String containerNo;
private String status;
@@ -12,18 +12,17 @@ public interface ContainerShippingGateMapper extends BaseMapper<ContainerShippin
@Select("SELECT * FROM ll_container_shipping_gate WHERE container_booking_id=#{containerBookingId} FOR UPDATE")
ContainerShippingGate lock(@Param("containerBookingId") String containerBookingId);
@Insert("INSERT IGNORE INTO ll_container_shipping_gate(container_booking_id,customer_id,ocean_booking_id," +
@Insert("INSERT IGNORE INTO ll_container_shipping_gate(container_booking_id,ocean_booking_id," +
"container_no,status,node_sequence,aggregate_version,handover_version,created_at,updated_at) VALUES(" +
"#{containerBookingId},#{customerId},#{oceanBookingId},#{containerNo},'AWAITING_PICKUP_READY',0," +
"#{containerBookingId},#{oceanBookingId},#{containerNo},'AWAITING_PICKUP_READY',0," +
"#{aggregateVersion},#{handoverVersion},NOW(3),NOW(3))")
int initialize(@Param("containerBookingId") String containerBookingId, @Param("customerId") String customerId,
int initialize(@Param("containerBookingId") String containerBookingId,
@Param("oceanBookingId") String oceanBookingId, @Param("containerNo") String containerNo,
@Param("aggregateVersion") long aggregateVersion, @Param("handoverVersion") long handoverVersion);
@Select("SELECT CASE WHEN COUNT(DISTINCT o.source_container_booking_id)=1 " +
"AND MIN(CASE WHEN g.status='RECEIVING_AVAILABLE' THEN 1 ELSE 0 END)=1 THEN 1 ELSE 0 END " +
"FROM ll_order o LEFT JOIN ll_container_shipping_gate g " +
"ON g.container_booking_id=o.source_container_booking_id AND g.customer_id=o.customer_id " +
"WHERE o.customer_id=#{customerId} AND o.container_no=#{containerNo}")
int receivingAvailable(@Param("customerId") String customerId, @Param("containerNo") String containerNo);
@Select("SELECT CASE WHEN COUNT(*)=1 AND MIN(CASE WHEN status='RECEIVING_AVAILABLE' THEN 1 ELSE 0 END)=1 " +
"THEN 1 ELSE 0 END FROM ll_container_shipping_gate " +
"WHERE container_booking_id=#{containerBookingId} AND container_no=#{containerNo}")
int receivingAvailable(@Param("containerBookingId") String containerBookingId,
@Param("containerNo") String containerNo);
}
@@ -49,11 +49,10 @@ public class ContainerShippingNodeConsumer {
return 1;
}
gates.initialize(payload.containerBookingId(), payload.customerId(), payload.oceanBookingId(),
gates.initialize(payload.containerBookingId(), payload.oceanBookingId(),
payload.containerNo(), 0, 1);
ContainerShippingGate gate = gates.lock(payload.containerBookingId());
if (gate == null || !payload.customerId().equals(gate.getCustomerId())
|| !payload.containerNo().equals(gate.getContainerNo())) {
if (gate == null || !payload.containerNo().equals(gate.getContainerNo())) {
throw new IllegalArgumentException("船节点柜号快照与装柜交接不一致");
}
apply(gate, envelope, payload);
@@ -108,9 +107,8 @@ public class ContainerShippingNodeConsumer {
}
private void validateEnvelopePayload(EventEnvelope envelope, FirstLegContainerShippingNodeRecordedV1 payload) {
if (!envelope.tenantId().equals(payload.customerId())
|| !envelope.aggregateId().equals(payload.containerBookingId())) {
throw new IllegalArgumentException("船节点交接客户或聚合标识不一致");
if (!"0".equals(envelope.tenantId()) || !envelope.aggregateId().equals(payload.containerBookingId())) {
throw new IllegalArgumentException("船节点交接平台作用域或聚合标识不一致");
}
}