From b2f3b896ccb64c5fd6f77943960cf51dcfadc125 Mon Sep 17 00:00:00 2001 From: windyboy Date: Sun, 6 Sep 2026 16:13:50 +0800 Subject: [PATCH] feat: scaffold Micronaut+Kotlin service per ACMA-8 v4 architecture - 4 modules (ingress/processing/delivery/reference) + codec + infra + jobs - Flyway V2.0.0: six aux tables (PROC_STATE/MSG_EVENT/REF_DATA/REQ_TRACK/PUMP_JOB/FLIGHT_STATE) - Lua: snapshot_replace (atomic cover+gen-diff-delete), batch_delete (3:30 sweep) - config: env-externalized secrets, phase A/B switch, ACMA-8 param-table defaults - logback: logstash TCP JSON + traceId MDC - pure-logic tests: identity (I3), schd aggregation max-by-EVENT_ID (FIX #7) - build verified: gradle compile+test green on JDK 25 (Micronaut 5.1 requires JVM 25+) Migrated from legacy repo subdirectory per repo-strategy decision (separate repo per service, org convention). Refs: ACMA-9, ACMA-8 v4, ACMA-6 --- .gitignore | 5 + README.md | 54 ++++++ build.gradle.kts | 55 ++++++ gradle.properties | 3 + gradle/libs.versions.toml | 34 ++++ settings.gradle.kts | 11 ++ .../omms/msgexchange/nextgen/Application.kt | 7 + .../msgexchange/nextgen/codec/XmlCodec.kt | 24 +++ .../nextgen/config/PipelineProps.kt | 47 ++++++ .../nextgen/delivery/Dispatcher.kt | 99 +++++++++++ .../nextgen/delivery/SchdAggregation.kt | 22 +++ .../msgexchange/nextgen/domain/Decision.kt | 40 +++++ .../nextgen/domain/DecodedMessage.kt | 34 ++++ .../msgexchange/nextgen/domain/MsgEvent.kt | 33 ++++ .../msgexchange/nextgen/domain/ProcState.kt | 25 +++ .../nextgen/infra/persistence/Repositories.kt | 120 +++++++++++++ .../nextgen/infra/redis/RedisScripts.kt | 26 +++ .../nextgen/ingress/InboxController.kt | 24 +++ .../nextgen/ingress/InboxService.kt | 26 +++ .../msgexchange/nextgen/jobs/JobExecutor.kt | 80 +++++++++ .../msgexchange/nextgen/processing/Handler.kt | 34 ++++ .../nextgen/processing/Identity.kt | 24 +++ .../msgexchange/nextgen/processing/Pump.kt | 158 ++++++++++++++++++ .../nextgen/processing/SnapshotFlow.kt | 63 +++++++ .../nextgen/reference/ReferenceService.kt | 18 ++ .../nextgen/reference/RequestCoordinator.kt | 57 +++++++ src/main/resources/application.yml | 60 +++++++ .../db/migration/V2.0.0__aux_tables.sql | 78 +++++++++ src/main/resources/logback.xml | 24 +++ src/main/resources/lua/batch_delete.lua | 9 + src/main/resources/lua/snapshot_replace.lua | 19 +++ .../nextgen/delivery/SchdAggregationTest.kt | 56 +++++++ .../nextgen/processing/IdentityTest.kt | 45 +++++ 33 files changed, 1414 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 build.gradle.kts create mode 100644 gradle.properties create mode 100644 gradle/libs.versions.toml create mode 100644 settings.gradle.kts create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/Application.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/codec/XmlCodec.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/config/PipelineProps.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/Dispatcher.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/SchdAggregation.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/Decision.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/DecodedMessage.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/MsgEvent.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/ProcState.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/persistence/Repositories.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/redis/RedisScripts.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxController.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxService.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/jobs/JobExecutor.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Handler.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Identity.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/SnapshotFlow.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/reference/ReferenceService.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/reference/RequestCoordinator.kt create mode 100644 src/main/resources/application.yml create mode 100644 src/main/resources/db/migration/V2.0.0__aux_tables.sql create mode 100644 src/main/resources/logback.xml create mode 100644 src/main/resources/lua/batch_delete.lua create mode 100644 src/main/resources/lua/snapshot_replace.lua create mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/SchdAggregationTest.kt create mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/IdentityTest.kt diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..dcd9b00 --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +.gradle/ +build/ +out/ +*.iml +.idea/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..1fd7e06 --- /dev/null +++ b/README.md @@ -0,0 +1,54 @@ +# msgexchange-api-v2（新一代消息交换服务） + +依据 **ACMA-8 v4 综合架构**（单写者严格 FIFO 管道 + 两阶段权威）与 **ACMA-6 技术选型** +（Micronaut 5.1 + Kotlin 2.3）搭建的新一代消息交换服务工程。本仓库独立于 legacy +`msgexchange-api`（Java 8 / Spring Boot 1.5 / Maven）——过渡期两套系统并存（影子对拍→ +切流→旧仓库冻结），legacy 维护不受本仓库影响。 + +> **JDK 口径实测修正**：Micronaut 5.1 系构件（如 micronaut-http-server-netty:5.1.10） +> 要求 JVM 25+，计划原定 JDK 21 不可行；工程已按 **JDK 25** 配置（ACMA-9 记录）。 + +## 包结构 → ACMA-8 架构映射 + +| 包 | 职责 | 对应 ACMA-8 | +|---|---|---| +| `ingress/` | Ingress & Inbox：接收事务（事务1），不解析报文 | 流程 1，I3 | +| `processing/` | Processing 主泵：严格 FIFO 领取、identity 绑定、纯函数决策、事务2 | 流程 2/4，I1/I2/I5 | +| `delivery/` | Delivery & Projection：每 target 严格 FIFO 投递、schd 聚合 | 流程 3 | +| `reference/` | Reference & Query：21 类同步 + 15 类请求状态机 | 流程 6 | +| `jobs/` | 泵作业：ARCHIVE / HISTORY_SWEEP / PROJECTION_REBUILD | 流程 4/5/7，I4 | +| `codec/` | XML codec（阶段 1 先 vendor 复用 legacy POJO，见 ACMA-6 选型） | 决策 4 前置 | +| `domain/` | 领域模型：Decision、事件、状态机枚举、Phase 开关 | I1–I5 | +| `infra/` | 仓储接口、Redis Lua 装载、配置 | 数据模型节 | + +## 资源 + +- `db/migration/V2.0.0__aux_tables.sql`：六表 DDL（PROC_STATE / MSG_EVENT / REF_DATA / + REQ_TRACK / PUMP_JOB / FLIGHT_STATE），与 ACMA-8 v4 数据模型节逐字一致；影子实例在 + 独立 schema 执行同一脚本。 +- `lua/snapshot_replace.lua`：同一 hash 原子“覆盖新代 + 按代差删”（流程 4，I4/I5）。 +- `lua/batch_delete.lua`：3:30 清场批量删除（仅 ES 写成功集，I4）。 +- `application.yml`：口令全部环境变量外置（零入库）；`msgx.phase` 为阶段 A/B 总开关； + pipeline 参数 = ACMA-8 参数表初值。 + +## 未完成（按计划属于后续阶段，不是本脚手架遗漏） + +1. **Handler 业务（3+29）**：`processing/HandlerRegistry` 仅注册骨架，翻译属阶段 2/3。 +2. **codec 实装**：vendor 复用 legacy `entity/msg` POJO + Jackson XML（ACMA-6 选型）， + 阶段 1 后续项。 +3. **依赖版本锁定**：`gradle/libs.versions.toml` 中版本为计划口径，需阶段 0 + 「Micronaut×现网 Eureka 互操作冒烟 + logstash + ES REST」通过后固化。 +4. **仓储实装**：`infra/persistence/Repositories.kt` 目前是接口（Micronaut Data JDBC + 实装属阶段 1 后续），主泵/调度循环以接口驱动，纯逻辑已抽离可单测。 + +## 构建 + +```bash +gradle build # 需网络拉取依赖；内网环境见 gradle.properties 注释 +gradle test # 纯逻辑单测（identity / schd 聚合） +``` + +## 关联 + +Plane `airport_chengdu_msgexchange_api`：ACMA-9（本阶段跟踪）、ACMA-8 v4（架构）、 +ACMA-6（技术选型）、ACMA-3（总计划）、ACMA-4（行为对拍基线）。 diff --git a/build.gradle.kts b/build.gradle.kts new file mode 100644 index 0000000..36855a6 --- /dev/null +++ b/build.gradle.kts @@ -0,0 +1,55 @@ +plugins { + alias(libs.plugins.kotlin.jvm) + alias(libs.plugins.micronaut.application) +} + +group = "com.gzzn.omms" +version = "2.0.0-SNAPSHOT" + +repositories { + mavenCentral() +} + +java { + toolchain { + // Micronaut 5.1 系构件要求 JVM 25+（实测依赖解析结论，修正计划的 JDK 21 口径） + languageVersion = JavaLanguageVersion.of(25) + } +} + +kotlin { + jvmToolchain(25) +} + +micronaut { + runtime("netty") + testRuntime("junit5") + processing { + incremental(true) + annotations("com.gzzn.omms.msgexchange.nextgen.*") + } +} + +dependencies { + implementation(libs.bundles.micronaut.runtime) + implementation(libs.micronaut.data.jdbc) + implementation(libs.micronaut.jdbc.hikari) + implementation(libs.micronaut.flyway) + implementation(libs.micronaut.redis.lettuce) + implementation(libs.micronaut.kafka) + implementation(libs.micronaut.discovery.eureka) + implementation(libs.jackson.dataformat.xml) + implementation(libs.jackson.module.kotlin) + implementation(libs.logstash.logback.encoder) + + runtimeOnly(libs.mysql.connector.j) + runtimeOnly("org.yaml:snakeyaml") + + testImplementation(libs.junit.jupiter) + testImplementation("io.micronaut.test:micronaut-test-junit5") + testImplementation("org.jetbrains.kotlin:kotlin-test") +} + +tasks.test { + useJUnitPlatform() +} diff --git a/gradle.properties b/gradle.properties new file mode 100644 index 0000000..2ff5376 --- /dev/null +++ b/gradle.properties @@ -0,0 +1,3 @@ +org.gradle.jvmargs=-Xmx1g +kotlin.code.style=official +# 阶段 0 锁定版本冒烟通过后，可将依赖镜像指向内网 Maven 仓库（沿用现网环境前提） diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml new file mode 100644 index 0000000..ae110de --- /dev/null +++ b/gradle/libs.versions.toml @@ -0,0 +1,34 @@ +# 版本为 ACMA-6 定稿口径（Micronaut 5.1 / Kotlin 2.3 / JDK 21）。 +# 注意：具体版本号在阶段 0「锁定版本互操作冒烟」（现网 Eureka/logstash/ES）通过后固化， +# 见 ACMA-8 遗留开放点与 ACMA-9 Checks。 +[versions] +micronaut = "5.1.0" +kotlin = "2.3.0" +junit = "5.11.4" + +[plugins] +kotlin-jvm = { id = "org.jetbrains.kotlin.jvm", version.ref = "kotlin" } +micronaut-application = { id = "io.micronaut.application", version = "4.6.0" } + +[libraries] +micronaut-http-server-netty = { module = "io.micronaut:micronaut-http-server-netty" } +micronaut-jackson-databind = { module = "io.micronaut:micronaut-jackson-databind" } +micronaut-data-jdbc = { module = "io.micronaut.data:micronaut-data-jdbc" } +micronaut-jdbc-hikari = { module = "io.micronaut.sql:micronaut-jdbc-hikari" } +micronaut-flyway = { module = "io.micronaut.flyway:micronaut-flyway" } +micronaut-redis-lettuce = { module = "io.micronaut.redis:micronaut-redis-lettuce" } +micronaut-kafka = { module = "io.micronaut.kafka:micronaut-kafka" } +micronaut-discovery-eureka = { module = "io.micronaut.discovery:micronaut-discovery-client" } +micronaut-management = { module = "io.micronaut:micronaut-management" } +jackson-dataformat-xml = { module = "com.fasterxml.jackson.dataformat:jackson-dataformat-xml", version = "2.18.2" } +jackson-module-kotlin = { module = "com.fasterxml.jackson.module:jackson-module-kotlin", version = "2.18.2" } +logstash-logback-encoder = { module = "net.logstash.logback:logstash-logback-encoder", version = "8.0" } +mysql-connector-j = { module = "com.mysql:mysql-connector-j", version = "9.1.0" } +junit-jupiter = { module = "org.junit.jupiter:junit-jupiter", version.ref = "junit" } + +[bundles] +micronaut-runtime = [ + "micronaut-http-server-netty", + "micronaut-jackson-databind", + "micronaut-management", +] diff --git a/settings.gradle.kts b/settings.gradle.kts new file mode 100644 index 0000000..cb930da --- /dev/null +++ b/settings.gradle.kts @@ -0,0 +1,11 @@ +rootProject.name = "msgexchange-api-v2" + +pluginManagement { + repositories { + gradlePluginPortal() + mavenCentral() + } +} + +// 新一代消息交换服务：ACMA-8 v4 架构（单写者严格 FIFO 管道 + 两阶段权威）。 +// 独立仓库，与 legacy msgexchange-api 并存至切流完成。 diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/Application.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/Application.kt new file mode 100644 index 0000000..64f5884 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/Application.kt @@ -0,0 +1,7 @@ +package com.gzzn.omms.msgexchange.nextgen + +import io.micronaut.runtime.Micronaut + +fun main(args: Array) { + Micronaut.run(*args) +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/codec/XmlCodec.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/codec/XmlCodec.kt new file mode 100644 index 0000000..c383c46 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/codec/XmlCodec.kt @@ -0,0 +1,24 @@ +package com.gzzn.omms.msgexchange.nextgen.codec + +import com.gzzn.omms.msgexchange.nextgen.domain.DecodedMessage +import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass + +/** + * ACMA-8 流程 2：解码失败分类（矩阵/状态机：MALFORMED 不重试；CODEC_ERROR 可一键重放）。 + */ +data class DecodeFailure(val errorClass: ErrorClass, val detail: String) + +sealed interface DecodeResult { + data class Ok(val message: DecodedMessage) : DecodeResult + data class Err(val failure: DecodeFailure) : DecodeResult +} + +/** + * XML codec：XXE 防护（禁 DTD/外部实体）；阶段 1 先 vendor 复用 legacy entity/msg POJO + + * jackson-dataformat-xml + JaxbAnnotationIntrospector（现役已验证组合，ACMA-6 选型）。 + * TODO(阶段1后续): vendor POJO 引入与 BDPB workaround 配置化（golden 固化）。 + */ +interface XmlCodec { + fun decode(rawXml: String): DecodeResult + fun encodeRqrd(kind: String, rangeJson: String): String +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/config/PipelineProps.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/config/PipelineProps.kt new file mode 100644 index 0000000..bf380d9 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/config/PipelineProps.kt @@ -0,0 +1,47 @@ +package com.gzzn.omms.msgexchange.nextgen.config + +import io.micronaut.context.annotation.ConfigurationProperties +import java.time.Duration + +/** + * ACMA-8 参数表（v4）初值；阶段 0 现网基线校准。 + */ +@ConfigurationProperties("msgx") +class PipelineProps { + var phase: Phase = Phase.A + var serviceName: String = "msgexchangeapi" + var registerEureka: Boolean = true + var pipeline: Pipeline = Pipeline() + var schd: Schd = Schd() + var identity: Identity = Identity() + var consistencyCheck: ConsistencyCheck = ConsistencyCheck() + + enum class Phase { A, B } + + class Pipeline { + var pollInterval: Duration = Duration.ofSeconds(1) // KEEP 现役节奏 + var claimBatch: Int = 50 + var maxAttempts: Int = 5 // 处理/投递同值 + var backoffMs: List = listOf(1000, 2000, 4000, 8000, 16000) + var backoffCapMs: Long = 60_000 + var headDeadline: Duration = Duration.ofMinutes(10) // 最坏 HOL 上界（毒丸升级） + + fun backoffFor(attempt: Int): Long = + backoffMs.drop(attempt - 1).firstOrNull()?.coerceAtMost(backoffCapMs) ?: backoffCapMs + } + + class Schd { + var flushPeriod: Duration = Duration.ofSeconds(3) // KEEP 现役节律 + var flushLimit: Int = 500 + } + + class Identity { + /** CONFIRM（矩阵 #11）：SEQN 重置作用域确认前保持 false，计算集中此处（I3）。 */ + var includeDayBoundary: Boolean = false + } + + class ConsistencyCheck { + var onStartup: Boolean = true // 阶段 A 必选哨兵 + var dailySampleRatio: Double = 0.01 + } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/Dispatcher.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/Dispatcher.kt new file mode 100644 index 0000000..1735331 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/Dispatcher.kt @@ -0,0 +1,99 @@ +package com.gzzn.omms.msgexchange.nextgen.delivery + +import com.gzzn.omms.msgexchange.nextgen.config.PipelineProps +import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass +import com.gzzn.omms.msgexchange.nextgen.domain.EventStatus +import com.gzzn.omms.msgexchange.nextgen.domain.MsgEvent +import com.gzzn.omms.msgexchange.nextgen.domain.Targets +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.MsgEventRepository +import jakarta.inject.Singleton +import java.time.Duration +import java.time.Instant + +/** 对外投递端口（Kafka 同步确认；阶段 B 追加 ES/Redis 投影写入）。 */ +interface DeliveryPort { + /** Kafka 发送（同步确认，at-least-once）。topic 由 target 映射：KAFKA:msg→"msg"，KAFKA:schd→"schd"。 */ + fun sendKafka(topic: String, payloadJson: String) + + /** 阶段 B：ES flight_hts 写入。 */ + fun indexFlightHts(payloadJson: String) + + /** 阶段 B：Redis 投影写（仅阶段 B；I5：Delivery 阶段 A 不写 Redis）。 */ + fun projectRedis(payloadJson: String) +} + +/** + * ACMA-8 流程 3：投递调度——每 target 严格 FIFO（I1 双层同策略）； + * 阶段 B（定案 2）：ES 投递成功后同线程同步 enqueue REDIS:flightInfo 删除事件。 + */ +@Singleton +class Dispatcher( + private val msgEvents: MsgEventRepository, + private val port: DeliveryPort, + private val props: PipelineProps, +) { + @Volatile + private var running = true + + private var lastFlush: Instant = Instant.EPOCH + + fun loop() { + while (running) { + tick() + Thread.sleep(200) + } + } + + internal fun tick() { + val targets = if (props.phase == PipelineProps.Phase.A) Targets.phaseA else Targets.phaseB + for (t in targets) { + val head = msgEvents.headUnsent(t) ?: continue + if (head.state == EventStatus.PENDING && head.nextAttemptAt != null && head.nextAttemptAt > Instant.now()) { + // 队头退避未到期：等待，不跳过（保序）；毒丸升级由实装补 attempts/deadline 判定 + continue + } + try { + deliver(t, head) + msgEvents.markSent(head.eventId!!) + if (t == Targets.ES_FLIGHT_HTS && props.phase == PipelineProps.Phase.B) { + // 定案 2：ES 投递成功 → 同线程同步 enqueue 删除事件（不轮询 ack） + msgEvents.insertSync(listOf(MsgEvent(target = Targets.REDIS_FLIGHT_INFO, payloadJson = deleteOf(head)))) + } + } catch (e: Exception) { + val attempts = head.attempts + 1 + if (attempts >= props.pipeline.maxAttempts) { + msgEvents.markDead(head.eventId!!, ErrorClass.EXHAUSTED, e.message ?: "unknown") + } else { + msgEvents.scheduleRetry( + head.eventId!!, + Instant.now().plusMillis(props.pipeline.backoffFor(attempts)), + attempts, + ) + } + } + } + if (Duration.between(lastFlush, Instant.now()) >= props.schd.flushPeriod) { + lastFlush = Instant.now() + flushSchd() + } + } + + private fun deliver(target: String, e: MsgEvent) = when (target) { + Targets.KAFKA_MSG -> port.sendKafka("msg", e.payloadJson) + Targets.KAFKA_SCHD -> Unit // schd 走 flushSchd 批量路径 + Targets.ES_FLIGHT_HTS -> port.indexFlightHts(e.payloadJson) + Targets.REDIS_FLIGHT_INFO -> port.projectRedis(e.payloadJson) + else -> error("unknown target $target") + } + + private fun deleteOf(e: MsgEvent) = """{"op":"delete","refs":${e.partitionKey ?: ""}}""" + + /** 流程 3 flushSchd：批上限 + 每 FLID 最新一态聚合（topic "schd"，wire=FLTR JSON 数组）。 */ + internal fun flushSchd() { + val pending = msgEvents.claimBatch(Targets.KAFKA_SCHD, limit = props.schd.flushLimit) + if (pending.isEmpty()) return + val payload = SchdAggregation.latestPerFlight(pending).joinToString(",", "[", "]") + port.sendKafka("schd", payload) // 失败整批退避（at-least-once） + msgEvents.markAllSent(pending.mapNotNull { it.eventId }) + } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/SchdAggregation.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/SchdAggregation.kt new file mode 100644 index 0000000..e329650 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/SchdAggregation.kt @@ -0,0 +1,22 @@ +package com.gzzn.omms.msgexchange.nextgen.delivery + +import com.gzzn.omms.msgexchange.nextgen.domain.MsgEvent +import com.gzzn.omms.msgexchange.nextgen.domain.SchdPush + +/** + * ACMA-8 流程 3 flushSchd 聚合（纯函数，可单测）： + * 按 PARTITION_KEY(=FLID) 分组、组内取 EVENT_ID 最大——v4 显式 max(EVENT_ID)， + * 不依赖集合遍历序（FIX：现役 buffer 无去重会重复投递旧值）。 + */ +object SchdAggregation { + fun latestPerFlight(pending: List): List = + pending + .groupBy { it.partitionKey ?: "" } + .map { (_, evs) -> evs.maxBy { it.eventId ?: 0 }.payloadJson } + + /** 由 Decision 直接构造时的等价聚合（测试与 dispatcher 共用语义）。 */ + fun latestPerFlightOfPush(pushes: List): List = + pushes + .groupBy { it.flid } + .map { (_, ps) -> ps.maxBy { it.eventSeq }.fltrJson } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/Decision.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/Decision.kt new file mode 100644 index 0000000..0736e61 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/Decision.kt @@ -0,0 +1,40 @@ +package com.gzzn.omms.msgexchange.nextgen.domain + +/** + * ACMA-8 流程 2：Handler 决策（纯函数）产物——状态与报文进，变更与事件出， + * 不直接触碰 Redis/Kafka。 + */ +data class Decision( + val flightChanges: List = emptyList(), + val msgNotifies: List = emptyList(), // → MSG_EVENT(KAFKA:msg) + val schdPush: List = emptyList(), // → MSG_EVENT(KAFKA:schd)，PARTITION_KEY=FLID + val outboundIntents: List = emptyList(), // → COUTMSGS（沿用既有列语义） + val refUpserts: List = emptyList(), // → REF_DATA +) + +/** 航班状态变更（阶段 A 由主泵线程 redisApply；阶段 B 落 FLIGHT_STATE 同事务）。 */ +data class FlightChange( + val flid: String, + val payloadJson: String, + val maid: String? = null, +) + +data class NotifyPayload(val payloadJson: String) + +data class SchdPush( + val flid: String, + val fltrJson: String, + val eventSeq: Long = 0, // 由事务插入时赋 EVENT_ID 语义序，聚合取 max +) + +data class OutboundIntent( + val coutmsgsXml: String, + val ackReqd: Boolean = true, +) + +data class RefUpsert( + val rtype: String, + val rkey: String, + val payloadJson: String, + val source: String, +) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/DecodedMessage.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/DecodedMessage.kt new file mode 100644 index 0000000..222d25b --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/DecodedMessage.kt @@ -0,0 +1,34 @@ +package com.gzzn.omms.msgexchange.nextgen.domain + +/** + * 解码后的入站报文（统一内部模型）。META 字段实名 SNDR/SEQN/DTTM（legacy META.java，I3 幂等键来源）。 + */ +data class MetaFields( + val sndr: String, + val type: String, + val styp: String, + val seqn: Long, + val dttm: Long, +) + +/** 消息分派（sealed + 穷尽 when，ACMA-6 选型；取代 legacy 反射 get{TYPE}()）。 */ +sealed interface MsgKind { + data class Schd(val subtype: SchdSubtype) : MsgKind + data class Flop(val subtype: String) : MsgKind // 29 类 STYP，阶段 2/3 逐类翻译 + + enum class SchdSubtype { RESP, DNLD, ADFT } +} + +data class DecodedMessage( + val meta: MetaFields, + val kind: MsgKind, + val rawXml: String, + /** 各类型载荷（vendor POJO 或 Kotlin 模型，阶段 1 后续统一）。 */ + val body: Any? = null, +) { + val typeTag: String + get() = when (val k = kind) { + is MsgKind.Schd -> "SCHD-${k.subtype.name}" + is MsgKind.Flop -> "FLOP-${k.subtype}" + } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/MsgEvent.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/MsgEvent.kt new file mode 100644 index 0000000..a4ac0f2 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/MsgEvent.kt @@ -0,0 +1,33 @@ +package com.gzzn.omms.msgexchange.nextgen.domain + +/** + * ACMA-8 MSG_EVENT（统一投递事件 / outbox）。 + * TARGET 阶段化：KAFKA:msg、KAFKA:schd 自阶段 A；ES:flight_hts、REDIS:flightInfo 仅阶段 B 投影期。 + */ +object Targets { + const val KAFKA_MSG = "KAFKA:msg" + const val KAFKA_SCHD = "KAFKA:schd" + const val ES_FLIGHT_HTS = "ES:flight_hts" + const val REDIS_FLIGHT_INFO = "REDIS:flightInfo" + + /** 阶段 A 投递目标（仅 Kafka）——I5：Delivery 阶段 A 不写 Redis。 */ + val phaseA: List = listOf(KAFKA_MSG, KAFKA_SCHD) + + /** 阶段 B 追加投影目标。 */ + val phaseB: List = phaseA + listOf(ES_FLIGHT_HTS, REDIS_FLIGHT_INFO) +} + +enum class EventStatus { PENDING, SENT, DEAD } + +data class MsgEvent( + val eventId: Long? = null, + val target: String, + val partitionKey: String? = null, // schd 事件恒为 FLID（v4 显式声明） + val payloadJson: String, + val state: EventStatus = EventStatus.PENDING, + val attempts: Int = 0, + val nextAttemptAt: java.time.Instant? = null, + val errorClass: ErrorClass? = null, + val lastError: String? = null, + val createdAt: java.time.Instant = java.time.Instant.now(), +) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/ProcState.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/ProcState.kt new file mode 100644 index 0000000..8e841d7 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/ProcState.kt @@ -0,0 +1,25 @@ +package com.gzzn.omms.msgexchange.nextgen.domain + +/** + * ACMA-8 数据模型 / PROC_STATE 状态机（六迁移表）。 + */ +enum class ProcStatus { PENDING, FAILED, SUCCEEDED, SKIPPED, DEAD } + +enum class ErrorClass { MALFORMED, CODEC_ERROR, EXHAUSTED, INFRA } + +data class ProcState( + val cminmsgsId: Long, + val state: ProcStatus, + val identityKey: String? = null, // decode 后首次绑定；FAILED 重试不重绑（I3） + val attempts: Int = 0, + val nextAttemptAt: java.time.Instant? = null, + val errorClass: ErrorClass? = null, + val lastError: String? = null, + val updatedAt: java.time.Instant = java.time.Instant.now(), +) { + val isTerminal: Boolean + get() = state == ProcStatus.SUCCEEDED || state == ProcStatus.SKIPPED || state == ProcStatus.DEAD + + /** 终态皆可归档（矩阵 #12：SUCCEEDED ∪ SKIPPED ∪ DEAD）；PENDING/FAILED 不迁。 */ + val archivable: Boolean get() = isTerminal +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/persistence/Repositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/persistence/Repositories.kt new file mode 100644 index 0000000..a6a06d9 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/persistence/Repositories.kt @@ -0,0 +1,120 @@ +package com.gzzn.omms.msgexchange.nextgen.infra.persistence + +import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass +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 java.time.Instant + +/** + * ACMA-8 v4 仓储接口（接口驱动，主泵/调度循环可单测；Micronaut Data JDBC 实装属阶段 1 后续）。 + */ +interface ProcStateRepository { + fun insert(cminmsgsId: Long, state: ProcStatus = ProcStatus.PENDING) + + /** I1：严格 FIFO 队头（最小未完成 CMINMSGS_ID）。 */ + fun headUnfinished(): ProcState? + + /** I3：identity 首次绑定；返回 false = 另一条消息已持有该键。 */ + fun tryBindIdentity(cminmsgsId: Long, identityKey: String): Boolean + + fun ownerOfIdentity(identityKey: String): Long? + + fun update( + cminmsgsId: Long, + state: ProcStatus, + nextAttemptAt: Instant? = null, + attempts: Int? = null, + errorClass: ErrorClass? = null, + lastError: String? = null, + ) +} + +interface MsgEventRepository { + fun insertAll(events: List): List + + /** I1 双层同策略：每 target 严格 FIFO 队头。 */ + fun headUnsent(target: String): MsgEvent? + + fun claimBatch(target: String, limit: Int): List // ORDER BY EVENT_ID ASC + + fun markSent(eventId: Long) + + fun markAllSent(eventIds: List) + + fun scheduleRetry(eventId: Long, nextAttemptAt: Instant, attempts: Int) + + fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String) + + /** 阶段 B（定案 2）：Delivery 同线程在 ES 投递成功后同步 enqueue 删除事件。 */ + fun insertSync(events: List) +} + +interface RefDataRepository { + data class GenMeta(val flids: List, val version: Long) + + fun getGen(day: String): GenMeta? + + /** 流程 4：版本 CAS（expected 未变才写，重放 no-op，不二次自增）。 */ + fun putGenIfVersion(day: String, expected: Long, new: GenMeta): Boolean + + fun upsertAll(rows: List>) // (rtype, rkey, payloadJson) +} + +interface ReqTrackRepository { + data class Req( + val reqId: Long, + val reqType: String, + val state: String, // REGISTERED/SENT/WAITING/DONE/EXPIRED + val sentAt: Instant? = null, + ) + + fun findOpenByKind(kind: String): Req? + + fun forceExpireOpenOf(kind: String) + + fun insert(kind: String, paramsJson: String): Long + + fun linkCoutmsgs(reqId: Long, coutmsgsId: Long) + + fun markSent(reqId: Long, sentAt: Instant) + + fun expireIfWaiting(reqId: Long) + + fun markDone(reqId: Long) +} + +interface PumpJobRepository { + data class Job(val jobId: Long, val kind: String) // ARCHIVE/HISTORY_SWEEP/PROJECTION_REBUILD + + fun enqueue(kind: String) + + fun headQueued(): Job? + + fun markRunning(jobId: Long) + + fun markDone(jobId: Long) + + fun markFailed(jobId: Long, lastError: String) +} + +interface FlightStateRepository { + /** 阶段 B 权威；replaceDay = 单事务删差集+写新代+版本提升。 */ + fun replaceDay(day: String, flights: List>) + + fun findByDay(day: String): List> +} + +interface CminmsgInboxRepository { + /** 事务 1：原文落库（沿用 SUBSYSTEM_* 列；仅 CLOB/DATE_RECEIVED，I3：无接收唯一约束）。 */ + fun insertRaw(rawXml: String): Long + + fun rawOf(cminmsgsId: Long): String? + + /** + * v4 回滚兼容关键：SUCCEEDED 时回填 SUBSYSTEM_*（FIX）+ DATE_PROCESSED/STATUS， + * 旧系统按自身 processed 语义可无缝接管（Runbook 第 7 步）。 + */ + fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/redis/RedisScripts.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/redis/RedisScripts.kt new file mode 100644 index 0000000..4d516d2 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/redis/RedisScripts.kt @@ -0,0 +1,26 @@ +package com.gzzn.omms.msgexchange.nextgen.infra.redis + +/** + * ACMA-8 流程 4 / I4 / I5：Redis Lua 脚本装载与执行入口。 + * 阶段 A 全部 flightInfo 写均经此处，且仅由主泵线程调用（I5）。 + */ +enum class RedisScript(val classpathLocation: String) { + /** 同一 hash 原子“覆盖新代 + 按代差删”（setArg 为 N 对 field/value，delArg 为差集）。 */ + SNAPSHOT_REPLACE("lua/snapshot_replace.lua"), + + /** 3:30 清场批量删除（仅 ES 归档成功集）。 */ + BATCH_DELETE("lua/batch_delete.lua"), +} + +interface FlightRedisClient { + /** + * 执行脚本。SNAPSHOT_REPLACE：setPairs 为新代全量 field/value，delFields 为按代差集； + * BATCH_DELETE：delFields 为待删 FLID 集。 + */ + fun eval(script: RedisScript, setPairs: List> = emptyList(), delFields: List = emptyList()) + + /** 阶段 A 权威读（处理决策 loadState、3:30 清场 findAll）。 */ + fun hgetAllFlightInfo(): Map +} + +// TODO(阶段1后续): 基于 micronaut-redis-lettuce 的实装（脚本自 classpath 装载并缓存 SHA）。 diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxController.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxController.kt new file mode 100644 index 0000000..7de4814 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxController.kt @@ -0,0 +1,24 @@ +package com.gzzn.omms.msgexchange.nextgen.ingress + +import io.micronaut.http.HttpResponse +import io.micronaut.http.MediaType +import io.micronaut.http.annotation.Body +import io.micronaut.http.annotation.Controller +import io.micronaut.http.annotation.Post +import io.micronaut.http.annotation.Produces + +/** + * ACMA-8 契约冻结：端点路径与响应语义不变。现役 Controller 返回记录 ID（原文落库后）。 + */ +@Controller +class InboxController(private val inbox: InboxService) { + + @Post("/cminmsgs/send") + @Produces(MediaType.TEXT_PLAIN) + fun send(@Body rawXml: String): HttpResponse { + val receipt = inbox.accept(rawXml) + return HttpResponse.ok(receipt.cminmsgsId.toString()) // TODO: 与现役响应体逐字对拍后固化 + } + + // TODO(阶段2): /schd/sync、/all/flights、/kafka/topics/{name}/msgs 按契约冻结清单补齐。 +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxService.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxService.kt new file mode 100644 index 0000000..7504fb6 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxService.kt @@ -0,0 +1,26 @@ +package com.gzzn.omms.msgexchange.nextgen.ingress + +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.CminmsgInboxRepository +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ProcStateRepository +import jakarta.inject.Singleton +import java.time.Instant + +/** ACMA-8 流程 1：入站接收（事务 1）。响应语义 = “已持久化”（与现役一致）；不解析报文（I3）。 */ +@Singleton +class InboxService( + private val inbox: CminmsgInboxRepository, + private val procState: ProcStateRepository, + // TODO(阶段1后续): 事务边界（@Transactional）随 Micronaut Data 实装补齐； + // pump 唤醒仅加速，崩溃后主泵 1s 轮询兜底。 +) { + data class Receipt(val cminmsgsId: Long, val receivedAt: Instant) + + fun accept(rawXml: String): Receipt { + val id = inbox.insertRaw(rawXml) + procState.insert(id) // 同事务（实装后）；接收层无唯一约束（I3） + wakePump() + return Receipt(id, Instant.now()) + } + + private fun wakePump() = Unit // TODO: 主泵唤醒（仅加速） +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/jobs/JobExecutor.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/jobs/JobExecutor.kt new file mode 100644 index 0000000..75075d1 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/jobs/JobExecutor.kt @@ -0,0 +1,80 @@ +package com.gzzn.omms.msgexchange.nextgen.jobs + +import com.gzzn.omms.msgexchange.nextgen.infra.redis.FlightRedisClient +import com.gzzn.omms.msgexchange.nextgen.infra.redis.RedisScript +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.FlightStateRepository +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.MsgEventRepository +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.PumpJobRepository +import com.gzzn.omms.msgexchange.nextgen.domain.MsgEvent +import com.gzzn.omms.msgexchange.nextgen.domain.Targets +import jakarta.inject.Singleton + +/** + * ACMA-8：泵作业执行器。cron 触发入队（PUMP_JOB），主泵 FIFO 执行（决策 1）—— + * 产物不绕过队头顺序；阶段 A 的清场同步链与归档在此落地（I4/I5）。 + */ +@Singleton +class JobExecutor( + private val historySweep: HistorySweepJob, + private val archive: ArchiveJob, + private val projectionRebuild: ProjectionRebuildJob, +) { + fun execute(job: PumpJobRepository.Job) = when (job.kind) { + "HISTORY_SWEEP" -> historySweep.run() + "ARCHIVE" -> archive.run() + "PROJECTION_REBUILD" -> projectionRebuild.run() + else -> error("unknown job ${job.kind}") + } +} + +/** + * 流程 4 runJob(HISTORY_SWEEP)：3:30 清场——主泵作业内同步链（I4，KEEP 现役 + * FlightHisScheduled 同步语义）：判史 → 同步写 ES（成功集）→ 仅删成功集。 + */ +@Singleton +class HistorySweepJob( + private val redis: FlightRedisClient, + // TODO(阶段2): esFlightHts.saveSync(history) 返回成功集（同步写，不经事件 ack 回查） +) { + fun run() { + val all = redis.hgetAllFlightInfo() + val history = pickHistory(all) // TODO(阶段2): 沿用现役判史规则（KEEP） + val success = emptyList() // TODO(阶段2): esFlightHts.saveSync(history) + if (success.isNotEmpty()) { + redis.eval(RedisScript.BATCH_DELETE, delFields = success) + } + } + + private fun pickHistory(all: Map): Map = all // TODO(阶段2) +} + +/** 流程 5 runJob(ARCHIVE)：3:00 归档——1 天前且仅终态可迁（矩阵 #12）。 */ +@Singleton +class ArchiveJob( + // TODO(阶段1后续): CMINMSGS⇆CMINMSGS_HST 迁移 SQL：JOIN PROC_STATE，STATE ∈ {SUCCEEDED,DEAD,SKIPPED} +) { + fun run() { + // TODO: 迁移（DEAD 带 ERROR_CLASS、SKIPPED 带 duplicate-of 审计随行保留） + } +} + +/** 流程 7：阶段 B 投影重建——切入阶段 B 时全量重建一次，此后增量走事件。 */ +@Singleton +class ProjectionRebuildJob( + private val flightState: FlightStateRepository, + private val msgEvents: MsgEventRepository, +) { + fun run() { + for (day in activeDays()) { + val flights = flightState.findByDay(day) + flights.forEach { (flid, payload) -> + msgEvents.insertSync( + listOf(MsgEvent(target = Targets.REDIS_FLIGHT_INFO, partitionKey = flid, payloadJson = payload)), + ) + } + } + // 重建完成前旧 Redis 视图继续服务（数据同源，仅短暂滞后） + } + + private fun activeDays(): List = emptyList() // TODO(阶段B) +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Handler.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Handler.kt new file mode 100644 index 0000000..b2eebaa --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Handler.kt @@ -0,0 +1,34 @@ +package com.gzzn.omms.msgexchange.nextgen.processing + +import com.gzzn.omms.msgexchange.nextgen.domain.DecodedMessage +import com.gzzn.omms.msgexchange.nextgen.domain.Decision +import com.gzzn.omms.msgexchange.nextgen.domain.MsgKind + +/** + * ACMA-8 流程 2：Handler = 纯函数（状态与报文进，Decision 出，不碰 Redis/Kafka）。 + * 32 个 Handler（flop 29 + schd 3）翻译属阶段 2/3，逐条对照 ACMA-4 基线与兼容矩阵 KEEP/FIX。 + */ +interface Handler { + val kind: MsgKind + + /** 在线状态以只读快照传入（阶段 A 读 Redis 权威、阶段 B 读 FLIGHT_STATE，由调用方装配）。 */ + fun decide(flightView: Map, msg: DecodedMessage): Decision +} + +/** + * sealed 穷尽分派（ACMA-6 选型：取代 legacy 反射 get{TYPE}()）。 + */ +class HandlerRegistry(handlers: List) { + private val byKind: Map = handlers.associateBy { keyOf(it.kind) } + + fun dispatcherFor(msg: DecodedMessage): Handler? = byKind[keyOf(msg.kind)] + + companion object { + fun keyOf(kind: MsgKind): String = when (kind) { + is MsgKind.Schd -> "SCHD-${kind.subtype.name}" + is MsgKind.Flop -> "FLOP-${kind.subtype}" + } + } +} + +// TODO(阶段2/3): 注册 3+29 Handler；本阶段仅含骨架（阶段 2 最小纵向链路先做 1 SCHD(DNLD) + 1 FLOP）。 diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Identity.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Identity.kt new file mode 100644 index 0000000..892b88e --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Identity.kt @@ -0,0 +1,24 @@ +package com.gzzn.omms.msgexchange.nextgen.processing + +import com.gzzn.omms.msgexchange.nextgen.config.PipelineProps +import com.gzzn.omms.msgexchange.nextgen.domain.DecodedMessage +import java.time.LocalDate + +/** + * ACMA-8 I3：幂等键 = SNDR|TYPE|STYP|SEQN。 + * 计算集中此处唯一入口；“是否含日边界”可配置且默认关闭（CONFIRM 矩阵 #11， + * SEQN 重置作用域确认前不改语义——上线后不改幂等键）。 + */ +object Identity { + fun of( + msg: DecodedMessage, + includeDayBoundary: Boolean, + day: LocalDate = LocalDate.now(), + ): String { + val base = "${msg.meta.sndr}|${msg.meta.type}|${msg.meta.styp}|${msg.meta.seqn}" + return if (includeDayBoundary) "$base|${day}" else base + } + + fun of(msg: DecodedMessage, props: PipelineProps.Identity): String = + of(msg, props.includeDayBoundary) +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt new file mode 100644 index 0000000..f89c8f0 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt @@ -0,0 +1,158 @@ +package com.gzzn.omms.msgexchange.nextgen.processing + +import com.gzzn.omms.msgexchange.nextgen.config.PipelineProps +import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass +import com.gzzn.omms.msgexchange.nextgen.domain.MsgEvent +import com.gzzn.omms.msgexchange.nextgen.domain.ProcStatus +import com.gzzn.omms.msgexchange.nextgen.domain.Targets +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.CminmsgInboxRepository +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.MsgEventRepository +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.redis.FlightRedisClient +import com.gzzn.omms.msgexchange.nextgen.jobs.JobExecutor +import jakarta.inject.Singleton +import java.time.Duration +import java.time.Instant + +/** + * ACMA-8 流程 2：处理主泵——严格 FIFO + HOL + 毒丸 DEAD 升级（I1）； + * 阶段 A 全部 Redis 写在本线程（I5），且先于事件创建（I2 happens-before）。 + */ +@Singleton +class Pump( + private val procState: ProcStateRepository, + private val pumpJobs: PumpJobRepository, + private val inbox: CminmsgInboxRepository, + private val processor: MessageProcessor, + private val jobExecutor: JobExecutor, + private val props: PipelineProps, +) { + @Volatile + private var running = true + + fun loop() { + while (running) { + tick() + } + } + + internal fun tick() { + val job = pumpJobs.headQueued() + val head = procState.headUnfinished() + + // 统一 FIFO：JOB 与消息同队列语义（定时任务产物不绕过队头顺序，决策 1）。 + val nextJob = job?.takeIf { jobBefore(job, head) } + when { + nextJob != null -> execute(nextJob) + head == null -> sleepQuietly(props.pipeline.pollInterval) + head.state == ProcStatus.FAILED && (head.nextAttemptAt ?: Instant.EPOCH) > Instant.now() -> + if (poisoned(head)) { + // 毒丸出队：DLQ + 告警，队列继续（I1） + procState.update( + head.cminmsgsId, ProcStatus.DEAD, + errorClass = ErrorClass.EXHAUSTED, + lastError = head.lastError ?: "head-deadline-exceeded", + ) + } else { + sleepQuietly(Duration.between(Instant.now(), head.nextAttemptAt)) + } + else -> processor.processOne(head.cminmsgsId) + } + } + + private fun poisoned(head: com.gzzn.omms.msgexchange.nextgen.domain.ProcState): Boolean = + head.attempts >= props.pipeline.maxAttempts || + Duration.between(head.updatedAt, Instant.now()) > props.pipeline.headDeadline + + /** JOB 与队头消息的先后由入队时间近似；实装以统一序号列保证（阶段 1 后续）。 */ + private fun jobBefore(job: PumpJobRepository.Job, head: com.gzzn.omms.msgexchange.nextgen.domain.ProcState?) = + head == null || head.state == ProcStatus.FAILED + + private fun execute(job: PumpJobRepository.Job) { + pumpJobs.markRunning(job.jobId) + try { + jobExecutor.execute(job) + pumpJobs.markDone(job.jobId) + } catch (e: Exception) { + pumpJobs.markFailed(job.jobId, e.message ?: "unknown") + } + } + + private fun sleepQuietly(d: Duration) { + if (!d.isNegative && !d.isZero) Thread.sleep(d.toMillis().coerceAtLeast(1)) + } +} + +/** ACMA-8 流程 2 processOne：解码→绑定→纯函数决策→提交。 */ +@Singleton +class MessageProcessor( + private val inbox: CminmsgInboxRepository, + private val procState: ProcStateRepository, + private val msgEvents: MsgEventRepository, + private val codecHolder: CodecHolder, + private val handlers: HandlerHolder, + private val redis: FlightRedisClient, + private val snapshotFlow: SnapshotFlow, + private val props: PipelineProps, +) { + fun processOne(cminmsgsId: Long) { + val raw = inbox.rawOf(cminmsgsId) ?: run { + procState.update(cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = "raw-missing") + return + } + val decoded = when (val r = codecHolder.codec.decode(raw)) { + is com.gzzn.omms.msgexchange.nextgen.codec.DecodeResult.Ok -> r.message + is com.gzzn.omms.msgexchange.nextgen.codec.DecodeResult.Err -> { + // MALFORMED 不重试；CODEC_ERROR 可一键重放（errorClass 入库） + procState.update(cminmsgsId, ProcStatus.DEAD, errorClass = r.failure.errorClass, lastError = r.failure.detail) + return + } + } + + val state = procState.headUnfinished() + // I3：identity 仅首次绑定；FAILED 重试不重绑（绑定失败 = 另一条同键消息 → SKIPPED） + // TODO(阶段1后续): 以行内 IDENTITY_KEY 判定而非 head 复查（含 FAILED 持久化字段读取） + if (state != null && state.cminmsgsId == cminmsgsId && state.identityKey == null) { + val identity = Identity.of(decoded, props.identity) + if (!procState.tryBindIdentity(cminmsgsId, identity)) { + val owner = procState.ownerOfIdentity(identity) ?: -1L + procState.update(cminmsgsId, ProcStatus.SKIPPED, lastError = "duplicate-of:$owner") + return + } + } + + // 快照消息走专属流程（流程 4：staging→Lua→putGenIfVersion CAS 同事务） + val handler = handlers.registry.dispatcherFor(decoded) + if (handler == null) { + procState.update(cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = "no-handler:${decoded.typeTag}") + return + } + val schdKind = decoded.kind as? com.gzzn.omms.msgexchange.nextgen.domain.MsgKind.Schd + if (schdKind != null && + schdKind.subtype == com.gzzn.omms.msgexchange.nextgen.domain.MsgKind.SchdSubtype.DNLD + ) { + snapshotFlow.publishSnapshot(cminmsgsId, decoded) + return + } + + val decision = handler.decide(redis.hgetAllFlightInfo(), decoded) // 纯函数 + + if (props.phase == PipelineProps.Phase.A) { + // 阶段 A：主泵线程先写 Redis（幂等），先于事件创建（I2）；TODO: redisApply(flightChanges) + } + // 事务 2：事件 + 出站意图 + REF_DATA + CMINMSGS 回填 + SUCCEEDED（实装后 @Transactional） + 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)) } + } + msgEvents.insertAll(events) + inbox.backfillOnSuccess(cminmsgsId, decoded.meta.sndr, decoded.meta.type, decoded.meta.styp, decoded.meta.seqn) + procState.update(cminmsgsId, ProcStatus.SUCCEEDED) + // 阶段 B：flightState.apply(decision.flightChanges) 进入同一事务；投影事件（ES/REDIS）追加。 + } +} + +/** 延迟装配占位（阶段 1 后续以 Micronaut Bean 替换直连构造）。 */ +class CodecHolder(val codec: com.gzzn.omms.msgexchange.nextgen.codec.XmlCodec) +class HandlerHolder(val registry: HandlerRegistry) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/SnapshotFlow.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/SnapshotFlow.kt new file mode 100644 index 0000000..92fb6fc --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/SnapshotFlow.kt @@ -0,0 +1,63 @@ +package com.gzzn.omms.msgexchange.nextgen.processing + +import com.gzzn.omms.msgexchange.nextgen.domain.DecodedMessage +import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass +import com.gzzn.omms.msgexchange.nextgen.domain.ProcStatus +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ProcStateRepository +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.RefDataRepository +import com.gzzn.omms.msgexchange.nextgen.infra.redis.FlightRedisClient +import com.gzzn.omms.msgexchange.nextgen.infra.redis.RedisScript +import jakarta.inject.Singleton + +/** + * ACMA-8 流程 4：日计划快照（generation，主泵内执行）。 + * staging（内存瞬态，崩溃从 raw 整包重放）→ Lua 原子“覆盖+按代差删”（I4/I5） + * → putGenIfVersion CAS + SUCCEEDED 同一 MySQL 事务（重放幂等，版本不二次自增）。 + * 整包失败 = 现役等价（KEEP）；跳坏行进 quarantine = CONFIRM（矩阵 #9）。 + */ +@Singleton +class SnapshotFlow( + private val procState: ProcStateRepository, + private val refData: RefDataRepository, + private val redis: FlightRedisClient, +) { + fun publishSnapshot(cminmsgsId: Long, msg: DecodedMessage) { + // 1) staging：流式解析 + 整包校验（TODO(阶段2): 流式 codec；千级 FLTR 为 MB 级，内存瞬态） + val staged = StageResult.stagingOf(msg) // 骨架：TODO 解析 FLTR 集与重组（KEEP 现役 MAFL/登机桥规则） + if (staged is StageResult.Invalid) { + procState.update(cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = staged.reason) + return + } + val normalized = (staged as StageResult.Ok).flights // (flid, payloadJson) + + // 2) 发布：Lua 原子覆盖 + 按代差删（删除集 = gen.flids − 新代，ADFT 自动存活） + val day = staged.day + val gen = refData.getGen(day) + val newFlids = normalized.map { it.first } + val delFields = gen?.flids?.minus(newFlids.toSet()) ?: emptyList() + redis.eval(RedisScript.SNAPSHOT_REPLACE, setPairs = normalized, delFields = delFields) + + // 3) 事务：putGen CAS + SUCCEEDED（实装后同 @Transactional；CAS 失败=并发，串行泵下不应发生→告警） + val expected = gen?.version ?: 0L + if (!refData.putGenIfVersion(day, expected, RefDataRepository.GenMeta(newFlids, expected + 1))) { + // 重放路径：version 已是目标值 → no-op 视为成功 + val again = refData.getGen(day) + if (again == null || again.version != expected + 1) { + procState.update(cminmsgsId, ProcStatus.FAILED, lastError = "gen-cas-conflict") + return + } + } + procState.update(cminmsgsId, ProcStatus.SUCCEEDED) + } + + /** staging 结果（骨架）。 */ + sealed interface StageResult { + data class Ok(val day: String, val flights: List>) : StageResult + data class Invalid(val reason: String) : StageResult + + companion object { + fun stagingOf(msg: DecodedMessage): StageResult = + Invalid("staging-not-implemented(${msg.typeTag})") // TODO(阶段2) + } + } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/reference/ReferenceService.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/reference/ReferenceService.kt new file mode 100644 index 0000000..fba3bb7 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/reference/ReferenceService.kt @@ -0,0 +1,18 @@ +package com.gzzn.omms.msgexchange.nextgen.reference + +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.RefDataRepository +import jakarta.inject.Singleton + +/** + * ACMA-8 决策 6：21 类 admin-api 基础数据同步（阶段 6）。 + * 只产 REF_DATA 行（权威写唯一入口）；影子实例不运行本模块（v4 影子二分）。 + */ +@Singleton +class ReferenceService( + private val refData: RefDataRepository, + // TODO(阶段6): @Client(id="ADMINAPI") 声明式客户端 + 21 类端点配置化清单 +) { + fun refresh(type: String) { + // TODO(阶段6): 拉取 → upsertAll(type, key, payload)；失败旧数据可用（REF_DATA 保留旧值） + } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/reference/RequestCoordinator.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/reference/RequestCoordinator.kt new file mode 100644 index 0000000..dea38d5 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/reference/RequestCoordinator.kt @@ -0,0 +1,57 @@ +package com.gzzn.omms.msgexchange.nextgen.reference + +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.RefDataRepository +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ReqTrackRepository +import jakarta.inject.Singleton +import java.time.Instant + +/** + * ACMA-8 流程 6：请求式查询状态机（15 类，决策树版 v4）。 + * 同类并发=1（注册新请求先强制旧请求 EXPIRED，终态不再被匹配）； + * 应答匹配：回显字段 CONFIRM（矩阵 #13）→ 精确匹配；否则退化模式（DTTM ≥ SENT 才应用 + 审计）。 + */ +@Singleton +class RequestCoordinator( + private val reqTrack: ReqTrackRepository, + private val refData: RefDataRepository, +) { + fun request(kind: String, rangeJson: String): Long { + reqTrack.forceExpireOpenOf(kind) // 防护① + val reqId = reqTrack.insert(kind, rangeJson) + // 出站走既有 COUTMSGS outbox（TODO(阶段6): codec.encodeRqrd + coutmsgs 落库 + link） + reqTrack.markSent(reqId, Instant.now()) + return reqId + // TODO(阶段6): timer.at(kind.timeout) { expireIfWaiting(reqId) } + } + + /** 应答处理（决策树）：msg.dttm 与回显字段由 codec 解出后传入。 */ + fun onResp( + kind: String, + dttm: Long, + echoSeqn: Long?, + records: List>, // (rtype, rkey, payloadJson) + ): String { + val req = reqTrack.findOpenByKind(kind) ?: return audit("late/unknown resp kind=$kind") + val echoConfirmed = false // CONFIRM（矩阵 #13）待机场方结论 + return when { + echoConfirmed && echoSeqn != null -> { // ① 精确匹配（TODO: matchBy echoSeqn） + applyResp(req.reqId, records); "matched" + } + dttm < epochMillis(req.sentAt) -> audit("stale resp kind=$kind") // ② DTTM < SENT → 丢弃 + else -> { // ③ 退化模式应用；残余跨代错配风险明示接受并审计 + applyResp(req.reqId, records); "applied-degraded" + } + } + } + + private fun applyResp(reqId: Long, records: List>) { + refData.upsertAll(records) // 大应答复用流程 4 staging 路径（阶段 6） + reqTrack.markDone(reqId) + } + + private fun audit(reason: String): String = reason.also { + // TODO(阶段1后续): 审计表/日志（跨代错配残余风险可观测） + } + + private fun epochMillis(i: Instant?): Long = i?.toEpochMilli() ?: 0 +} diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml new file mode 100644 index 0000000..a9b2522 --- /dev/null +++ b/src/main/resources/application.yml @@ -0,0 +1,60 @@ +# ===================================================================== +# msgexchange-nextgen 配置（ACMA-8 v4） +# 口令/端点全部环境变量外置——零入库（ACMA-8 契约冻结 + 安全基线） +# ===================================================================== +msgx: + phase: A # A=Redis 权威（主泵单线程写，I5）；B=FLIGHT_STATE 权威+投影 + service-name: msgexchangeapi # 契约冻结；影子实例用 msgexchangeapi-shadow + 独立表/key前缀/topic + register-eureka: true # 影子对拍期置 false 或独立服务名 + pipeline: # —— ACMA-8 参数表初值（阶段 0 基线校准）—— + poll-interval: 1s # KEEP 现役 1s 轮询节奏 + claim-batch: 50 + max-attempts: 5 # 处理/投递同值 + backoff-ms: [1000, 2000, 4000, 8000, 16000] # 指数退避，单次封顶 60s + backoff-cap-ms: 60000 + head-deadline: 10m # 队头滞留上界 = 最坏 HOL 时长（毒丸升级） + schd: + flush-period: 3s # KEEP 现役推送节律 + flush-limit: 500 # 批上限，防积压尖峰 + identity: + include-day-boundary: false # CONFIRM（矩阵 #11）：SEQN 重置作用域确认前保持关闭 + consistency-check: # 阶段 A 唯一主动一致性哨兵（必选） + on-startup: true + daily-sample-ratio: 0.01 + +micronaut: + application: + name: msgexchange-nextgen + server: + port: 8080 + +datasource: + url: ${MSGX_DB_URL} + username: ${MSGX_DB_USER} + password: ${MSGX_DB_PASSWORD} + driver-class-name: com.mysql.cj.jdbc.Driver + +flyway: + enabled: true + locations: classpath:db/migration + +redis: + uri: ${MSGX_REDIS_URI} # 阶段 A 权威存储；仅主泵线程写（I5） + +kafka: + bootstrap: + servers: ${MSGX_KAFKA_SERVERS} + producers: + acks: all + enable-idempotence: true + +eureka: + client: + registration: + enabled: ${msgx.register-eureka} + default-zone: ${MSGX_EUREKA_URL:http://127.0.0.1:8761/eureka} + +# logstash TCP JSON 通道保留（字段兼容现网，ACMA-3 部署前提）——见 logback.xml +logstash: + host: ${MSGX_LOGSTASH_HOST:127.0.0.1} + port: ${MSGX_LOGSTASH_PORT:5044} diff --git a/src/main/resources/db/migration/V2.0.0__aux_tables.sql b/src/main/resources/db/migration/V2.0.0__aux_tables.sql new file mode 100644 index 0000000..c5da4cd --- /dev/null +++ b/src/main/resources/db/migration/V2.0.0__aux_tables.sql @@ -0,0 +1,78 @@ +-- ===================================================================== +-- ACMA-8 v4 数据模型（六表）· Flyway 基线迁移 +-- 与 issue「综合架构设计」数据模型节逐字一致；影子实例在独立 schema 执行同一脚本。 +-- ===================================================================== + +-- ① 处理状态伴生表（不动 CMINMSGS 旧列； +-- 回滚兼容：SUCCEEDED 时回填 CMINMSGS.DATE_PROCESSED/STATUS，旧系统可接管续跑） +CREATE TABLE PROC_STATE ( + CMINMSGS_ID BIGINT PRIMARY KEY, + STATE VARCHAR(16) NOT NULL, -- PENDING/FAILED/SUCCEEDED/SKIPPED/DEAD + IDENTITY_KEY VARCHAR(200) NULL, -- SNDR|TYPE|STYP|SEQN，decode 后首次绑定（I3） + ATTEMPTS INT NOT NULL DEFAULT 0, + NEXT_ATTEMPT_AT TIMESTAMP NULL, + ERROR_CLASS VARCHAR(20) NULL, -- MALFORMED/CODEC_ERROR/EXHAUSTED/INFRA + LAST_ERROR VARCHAR(1000) NULL, + UPDATED_AT TIMESTAMP NOT NULL, + UNIQUE KEY UK_PROC_IDENTITY (IDENTITY_KEY), + INDEX IDX_PROC_HEAD (STATE, CMINMSGS_ID) -- 主泵队头查询（I1） +); + +-- ② 统一投递事件（outbox） +CREATE TABLE MSG_EVENT ( + EVENT_ID BIGINT AUTO_INCREMENT PRIMARY KEY, + TARGET VARCHAR(30) NOT NULL, -- KAFKA:msg / KAFKA:schd（A起）；ES:flight_hts / REDIS:flightInfo（B起） + PARTITION_KEY VARCHAR(64) NULL, -- schd 事件恒为 FLID + PAYLOAD_JSON MEDIUMTEXT NOT NULL, + STATE VARCHAR(16) NOT NULL, -- PENDING/SENT/DEAD + ATTEMPTS INT NOT NULL DEFAULT 0, + NEXT_ATTEMPT_AT TIMESTAMP NULL, + ERROR_CLASS VARCHAR(20) NULL, + LAST_ERROR VARCHAR(1000) NULL, + CREATED_AT TIMESTAMP NOT NULL, + INDEX IDX_EVT_HEAD (TARGET, STATE, EVENT_ID) -- 每 target 队头查询（I1 双层同策略） +); + +-- ③ 参考数据 + SCHD generation 元数据（21 类；SCHD_GEN 行 RKEY=日期） +CREATE TABLE REF_DATA ( + RTYPE VARCHAR(20) NOT NULL, -- AIRL/ARPT/... | SCHD_GEN + RKEY VARCHAR(64) NOT NULL, + PAYLOAD_JSON MEDIUMTEXT NOT NULL, -- SCHD_GEN: {"flids":[...],"version":N} + VERSION BIGINT NOT NULL DEFAULT 0, -- putGenIfVersion CAS 依据（流程 4） + SOURCE VARCHAR(20) NOT NULL, -- ADMINAPI / AODB / PIPELINE + REFRESHED_AT TIMESTAMP NOT NULL, + PRIMARY KEY (RTYPE, RKEY) +); + +-- ④ 请求状态机（15 类，流程 6） +CREATE TABLE REQ_TRACK ( + REQ_ID BIGINT AUTO_INCREMENT PRIMARY KEY, + REQ_TYPE VARCHAR(20) NOT NULL, + PARAMS_JSON VARCHAR(500) NOT NULL, + STATE VARCHAR(16) NOT NULL, -- REGISTERED/SENT/WAITING/DONE/EXPIRED + COUTMSGS_ID INT NULL, + SENT_AT TIMESTAMP NULL, + COMPLETED_AT TIMESTAMP NULL, + INDEX IDX_REQ_OPEN (REQ_TYPE, STATE) +); + +-- ⑤ 泵作业队列（定时任务统一入口：cron 触发入队，主泵 FIFO 执行） +CREATE TABLE PUMP_JOB ( + JOB_ID BIGINT AUTO_INCREMENT PRIMARY KEY, + KIND VARCHAR(20) NOT NULL, -- ARCHIVE / HISTORY_SWEEP / PROJECTION_REBUILD + STATE VARCHAR(16) NOT NULL, -- QUEUED/RUNNING/DONE/FAILED + CREATED_AT TIMESTAMP NOT NULL, + UPDATED_AT TIMESTAMP NOT NULL, + LAST_ERROR VARCHAR(1000) NULL +); + +-- ⑥ 航班状态权威（阶段 B 启用；此前 Redis 为权威，I5） +CREATE TABLE FLIGHT_STATE ( + FLID VARCHAR(32) PRIMARY KEY, + FDAY DATE NOT NULL, + VERSION BIGINT NOT NULL, + MAID VARCHAR(32) NULL, + PAYLOAD_JSON MEDIUMTEXT NOT NULL, -- 投影兼容格式的 FLTR JSON + UPDATED_AT TIMESTAMP NOT NULL, + INDEX IDX_FS_DAY (FDAY, VERSION) +); diff --git a/src/main/resources/logback.xml b/src/main/resources/logback.xml new file mode 100644 index 0000000..b0c4527 --- /dev/null +++ b/src/main/resources/logback.xml @@ -0,0 +1,24 @@ + + + + + + %d{HH:mm:ss.SSS} [%thread] %-5level [%X{traceId:-}] %logger{36} - %msg%n + + + + + ${LOGSTASH_HOST:-127.0.0.1}:${LOGSTASH_PORT:-5044} + + traceId + + 10 seconds + + + + + + + + + diff --git a/src/main/resources/lua/batch_delete.lua b/src/main/resources/lua/batch_delete.lua new file mode 100644 index 0000000..bdca89b --- /dev/null +++ b/src/main/resources/lua/batch_delete.lua @@ -0,0 +1,9 @@ +-- ACMA-8 流程 4（runJob HISTORY_SWEEP）/ I4：3:30 清场批量删除。 +-- 仅删除“ES 归档成功集”（调用方以 saveSync 成功返回集为参），同一泵线程程序序保证依赖。 +-- KEYS[1] = flightInfo；ARGV = 待删除 FLID 列表 +local n = 0 +for i = 1, #ARGV do + redis.call('HDEL', KEYS[1], ARGV[i]) + n = n + 1 +end +return n diff --git a/src/main/resources/lua/snapshot_replace.lua b/src/main/resources/lua/snapshot_replace.lua new file mode 100644 index 0000000..9aaa492 --- /dev/null +++ b/src/main/resources/lua/snapshot_replace.lua @@ -0,0 +1,19 @@ +-- ACMA-8 流程 4 / I4 / I5：日计划快照“覆盖新代 + 按代差删”，同一 hash 原子执行。 +-- KEYS[1] = flightInfo +-- ARGV[1] = set 对数 N，其后为 N 对 field/value；再后为待删除 field（按代差集） +-- 删除集边界：仅上一代 gen.flids − 新代 flids（按代，非全 hash）→ ADFT 自动存活。 +local key = KEYS[1] +local nSet = tonumber(ARGV[1]) +local i = 2 +while i <= 1 + nSet * 2 do + redis.call('HSET', key, ARGV[i], ARGV[i + 1]) + i = i + 2 +end +local j = 2 + nSet * 2 +local deleted = 0 +while j <= #ARGV do + redis.call('HDEL', key, ARGV[j]) + deleted = deleted + 1 + j = j + 1 +end +return deleted diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/SchdAggregationTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/SchdAggregationTest.kt new file mode 100644 index 0000000..72cd9fc --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/SchdAggregationTest.kt @@ -0,0 +1,56 @@ +package com.gzzn.omms.msgexchange.nextgen.delivery + +import com.gzzn.omms.msgexchange.nextgen.domain.EventStatus +import com.gzzn.omms.msgexchange.nextgen.domain.MsgEvent +import com.gzzn.omms.msgexchange.nextgen.domain.Targets +import org.junit.jupiter.api.Test +import kotlin.test.assertEquals + +/** + * ACMA-8 流程 3 / 兼容矩阵 #7（FIX）：同一 FLID 多次变更只发最新一态； + * v4 显式“组内取 EVENT_ID 最大”，不依赖集合遍历序。 + */ +class SchdAggregationTest { + + private fun ev(flid: String, eventId: Long, payload: String) = MsgEvent( + eventId = eventId, + target = Targets.KAFKA_SCHD, + partitionKey = flid, + payloadJson = payload, + state = EventStatus.PENDING, + ) + + @Test + fun `same flight collapsed to max EVENT_ID`() { + val pending = listOf( + ev("F1", 1, """{"FLID":"F1","v":"old"}"""), + ev("F2", 2, """{"FLID":"F2","v":"only"}"""), + ev("F1", 3, """{"FLID":"F1","v":"new"}"""), + ) + val latest = SchdAggregation.latestPerFlight(pending) + assertEquals(2, latest.size) + assertEquals("""{"FLID":"F1","v":"new"}""", latest.first { it.contains("F1") }) + assertEquals("""{"FLID":"F2","v":"only"}""", latest.first { it.contains("F2") }) + } + + @Test + fun `order-independent grouping yields same latest per flight`() { + val a = listOf( + ev("F1", 1, "old"), + ev("F1", 2, "new"), + ) + val b = a.reversed() + assertEquals( + SchdAggregation.latestPerFlight(a).toSet(), + SchdAggregation.latestPerFlight(b).toSet(), + ) + } + + @Test + fun `wire shape is FLTR JSON array`() { + val payload = SchdAggregation + .latestPerFlight(listOf(ev("F1", 1, "a"), ev("F2", 2, "b"))) + .joinToString(",", "[", "]") + assertEquals("[a,b]", payload) + } +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/IdentityTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/IdentityTest.kt new file mode 100644 index 0000000..fe060a6 --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/IdentityTest.kt @@ -0,0 +1,45 @@ +package com.gzzn.omms.msgexchange.nextgen.processing + +import com.gzzn.omms.msgexchange.nextgen.config.PipelineProps +import com.gzzn.omms.msgexchange.nextgen.domain.DecodedMessage +import com.gzzn.omms.msgexchange.nextgen.domain.MetaFields +import com.gzzn.omms.msgexchange.nextgen.domain.MsgKind +import org.junit.jupiter.api.Test +import java.time.LocalDate +import kotlin.test.assertEquals + +/** + * ACMA-8 I3：identity = SNDR|TYPE|STYP|SEQN；日边界含否集中可配（CONFIRM 矩阵 #11，默认关）。 + */ +class IdentityTest { + + private fun msg(seqn: Long, dttm: Long = 20260906120000) = DecodedMessage( + meta = MetaFields(sndr = "AODB", type = "FLOP", styp = "DELY", seqn = seqn, dttm = dttm), + kind = MsgKind.Flop("DELY"), + rawXml = "", + ) + + @Test + fun `identity is pipe-joined SNDR TYPE STYP SEQN`() { + val identity = Identity.of(msg(42), includeDayBoundary = false) + assertEquals("AODB|FLOP|DELY|42", identity) + } + + @Test + fun `day boundary included only when configured`() { + val day = LocalDate.of(2026, 9, 6) + assertEquals( + "AODB|FLOP|DELY|42|2026-09-06", + Identity.of(msg(42), includeDayBoundary = true, day = day), + ) + assertEquals( + "AODB|FLOP|DELY|42", + Identity.of(msg(42), includeDayBoundary = false, day = day), + ) + } + + @Test + fun `props default keeps day boundary off`() { + assertEquals(false, PipelineProps.Identity().includeDayBoundary) + } +}