fix: align actual port arrival with latest observation

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

@@ -63,6 +63,7 @@ public interface CarrierScheduleQueryMapper extends BaseMapper<CarrierScheduleQu
         """)
     CarrierScheduleQueryEntity findByContainerBookingId(@Param("containerBookingId") long containerBookingId);
 
+    /** 到港与日历统一取最新观测的实际记录，后续修正为更早时间时不能被历史最大值覆盖。 */
     @Select("""
         <script>
         SELECT b.id AS containerBookingId,q.id AS queryId,q.version AS queryVersion,
@@ -70,7 +71,7 @@ public interface CarrierScheduleQueryMapper extends BaseMapper<CarrierScheduleQu
                current.actual_at AS latestActualAt,
                MAX(CASE WHEN f.standard_node='DEPARTED_WAREHOUSE' THEN f.actual_at END) AS departedWarehouseAt,
                MAX(CASE WHEN f.standard_node='DEPARTED_ORIGIN_PORT' THEN f.actual_at END) AS departedOriginPortAt,
-               MAX(CASE WHEN f.standard_node='ARRIVED_DESTINATION_PORT' THEN f.actual_at END) AS arrivedDestinationPortAt,
+               MAX(arrival.actual_at) AS arrivedDestinationPortAt,
                MAX(CASE WHEN f.standard_node='PICKUP_READY' THEN f.actual_at END) AS pickupReadyAt,
                MAX(CASE WHEN f.standard_node='PICKED_UP' THEN f.actual_at END) AS pickedUpAt,
                MAX(CASE WHEN f.standard_node='EMPTY_RETURNED' THEN f.actual_at END) AS emptyReturnedAt
@@ -81,6 +82,10 @@ public interface CarrierScheduleQueryMapper extends BaseMapper<CarrierScheduleQu
                  ORDER BY f2.observed_at DESC,f2.id DESC LIMIT 1)
           LEFT JOIN fl_carrier_schedule_query q ON q.id=current.query_id
           LEFT JOIN fl_carrier_schedule_node_fact f ON f.container_booking_id=b.id AND f.actual_at IS NOT NULL
+          LEFT JOIN fl_carrier_schedule_node_fact arrival ON arrival.id=(
+                SELECT a.id FROM fl_carrier_schedule_node_fact a
+                 WHERE a.container_booking_id=b.id AND a.standard_node='ARRIVED_DESTINATION_PORT' AND a.actual_at IS NOT NULL
+                 ORDER BY a.observed_at DESC,a.id DESC LIMIT 1)
          WHERE b.id IN
          <foreach collection="containerBookingIds" item="id" open="(" separator="," close=")">#{id}</foreach>
          GROUP BY b.id,q.id,q.version,q.status,current.standard_node,current.actual_at

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

@@ -44,8 +44,9 @@ class CarrierScheduleQuerySqlContractTest {
         }
     }
 
+    /** 保留批量读取，实际到港与日历采用相同的最新观测顺序。 */
     @Test
