package com.baiyue.lastleg.service; import com.baiyue.common.event.*; import com.baiyue.lastleg.entity.*; import com.baiyue.lastleg.mapper.*; import com.baiyue.lastleg.dropship.dto.DropshipRequests; import com.baiyue.lastleg.dropship.service.DropshipOrderService; import com.baiyue.lastleg.inboundplan.entity.InboundPlanCarton; import com.baiyue.lastleg.inboundplan.mapper.InboundPlanCartonMapper; import com.baiyue.lastleg.inventory.command.dto.InventoryCommandRequests; import com.baiyue.lastleg.inventory.command.service.InventoryCommandService; import com.baiyue.lastleg.shipping.mapper.ContainerShippingGateMapper; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.DeserializationFeature; import jakarta.annotation.Resource; import org.springframework.dao.DuplicateKeyException; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Isolation; import com.baomidou.mybatisplus.core.toolkit.IdWorker; import java.math.BigDecimal; import java.time.LocalDateTime; import java.util.Objects; import java.util.ArrayList; import java.util.HashSet; import java.util.List; import java.util.Set; import com.fasterxml.jackson.databind.JsonNode; @Service public class LoadingCompletedConsumer { @Resource private LastlegEventInboxMapper inboxMapper; @Resource private LastlegOrderMapper orderMapper; @Resource private LastlegInboundOrderMapper inboundOrderMapper; @Resource private LastlegOrderDetailMapper detailMapper; @Resource private LastlegOrderStatusHistoryMapper historyMapper; @Resource private InboundPlanCartonMapper cartonMapper; @Resource private ObjectMapper objectMapper; @Resource private LastlegOrderNoGenerator orderNoGenerator; @Resource private DropshipOrderService dropshipOrders; @Resource private InventoryCommandService inventory; @Resource private ContainerShippingGateMapper shippingGates; @Transactional(isolation = Isolation.READ_COMMITTED) public int consume(EventEnvelope envelope) { envelope.requireValid(); if (!EventTypes.FIRST_LEG_LOADING_PLAN_COMPLETED.equals(envelope.eventType()) || (envelope.eventVersion() != 1 && envelope.eventVersion() != 2)) throw new IllegalArgumentException("不支持的尾程交接事件"); if (envelope.sourceRegion() != RegionCode.CN || envelope.targetRegion() != RegionCode.OVERSEAS) throw new IllegalArgumentException("尾程交接事件区域无效"); if (!"LOADING_PLAN".equals(envelope.aggregateType())) throw new IllegalArgumentException("尾程交接事件聚合类型无效"); if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(envelope.payload()), envelope.payloadHash())) throw new IllegalArgumentException("payloadHash 与尾程事件负载不一致"); FirstLegLoadingPlanCompletedV1 payload = parsePayload(envelope); if(payload.containerNo()==null||payload.containerNo().isBlank()) throw new IllegalArgumentException("事件没有实际集装箱号 containerNo"); if(payload.orders()==null||payload.orders().isEmpty()) throw new IllegalArgumentException("事件没有入仓单"); if (envelope.eventVersion() == 2) validateCartonSkuMappings(payload); LastlegEventInbox existing=inboxMapper.selectById(envelope.eventId()); if(existing!=null) { if(!existing.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) throw new IllegalArgumentException("事件摘要冲突"); return countOrders(envelope,payload.containerNo()); } LastlegEventInbox inbox=new LastlegEventInbox(); inbox.setEventId(envelope.eventId()); inbox.setCustomerId(envelope.tenantId()); inbox.setEventType(envelope.eventType()); inbox.setEventVersion(envelope.eventVersion()); inbox.setPayloadHash(envelope.payloadHash()); inbox.setPayloadJson(envelope.payload().toString()); inbox.setStatus("RECEIVED"); inbox.setReceivedAt(LocalDateTime.now()); try { inboxMapper.insert(inbox); } catch(DuplicateKeyException exception) { LastlegEventInbox raced = inboxMapper.selectById(envelope.eventId()); if (raced == null || raced.getPayloadHash() == null || !raced.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) { throw new IllegalArgumentException("事件摘要冲突", exception); } return countOrders(envelope,payload.containerNo()); } // The loading-plan and container-shipping events are independent aggregate streams. // Seeding the shipping version from the loading-plan version makes a valid first node // (aggregateVersion=0) look stale whenever the loading plan has a higher version. shippingGates.initialize(payload.containerBookingId(), envelope.tenantId(), "", payload.containerNo(), 0, envelope.handoverVersion()); String orderNo = orderNoGenerator.forContainerNo(payload.containerNo()); JsonNode rawOrders = envelope.payload().path("orders"); for (int index = 0; index < payload.orders().size(); index++) { var source = payload.orders().get(index); JsonNode rawSource = rawOrders.isArray() && index < rawOrders.size() ? rawOrders.get(index) : objectMapper.createObjectNode(); createOrder(envelope, payload, source, orderNo, rawSource); } inbox.setStatus("PROCESSED"); inbox.setProcessedAt(LocalDateTime.now()); inboxMapper.updateById(inbox); return payload.orders().size(); } private FirstLegLoadingPlanCompletedV1 parsePayload(EventEnvelope envelope) { try { ObjectMapper compatibleMapper = objectMapper.copy() .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); // V2 only adds immutable item remarks; the current tail-leg snapshot model // intentionally ignores those fields until its schema exposes them. return compatibleMapper.treeToValue(envelope.payload(), FirstLegLoadingPlanCompletedV1.class); } catch(Exception exception) { throw new IllegalArgumentException("尾程事件负载无效",exception); } } private int countOrders(EventEnvelope envelope,String containerNo) { return Math.toIntExact(orderMapper.countContainer(envelope.tenantId(),containerNo)); } private void validateCartonSkuMappings(FirstLegLoadingPlanCompletedV1 payload) { for (FirstLegLoadingPlanCompletedV1.Order order : payload.orders()) { Set orderCartonMarks = new HashSet<>(); if ("ONE_CLICK_DROPSHIP".equals(order.orderType())) { for (FirstLegLoadingPlanCompletedV1.PalletSnapshot pallet : order.pallets()) { for (FirstLegLoadingPlanCompletedV1.SkuPackageSnapshot sku : pallet.skuPackages()) { requireSkuCartonMapping(sku.skuCode(), sku.packageCount(), sku.cartonMarks(), orderCartonMarks); } } continue; } String destinationType = destinationType(payload, order); for (FirstLegLoadingPlanCompletedV1.Item item : order.items()) { if ("OVERSEAS".equals(destinationType)) { requireSkuCartonMapping(item.skuCode(), item.packageCount(), item.cartonMarks(), orderCartonMarks); } else { validateOptionalFbaCartonMapping(item.skuCode(), item.packageCount(), item.cartonMarks(), orderCartonMarks); } } } } private void validateOptionalFbaCartonMapping(String skuCode, int loadedPackages, List cartonMarks, Set orderCartonMarks) { List marks = cartonMarks == null ? List.of() : cartonMarks; if (skuCode == null || skuCode.isBlank()) { throw new IllegalArgumentException("V2 尾程交接事件缺少 SKU"); } if (marks.isEmpty()) return; requireSkuCartonMapping(skuCode, loadedPackages, marks, orderCartonMarks); } private void requireSkuCartonMapping(String skuCode, int loadedPackages, List cartonMarks, Set orderCartonMarks) { if (skuCode == null || skuCode.isBlank()) { throw new IllegalArgumentException("V2 尾程交接事件的箱唛映射缺少 SKU"); } List marks = cartonMarks == null ? List.of() : cartonMarks; if (marks.size() != loadedPackages) { throw new IllegalArgumentException("V2 尾程交接事件的 SKU 箱唛数量必须等于实装箱数: " + skuCode); } for (String mark : marks) { if (mark == null || mark.isBlank() || !orderCartonMarks.add(mark)) { throw new IllegalArgumentException("V2 尾程交接事件包含空值或重复箱唛: " + skuCode); } } } private void createOrder(EventEnvelope event,FirstLegLoadingPlanCompletedV1 payload, FirstLegLoadingPlanCompletedV1.Order source,String orderNo, JsonNode rawSource) { String orderHash=EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(rawSource)); LastlegOrder existing=orderMapper.findSource(event.tenantId(),event.sourceRegion().name(),source.sourceInboundId()); if(existing!=null) { if(!Objects.equals(existing.getSourcePayloadHash(),orderHash)) throw new IllegalStateException("来源入仓单快照冲突: "+source.sourceInboundId()); return; } LastlegOrder order=new LastlegOrder(); String orderType = source.orderType(); String destinationType = destinationType(payload, source); String customerId = source.customerId() == null || source.customerId().isBlank() ? event.tenantId() : source.customerId(); if (!event.tenantId().equals(customerId)) { throw new IllegalArgumentException("Order customerId does not match the trusted event customer"); } List dropshipPallets = "ONE_CLICK_DROPSHIP".equals(orderType) ? pallets(source, rawSource) : List.of(); if ("ONE_CLICK_DROPSHIP".equals(orderType) && dropshipPallets.isEmpty()) { throw new IllegalArgumentException("One-click dropship event has no pallet snapshot"); } order.setId(IdWorker.getId()); order.setOrderNo(orderNo); order.setCustomerId(customerId); order.setOrderType(orderType); order.setDestinationType(destinationType); order.setInboundCode(source.inboundCode()); order.setSourceRegion(event.sourceRegion().name()); order.setSourceInboundId(source.sourceInboundId()); order.setRootInboundId(source.rootInboundId()); order.setSourceOrderId(source.sourceOrderId()); order.setSourceOrderNo(source.sourceOrderNo()); order.setRootOrderId(source.rootOrderId()); order.setSourceWarehouseId(payload.warehouseId()); order.setSourceWarehouseName(payload.warehouseName()); order.setOverseasWarehouseId(null); order.setOverseasWarehouseName(null); order.setFbaWarehouseCode(source.destinationWarehouseCode()); order.setContainerNo(payload.containerNo()); order.setContainerStatus("待转柜"); order.setAbnormal(false); order.setDetained(false); order.setPackageCount(source.packageCount()); order.setPalletCount(dropshipPallets.size()); order.setProductQuantity(source.productQuantity()); order.setVolumeCbm(source.volumeCbm()); order.setVolumeWeightKg(source.items().stream().map(FirstLegLoadingPlanCompletedV1.Item::dimensionalWeightKg) .filter(Objects::nonNull).reduce(BigDecimal.ZERO,BigDecimal::add)); order.setCurrentStatus("PENDING_ASSIGNMENT"); order.setSourceEventId(event.eventId()); order.setSourceLoadingPlanId(payload.loadingPlanId()); order.setSourceContainerBookingId(payload.containerBookingId()); order.setSourceLoadingPlanVersion(event.aggregateVersion()); order.setSourceHandoverVersion(event.handoverVersion()); order.setSourcePayloadHash(orderHash); order.setCreatedAt(LocalDateTime.now()); order.setUpdatedAt(LocalDateTime.now()); try { orderMapper.insert(order); } catch (DuplicateKeyException exception) { LastlegOrder raced = orderMapper.findSource(event.tenantId(), event.sourceRegion().name(), source.sourceInboundId()); if (raced != null && Objects.equals(raced.getSourcePayloadHash(), orderHash)) return; throw exception; } createInboundOrderSnapshot(event, payload, source, order, orderHash); for(var item:source.items()) { LastlegOrderDetail detail=new LastlegOrderDetail(); detail.setCustomerId(customerId); detail.setOrderId(order.getId()); detail.setSourceInboundDetailId(item.sourceInboundDetailId()); detail.setSkuCode(item.skuCode()); detail.setProductNameZh(item.productNameZh()); detail.setProductNameEn(item.productNameEn()); detail.setPackageCount(item.packageCount()); detail.setProductQuantity(item.productQuantity()); detail.setPalletCount(item.palletCount()); detail.setGrossWeightKg(item.grossWeightKg()); detail.setNetWeightKg(item.netWeightKg()); detail.setDimensionalWeightKg(item.dimensionalWeightKg()); detail.setVolumeCbm(item.volumeCbm()); detail.setDestinationWarehouseId(item.destinationWarehouseId()); detail.setDestinationWarehouseCode(item.destinationWarehouseCode()); detail.setDestinationWarehouseName(item.destinationWarehouseName()); detail.setFbaTrackingNo(item.fbaTrackingNo()); detail.setShipmentNo(item.shipmentNo()); detail.setShippingMark(item.shippingMark()); detailMapper.insert(detail); createCartonLabels(customerId, order.getId(), destinationType, item); } LastlegOrderStatusHistory history=new LastlegOrderStatusHistory(); history.setCustomerId(customerId); history.setOrderId(order.getId()); history.setToStatus("PENDING_ASSIGNMENT"); history.setEventId(event.eventId()); history.setRemark("头程装柜完成自动生成"); history.setCreatedAt(LocalDateTime.now()); historyMapper.insert(history); if ("FIRST_LEG".equals(orderType)) { // The cross-region handover freezes the source inbound code only. An inbound plan is // created later by an explicit platform conversion, which replaces this display snapshot. registerInTransit(event, payload, source, order, source.inboundCode()); } if ("ONE_CLICK_DROPSHIP".equals(orderType)) { long warehouseId; try { warehouseId = Long.parseLong(payload.warehouseId()); } catch (Exception exception) { throw new IllegalArgumentException("One-click dropship warehouseId is invalid", exception); } dropshipOrders.receive(new DropshipRequests.Receive(customerId, order.getId(), warehouseId, event.eventId() + ":DROPSHIP", dropshipPallets)); } } private void createInboundOrderSnapshot(EventEnvelope event, FirstLegLoadingPlanCompletedV1 payload, FirstLegLoadingPlanCompletedV1.Order source, LastlegOrder order, String orderHash) { LastlegInboundOrder snapshot = new LastlegInboundOrder(); snapshot.setId(order.getId()); snapshot.setCustomerId(order.getCustomerId()); snapshot.setSourceRegion(event.sourceRegion().name()); snapshot.setSourceInboundId(source.sourceInboundId()); snapshot.setRootInboundId(source.rootInboundId()); snapshot.setSourceOrderId(source.sourceOrderId()); snapshot.setSourceOrderNo(source.sourceOrderNo()); snapshot.setRootOrderId(source.rootOrderId()); snapshot.setInboundCode(source.inboundCode()); snapshot.setDestinationWarehouseId(source.destinationWarehouseId()); snapshot.setDestinationWarehouseCode(source.destinationWarehouseCode()); snapshot.setDestinationWarehouseName(source.destinationWarehouseName()); snapshot.setDestinationType(order.getDestinationType()); snapshot.setPackageCount(source.packageCount()); snapshot.setProductQuantity(source.productQuantity()); snapshot.setPalletCount(source.palletCount()); snapshot.setGrossWeightKg(source.grossWeightKg()); snapshot.setNetWeightKg(source.netWeightKg()); snapshot.setVolumeCbm(source.volumeCbm()); snapshot.setSourceEventId(event.eventId()); snapshot.setSourceLoadingPlanId(payload.loadingPlanId()); snapshot.setSourceLoadingPlanVersion(event.aggregateVersion()); snapshot.setHandoverVersion(event.handoverVersion()); snapshot.setPayloadHash(orderHash); snapshot.setStatus("READY"); snapshot.setExceptionFlag("NONE"); snapshot.setCreatedAt(order.getCreatedAt()); snapshot.setUpdatedAt(order.getUpdatedAt()); inboundOrderMapper.insert(snapshot); } private void registerInTransit(EventEnvelope event, FirstLegLoadingPlanCompletedV1 payload, FirstLegLoadingPlanCompletedV1.Order source, LastlegOrder order, String inboundOrderNo) { for (int index = 0; index < source.items().size(); index++) { FirstLegLoadingPlanCompletedV1.Item item = source.items().get(index); if (item.packageCount() <= 0) continue; inventory.registerInTransit(order.getCustomerId(), "SYSTEM", new InventoryCommandRequests.InTransit(order.getCustomerName(), payload.warehouseId(), null, payload.warehouseName(), item.destinationWarehouseId(), item.destinationWarehouseCode(), item.destinationWarehouseName(), order.getId(), inboundOrderNo, source.inboundCode(), item.skuCode(), item.productNameZh(), item.productNameEn(), item.packageCount(), order.getOrderNo(), event.eventId(), "TAIL_IN_TRANSIT:" + event.eventId() + ":" + order.getId() + ":" + index, "Tail-leg order created from completed loading plan")); } } private void createCartonLabels(String customerId, Long orderId, String destinationType, FirstLegLoadingPlanCompletedV1.Item item) { if (!"OVERSEAS".equals(destinationType)) return; List marks = item.cartonMarks() == null ? List.of() : item.cartonMarks(); if (marks.isEmpty()) return; if (marks.stream().anyMatch(mark -> mark == null || mark.isBlank()) || marks.stream().distinct().count() != marks.size()) { throw new IllegalArgumentException("装柜事件的箱唛池不可包含空值或重复箱唛"); } for (String mark : marks) { InboundPlanCarton carton = new InboundPlanCarton(); carton.setId(IdWorker.getId()); carton.setCustomerId(customerId); carton.setOrderId(orderId); carton.setSourceInboundDetailId(item.sourceInboundDetailId()); carton.setSkuCode(item.skuCode()); carton.setCartonMark(mark); carton.setDestinationWarehouseId(item.destinationWarehouseId()); carton.setDestinationWarehouseCode(item.destinationWarehouseCode()); carton.setDestinationWarehouseName(item.destinationWarehouseName()); carton.setStatus("UNASSIGNED"); carton.setVersion(0); carton.setUpdatedAt(LocalDateTime.now()); try { cartonMapper.insert(carton); } catch (DuplicateKeyException ignored) { } } } private String destinationType(FirstLegLoadingPlanCompletedV1 payload, FirstLegLoadingPlanCompletedV1.Order order) { String explicit = normalizeDestinationType(order.destinationType()); if (explicit != null) return explicit; boolean fbaEvidence = order.items().stream().anyMatch(item -> hasText(item.fbaTrackingNo()) || hasText(item.shipmentNo()) || startsWithFba(item.destinationWarehouseName())); boolean overseasWarehouseIdentity = hasText(payload.warehouseId()) && Objects.equals(payload.warehouseId(), order.destinationWarehouseId()); if (fbaEvidence && overseasWarehouseIdentity) { throw new IllegalArgumentException("历史尾程事件目的类型证据冲突: " + order.sourceInboundId()); } if (fbaEvidence) return "FBA"; if (overseasWarehouseIdentity) return "OVERSEAS"; throw new IllegalArgumentException("历史尾程事件缺少 destinationType 且无法安全推断: " + order.sourceInboundId()); } private String normalizeDestinationType(String value) { if (!hasText(value)) return null; String normalized = value.trim().toUpperCase(java.util.Locale.ROOT); if (!"FBA".equals(normalized) && !"OVERSEAS".equals(normalized)) { throw new IllegalArgumentException("尾程事件 destinationType 必须为 FBA 或 OVERSEAS"); } return normalized; } private boolean startsWithFba(String value) { return hasText(value) && value.trim().toUpperCase(java.util.Locale.ROOT).startsWith("FBA"); } private boolean hasText(String value) { return value != null && !value.isBlank(); } private List pallets(FirstLegLoadingPlanCompletedV1.Order source, JsonNode rawSource) { if (source.pallets() != null && !source.pallets().isEmpty()) { return source.pallets().stream().map(pallet -> new DropshipRequests.Pallet(pallet.palletNo(), pallet.skuPackages().stream().map(sku -> new DropshipRequests.SkuPackages( sku.skuCode(), sku.productName(), sku.packageCount(), sku.productQuantity(), sku.cartonMarks())).toList())).toList(); } return legacyPallets(rawSource); } private List legacyPallets(JsonNode source) { List result = new ArrayList<>(); JsonNode nodes = source.path("pallets"); if (nodes.isArray()) { for (JsonNode pallet : nodes) { String palletNo = pallet.path("palletNo").asText(pallet.path("sourcePalletNo").asText(null)); List skuRows = skuPackages(pallet); if (palletNo != null && !palletNo.isBlank() && !skuRows.isEmpty()) { result.add(new DropshipRequests.Pallet(palletNo, skuRows)); } } return result; } String palletNo = source.path("palletNo").asText(source.path("sourcePalletNo").asText(null)); List skuRows = skuPackages(source); if (palletNo != null && !palletNo.isBlank() && !skuRows.isEmpty()) { result.add(new DropshipRequests.Pallet(palletNo, skuRows)); } return result; } private List skuPackages(JsonNode source) { List result = new ArrayList<>(); JsonNode nodes = source.path("skuPackages"); if (!nodes.isArray()) nodes = source.path("items"); if (!nodes.isArray()) return result; for (JsonNode item : nodes) { String sku = item.path("skuCode").asText(null); long packageCount = item.path("packageCount").asLong(0); long productQuantity = item.path("productQuantity").asLong(0); JsonNode markNodes = item.path("cartonMarks"); if (!markNodes.isArray()) markNodes = item.path("boxMarks"); if (!markNodes.isArray()) markNodes = item.path("cartonNos"); List marks = new ArrayList<>(); if (markNodes.isArray()) { for (JsonNode mark : markNodes) if (!mark.asText().isBlank()) marks.add(mark.asText()); } if (sku != null && !sku.isBlank() && packageCount > 0 && productQuantity > 0 && !marks.isEmpty()) { String productName = item.path("productName").asText(item.path("productNameZh").asText(null)); result.add(new DropshipRequests.SkuPackages(sku, productName, packageCount, productQuantity, marks)); } } return result; } }