fix(shipping): replay immutable carrier facts without updates

firstleg/src/main/java/com/baiyue/firstleg/portschedule/mapper/CarrierScheduleExceptionFactMapper.java

@@ -17,7 +17,6 @@ public interface CarrierScheduleExceptionFactMapper extends BaseMapper<CarrierSc
            #{value.providerStatus},#{value.exceptionCode},#{value.active},#{value.observedAt},
            #{value.locationCode},#{value.locationName},#{value.sourceReference},#{value.rawPayloadHash},
            CURRENT_TIMESTAMP(3))
-        ON DUPLICATE KEY UPDATE id=id
         """)
-    int insertIfAbsent(@Param("value") CarrierScheduleExceptionFactEntity value);
+    int insertStrict(@Param("value") CarrierScheduleExceptionFactEntity value);
 }

firstleg/src/main/java/com/baiyue/firstleg/portschedule/mapper/CarrierScheduleNodeFactMapper.java

@@ -20,14 +20,13 @@ public interface CarrierScheduleNodeFactMapper extends BaseMapper<CarrierSchedul
            #{value.standardNode},#{value.nodeSequence},#{value.observedAt},#{value.estimatedAt},#{value.actualAt},
            #{value.locationCode},#{value.locationName},#{value.sourceReference},#{value.rawPayloadHash},
            CURRENT_TIMESTAMP(3))
-        ON DUPLICATE KEY UPDATE id=id
         """)
-    int insertIgnore(@Param("value") CarrierScheduleNodeFactEntity value);
+    int insertStrict(@Param("value") CarrierScheduleNodeFactEntity value);
 
     @Select("SELECT * FROM fl_carrier_schedule_node_fact WHERE container_booking_id=#{containerBookingId} ORDER BY observed_at,id")
     List<CarrierScheduleNodeFactEntity> selectByContainer(@Param("containerBookingId") long containerBookingId);
 
-    @Select("SELECT * FROM fl_carrier_schedule_node_fact WHERE container_booking_id=#{containerBookingId} AND event_key=#{eventKey} LIMIT 1")
+    @Select("SELECT * FROM fl_carrier_schedule_node_fact WHERE container_booking_id=#{containerBookingId} AND event_key=#{eventKey} LIMIT 1 FOR UPDATE")
     CarrierScheduleNodeFactEntity selectByContainerEvent(@Param("containerBookingId") long containerBookingId,
                                                           @Param("eventKey") String eventKey);
 }

firstleg/src/main/java/com/baiyue/firstleg/portschedule/repository/CarrierScheduleQueryRepository.java

@@ -20,8 +20,10 @@ import com.baiyue.firstleg.portschedule.mapper.CarrierScheduleNodeFactMapper;
 import com.baiyue.firstleg.portschedule.mapper.CarrierScheduleExceptionFactMapper;
 import jakarta.annotation.Resource;
 import org.springframework.stereotype.Repository;
+import org.springframework.dao.DuplicateKeyException;
 
 import java.util.List;
+import java.util.Objects;
 import java.time.LocalDateTime;
 
 @Repository
@@ -76,7 +78,20 @@ public class CarrierScheduleQueryRepository {
             .orderByAsc(CarrierScheduleNodeFactEntity::getObservedAt)
             .orderByAsc(CarrierScheduleNodeFactEntity::getId));
     }