-    void batchesActualContainerFactsWithoutEstimatedOrPerNodeQueries() throws Exception {
+    void batchesActualContainerFactsWithLatestObservedArrivalAndWithoutEstimatedTimes() throws Exception {
         String sql = String.join("\n", CarrierScheduleQueryMapper.class
             .getMethod("selectContainerActualFacts", List.class)
             .getAnnotation(Select.class).value());
@@ -55,13 +56,16 @@ class CarrierScheduleQuerySqlContractTest {
             .contains("f.actual_at IS NOT NULL")
             .contains("MAX(CASE WHEN f.standard_node='DEPARTED_WAREHOUSE'")
             .contains("MAX(CASE WHEN f.standard_node='DEPARTED_ORIGIN_PORT'")
-            .contains("MAX(CASE WHEN f.standard_node='ARRIVED_DESTINATION_PORT'")
+            .contains("MAX(arrival.actual_at) AS arrivedDestinationPortAt")
+            .contains("a.standard_node='ARRIVED_DESTINATION_PORT' AND a.actual_at IS NOT NULL")
+            .contains("ORDER BY a.observed_at DESC,a.id DESC LIMIT 1")
+            .doesNotContain("MAX(CASE WHEN f.standard_node='ARRIVED_DESTINATION_PORT'")
             .contains("MAX(CASE WHEN f.standard_node='PICKUP_READY'")
             .contains("MAX(CASE WHEN f.standard_node='PICKED_UP'")
             .contains("MAX(CASE WHEN f.standard_node='EMPTY_RETURNED'")
             .contains("ORDER BY b.id")
             .doesNotContain("estimated_at");
-        assertThat(count(sql, "fl_carrier_schedule_node_fact")).isEqualTo(3);
+        assertThat(count(sql, "fl_carrier_schedule_node_fact")).isEqualTo(5);
         assertThat(count(sql, "SELECT f2.id FROM fl_carrier_schedule_node_fact f2")).isEqualTo(1);
     }
 

firstleg/src/test/java/com/baiyue/firstleg/terminaldrayage/mapper/ContainerCalendarSourceMapperTest.java

@@ -2,6 +2,7 @@ package com.baiyue.firstleg.terminaldrayage.mapper;
 
 import com.baomidou.mybatisplus.core.MybatisConfiguration;
 import com.baomidou.mybatisplus.core.MybatisSqlSessionFactoryBuilder;
+import com.baiyue.firstleg.portschedule.mapper.CarrierScheduleQueryMapper;
 import org.apache.ibatis.mapping.Environment;
 import org.apache.ibatis.transaction.jdbc.JdbcTransactionFactory;
 import org.h2.jdbcx.JdbcDataSource;
@@ -10,6 +11,51 @@ import java.time.LocalDateTime;
 import static org.junit.jupiter.api.Assertions.*;
 
 class ContainerCalendarSourceMapperTest {
+    /** 后续观测可将实际到港修正为更早日期，两入口须取同一条最新实际记录。 */
+    @Test
+    void usesLatestObservedArrivalIncludingSameObservationIdTieBreakAcrossBothEntrypoints() throws Exception {
+        var ds = new JdbcDataSource();
+        ds.setURL("jdbc:h2:mem:arrival-source-correction;MODE=MySQL;DB_CLOSE_DELAY=-1");
+        try (var connection = ds.getConnection(); var sql = connection.createStatement()) {
+            sql.execute("CREATE TABLE fl_container_booking(id BIGINT PRIMARY KEY,cabinet_no VARCHAR,container_no VARCHAR," +
+                "vessel_name VARCHAR,voyage_no VARCHAR,destination_port_name VARCHAR,destination_warehouse_code VARCHAR," +
+                "actual_port_arrival_at TIMESTAMP,estimated_port_arrival_at TIMESTAMP,unloading_completed_at TIMESTAMP,record_status VARCHAR)");
+            sql.execute("CREATE TABLE fl_terminal_drayage(container_booking_id BIGINT,actual_empty_return_at TIMESTAMP)");
+            sql.execute("CREATE TABLE fl_carrier_schedule_query(id BIGINT,version INT,status VARCHAR)");
+            sql.execute("CREATE TABLE fl_carrier_schedule_node_fact(id BIGINT,query_id BIGINT,container_booking_id BIGINT," +
+                "standard_node VARCHAR,actual_at TIMESTAMP,observed_at TIMESTAMP)");
+            sql.execute("CREATE TABLE fl_direct_delivery_unloading(container_booking_id BIGINT,actual_at TIMESTAMP)");
+            sql.execute("CREATE TABLE fl_outbound_dispatch(container_booking_id BIGINT,confirmed_at TIMESTAMP,reversed_at TIMESTAMP,status VARCHAR)");
+            sql.execute("INSERT INTO fl_container_booking(id,cabinet_no,estimated_port_arrival_at,record_status) VALUES " +
+                "(1,'D1','2026-10-29 08:00:00','ACTIVE'),(2,'D2','2026-10-29 08:00:00','ACTIVE')");
+            sql.execute("INSERT INTO fl_carrier_schedule_query VALUES(1,0,'READY'),(2,0,'READY')");
+            sql.execute("INSERT INTO fl_outbound_dispatch SELECT id,'2026-10-01 00:00:00',NULL,'WAITING_FOR_LOADING' FROM fl_container_booking");
+            sql.execute("INSERT INTO fl_carrier_schedule_node_fact VALUES " +
+                "(1,1,1,'ARRIVED_DESTINATION_PORT','2026-10-23 14:00:00','2026-10-24 14:00:00')," +
+                "(2,1,1,'ARRIVED_DESTINATION_PORT','2026-10-21 14:00:00','2026-10-25 14:00:00')," +
+                "(3,1,1,'ARRIVED_DESTINATION_PORT',NULL,'2026-10-26 14:00:00')," +
+                "(4,2,2,'ARRIVED_DESTINATION_PORT','2026-10-24 14:00:00','2026-10-25 14:00:00')," +
+                "(5,2,2,'ARRIVED_DESTINATION_PORT','2026-10-22 14:00:00','2026-10-25 14:00:00')");
+        }
+        var config = new MybatisConfiguration();
+        config.setEnvironment(new Environment("test", new JdbcTransactionFactory(), ds));
+        config.addMapper(ContainerCalendarSourceMapper.class);
+        config.addMapper(CarrierScheduleQueryMapper.class);
+        try (var session = new MybatisSqlSessionFactoryBuilder().build(config).openSession()) {
+            var from = LocalDateTime.parse("2026-10-01T00:00:00");
+            var calendar = session.getMapper(ContainerCalendarSourceMapper.class).sources(from, from.minusDays(2), from.minusDays(4), from.plusMonths(1));
+            var workbench = session.getMapper(CarrierScheduleQueryMapper.class).selectContainerActualFacts(java.util.List.of(1L, 2L));
+            assertEquals(2, calendar.size());
+            assertEquals(2, workbench.size());
+            for (int index = 0; index < calendar.size(); index++) {
+                var expected = LocalDateTime.parse(index == 0 ? "2026-10-21T14:00:00" : "2026-10-22T14:00:00");
+                assertEquals(expected, calendar.get(index).arrivalAt());
+                assertEquals("ACTUAL", calendar.get(index).arrivalType());
+                assertEquals(expected, workbench.get(index).arrivedDestinationPortAt());
+                assertEquals(calendar.get(index).arrivalAt().plusDays(2).toLocalDate(), workbench.get(index).arrivedDestinationPortAt().plusDays(2).toLocalDate());
+            }
+        }
+    }
     /** 使用实际 Mapper 核对跨月默认日期候选、缺日期与真实出仓边界。 */
     @Test
     void selectsShiftedArrivalUndatedAndOlderReturnCandidatesWithoutInactiveOrUndepartedContainers() throws Exception {