fix(processing): 接通 Redis 投影并修正验证缺口

2026-09-21 17:03:26 +08:00
parent 6980926a7c
commit 43bb561f47
17 changed files with 162 additions and 43 deletions
@@ -121,7 +121,7 @@ class PipelineProps {
        /** 出站请求落信后等待应答的最长时限（`US-09`；具体取值待 Q）。 */
        var responseTimeout: Duration = Duration.ofMinutes(30)

-        var routingRqfd: String = "OMMSRQFD"
+        var routingRqfd: String = "OSH5RQFD"
        var routingRqrd: String = "OMMSRQRD"
    }

@@ -5,7 +5,7 @@ import io.micronaut.context.annotation.ConfigurationProperties
 /** 航班查询投影所在的 Redis，对应 `msgx.redis.*`；键位沿用旧系统（`C-11`、`Q6`）。 */
@ConfigurationProperties("msgx.redis")
 class RedisProps {
-    /** 有没有 Redis 可用。关着的时候投影写是空操作（`G-REDIS-PROJECTION`）。 */
+    /** 有没有 Redis 可用。关着的时候投影写是空操作。 */
    var enabled: Boolean = false

    /** 航班快照所在的哈希键，字段名是 `FLID`。 */
@@ -965,11 +965,18 @@ class JdbcFlightStateRepository(
        } else {
            "SELECT * FROM ${spec.table} WHERE flid = ? ORDER BY ordinal ASC"
        }
-        return ds.query(sql, { ps -> ps.setString(1, flid); spec.routeKind?.let { ps.setString(2, it) } }, ::detailItem)
+        return ds.query(
+            sql,
+            { ps -> ps.setString(1, flid); spec.routeKind?.let { ps.setString(2, it) } },
+            { rs -> detailItem(rs, spec) },
+        )
    }

