fix(firstleg): compare stable carrier facts separately from response audit hashes

2026-10-03 00:39:57 +08:00
parent 9803dcfd2c
commit 4cd51a4955
9 changed files with 266 additions and 23 deletions
@@ -575,8 +575,8 @@ public class CarrierScheduleQueryService {
        fact.setSourceReference(trim(sourceReference)); fact.setRawPayloadHash(rawPayloadHash);
        if (repository.insertIfAbsent(fact)) return fact;
        CarrierScheduleNodeFactEntity existing = repository.nodeByContainer(containerBookingId, fact.getEventKey());
-        if (existing != null && !Objects.equals(existing.getRawPayloadHash(), rawPayloadHash)) {
-            throw conflict("相同providerEventId对应不同rawPayloadHash，拒绝覆盖既有节点事实");
+        if (existing != null && !nodeFactHash(existing).equals(nodeFactHash(fact))) {
+            throw conflict("相同providerEventId对应不同事件事实摘要，拒绝覆盖既有节点事实");
        }
        return null;
    }
@@ -595,12 +595,32 @@ public class CarrierScheduleQueryService {
        fact.setSourceReference(trim(sourceReference)); fact.setRawPayloadHash(rawPayloadHash);
        if (repository.insertIfAbsent(fact)) return true;
        CarrierScheduleExceptionFactEntity existing = repository.exception(queryId, fact.getEventKey());
-        if (existing != null && !Objects.equals(existing.getRawPayloadHash(), rawPayloadHash)) {
-            throw conflict("相同providerEventId对应不同rawPayloadHash，拒绝覆盖既有异常事实");
+        if (existing != null && !exceptionFactHash(existing).equals(exceptionFactHash(fact))) {
+            throw conflict("相同providerEventId对应不同事件事实摘要，拒绝覆盖既有异常事实");
        }
        return false;
    }

+    // rawPayloadHash audits the whole provider response. Sibling events, ETA and query time
+    // may change between polls; compare the immutable event projection instead of that audit hash.
+    private String nodeFactHash(CarrierScheduleNodeFactEntity fact) {
+        return hash("PROVIDER_NODE_FACT_V1", fact.getContainerBookingId(), java.util.Arrays.asList(
+            fact.getProviderCode(), fact.getProviderEventId(), fact.getProviderNodeCode(), fact.getStandardNode(),
+            factTimestamp(fact.getEstimatedAt()), factTimestamp(fact.getActualAt()),
+            fact.getLocationCode(), fact.getLocationName(), fact.getSourceReference()));
+    }
+
+    private String exceptionFactHash(CarrierScheduleExceptionFactEntity fact) {
+        return hash("PROVIDER_EXCEPTION_FACT_V1", fact.getQueryId(), java.util.Arrays.asList(
+            fact.getProviderCode(), fact.getProviderEventId(), fact.getProviderStatus(), fact.getExceptionCode(),
+            fact.getActive(), fact.getLocationCode(), fact.getLocationName(), fact.getSourceReference()));
+    }
+
+    private static LocalDateTime factTimestamp(LocalDateTime value) {
+        // Match DATETIME(3) storage rounding, including a carry into the next second.
+        return value == null ? null : value.plusNanos(500_000).truncatedTo(java.time.temporal.ChronoUnit.MILLIS);
+    }
+
    private String requireContainerNo(long containerBookingId) {
        ContainerBookingEntity container = portSchedules.containerBooking(containerBookingId);
        if (container == null) {
@@ -11,6 +11,8 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import org.h2.jdbcx.JdbcDataSource;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 import org.mybatis.spring.SqlSessionTemplate;
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.jdbc.core.JdbcTemplate;
@@ -23,6 +25,7 @@ import java.lang.reflect.Modifier;
 import java.time.LocalDateTime;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 import java.util.UUID;
 import java.util.stream.Collectors;

@@ -43,12 +46,21 @@ class CarrierScheduleExceptionPersistenceTest {
        for (Class<?> entity : List.of(CarrierScheduleQueryEntity.class, CarrierScheduleQueryHistoryEntity.class,
            CarrierScheduleExceptionFactEntity.class, CarrierScheduleNodeFactEntity.class)) createTable(entity);
        jdbc.execute("ALTER TABLE fl_carrier_schedule_exception_fact ADD CONSTRAINT uq_exception UNIQUE(query_id,event_key)");
+        jdbc.execute("ALTER TABLE fl_carrier_schedule_node_fact ADD CONSTRAINT uq_node UNIQUE(container_booking_id,event_key)");
        jdbc.update("INSERT INTO fl_carrier_schedule_query(id,query_no,customer_id,ocean_booking_id,container_booking_id,"
            + "carrier_code,tracking_type,tracking_no,status,trigger_source,latest_standard_node,latest_node_observed_at,version)"
            + " VALUES(1,'CQ-TEST',0,2,3,'MATSON','CONTAINER_NO','TSTU0000010','READY','OUTBOUND_CONFIRM','IN_TRANSIT',?,7)", TIME);
        jdbc.update("INSERT INTO fl_carrier_schedule_node_fact(id,query_id,container_booking_id,event_key,provider_code,"
            + "provider_event_id,provider_node_code,standard_node,node_sequence,observed_at,actual_at,raw_payload_hash)"
            + " VALUES(4,1,3,'MATSON:normal','MATSON','normal','TRANSIT','IN_TRANSIT',50,?,?,?)", TIME, TIME, "c".repeat(64));
+        jdbc.update("INSERT INTO fl_carrier_schedule_node_fact(id,query_id,container_booking_id,event_key,provider_code,"
+            + "provider_event_id,provider_node_code,standard_node,node_sequence,observed_at,estimated_at,raw_payload_hash)"
+            + " VALUES(5,1,3,'MATSON:estimated','MATSON','estimated','ETA','ARRIVED_DESTINATION_PORT',60,?,?,?)",
+            TIME, TIME.plusDays(14), "d".repeat(64));
+        jdbc.update("INSERT INTO fl_carrier_schedule_query_history(id,query_id,action,from_version,to_version,reason,"
+            + "operator_id,operator_name,created_at) VALUES(6,1,'NODE_OBSERVED',6,7,'IN_TRANSIT',0,'carrier-scheduler',?)", TIME);
+        jdbc.update("INSERT INTO fl_carrier_schedule_query_history(id,query_id,action,from_version,to_version,reason,"
+            + "operator_id,operator_name,created_at) VALUES(7,1,'NODE_OBSERVED',7,7,'ARRIVED_DESTINATION_PORT',0,'carrier-scheduler',?)", TIME);
        MybatisConfiguration configuration = new MybatisConfiguration();
        configuration.setMapUnderscoreToCamelCase(true);
        for (Class<?> mapper : List.of(CarrierScheduleQueryMapper.class, CarrierScheduleQueryHistoryMapper.class,
@@ -62,6 +74,11 @@ class CarrierScheduleExceptionPersistenceTest {
        ReflectionTestUtils.setField(repository, "exceptions", session.getMapper(CarrierScheduleExceptionFactMapper.class));
        ReflectionTestUtils.setField(repository, "nodes", session.getMapper(CarrierScheduleNodeFactMapper.class));
        CarrierScheduleQueryService target = new CarrierScheduleQueryService();
+        var ports = org.mockito.Mockito.mock(com.baiyue.firstleg.portschedule.repository.PortScheduleRepository.class);
+        var container = new com.baiyue.firstleg.entity.ContainerBookingEntity(); container.setContainerNo("ZCSU7881970");
+        org.mockito.Mockito.when(ports.containerBooking(3L)).thenReturn(container);
+        ReflectionTestUtils.setField(target, "portSchedules", ports);
+        ReflectionTestUtils.setField(target, "nodeOutbox", org.mockito.Mockito.mock(CarrierScheduleNodeOutboxService.class));
        ReflectionTestUtils.setField(target, "repository", repository);
        ReflectionTestUtils.setField(target, "objectMapper", new ObjectMapper().findAndRegisterModules());
        ProxyFactory proxy = new ProxyFactory(target);
@@ -72,47 +89,140 @@ class CarrierScheduleExceptionPersistenceTest {

    @Test
    void fourHoldAndReleaseFactsPersistIndependentlyAlongsideUnchangedNormalProgress() {
+        var normalBefore = normalSnapshot();
        for (String code : List.of("CUSTOMS_HOLD", "TERMINAL_HOLD", "DOCUMENT_HOLD", "UNKNOWN_HOLD")) {
+            var existingExceptions = exceptionRows();
            service.applyProviderException(1, request(code, code + "-ON", true, "a"));
+            var activeSnapshot = exceptionRows();
+            assertThat(activeSnapshot).containsAll(existingExceptions).hasSize(existingExceptions.size() + 1);
            service.applyProviderException(1, request(code, code + "-OFF", false, "b"));
+            assertThat(exceptionRows()).containsAll(activeSnapshot).hasSize(activeSnapshot.size() + 1);
+            assertThat(normalSnapshot()).isEqualTo(normalBefore);
+            assertThat(service.detail(1).exceptions().stream().filter(fact -> code.equals(fact.exceptionCode())).toList())
+                .hasSize(2).anyMatch(fact -> fact.active()).anyMatch(fact -> !fact.active())
+                .allSatisfy(fact -> {
+                    assertThat(fact.providerCode()).isEqualTo("MATSON");
+                    assertThat(fact.providerStatus()).isEqualTo(fact.active() ? code : code + " RELEASED");
+                    assertThat(fact.observedAt()).isEqualTo(fact.active() ? TIME : TIME.plusMinutes(1));
+                    assertThat(fact.locationCode()).isEqualTo("USLAX");
+                    assertThat(fact.locationName()).isEqualTo("Los Angeles");
+                    assertThat(fact.sourceReference()).isEqualTo("synthetic-test:" + fact.providerEventId());
+                    assertThat(fact.rawPayloadHash()).isEqualTo((fact.active() ? "a" : "b").repeat(64));
+                });
        }
        var detail = service.detail(1);
        assertThat(detail.exceptions()).hasSize(8);
        assertThat(detail.exceptions().stream().filter(fact -> fact.active()).count()).isEqualTo(4);
-        assertThat(detail.histories()).hasSize(8).allMatch(history -> history.action().equals("EXCEPTION_OBSERVED"));
-        assertThat(detail.nodes()).hasSize(1);
+        assertThat(detail.histories()).hasSize(10);
+        assertThat(detail.histories().stream().filter(history -> history.action().equals("EXCEPTION_OBSERVED")).toList())
+            .hasSize(8).allSatisfy(history -> {
+                assertThat(history.fromVersion()).isEqualTo(7);
+                assertThat(history.toVersion()).isEqualTo(7);
+            });
+        assertThat(detail.nodes()).hasSize(2)
+            .anyMatch(node -> node.actualAt() != null && node.estimatedAt() == null)
+            .anyMatch(node -> node.estimatedAt() != null && node.actualAt() == null);
        assertThat(detail.query().query().latestActualNode()).isEqualTo("IN_TRANSIT");
        assertThat(detail.query().query().version()).isEqualTo(7);
        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM fl_carrier_schedule_exception_fact WHERE source_reference IS NOT NULL AND raw_payload_hash IS NOT NULL", Integer.class)).isEqualTo(8);
    }

+    @ParameterizedTest
+    @ValueSource(strings = {"CUSTOMS_HOLD", "TERMINAL_HOLD", "DOCUMENT_HOLD", "UNKNOWN_HOLD"})
+    void sameFactsReplayPreservesAuditAndChangedFactsCannotOverwrite(String code) {
+        var normalBefore = normalSnapshot();
+        for (boolean active : List.of(true, false)) {
+            var original = request(code, code + (active ? "-ON" : "-OFF"), active, active ? "a" : "b");
+            service.applyProviderException(1, original);
+            var persisted = allSnapshot();
+            service.applyProviderException(1, original);
+            assertThat(allSnapshot()).as("same event/hash must not change any persisted row").isEqualTo(persisted);
+            service.applyProviderException(1, new ProviderExceptionRequest(original.providerCode(), original.providerEventId(),
+                original.providerStatus(), original.exceptionCode(), active, TIME.plusHours(3), original.locationCode(),
+                original.locationName(), original.sourceReference(), "e".repeat(64)));
+            assertThat(allSnapshot()).as("response audit changes do not mutate an existing event").isEqualTo(persisted);
+            assertThatThrownBy(() -> service.applyProviderException(1,
+                new ProviderExceptionRequest(original.providerCode(), original.providerEventId(), original.providerStatus(),
+                    original.exceptionCode(), !active, original.observedAt(), original.locationCode(), original.locationName(),
+                    original.sourceReference(), original.rawPayloadHash())))
+                .isInstanceOf(com.baiyue.common.core.exception.BizException.class)
+                .hasMessageContaining("事件事实摘要");
+            assertThat(allSnapshot()).as("conflicting hash must roll back without changing any persisted row").isEqualTo(persisted);
+            assertThat(normalSnapshot()).isEqualTo(normalBefore);
+        }
+        assertThat(service.detail(1).exceptions()).hasSize(2);
+        assertThat(service.detail(1).histories()).hasSize(4);
+    }
+
    @Test
-    void sameEventHashReplayAddsNoFactOrHistoryAndDifferentHashCannotOverwrite() {
-        var request = request("CUSTOMS_HOLD", "hold-original", true, "a");
-        service.applyProviderException(1, request);
-        service.applyProviderException(1, request);
-        assertThat(service.detail(1).exceptions()).hasSize(1);
-        assertThat(service.detail(1).histories()).hasSize(1);
-        assertThatThrownBy(() -> service.applyProviderException(1,
-            request("CUSTOMS_HOLD", "hold-original", true, "b")))
-            .hasMessageContaining("rawPayloadHash");
-        assertThat(service.detail(1).exceptions()).hasSize(1);
-        assertThat(service.detail(1).exceptions().get(0).rawPayloadHash()).isEqualTo("a".repeat(64));
-        assertThat(service.detail(1).histories()).hasSize(1);
+    void nextPollReplaysOldNodesThenAppendsIndependentEventAndRevisedEta() {
+        var oldNodes = normalSnapshot().get("nodes");
+        service.applyProviderObservation(1, new com.baiyue.firstleg.portschedule.dto.CarrierScheduleQueryDtos.ProviderObservationRequest(
+            "MATSON", "normal", "TRANSIT", "IN_TRANSIT", TIME.plusHours(3), null, TIME, null, null, null, "e".repeat(64)));
+        service.applyProviderObservation(1, new com.baiyue.firstleg.portschedule.dto.CarrierScheduleQueryDtos.ProviderObservationRequest(
+            "MATSON", "estimated", "ETA", "ARRIVED_DESTINATION_PORT", TIME.plusHours(3), TIME.plusDays(14), null, null, null, null, "e".repeat(64)));
+        assertThat(normalSnapshot().get("nodes")).isEqualTo(oldNodes);
+        service.applyProviderObservation(1, new com.baiyue.firstleg.portschedule.dto.CarrierScheduleQueryDtos.ProviderObservationRequest(
+            "MATSON", "estimated-next", "ETA", "ARRIVED_DESTINATION_PORT", TIME.plusHours(3), TIME.plusDays(15), null, null, null, null, "e".repeat(64)));
+        service.applyProviderObservation(1, new com.baiyue.firstleg.portschedule.dto.CarrierScheduleQueryDtos.ProviderObservationRequest(
+            "MATSON", "late-gate", "INGATE", "AT_ORIGIN_PORT", TIME.plusHours(3), null, TIME.minusDays(2), null, null, null, "e".repeat(64)));
+        assertThat(normalSnapshot().get("nodes")).containsAll(oldNodes).hasSize(4);
+        assertThat(service.detail(1).query().query().latestActualNode()).isEqualTo("IN_TRANSIT");
+        assertThat(service.detail(1).query().query().version()).isEqualTo(7);
+        var before = allSnapshot();
+        assertThatThrownBy(() -> service.applyProviderObservation(1,
+            new com.baiyue.firstleg.portschedule.dto.CarrierScheduleQueryDtos.ProviderObservationRequest(
+                "MATSON", "normal", "TRANSIT", "IN_TRANSIT", TIME, null, TIME.plusDays(1), null, null, null, "c".repeat(64))))
+            .isInstanceOf(com.baiyue.common.core.exception.BizException.class).hasMessageContaining("事件事实摘要");
+        assertThat(allSnapshot()).isEqualTo(before);
    }

    @Test
    void failedHistoryWriteRollsBackExceptionAndOriginalRequestCanCommitAfterRepair() {
+        var before = allSnapshot();
        jdbc.execute("ALTER TABLE fl_carrier_schedule_query_history ADD CONSTRAINT reject_observation CHECK(action <> 'EXCEPTION_OBSERVED')");
        var request = request("DOCUMENT_HOLD", "rollback-original", true, "a");
        assertThatThrownBy(() -> service.applyProviderException(1, request)).isInstanceOf(RuntimeException.class);
        assertThat(service.detail(1).exceptions()).isEmpty();
-        assertThat(service.detail(1).histories()).isEmpty();
+        assertThat(allSnapshot()).isEqualTo(before);
        jdbc.execute("ALTER TABLE fl_carrier_schedule_query_history DROP CONSTRAINT reject_observation");
        service.applyProviderException(1, request);
        assertThat(service.detail(1).exceptions()).hasSize(1);
-        assertThat(service.detail(1).histories()).hasSize(1);
+        assertThat(service.detail(1).histories()).hasSize(3);
        assertThat(service.detail(1).query().query().latestActualNode()).isEqualTo("IN_TRANSIT");
+        assertThat(normalSnapshot()).isEqualTo(before.get("normal"));
+    }
+
+    private List<Map<String, Object>> exceptionRows() {
+        return jdbc.queryForList("SELECT * FROM fl_carrier_schedule_exception_fact ORDER BY id");
+    }
+
+    @Test
+    void identicalSubMillisecondProviderTimeReplaysAfterDatabasePrecisionRounding() {
+        var request = new com.baiyue.firstleg.portschedule.dto.CarrierScheduleQueryDtos.ProviderObservationRequest(
+            "MATSON", "fractional", "INGATE", "AT_ORIGIN_PORT", TIME, null,
+            TIME.minusDays(1).withNano(999_700_000), null, null, null, "a".repeat(64));
+        service.applyProviderObservation(1, request);
+        var before = allSnapshot();
+        service.applyProviderObservation(1, request);
+        assertThat(allSnapshot()).isEqualTo(before);
+        assertThatThrownBy(() -> service.applyProviderObservation(1,
+            new com.baiyue.firstleg.portschedule.dto.CarrierScheduleQueryDtos.ProviderObservationRequest(
+                request.providerCode(), request.providerEventId(), request.providerNodeCode(), request.standardNode(), TIME,
+                null, request.actualAt().plusNanos(1_000_000), null, null, null, request.rawPayloadHash())))
+            .isInstanceOf(com.baiyue.common.core.exception.BizException.class).hasMessageContaining("事件事实摘要");
+        assertThat(allSnapshot()).isEqualTo(before);
+    }
+
+    private Map<String, List<Map<String, Object>>> normalSnapshot() {
+        return Map.of("query", jdbc.queryForList("SELECT * FROM fl_carrier_schedule_query ORDER BY id"),
+            "nodes", jdbc.queryForList("SELECT * FROM fl_carrier_schedule_node_fact ORDER BY id"),
+            "normalHistory", jdbc.queryForList("SELECT * FROM fl_carrier_schedule_query_history WHERE action <> 'EXCEPTION_OBSERVED' ORDER BY id"));
+    }
+
+    private Map<String, Object> allSnapshot() {
+        return Map.of("normal", normalSnapshot(), "exceptions", exceptionRows(),
+            "histories", jdbc.queryForList("SELECT * FROM fl_carrier_schedule_query_history ORDER BY id"));
    }

    private ProviderExceptionRequest request(String code, String eventId, boolean active, String hash) {
@@ -566,11 +566,15 @@ class CarrierScheduleQueryServiceTest {
    }

    @Test
-    void providerEventIdReplayRequiresTheSameRawPayloadHash() {
+    void providerReplayAllowsChangedResponseAuditButRejectsChangedEventFacts() {
        CarrierScheduleQueryEntity task = task(0, 1);
        task.setLatestStandardNode("IN_TRANSIT");
        CarrierScheduleNodeFactEntity existing = new CarrierScheduleNodeFactEntity();
        existing.setEventKey("ZIM:evt-duplicate"); existing.setRawPayloadHash("a".repeat(64));
+        existing.setContainerBookingId(task.getContainerBookingId());
+        existing.setProviderCode("ZIM"); existing.setProviderEventId("evt-duplicate");
+        existing.setProviderNodeCode("GATE_IN"); existing.setStandardNode("AT_ORIGIN_PORT");
+        existing.setActualAt(NOW.minusDays(1)); existing.setObservedAt(NOW); existing.setSourceReference("ZIM:evt");
        when(repository.lock(1L)).thenReturn(task);
        when(repository.insertIfAbsent(any(CarrierScheduleNodeFactEntity.class))).thenReturn(false);
        when(repository.nodeByContainer(task.getContainerBookingId(), "ZIM:evt-duplicate")).thenReturn(existing);
@@ -578,12 +582,15 @@ class CarrierScheduleQueryServiceTest {
            "AT_ORIGIN_PORT", NOW, null, NOW.minusDays(1), null, null, "ZIM:evt", "a".repeat(64));

        service.applyProviderObservation(1L, replay);
+        service.applyProviderObservation(1L, new ProviderObservationRequest("ZIM", "evt-duplicate", "GATE_IN",
+            "AT_ORIGIN_PORT", NOW.plusHours(3), null, NOW.minusDays(1), null, null, "ZIM:evt", "b".repeat(64)));
+        assertThat(existing.getRawPayloadHash()).isEqualTo("a".repeat(64));

        verify(nodeOutbox, never()).publish(any(), any(), anyString());
        ProviderObservationRequest conflict = new ProviderObservationRequest("ZIM", "evt-duplicate", "GATE_IN",
-            "AT_ORIGIN_PORT", NOW, null, NOW.minusDays(1), null, null, "ZIM:evt", "b".repeat(64));
+            "AT_ORIGIN_PORT", NOW, null, NOW.minusDays(2), null, null, "ZIM:evt", "a".repeat(64));
        assertThatThrownBy(() -> service.applyProviderObservation(1L, conflict))
-            .isInstanceOf(BizException.class).hasMessageContaining("不同rawPayloadHash");
+            .isInstanceOf(BizException.class).hasMessageContaining("事件事实摘要");
    }

    @Test