test(jdbc): make PG integration tests self-contained on env-configured port

Restore 3 v2 tests lost in file corruption (persistNextStates roundtrip,
legacy columns all-null, schd_gen last_message_id CAS) — 13 JDBC cases.
Add idempotent Flyway migrate in @BeforeEach so an empty container works.
Parameterize compose port as ${MSGX_PG_PORT:-5432}. Verified 14/14 green
(0 skipped) against real PostgreSQL on :5433.

src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlightSchdJdbcPgTest.kt

@@ -47,6 +47,12 @@ class FlightSchdJdbcPgTest {
             driverClassName = "org.postgresql.Driver"
             maximumPoolSize = 2
         }
+        // 幂等前置迁移：全新容器（或库缺失 schema）时补齐 V1.0.0→V1.3.0
+        org.flywaydb.core.Flyway.configure()
+            .dataSource(PgTestSupport.jdbcUrl, PgTestSupport.user, PgTestSupport.password)
+            .locations("classpath:db/migration")
+            .load()
+            .migrate()
         repo = JdbcFlightSchdRepository(ds)
         cleanup()
     }
@@ -409,4 +415,124 @@ class FlightSchdJdbcPgTest {
         assertEquals(writeInstant, instantUtc)
         assertEquals(instantChicago, instantUtc)
     }
