feat(ref): 21 类静态主数据独立 PostgreSQL 参考库边界就位（ACM2-11 落地）

按 ACM2-11 定案调整（业务事务库 MySQL + 航班动态 Redis 不变）： - 接口拆分：RefDataRepository 收为 gen-only（SCHD_GEN/流程 4 CAS，留守业务库）；新增 StaticRefRepository + RefUpsert 对齐 SOURCE 审计（N19）。 - 调用点：RequestCoordinator 应答落库、ReferenceService(21 类同步 TODO) 指向 StaticRefRepository；Decision.refUpserts 与 Pump 事务 2 注释同步（静态写为弱事务，不入主泵事务 2）。 - Stub：StubRefData 仅 gen；新增 StubStaticRef（内存，保留 source）。 - 配置：datasources.reference / flyway.datasources.reference 占位（PostgreSQL，enabled=false 不建连；驱动 org.postgresql 随阶段 6 实装引入，版本由 platform BOM 约束 ~42.7）；catalog 补 postgresql。 - 文档：README / docs/architecture.md §2/§6 / docs/design.md §2/§6/§9 数据边界全部按"21 类 → PG 参考库、gen 留守 MySQL、Redis 动态不变"更新。验证：37 测试全绿（./gradlew test，接口拆分后无破坏）。表结构/迁移/ReferenceService 实装/应答接线属阶段 6（U05 批次后）。
2026-09-07 07:56:54 +08:00
parent 7e3bf383a5
commit a3f1da4bcc
13 changed files with 100 additions and 24 deletions
@@ -9,7 +9,7 @@ data class Decision(
    val msgNotifies: List<NotifyPayload> = emptyList(),      // → MSG_EVENT(KAFKA:msg)
    val schdPush: List<SchdPush> = emptyList(),              // → MSG_EVENT(KAFKA:schd)，PARTITION_KEY=FLID
    val outboundIntents: List<OutboundIntent> = emptyList(), // → COUTMSGS（沿用既有列语义）
-    val refUpserts: List<RefUpsert> = emptyList(),           // → REF_DATA
+    val refUpserts: List<RefUpsert> = emptyList(),           // → 静态主数据（独立 PG reference 库，ACM2-11）
 )

 /** 航班状态变更（阶段 A 由主泵线程 redisApply；阶段 B 落 FLIGHT_STATE 同事务）。 */
@@ -5,6 +5,7 @@ import com.gzzn.omms.msgexchange.nextgen.domain.EventStatus
 import com.gzzn.omms.msgexchange.nextgen.domain.MsgEvent
 import com.gzzn.omms.msgexchange.nextgen.domain.ProcState
 import com.gzzn.omms.msgexchange.nextgen.domain.ProcStatus
