baiyue-firstleg/lastleg-service/src/main/java/com/baiyue/lastleg/service/LoadingCompletedConsumer.java

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.inboundplan.service.InboundPlanService;
import com.baiyue.lastleg.inboundplan.dto.InboundPlanResponse;
import com.baiyue.lastleg.inventory.command.dto.InventoryCommandRequests;
import com.baiyue.lastleg.inventory.command.service.InventoryCommandService;
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 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 InboundPlanService inboundPlans;
    @Resource private ObjectMapper objectMapper;
    @Resource private LastlegOrderNoGenerator orderNoGenerator;
    @Resource private DropshipOrderService dropshipOrders;
    @Resource private InventoryCommandService inventory;

    @Transactional
    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("尾程交接事件聚合类型无效");
        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) {
            return countOrders(envelope,payload.containerNo());
        }

        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<String> 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;
            }
            for (FirstLegLoadingPlanCompletedV1.Item item : order.items()) {
                requireSkuCartonMapping(item.skuCode(), item.packageCount(), item.cartonMarks(), orderCartonMarks);
            }
        }
    }

    private void requireSkuCartonMapping(String skuCode, int loadedPackages, List<String> cartonMarks,
                                         Set<String> orderCartonMarks) {
        if (skuCode == null || skuCode.isBlank()) {
            throw new IllegalArgumentException("V2 尾程交接事件的箱唛映射缺少 SKU");
        }
        List<String> 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 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<DropshipRequests.Pallet> 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.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.setOverseasWarehouseId(payload.warehouseId()); order.setOverseasWarehouseName(payload.warehouseName());
        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(), 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)) {
            InboundPlanResponse plan = inboundPlans.convert(customerId, order.getId());
            registerInTransit(event, payload, source, order, plan.planNo());
        }
        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.setTenantId("BAIYUE");
        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.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, FirstLegLoadingPlanCompletedV1.Item item) {
        List<String> marks = item.cartonMarks() == null ? List.of() : item.cartonMarks();
        if (marks.isEmpty()) return; // 历史 V1 事件保持原有按箱数入库路径。
        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 List<DropshipRequests.Pallet> 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<DropshipRequests.Pallet> legacyPallets(JsonNode source) {
        List<DropshipRequests.Pallet> 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<DropshipRequests.SkuPackages> 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<DropshipRequests.SkuPackages> skuRows = skuPackages(source);
        if (palletNo != null && !palletNo.isBlank() && !skuRows.isEmpty()) {
            result.add(new DropshipRequests.Pallet(palletNo, skuRows));
        }
        return result;
    }

    private List<DropshipRequests.SkuPackages> skuPackages(JsonNode source) {
        List<DropshipRequests.SkuPackages> 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<String> 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;
    }
}