-    private fun detailItem(rs: ResultSet): Map<String, String> {
+    private fun detailItem(rs: ResultSet, spec: DetailSpec): Map<String, String> {
        val item = linkedMapOf<String, String>()
+        spec.seqAttr?.let { seqAttr ->
+            rs.getString("source_seq")?.takeIf { it.isNotEmpty() }?.let { item[seqAttr] = it }
+        }
        val md = rs.metaData
        for (i in 1..md.columnCount) {
            val col = md.getColumnName(i).uppercase()
@@ -997,7 +1004,7 @@ class JdbcFlightStateRepository(

        /** 明细读取时排除的追踪列（业务键以报文线格式大写键回传）。 */
        private val IGNORED_DETAIL_COLUMNS =
-            setOf("CREATED_AT", "UPDATED_AT", "FLID", "ORDINAL", "ROUTE_KIND")
+            setOf("CREATED_AT", "UPDATED_AT", "FLID", "ORDINAL", "ROUTE_KIND", "SOURCE_SEQ")

        private const val SELECT_MAIN =
            "SELECT flid, operation_day, state, state_version, last_msg_id, updated_at FROM flight_schd"
@@ -4,42 +4,52 @@ import com.gzzn.omms.msgexchange.config.RedisProps
 import com.gzzn.omms.msgexchange.processing.FlightProjectionPort
 import com.gzzn.omms.msgexchange.processing.FlightProjectionWrite
 import io.micronaut.context.annotation.Requires
+import io.lettuce.core.api.StatefulRedisConnection
 import jakarta.inject.Singleton

 /**
- * Redis 投影适配器骨架：键位与写法在这里定死，**但客户端还没接进来**（`G-REDIS-PROJECTION`）。
+ * Redis 投影适配器：一个哈希键保存全部航班，批次用 Redis 事务原子提交。
 *
 * 写入形态沿用旧系统：一个哈希键（`PARAM:msgx.redis.flight-key`）装全部航班，
 * field 是 `FLID`，value 是整态 JSON，不设过期；删除航班就删掉这个 field（`INV-8`）。
 *
- * 开了 `msgx.redis.enabled` 却没有客户端时，写入直接抛异常而不是假装成功：
- * 按 `INV-10`，这条消息就停在未完成、下轮重试，不会被当成已处理写回信箱。
+ * Redis 调用失败会直接抛异常；按 `INV-10`，消息保持未完成并在下轮重试。
 */
@Requires(property = "msgx.stubs", notEquals = "true")
@Requires(property = "msgx.redis.enabled", value = "true")
@Singleton
 class RedisFlightProjectionPort(
    private val props: RedisProps,
+    private val connection: StatefulRedisConnection<String, String>,
 ) : FlightProjectionPort {

    override fun write(writes: List<FlightProjectionWrite>) {
-        throw UnsupportedOperationException(
-            "redis client not wired yet (G-REDIS-PROJECTION); pending writes=${writes.size} key=${props.flightKey}",
-        )
+        if (writes.isEmpty()) return
+        val commands = connection.sync()
+        commands.multi()
+        try {
+            writes.forEach { write ->
+                when (write) {
+                    is FlightProjectionWrite.Upsert -> commands.hset(props.flightKey, write.flid, write.payloadJson)
+                    is FlightProjectionWrite.Delete -> commands.hdel(props.flightKey, write.flid)
+                }
+            }
+            commands.exec()
+        } catch (failure: RuntimeException) {
+            runCatching { commands.discard() }
+            throw failure
+        }
    }

-    override fun readAll(): List<String> = throw UnsupportedOperationException(
-        "redis client not wired yet (G-REDIS-PROJECTION); key=${props.flightKey}",
-    )
+    override fun readAll(): List<String> = connection.sync().hvals(props.flightKey)

-    override fun ping(): Boolean = false
+    override fun ping(): Boolean = connection.sync().ping().equals("PONG", ignoreCase = true)
 }

 /**
 * 没接 Redis 时的投影出口：什么都不做。
 *
- * 这不是"投影写成功"的承诺——`INV-10` 在真实客户端接通之前只能空转（`G-REDIS-PROJECTION`）。
- * 摆这个 bean 是为了让三步提交的时序在没有 Redis 的环境里也照常跑，而不是让每条航班报文都失败。
+ * 这不是"投影写成功"的承诺。这个 bean 只让明确关闭投影的环境继续运行。
 */
@Requires(property = "msgx.stubs", notEquals = "true")
@Requires(property = "msgx.redis.enabled", notEquals = "true")
@@ -48,7 +58,7 @@ class NoopFlightProjectionPort : FlightProjectionPort {
    private val log = org.slf4j.LoggerFactory.getLogger(NoopFlightProjectionPort::class.java)

    override fun write(writes: List<FlightProjectionWrite>) {
-        log.debug("redis projection disabled, skipping {} write(s) [G-REDIS-PROJECTION]", writes.size)
+        log.debug("redis projection disabled, skipping {} write(s)", writes.size)
    }

    /** 没有投影可读时报错而不是回空列表：空列表会被查询方当成"现在没有航班"（`INV-11`）。 */
@@ -14,7 +14,7 @@ import java.time.ZoneId
 import java.time.format.DateTimeFormatter

 /**
- * 出站 REQ_TRACK 协调：登记、COUTMSGS 落信、超时与 EROR 失败（`G-REQ-TRACK`、`C-4`）。
+ * 出站 REQ_TRACK 协调：登记、COUTMSGS 落信、超时与 EROR 失败（`C-4`）。
 */
@Singleton
 class OutboundRequestService(
@@ -206,7 +206,7 @@ class MessageProcessor(
                        scheduleProcessor.applyScheduleRecords(head, decoded)
                    MsgKind.SchdSubtype.RESP -> {
                        if (!outbound.hasOpenSentRqfd()) {
-                            log.info("SCHD-RESP without open RQFD -> SKIPPED msgId={} [G-RESP-GUARD]", head.msgId)
+                            log.info("SCHD-RESP without open RQFD -> SKIPPED msgId={}", head.msgId)
                            procState.markTerminal(
                                head.msgId, ProcStatus.SKIPPED,
                                lastError = "resp-guard:no-open-req",
@@ -36,10 +36,15 @@ msgx:
    arrived-hours: 1               # 条件 5：实际到港早于当前超过 1 小时
    snap-log-retention-days: 90
    history-store-enabled: false   # 未接通历史存储时 HISTORY_SWEEP 删 0 条
-  redis:                           # 航班查询投影（C-11）；真实客户端未接通（G-REDIS-PROJECTION）
-    enabled: false                 # 关闭时投影写是空操作；开启但无客户端时写入直接失败，不伪装成功
+  redis:                           # 航班查询投影（C-11）
+    enabled: false                 # 关闭时投影写是空操作
    flight-key: flightInfo         # 航班快照哈希键，field = FLID（沿用旧系统）

+# Lettuce 连接由 msgx.redis.enabled 同步启停；地址必须由环境覆盖后再启用。
+redis:
+  enabled: ${msgx.redis.enabled}
+  uri: ${MSGX_REDIS_URI:redis://127.0.0.1:6379}
+
 micronaut:
  application:
    name: msgexchange-nextgen
@@ -11,7 +11,7 @@ import java.sql.DriverManager

 /**
 * 在真实 PostgreSQL 上跑一遍迁移链，确认结果符合预期：`V1__flight_state_baseline.sql`
- * 基线加 `V2__flight_chute_class_type_rename.sql` 更名、`V3__msg_event_hold.sql` 事件归属列
+ * 基线加 V2～V6 增量迁移
 * 依序执行成功，该建的表和 PIPELINE_LOCK 单行种子都在，回填事实落在基线里，`INBOX_CURSOR`、
 * `BACKFILL_TODO`、`idx_evt_flid`、`PROC_STATE` 的处理开始时间列都不复存在；FLIGHT_CHUTE
 * 的类字段列已由 V2 更名为 CCLS/CTYP（SIS 口径）。
@@ -51,13 +51,17 @@ class FlywayMigrationTest {
                    while (rs.next()) {
                        records.add(Triple(rs.getString("version"), rs.getString("script"), rs.getBoolean("success")))
                    }
-                    assertEquals(3, records.size, "迁移链应为 V1 基线 + V2 更名 + V3 事件归属三条")
-                    assertEquals("1", records[0].first)
-                    assertEquals("V1__flight_state_baseline.sql", records[0].second)
-                    assertEquals("2", records[1].first)
-                    assertEquals("V2__flight_chute_class_type_rename.sql", records[1].second)
-                    assertEquals("3", records[2].first)
-                    assertEquals("V3__msg_event_hold.sql", records[2].second)
+                    assertEquals(
+                        listOf(
+                            "1" to "V1__flight_state_baseline.sql",
+                            "2" to "V2__flight_chute_class_type_rename.sql",
+                            "3" to "V3__msg_event_hold.sql",
+                            "4" to "V4__req_track_outbound_seqn.sql",
+                            "5" to "V5__unmapped_field_srvt_vipf.sql",
+                            "6" to "V6__basicdata_ref_data.sql",
+                        ),
+                        records.map { it.first to it.second },
+                    )
                    assertTrue(records.all { it.third })
                }

@@ -79,6 +79,26 @@ class JdbcFlightStateRoutePgTest {
        assertEquals(4, loaded.collections["ERUT"]!!.size)
    }

+    @Test
+    fun `CHDT persists and loads SIS class fields`() {
+        assumeTrue(PgTestSupport.canConnect(), PgTestSupport.skipMessage())
+        val repo = JdbcFlightStateRepository(dataSource(), Clock.fixed(t0, ZoneOffset.UTC))
+        val flid = "CH-" + UUID.randomUUID().toString().take(8)
+        val chdt = mapOf("CHNO" to "1", "CHUT" to "C01", "CCLS" to "A", "CTYP" to "IN")
+
+        repo.persistFullState(
+            FlightSnapshot(
+                flid, LocalDate.of(2026, 9, 12), FlightState.ACTIVE, 1,
+                mapOf("SODT" to "12Sep261200"),
+                mapOf("CHDT" to listOf(chdt)),
+            ),
+            msgId = 2,
+            now = t0,
+        )
+
+        assertEquals(listOf(chdt), repo.loadFullSnapshot(flid)!!.collections["CHDT"])
+    }
+
    private fun dataSource(): javax.sql.DataSource {
        Flyway.configure()
            .dataSource(PgTestSupport.jdbcUrl, PgTestSupport.user, PgTestSupport.password)
@@ -0,0 +1,40 @@
+package com.gzzn.omms.msgexchange.infra.projection
+
+import com.gzzn.omms.msgexchange.config.RedisProps
+import com.gzzn.omms.msgexchange.processing.FlightProjectionWrite
+import com.gzzn.omms.msgexchange.support.RedisTestSupport
+import io.lettuce.core.RedisClient
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Assumptions.assumeTrue
+import org.junit.jupiter.api.Test
+import java.util.UUID
+
+class RedisFlightProjectionPortTest {
+
+    @Test
+    fun `writes reads and deletes flight hash in real Valkey`() {
+        assumeTrue(RedisTestSupport.canConnect(), RedisTestSupport.skipMessage())
+        val client = RedisClient.create(RedisTestSupport.uri)
+        client.connect().use { connection ->
+            val key = "flightInfo:test:${UUID.randomUUID()}"
+            val port = RedisFlightProjectionPort(RedisProps().apply { flightKey = key }, connection)
+            try {
+                port.write(
+                    listOf(
+                        FlightProjectionWrite.Upsert("F1", 1, """{"flid":"F1","stateVersion":1}"""),
+                        FlightProjectionWrite.Upsert("F2", 1, """{"flid":"F2","stateVersion":1}"""),
+                    ),
+                )
+                assertEquals(setOf("F1", "F2"), port.readAll().map { Regex("F\\d").find(it)!!.value }.toSet())
+
+                port.write(listOf(FlightProjectionWrite.Delete("F1", 2)))
+                assertEquals(listOf("""{"flid":"F2","stateVersion":1}"""), port.readAll())
+                assertTrue(port.ping())
+            } finally {
+                connection.sync().del(key)
+            }
+        }
+        client.shutdown()
+    }
+}
@@ -0,0 +1,31 @@
+package com.gzzn.omms.msgexchange.support
+
+import org.testcontainers.containers.GenericContainer
+
+private class ValkeyContainer(image: String) : GenericContainer<ValkeyContainer>(image)
+
+/** Real Valkey endpoint for projection adapter tests; skips when neither env nor Docker is available. */
+object RedisTestSupport {
+    private val envUri = System.getenv("MSGX_REDIS_URI")
+
+    private val container: ValkeyContainer? by lazy {
+        if (envUri != null) null else runCatching {
+            ValkeyContainer("valkey/valkey:8-alpine").withExposedPorts(6379).also { it.start() }
+        }.getOrNull()
+    }
+
+    val uri: String
+        get() = envUri ?: container?.let { "redis://${it.host}:${it.getMappedPort(6379)}" }
+            ?: "redis://127.0.0.1:6379"
+
+    fun canConnect(): Boolean = runCatching {
+        val client = io.lettuce.core.RedisClient.create(uri)
+        try {
+            client.connect().use { it.sync().ping() }
+        } finally {
+            client.shutdown()
+        }
+    }.isSuccess
+
+    fun skipMessage(): String = "Valkey not accessible (set MSGX_REDIS_URI or enable Docker for Testcontainers)"
+}