+import com.gzzn.omms.msgexchange.nextgen.domain.RefUpsert
 import java.time.Instant

 /**
@@ -58,6 +59,11 @@ interface MsgEventRepository {
    fun insertSync(events: List<MsgEvent>)
 }

+/**
+ * 快照 generation（SCHD_GEN）协议——只留 gen（ACM2-11 拆分后）。
+ * 位置：业务事务库（MySQL，与 PROC_STATE/SUCCEEDED 同事务，流程 4 重放幂等依据）；
+ * 21 类静态主数据已迁独立 PostgreSQL 参考库（见 [StaticRefRepository]，阶段 6 实装）。
+ */
 interface RefDataRepository {
    data class GenMeta(val flids: List<String>, val version: Long)

@@ -65,8 +71,18 @@ interface RefDataRepository {

    /** 流程 4：版本 CAS（expected 未变才写，重放 no-op，不二次自增）。 */
    fun putGenIfVersion(day: String, expected: Long, new: GenMeta): Boolean
+}

-    fun upsertAll(rows: List<Triple<String, String, String>>)   // (rtype, rkey, payloadJson)
+/**
+ * 21 类静态主数据（航空公司/航线/机位/登机桥等）——独立 PostgreSQL 参考库
+ * （`datasources.reference`，ACM2-11 定案；SOURCE=ADMINAPI/AODB/PIPELINE，N19 对齐）。
+ * 与主处理链路弱事务耦合：写入者为 ReferenceService（21 类同步）与请求应答路径，
+ * 不参与主泵事务 2；Redis 只作只读热点投影（legacy orms_stand 语义延续，阶段 6）。
+ */
+interface StaticRefRepository {
+    fun upsertAll(refs: List<RefUpsert>)
+
+    fun findByType(type: String): List<RefUpsert>
 }

 interface ReqTrackRepository {
@@ -5,6 +5,7 @@ import com.gzzn.omms.msgexchange.nextgen.domain.EventStatus
 import com.gzzn.omms.msgexchange.nextgen.domain.MsgEvent
 import com.gzzn.omms.msgexchange.nextgen.domain.ProcState
 import com.gzzn.omms.msgexchange.nextgen.domain.ProcStatus
+import com.gzzn.omms.msgexchange.nextgen.domain.RefUpsert
 import com.gzzn.omms.msgexchange.nextgen.infra.persistence.CminmsgInboxRepository
 import com.gzzn.omms.msgexchange.nextgen.infra.persistence.FlightStateRepository
 import com.gzzn.omms.msgexchange.nextgen.infra.persistence.MsgEventRepository
@@ -12,6 +13,7 @@ import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ProcStateRepository
 import com.gzzn.omms.msgexchange.nextgen.infra.persistence.PumpJobRepository
 import com.gzzn.omms.msgexchange.nextgen.infra.persistence.RefDataRepository
 import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ReqTrackRepository
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.StaticRefRepository
 import io.micronaut.context.annotation.Requires
 import jakarta.inject.Singleton
 import java.time.Instant
@@ -166,13 +168,13 @@ class StubPumpJobs : PumpJobRepository {
    override fun markFailed(jobId: Long, lastError: String) { rows[jobId]?.state = "FAILED"; rows[jobId]?.lastError = lastError }
 }

+/** 快照 gen 协议 stub（业务库 REF_DATA 的 SCHD_GEN 行；ACM2-11 拆分后 21 类不在本实现）。 */
@Requires(property = "msgx.stubs", value = "true")
@Singleton
 class StubRefData : RefDataRepository {
    private val gens = mutableMapOf<String, RefDataRepository.GenMeta>()
-    private val flat = mutableMapOf<Pair<String, String>, String>()

-    fun clear() { gens.clear(); flat.clear() }
+    fun clear() { gens.clear() }

    override fun getGen(day: String): RefDataRepository.GenMeta? = gens[day]

@@ -182,10 +184,22 @@ class StubRefData : RefDataRepository {
        gens[day] = new
        return true
    }
+}

-    override fun upsertAll(rows: List<Triple<String, String, String>>) {
-        rows.forEach { flat[it.first to it.second] = it.third }
+/** 21 类静态主数据 stub（独立 PG reference 库；内存实现，source 保留供审计断言）。 */
+@Requires(property = "msgx.stubs", value = "true")
+@Singleton
+class StubStaticRef : StaticRefRepository {
+    private val rows = mutableMapOf<Pair<String, String>, RefUpsert>()
+
+    fun clear() { rows.clear() }
+
+    override fun upsertAll(refs: List<RefUpsert>) {
+        refs.forEach { rows[it.rtype to it.rkey] = it }
    }
+
+    override fun findByType(type: String): List<RefUpsert> =
+        rows.values.filter { it.rtype == type }
 }

@Requires(property = "msgx.stubs", value = "true")
@@ -198,7 +198,8 @@ class MessageProcessor(
        if (props.phase == PipelineProps.Phase.A) {
            // 阶段 A：主泵线程先写 Redis（幂等），先于事件创建（I2）；TODO: redisApply(flightChanges)
        }
-        // 事务 2：事件 + 出站意图 + REF_DATA + CMINMSGS 回填 + SUCCEEDED（实装后 @Transactional）
+        // 事务 2：事件 + CMINMSGS 回填 + SUCCEEDED（实装后 @Transactional）；
+        // 静态主数据（refUpserts）落独立 PG reference 库，弱事务不入本事务（ACM2-11）
        val events = buildList {
            decision.msgNotifies.forEach { add(MsgEvent(target = Targets.KAFKA_MSG, payloadJson = it.payloadJson)) }
            decision.schdPush.forEach { add(MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = it.flid, payloadJson = it.fltrJson)) }
@@ -1,18 +1,19 @@
 package com.gzzn.omms.msgexchange.nextgen.reference

-import com.gzzn.omms.msgexchange.nextgen.infra.persistence.RefDataRepository
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.StaticRefRepository
 import jakarta.inject.Singleton

 /**
 * ACMA-8 决策 6：21 类 admin-api 基础数据同步（阶段 6）。
- * 只产 REF_DATA 行（权威写唯一入口）；影子实例不运行本模块（v4 影子二分）。
+ * 落点 = 21 类静态主数据独立 PostgreSQL 参考库（StaticRefRepository，ACM2-11）；
+ * 只产静态主数据行（权威写唯一入口）；影子实例不运行本模块（v4 影子二分）。
 */
@Singleton
 class ReferenceService(
-    private val refData: RefDataRepository,
+    private val staticRef: StaticRefRepository,
    // TODO(阶段6): @Client(id="ADMINAPI") 声明式客户端 + 21 类端点配置化清单
 ) {
    fun refresh(type: String) {
-        // TODO(阶段6): 拉取 → upsertAll(type, key, payload)；失败旧数据可用（REF_DATA 保留旧值）
+        // TODO(阶段6): 拉取 → staticRef.upsertAll(rows, source=ADMINAPI)；失败旧数据可用（参考库保留旧值）
    }
 }
@@ -1,7 +1,8 @@
 package com.gzzn.omms.msgexchange.nextgen.reference

-import com.gzzn.omms.msgexchange.nextgen.infra.persistence.RefDataRepository
+import com.gzzn.omms.msgexchange.nextgen.domain.RefUpsert
 import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ReqTrackRepository
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.StaticRefRepository
 import jakarta.inject.Singleton
 import java.time.Instant

@@ -9,11 +10,13 @@ import java.time.Instant
 * ACMA-8 流程 6：请求式查询状态机（15 类，决策树版 v4）。
 * 同类并发=1（注册新请求先强制旧请求 EXPIRED，终态不再被匹配）；
 * 应答匹配：回显字段 CONFIRM（矩阵 #13）→ 精确匹配；否则退化模式（DTTM ≥ SENT 才应用 + 审计）。
+ * 应答落库目标 = 21 类静态主数据（独立 PG 参考库，StaticRefRepository，ACM2-11）；
+ * 注：N04 量纲缺陷（dttm vs epochMillis）与 N16 死分支未修，接线阶段 6 前处理（U19–U21）。
 */
@Singleton
 class RequestCoordinator(
    private val reqTrack: ReqTrackRepository,
-    private val refData: RefDataRepository,
+    private val staticRef: StaticRefRepository,
 ) {
    fun request(kind: String, rangeJson: String): Long {
        reqTrack.forceExpireOpenOf(kind)                       // 防护①
@@ -29,7 +32,7 @@ class RequestCoordinator(
        kind: String,
        dttm: Long,
        echoSeqn: Long?,
-        records: List<Triple<String, String, String>>,         // (rtype, rkey, payloadJson)
+        records: List<RefUpsert>,               // 应答载荷 → 静态主数据（source=AODB）
    ): String {
        val req = reqTrack.findOpenByKind(kind) ?: return audit("late/unknown resp kind=$kind")
        val echoConfirmed = false                               // CONFIRM（矩阵 #13）待机场方结论
@@ -44,8 +47,8 @@ class RequestCoordinator(
        }
    }

-    private fun applyResp(reqId: Long, records: List<Triple<String, String, String>>) {
-        refData.upsertAll(records)                              // 大应答复用流程 4 staging 路径（阶段 6）
+    private fun applyResp(reqId: Long, records: List<RefUpsert>) {
+        staticRef.upsertAll(records)                            // 独立 PG 参考库（阶段 6 接线）
        reqTrack.markDone(reqId)
    }

@@ -38,6 +38,15 @@ datasources:
    username: ${MSGX_DB_USER}
    password: ${MSGX_DB_PASSWORD}
    driver-class-name: com.mysql.cj.jdbc.Driver
+  # ACM2-11：21 类静态主数据独立 PostgreSQL 参考库（reference datasource）。
+  # 数据层实装（阶段 6 / U05 批次）前默认 disabled——不建连不解析占位符；
+  # 接入时置 enabled=true 并设 MSGX_REF_DB_URL（或 host/port/name）等环境变量。
+  reference:
+    enabled: false
+    url: ${MSGX_REF_DB_URL}
+    username: ${MSGX_REF_DB_USER:reference}
+    password: ${MSGX_REF_DB_PASSWORD:}
+    driver-class-name: org.postgresql.Driver

 flyway:
  datasources:
@@ -46,6 +55,10 @@ flyway:
      locations: classpath:db/migration
      # U04（R01）：V2.0.0 只建六张辅助表，假定 CMINMSGS/CMINMSGS_HST 等 legacy 旧表已存在；
      # 全新库请按 README「数据库初始化」先落 legacy schema（或 baseline），否则收报首句 SQL 会报表不存在。
+    reference:
+      # ACM2-11：21 类静态主数据 PG 库迁移（类表 DDL 草案见 ACM2-11 Checks，阶段 6 落库）
+      enabled: false
+      locations: classpath:db/ref-migration

 redis:
  uri: ${MSGX_REDIS_URI}            # 阶段 A 权威存储；仅主泵线程写（I5）
@@ -144,7 +144,6 @@ class MessageProcessorTest {
    private class FakeRefData : RefDataRepository {
        override fun getGen(day: String): RefDataRepository.GenMeta? = null
        override fun putGenIfVersion(day: String, expected: Long, new: RefDataRepository.GenMeta): Boolean = true
-        override fun upsertAll(rows: List<Triple<String, String, String>>) = Unit
    }

    // ---------- helpers ----------