+
+    @Test
+    fun `PG dialect - persistNextStates roundtrip preserves collections source_seq and state_version`() {
+        val day = "2026-09-07"
+        val flid = "TEST_V2_ROUNDTRIP"
+        val gates = listOf(
+            mapOf("GTNO" to "1", "GATE" to "G28", "GTYP" to "D"),
+            mapOf("GTNO" to "3", "GATE" to "G33", "GTYP" to "I"),
+            mapOf("GTNO" to "5", "GATE" to "G23", "GTYP" to "D"),
+        )
+        val state = com.gzzn.omms.msgexchange.domain.flight.FlightNextState(
+            flid = flid,
+            scalars = mapOf("FLNO" to "CA888"),
+            collections = mapOf("GTDT" to gates),
+            stateVersion = 1L,
+            lastMessageId = "msg-dnld-001",
+        )
+
+        repo.persistNextStates(day, listOf(state), snapshotReplace = true)
+        assertLegacyCollectionColumnsAllNull(flid)
+
+        val roundtrip = repo.findNextStateByFlid(flid)!!
+        assertEquals(1L, roundtrip.stateVersion)
+        assertEquals("msg-dnld-001", roundtrip.lastMessageId)
+        // 明细表读回保序保源序号；未填属性以空串回填，按语义字段断言
+        val roundtripGates = roundtrip.collections["GTDT"]!!
+        assertEquals(3, roundtripGates.size)
+        gates.forEachIndexed { i, expected ->
+            val actual = roundtripGates[i]
+            assertEquals(expected["GTNO"], actual["GTNO"])
+            assertEquals(expected["GATE"], actual["GATE"])
+            assertEquals(expected["GTYP"], actual["GTYP"])
+        }
+
+        val replaced = state.copy(
+            collections = mapOf("GTDT" to listOf(mapOf("GTNO" to "2", "GATE" to "Z9"))),
+            stateVersion = 2L,
+            lastMessageId = "msg-flop-002",
+        )
+        repo.persistNextStates(null, listOf(replaced), snapshotReplace = false)
+        assertEquals("Z9", repo.findNextStateByFlid(flid)!!.collections["GTDT"]!![0]["GATE"])
+        assertEquals(2L, repo.findNextStateByFlid(flid)!!.stateVersion)
+        val detailRows = ds.query(
+            "SELECT ordinal, source_seq, gate FROM flight_gate WHERE flid = ? ORDER BY ordinal",
+            { ps -> ps.setString(1, flid) },
+        ) { rs -> Triple(rs.getInt("ordinal"), rs.getString("source_seq"), rs.getString("gate")) }
+        assertEquals(1, detailRows.size)
+        assertEquals(Triple(1, "2", "Z9"), detailRows[0])
+    }
+
+    @Test
+    fun `PG dialect - persistNextStates leaves all legacy collection storage columns null`() {
+        val day = "2026-09-07"
+        val flid = "TEST_V2_ALL_COLS"
+        val state = com.gzzn.omms.msgexchange.domain.flight.FlightNextState(
+            flid = flid,
+            scalars = mapOf("FLNO" to "MU123"),
+            collections = mapOf(
+                "GTDT" to listOf(mapOf("GTNO" to "1", "GATE" to "G1", "GTYP" to "D")),
+                "CKDT" to listOf(mapOf("CKNO" to "1", "CHKC" to "C1", "CTYP" to "Eco")),
+                "CLDT" to listOf(mapOf("CLNO" to "1", "BELT" to "B1", "BTYP" to "D")),
+                "DELY" to listOf(mapOf("DLNO" to "1", "CODE" to "01", "STRT" to "1200", "DURA" to "30")),
+                "ABTM" to listOf(mapOf("ASNO" to "1", "ABDG" to "B12", "ABOP" to "A", "AOTM" to "0730")),
+                "CHOT" to listOf(mapOf("CSNO" to "1", "CHID" to "ON", "CHTM" to "0800")),
+                "ROUT" to listOf(mapOf("RTNO" to "1", "APCD" to "CTU")),
+            ),
+            stateVersion = 1L,
+            lastMessageId = "msg-dnld-002",
+        )
+
+        repo.persistNextStates(day, listOf(state), snapshotReplace = true)
+        assertLegacyCollectionColumnsAllNull(flid)
+        assertEquals("MU123", repo.findByFlid(flid)!!["FLNO"])
+        assertEquals(1, ds.query("SELECT COUNT(*) FROM flight_gate WHERE flid = ?", { ps -> ps.setString(1, flid) }) { rs -> rs.getInt(1) }.first())
+        assertEquals(1, ds.query("SELECT COUNT(*) FROM flight_delay WHERE flid = ?", { ps -> ps.setString(1, flid) }) { rs -> rs.getInt(1) }.first())
+    }
+
+    @Test
+    fun `PG dialect - schd_gen stores last_message_id and rejects stale CAS without advancing`() {
+        val day = "2026-09-07"
+
+        // 初始提交：last_message_id = "m1"，版本 1
+        val ok1 = repo.putGenIfVersion(
+            day, 0L,
+            FlightSchdRepository.GenMeta(day, 1L, setOf("TEST_CAS_M"), lastMessageId = "m1"),
+        )
+        assertTrue(ok1)
+        assertEquals("m1", repo.getGen(day)!!.lastMessageId)
+
+        // 重放判定：以过期 expected=0 重试必须失败（同一消息的身份凭 getGen().lastMessageId 判定，不是靠版本相等）
+        val replayAttempt = repo.putGenIfVersion(
+            day, 0L,
+            FlightSchdRepository.GenMeta(day, 1L, setOf("TEST_CAS_M"), lastMessageId = "m1"),
+        )
+        assertFalse(replayAttempt)
+        assertEquals(1L, repo.getGen(day)!!.version)
+
+        // 正常推进 expected=1 → 版本 2，last_message_id 换为 "m2"
+        val ok2 = repo.putGenIfVersion(
+            day, 1L,
+            FlightSchdRepository.GenMeta(day, 2L, setOf("TEST_CAS_M"), lastMessageId = "m2"),
+        )
+        assertTrue(ok2)
+        val gen = repo.getGen(day)!!
+        assertEquals(2L, gen.version)
+        assertEquals("m2", gen.lastMessageId)
+    }
+
+    private fun assertLegacyCollectionColumnsAllNull(flid: String) {
+        val columnList = com.gzzn.omms.msgexchange.support.LEGACY_COLLECTION_STORAGE_COLUMNS.joinToString(", ")
+        ds.queryOne(
+            "SELECT $columnList FROM flight_schd WHERE flid = ?",
+            { ps -> ps.setString(1, flid) },
+        ) { rs ->
+            com.gzzn.omms.msgexchange.support.LEGACY_COLLECTION_STORAGE_COLUMNS.forEach { column ->
+                assertNull(rs.getString(column), "legacy column $column must be null for v2 persistNextStates (flid=$flid)")
+            }
+            true
+        }
+    }
 }