-    public boolean insertIfAbsent(CarrierScheduleNodeFactEntity value) { return nodes.insertIgnore(value) == 1; }
+    public boolean insertIfAbsent(CarrierScheduleNodeFactEntity value) {
+        try {
+            return nodes.insertStrict(value) == 1;
+        } catch (DuplicateKeyException duplicate) {
+            // Current reads also see a concurrent committed winner under MySQL REPEATABLE READ.
+            var idCollision = nodes.selectOne(new LambdaQueryWrapper<CarrierScheduleNodeFactEntity>()
+                .eq(CarrierScheduleNodeFactEntity::getId, value.getId()).last("FOR UPDATE"));
+            if (idCollision != null && (!Objects.equals(idCollision.getContainerBookingId(), value.getContainerBookingId())
+                || !Objects.equals(idCollision.getEventKey(), value.getEventKey()))) throw duplicate;
+            if (nodeByContainer(value.getContainerBookingId(), value.getEventKey()) == null) throw duplicate;
+            // The service retains the immutable fact comparison and rejects changed payloads.
+            return false;
+        }
+    }
     public CarrierScheduleNodeFactEntity node(long queryId, String eventKey) {
         return nodes.selectOne(new LambdaQueryWrapper<CarrierScheduleNodeFactEntity>()
             .eq(CarrierScheduleNodeFactEntity::getQueryId, queryId)
@@ -101,12 +116,21 @@ public class CarrierScheduleQueryRepository {
             .orderByAsc(CarrierScheduleExceptionFactEntity::getId));
     }
     public boolean insertIfAbsent(CarrierScheduleExceptionFactEntity value) {
-        return exceptions.insertIfAbsent(value) == 1;
+        try {
+            return exceptions.insertStrict(value) == 1;
+        } catch (DuplicateKeyException duplicate) {
+            var idCollision = exceptions.selectOne(new LambdaQueryWrapper<CarrierScheduleExceptionFactEntity>()
+                .eq(CarrierScheduleExceptionFactEntity::getId, value.getId()).last("FOR UPDATE"));
+            if (idCollision != null && (!Objects.equals(idCollision.getQueryId(), value.getQueryId())
+                || !Objects.equals(idCollision.getEventKey(), value.getEventKey()))) throw duplicate;
+            if (exception(value.getQueryId(), value.getEventKey()) == null) throw duplicate;
+            return false;
+        }
     }
     public CarrierScheduleExceptionFactEntity exception(long queryId, String eventKey) {
         return exceptions.selectOne(new LambdaQueryWrapper<CarrierScheduleExceptionFactEntity>()
             .eq(CarrierScheduleExceptionFactEntity::getQueryId, queryId)
-            .eq(CarrierScheduleExceptionFactEntity::getEventKey, eventKey));
+            .eq(CarrierScheduleExceptionFactEntity::getEventKey, eventKey).last("FOR UPDATE"));
     }
     public boolean claimOperation(Long queryId, long operatorId, String operation, String key, String hash) {
         return operations.claim(IdWorker.getId(), queryId, operatorId, operation, key, hash) == 1;

firstleg/src/test/java/com/baiyue/firstleg/portschedule/repository/ImmutableCarrierFactPersistenceTest.java

@@ -0,0 +1,184 @@
+package com.baiyue.firstleg.portschedule.repository;
+
+import com.baomidou.mybatisplus.annotation.TableName;
+import com.baomidou.mybatisplus.core.MybatisConfiguration;
+import com.baomidou.mybatisplus.extension.spring.MybatisSqlSessionFactoryBean;
+import com.baiyue.firstleg.portschedule.entity.CarrierScheduleNodeFactEntity;
+import com.baiyue.firstleg.portschedule.entity.CarrierScheduleExceptionFactEntity;
+import com.baiyue.firstleg.portschedule.mapper.CarrierScheduleNodeFactMapper;
+import com.baiyue.firstleg.portschedule.mapper.CarrierScheduleExceptionFactMapper;
+import org.apache.ibatis.annotations.Insert;
+import org.h2.api.Trigger;
+import org.h2.jdbcx.JdbcDataSource;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mybatis.spring.SqlSessionTemplate;
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.lang.reflect.Modifier;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.time.LocalDateTime;
+import java.util.Arrays;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.*;
+
+/** H2 executes production Mapper SQL with an UPDATE-rejecting guard. It does not replace MySQL acceptance. */
+public class ImmutableCarrierFactPersistenceTest {
+    private JdbcTemplate jdbc;
+    private CarrierScheduleQueryRepository repository;
+
+    public static class RejectUpdates implements Trigger {
+        @Override public void fire(Connection connection, Object[] oldRow, Object[] newRow) throws SQLException {
+            throw new SQLException("immutable carrier fact", "45000", 1644);
+        }
+    }
+
+    @BeforeEach
+    void setup() throws Exception {
+        var dataSource = new JdbcDataSource();
+        dataSource.setURL("jdbc:h2:mem:immutable-facts-" + UUID.randomUUID() + ";MODE=MySQL;DB_CLOSE_DELAY=-1");
+        jdbc = new JdbcTemplate(dataSource);
+        table(CarrierScheduleNodeFactEntity.class); table(CarrierScheduleExceptionFactEntity.class);
+        jdbc.execute("ALTER TABLE fl_carrier_schedule_node_fact ADD CONSTRAINT uq_node UNIQUE(container_booking_id,event_key)");
+        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 ALTER COLUMN provider_code SET NOT NULL");
+        jdbc.execute("ALTER TABLE fl_carrier_schedule_exception_fact ADD CONSTRAINT valid_query CHECK(query_id>0)");
+        for (String table : List.of("fl_carrier_schedule_node_fact", "fl_carrier_schedule_exception_fact"))
+            jdbc.execute("CREATE TRIGGER guard_" + table + " BEFORE UPDATE ON " + table
+                + " FOR EACH ROW CALL '" + RejectUpdates.class.getName() + "'");
+        var configuration = new MybatisConfiguration(); configuration.setMapUnderscoreToCamelCase(true);
+        configuration.addMapper(CarrierScheduleNodeFactMapper.class);
+        configuration.addMapper(CarrierScheduleExceptionFactMapper.class);
+        var factory = new MybatisSqlSessionFactoryBean(); factory.setDataSource(dataSource); factory.setConfiguration(configuration);
+        var session = new SqlSessionTemplate(factory.getObject());
+        repository = new CarrierScheduleQueryRepository();
+        ReflectionTestUtils.setField(repository, "nodes", session.getMapper(CarrierScheduleNodeFactMapper.class));
+        ReflectionTestUtils.setField(repository, "exceptions", session.getMapper(CarrierScheduleExceptionFactMapper.class));
+    }
+
+    @Test
+    void bothImmutableFactMappersUseStrictInsertWithoutAnyUpdateOrIgnoreClause() {
+        for (Class<?> mapper : List.of(CarrierScheduleNodeFactMapper.class, CarrierScheduleExceptionFactMapper.class)) {
+            var sql = Arrays.stream(mapper.getDeclaredMethods()).filter(m -> m.isAnnotationPresent(Insert.class))
+                .map(m -> String.join(" ", m.getAnnotation(Insert.class).value())).collect(Collectors.joining(" "));
+            assertThat(sql).contains("INSERT INTO").doesNotContainIgnoringCase("ON DUPLICATE", "INSERT IGNORE", "UPDATE");
+        }
+    }
+
+    @Test
+    void nodeReplayAvoidsUpdateGuardAndPreservesOriginalFact() {
+        assertThat(repository.insertIfAbsent(node(1,3,"same"))).isTrue();
+        var before = jdbc.queryForList("SELECT * FROM fl_carrier_schedule_node_fact");
+        assertThat(repository.insertIfAbsent(node(2,3,"same"))).isFalse();
+        assertThat(jdbc.queryForList("SELECT * FROM fl_carrier_schedule_node_fact")).isEqualTo(before);
+    }
+
+    @Test
+    void exceptionReplayAvoidsUpdateGuardAndPreservesOriginalFact() {
+        assertThat(repository.insertIfAbsent(exception(1,5,"same"))).isTrue();
+        var before = jdbc.queryForList("SELECT * FROM fl_carrier_schedule_exception_fact");
+        assertThat(repository.insertIfAbsent(exception(2,5,"same"))).isFalse();
+        assertThat(jdbc.queryForList("SELECT * FROM fl_carrier_schedule_exception_fact")).isEqualTo(before);
+    }
+
+    @Test
+    void primaryKeyCollisionCannotBeMistakenForSameBusinessEventEvenWhenThatEventAlsoExists() {
+        repository.insertIfAbsent(node(1,3,"one")); repository.insertIfAbsent(node(2,4,"two"));
+        assertThatThrownBy(() -> repository.insertIfAbsent(node(1,4,"unrelated"))).isInstanceOf(DuplicateKeyException.class);
+        assertThatThrownBy(() -> repository.insertIfAbsent(node(1,4,"two"))).isInstanceOf(DuplicateKeyException.class);
+        repository.insertIfAbsent(exception(1,5,"one")); repository.insertIfAbsent(exception(2,6,"two"));
+        assertThatThrownBy(() -> repository.insertIfAbsent(exception(1,6,"unrelated"))).isInstanceOf(DuplicateKeyException.class);
+        assertThatThrownBy(() -> repository.insertIfAbsent(exception(1,6,"two"))).isInstanceOf(DuplicateKeyException.class);
+    }
+
+    @Test
+    void nonDuplicateConstraintErrorsContinueToPropagate() {
+        var invalid = node(1,3,"invalid"); invalid.setProviderCode(null);
+        assertThatThrownBy(() -> repository.insertIfAbsent(invalid)).isInstanceOf(DataIntegrityViolationException.class);
+        assertThatThrownBy(() -> repository.insertIfAbsent(exception(1,0,"invalid"))).isInstanceOf(DataIntegrityViolationException.class);
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM fl_carrier_schedule_node_fact",Integer.class)).isZero();
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM fl_carrier_schedule_exception_fact",Integer.class)).isZero();
+    }
+
+    @Test
+    void nonDuplicateSqlFailureIsNeverConvertedIntoAReplay() {
+        var nodes = org.mockito.Mockito.mock(CarrierScheduleNodeFactMapper.class);
+        var exceptions = org.mockito.Mockito.mock(CarrierScheduleExceptionFactMapper.class);
+        var target = new CarrierScheduleQueryRepository();
+        ReflectionTestUtils.setField(target,"nodes",nodes);
+        ReflectionTestUtils.setField(target,"exceptions",exceptions);
+        var failure = new org.springframework.jdbc.UncategorizedSQLException("insert","INSERT",
+            new SQLException("immutable guard failure","45000",1644));
+        org.mockito.Mockito.when(nodes.insertStrict(org.mockito.ArgumentMatchers.any())).thenThrow(failure);
+        org.mockito.Mockito.when(exceptions.insertStrict(org.mockito.ArgumentMatchers.any())).thenThrow(failure);
+        assertThatThrownBy(() -> target.insertIfAbsent(node(1,3,"event"))).isSameAs(failure);
+        assertThatThrownBy(() -> target.insertIfAbsent(exception(1,5,"event"))).isSameAs(failure);
+        org.mockito.Mockito.verify(nodes,org.mockito.Mockito.never()).selectOne(org.mockito.ArgumentMatchers.any());
+        org.mockito.Mockito.verify(exceptions,org.mockito.Mockito.never()).selectOne(org.mockito.ArgumentMatchers.any());
+    }
+
+    @Test
+    void immutableUpdateGuardsRemainEffectiveForBothTables() {
+        repository.insertIfAbsent(node(1,3,"one")); repository.insertIfAbsent(exception(1,5,"one"));
+        assertThatThrownBy(() -> jdbc.update("UPDATE fl_carrier_schedule_node_fact SET id=id WHERE id=1"))
+            .hasMessageContaining("immutable carrier fact");
+        assertThatThrownBy(() -> jdbc.update("UPDATE fl_carrier_schedule_exception_fact SET id=id WHERE id=1"))
+            .hasMessageContaining("immutable carrier fact");
+    }
+
+    @Test
+    void concurrentNodeDuplicateHasExactlyOneInsertAndOneConfirmedReplay() throws Exception {
+        concurrent(() -> repository.insertIfAbsent(node(1,3,"race")), () -> repository.insertIfAbsent(node(2,3,"race")));
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM fl_carrier_schedule_node_fact",Integer.class)).isEqualTo(1);
+    }
+
+    @Test
+    void concurrentExceptionDuplicateHasExactlyOneInsertAndOneConfirmedReplay() throws Exception {
+        concurrent(() -> repository.insertIfAbsent(exception(1,5,"race")), () -> repository.insertIfAbsent(exception(2,5,"race")));
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM fl_carrier_schedule_exception_fact",Integer.class)).isEqualTo(1);
+    }
+
+    private void concurrent(java.util.concurrent.Callable<Boolean> left, java.util.concurrent.Callable<Boolean> right) throws Exception {
+        var executor = Executors.newFixedThreadPool(2);
+        var ready = new CountDownLatch(2); var start = new CountDownLatch(1);
+        try {
+            var first = executor.submit(() -> { ready.countDown(); start.await(); return left.call(); });
+            var second = executor.submit(() -> { ready.countDown(); start.await(); return right.call(); });
+            assertThat(ready.await(5,TimeUnit.SECONDS)).isTrue(); start.countDown();
+            assertThat(List.of(first.get(10,TimeUnit.SECONDS),second.get(10,TimeUnit.SECONDS))).containsExactlyInAnyOrder(true,false);
+        } finally { start.countDown(); executor.shutdownNow(); }
+    }
+
+    private CarrierScheduleNodeFactEntity node(long id,long container,String event) {
+        var row=new CarrierScheduleNodeFactEntity(); row.setId(id); row.setQueryId(5L); row.setContainerBookingId(container);
+        row.setEventKey("MANUAL:"+event); row.setProviderCode("MANUAL"); row.setProviderEventId(event);
+        row.setProviderNodeCode("AT_ORIGIN_PORT"); row.setStandardNode("AT_ORIGIN_PORT"); row.setNodeSequence(20);
+        row.setObservedAt(LocalDateTime.of(2026,10,5,1,40)); row.setActualAt(LocalDateTime.of(2026,10,5,1,44));
+        row.setRawPayloadHash("a".repeat(64)); return row;
+    }
+
+    private CarrierScheduleExceptionFactEntity exception(long id,long query,String event) {
+        var row=new CarrierScheduleExceptionFactEntity(); row.setId(id); row.setQueryId(query); row.setEventKey("MATSON:"+event);
+        row.setProviderCode("MATSON"); row.setProviderEventId(event); row.setProviderStatus("HOLD");
+        row.setExceptionCode("CUSTOMS_HOLD"); row.setActive(true); row.setObservedAt(LocalDateTime.of(2026,10,5,1,44));
+        row.setRawPayloadHash("a".repeat(64)); return row;
+    }
+
+    private void table(Class<?> entity) {
+        var columns=Arrays.stream(entity.getDeclaredFields()).filter(f -> !Modifier.isStatic(f.getModifiers()))
+            .map(f -> f.getName().replaceAll("([a-z0-9])([A-Z])","$1_$2").toLowerCase()+" "
+                +(f.getType()==Long.class?"BIGINT":f.getType()==Integer.class?"INT":f.getType()==Boolean.class?"BOOLEAN":f.getType()==LocalDateTime.class?"TIMESTAMP(3)":"VARCHAR(16000)")
+                +(f.getName().equals("id")?" PRIMARY KEY":"")).collect(Collectors.joining(","));
+        jdbc.execute("CREATE TABLE "+entity.getAnnotation(TableName.class).value()+"("+columns+")");
+    }
+}

firstleg/src/test/java/com/baiyue/firstleg/portschedule/service/CarrierScheduleExceptionPersistenceTest.java

@@ -61,6 +61,12 @@ class CarrierScheduleExceptionPersistenceTest {
             + "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);
+        for (String table : List.of("fl_carrier_schedule_node_fact", "fl_carrier_schedule_exception_fact")) {
+            jdbc.execute("CREATE TRIGGER immutable_" + table + " BEFORE UPDATE ON " + table
+                + " FOR EACH ROW CALL '"
+                + com.baiyue.firstleg.portschedule.repository.ImmutableCarrierFactPersistenceTest.RejectUpdates.class.getName()
+                + "'");
+        }
         MybatisConfiguration configuration = new MybatisConfiguration();
         configuration.setMapUnderscoreToCamelCase(true);
         for (Class<?> mapper : List.of(CarrierScheduleQueryMapper.class, CarrierScheduleQueryHistoryMapper.class,
@@ -127,6 +133,23 @@ class CarrierScheduleExceptionPersistenceTest {
         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);
     }
 
+    @Test
+    void originalContainerManualBodyReplaysWithoutUpdateHistoryOrFactMutationAndRejectsChangedFact() {
+        var body = new com.baiyue.firstleg.portschedule.dto.CarrierScheduleQueryDtos.ContainerManualNodeRequest(
+            "ARRIVED_DESTINATION_PORT", TIME.plusDays(2), null, TIME.plusDays(2), "USLAX", "Los Angeles",
+            "synthetic container manual fact");
+        service.addContainerManualNode(3,10001,"test-operator","original-body-key",body);
+        var saved = allSnapshot();
+        service.addContainerManualNode(3,10001,"test-operator","original-body-key",body);
+        assertThat(allSnapshot()).isEqualTo(saved);
+        assertThatThrownBy(() -> service.addContainerManualNode(3,10001,"test-operator","original-body-key",
+            new com.baiyue.firstleg.portschedule.dto.CarrierScheduleQueryDtos.ContainerManualNodeRequest(
+                body.standardNode(),body.observedAt(),null,body.actualAt().plusSeconds(1),body.locationCode(),
+                body.locationName(),body.reason())))
+            .isInstanceOf(com.baiyue.common.core.exception.BizException.class).hasMessageContaining("事件事实摘要");
+        assertThat(allSnapshot()).isEqualTo(saved);
+    }
+
     @ParameterizedTest
     @ValueSource(strings = {"CUSTOMS_HOLD", "TERMINAL_HOLD", "DOCUMENT_HOLD", "UNKNOWN_HOLD"})
     void sameFactsReplayPreservesAuditAndChangedFactsCannotOverwrite(String code) {