feat: gate overseas receiving and harden phase4 workflows

2026-08-25 10:31:04 +08:00
parent cb78bc6b41
commit 52b9fc382f
93 changed files with 2779 additions and 201 deletions
@@ -0,0 +1,123 @@
+package com.baiyue.lastleg.shipping.service;
+
+import com.baiyue.common.event.EventEnvelope;
+import com.baiyue.common.event.EventTypes;
+import com.baiyue.common.event.FirstLegContainerShippingNodeRecordedV1;
+import com.baiyue.common.event.RegionCode;
+import com.baiyue.lastleg.entity.LastlegEventInbox;
+import com.baiyue.lastleg.mapper.LastlegEventInboxMapper;
+import com.baiyue.lastleg.shipping.entity.ContainerShippingGate;
+import com.baiyue.lastleg.shipping.mapper.ContainerShippingGateMapper;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import jakarta.annotation.Resource;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Isolation;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+
+@Service
+public class ContainerShippingNodeConsumer {
+    public static final int PICKUP_READY_SEQUENCE = 80;
+    @Resource private LastlegEventInboxMapper inboxes;
+    @Resource private ContainerShippingGateMapper gates;
+    @Resource private ObjectMapper objectMapper;
+
+    @Transactional(isolation = Isolation.READ_COMMITTED)
+    public int consume(EventEnvelope envelope) {
+        validate(envelope);
+        FirstLegContainerShippingNodeRecordedV1 payload = payload(envelope);
+        validateEnvelopePayload(envelope, payload);
+        LastlegEventInbox existing = inboxes.selectById(envelope.eventId());
+        if (existing != null) {
+            if (!existing.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) {
+                throw new IllegalArgumentException("船节点事件摘要冲突");
+            }
+            return 1;
+        }
+        LastlegEventInbox inbox = inbox(envelope);
+        try { inboxes.insert(inbox); }
+        catch (DuplicateKeyException raced) {
+            LastlegEventInbox concurrent = inboxes.selectById(envelope.eventId());
+            if (concurrent == null || !concurrent.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) {
+                throw new IllegalArgumentException("船节点事件摘要冲突", raced);
+            }
+            return 1;
+        }
+
+        gates.initialize(payload.containerBookingId(), payload.customerId(), 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())) {
+            throw new IllegalArgumentException("船节点柜号快照与装柜交接不一致");
+        }
+        apply(gate, envelope, payload);
+        gates.updateById(gate);
+        inbox.setStatus("PROCESSED"); inbox.setProcessedAt(LocalDateTime.now()); inboxes.updateById(inbox);
+        return 1;
+    }
+
+    private void apply(ContainerShippingGate gate, EventEnvelope envelope,
+                       FirstLegContainerShippingNodeRecordedV1 payload) {
+        long currentVersion = gate.getAggregateVersion() == null ? 0 : gate.getAggregateVersion();
+        int currentSequence = gate.getNodeSequence() == null ? 0 : gate.getNodeSequence();
+        LocalDateTime actualAt = LocalDateTime.ofInstant(payload.actualAt(), ZoneOffset.UTC);
+        boolean newerVersion = envelope.aggregateVersion() >= currentVersion;
+        boolean newerNode = payload.nodeSequence() > currentSequence;
+        boolean laterSameNode = payload.nodeSequence() == currentSequence
+            && (gate.getActualAt() == null || actualAt.isAfter(gate.getActualAt()));
+        boolean accepted = newerVersion && (newerNode || laterSameNode);
+        if (accepted) {
+            gate.setOceanBookingId(payload.oceanBookingId());
+            gate.setStandardNode(payload.standardNode()); gate.setNodeSequence(payload.nodeSequence());
+            gate.setObservedAt(LocalDateTime.ofInstant(payload.observedAt(), ZoneOffset.UTC)); gate.setActualAt(actualAt);
+            gate.setLocationCode(payload.locationCode()); gate.setLocationName(payload.locationName());
+            gate.setProviderCode(payload.providerCode()); gate.setProviderEventId(payload.providerEventId());
+            gate.setProviderNodeCode(payload.providerNodeCode()); gate.setSourceReference(payload.sourceReference());
+            gate.setRawPayloadHash(payload.rawPayloadHash()); gate.setLastEventId(envelope.eventId());
+            gate.setPayloadHash(envelope.payloadHash());
+        }
+        if (accepted && payload.nodeSequence() >= PICKUP_READY_SEQUENCE) {
+            gate.setStatus("RECEIVING_AVAILABLE");
+        }
+        gate.setAggregateVersion(Math.max(currentVersion, envelope.aggregateVersion()));
+        gate.setHandoverVersion(Math.max(gate.getHandoverVersion() == null ? 0 : gate.getHandoverVersion(),
+            envelope.handoverVersion()));
+        gate.setUpdatedAt(LocalDateTime.now());
+    }
+
+    private void validate(EventEnvelope envelope) {
+        envelope.requireValid();
+        if (!EventTypes.FIRST_LEG_CONTAINER_SHIPPING_NODE_RECORDED.equals(envelope.eventType())
+            || envelope.eventVersion() != 1 || envelope.sourceRegion() != RegionCode.CN
+            || envelope.targetRegion() != RegionCode.OVERSEAS
+            || !"CONTAINER_BOOKING".equals(envelope.aggregateType())
+            || envelope.handoverVersion() == null
+            || envelope.handoverVersion() != envelope.aggregateVersion() + 1) {
+            throw new IllegalArgumentException("船节点交接信封无效");
+        }
+    }
+
+    private void validateEnvelopePayload(EventEnvelope envelope, FirstLegContainerShippingNodeRecordedV1 payload) {
+        if (!envelope.tenantId().equals(payload.customerId())
+            || !envelope.aggregateId().equals(payload.containerBookingId())) {
+            throw new IllegalArgumentException("船节点交接客户或聚合标识不一致");
+        }
+    }
+
+    private FirstLegContainerShippingNodeRecordedV1 payload(EventEnvelope envelope) {
+        try { return objectMapper.treeToValue(envelope.payload(), FirstLegContainerShippingNodeRecordedV1.class); }
+        catch (Exception exception) { throw new IllegalArgumentException("船节点交接负载无效", exception); }
+    }
+
+    private LastlegEventInbox inbox(EventEnvelope envelope) {
+        LastlegEventInbox value = new LastlegEventInbox(); value.setEventId(envelope.eventId());
+        value.setCustomerId(envelope.tenantId()); value.setEventType(envelope.eventType());
+        value.setEventVersion(envelope.eventVersion()); value.setPayloadHash(envelope.payloadHash());
+        value.setPayloadJson(envelope.payload().toString()); value.setStatus("RECEIVED");
+        value.setReceivedAt(LocalDateTime.now()); return value;
+    }
+}