diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodec.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodec.kt new file mode 100644 index 0000000..f682731 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodec.kt @@ -0,0 +1,175 @@ +package com.gzzn.omms.msgexchange.codec + +import com.gzzn.omms.msgexchange.domain.DecodedMessage +import com.gzzn.omms.msgexchange.domain.ErrorClass +import com.gzzn.omms.msgexchange.domain.MetaFields +import com.gzzn.omms.msgexchange.domain.MsgKind +import jakarta.inject.Singleton +import org.w3c.dom.Element +import org.w3c.dom.Node +import org.xml.sax.InputSource +import java.io.StringReader +import javax.xml.XMLConstants +import javax.xml.parsers.DocumentBuilderFactory + +/** + * SIS XML 解码（XXE 防护；META + TYPE 分派 + FLOP 集合解析）。 + * 阶段 2 先覆盖 FLOP-GTDT 纵向链路所需字段；其余 STYP 仅完成 META/FLID 提取。 + */ +@Singleton +class JacksonXmlCodec : XmlCodec { + + private val builderFactory: DocumentBuilderFactory = DocumentBuilderFactory.newInstance().apply { + isNamespaceAware = false + isExpandEntityReferences = false + setFeature(XMLConstants.FEATURE_SECURE_PROCESSING, true) + setFeature("http://apache.org/xml/features/disallow-doctype-decl", true) + setFeature("http://xml.org/sax/features/external-general-entities", false) + setFeature("http://xml.org/sax/features/external-parameter-entities", false) + try { + setAttribute(XMLConstants.ACCESS_EXTERNAL_DTD, "") + setAttribute(XMLConstants.ACCESS_EXTERNAL_SCHEMA, "") + } catch (_: IllegalArgumentException) { + // JDK 实现差异：忽略不支持的属性 + } + } + + override fun decode(rawXml: String): DecodeResult { + val trimmed = rawXml.trim() + if (trimmed.isEmpty() || !trimmed.startsWith("<")) { + return DecodeResult.Err(DecodeFailure(ErrorClass.MALFORMED, "empty-or-non-xml")) + } + return try { + val doc = builderFactory.newDocumentBuilder().parse(InputSource(StringReader(trimmed))) + val root = doc.documentElement ?: return DecodeResult.Err( + DecodeFailure(ErrorClass.MALFORMED, "missing-root"), + ) + if (!root.tagName.equals("MSG", ignoreCase = true)) { + return DecodeResult.Err(DecodeFailure(ErrorClass.MALFORMED, "root-not-msg:${root.tagName}")) + } + + val metaEl = firstChildElement(root, "META") + ?: return DecodeResult.Err(DecodeFailure(ErrorClass.MALFORMED, "missing-meta")) + val sndr = childText(metaEl, "SNDR") + ?: return DecodeResult.Err(DecodeFailure(ErrorClass.MALFORMED, "missing-sndr")) + val seqn = childText(metaEl, "SEQN")?.toLongOrNull() + ?: return DecodeResult.Err(DecodeFailure(ErrorClass.MALFORMED, "missing-or-invalid-seqn")) + val dttm = childText(metaEl, "DTTM")?.toLongOrNull() + ?: return DecodeResult.Err(DecodeFailure(ErrorClass.MALFORMED, "missing-or-invalid-dttm")) + val type = childText(metaEl, "TYPE")?.uppercase() + ?: return DecodeResult.Err(DecodeFailure(ErrorClass.MALFORMED, "missing-type")) + val styp = childText(metaEl, "STYP")?.uppercase() + ?: return DecodeResult.Err(DecodeFailure(ErrorClass.MALFORMED, "missing-styp")) + + val kind = kindOf(type, styp) + val body = when (kind) { + is MsgKind.Flop -> parseFlopBody(root, type) + is MsgKind.Schd -> null + } + + DecodeResult.Ok( + DecodedMessage( + meta = MetaFields(sndr = sndr, type = type, styp = styp, seqn = seqn, dttm = dttm), + kind = kind, + rawXml = trimmed, + body = body, + ), + ) + } catch (e: NumberFormatException) { + DecodeResult.Err(DecodeFailure(ErrorClass.MALFORMED, "invalid-number:${e.message}")) + } catch (e: Exception) { + DecodeResult.Err(DecodeFailure(ErrorClass.MALFORMED, "xml-parse:${e.message ?: e.javaClass.simpleName}")) + } + } + + override fun encodeRqrd(kind: String, rangeJson: String): String = + """RQFD""" + + private fun kindOf(type: String, styp: String): MsgKind = when (type) { + "SCHD" -> MsgKind.Schd( + when (styp) { + "RESP" -> MsgKind.SchdSubtype.RESP + "DNLD" -> MsgKind.SchdSubtype.DNLD + "ADFT" -> MsgKind.SchdSubtype.ADFT + else -> MsgKind.SchdSubtype.DNLD + }, + ) + "FLOP" -> MsgKind.Flop(styp) + else -> MsgKind.Flop(styp) + } + + private fun parseFlopBody(root: Element, type: String): FlopPayload? { + val section = firstChildElement(root, type) ?: return null + val scalars = linkedMapOf() + val collections = linkedMapOf>>() + + section.childNodes.let { nodes -> + for (i in 0 until nodes.length) { + val node = nodes.item(i) + if (node.nodeType != Node.ELEMENT_NODE) continue + val el = node as Element + val tag = el.tagName.uppercase() + when { + tag in COLLECTION_TAGS -> { + collections.getOrPut(tag) { mutableListOf() }.add(parseCollectionItem(el)) + } + tag == "FLID" -> scalars["FLID"] = el.textContent.trim() + hasElementChildren(el) -> Unit + else -> scalars[tag] = el.textContent.trim() + } + } + } + + val flid = scalars.remove("FLID") ?: return null + return FlopPayload(flid = flid, scalars = scalars, collections = collections.mapValues { it.value.toList() }) + } + + private fun parseCollectionItem(el: Element): Map { + val item = linkedMapOf() + if (el.hasAttributes()) { + for (i in 0 until el.attributes.length) { + val attr = el.attributes.item(i) + item[attr.nodeName.uppercase()] = attr.nodeValue.trim() + } + } + el.childNodes.let { nodes -> + for (i in 0 until nodes.length) { + val node = nodes.item(i) + if (node.nodeType != Node.ELEMENT_NODE) continue + val child = node as Element + item[child.tagName.uppercase()] = child.textContent.trim() + } + } + return item + } + + private fun firstChildElement(parent: Element, tag: String): Element? { + parent.childNodes.let { nodes -> + for (i in 0 until nodes.length) { + val node = nodes.item(i) + if (node.nodeType == Node.ELEMENT_NODE && (node as Element).tagName.equals(tag, ignoreCase = true)) { + return node + } + } + } + return null + } + + private fun childText(parent: Element, tag: String): String? = + firstChildElement(parent, tag)?.textContent?.trim()?.takeIf { it.isNotEmpty() } + + private fun hasElementChildren(el: Element): Boolean { + el.childNodes.let { nodes -> + for (i in 0 until nodes.length) { + if (nodes.item(i).nodeType == Node.ELEMENT_NODE) return true + } + } + return false + } + + companion object { + val COLLECTION_TAGS: Set = setOf( + "GTDT", "CKDT", "CLDT", "PSDT", "CHDT", "DELY", "ABTM", "CHOT", "ROUT", "ERUT", + ) + } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/codec/SisMessageBody.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/codec/SisMessageBody.kt new file mode 100644 index 0000000..8e84068 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/codec/SisMessageBody.kt @@ -0,0 +1,11 @@ +package com.gzzn.omms.msgexchange.codec + +/** + * FLOP 报文解析体（v2 §4 命令化输入）。 + * scalars：FLID/FFID 等标量；collections：GTDT/CKDT 等重复集合（含源序号属性）。 + */ +data class FlopPayload( + val flid: String, + val scalars: Map = emptyMap(), + val collections: Map>> = emptyMap(), +) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/pipeline/PipelineHolderFactory.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/pipeline/PipelineHolderFactory.kt new file mode 100644 index 0000000..dca40d4 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/pipeline/PipelineHolderFactory.kt @@ -0,0 +1,23 @@ +package com.gzzn.omms.msgexchange.infra.pipeline + +import com.gzzn.omms.msgexchange.codec.JacksonXmlCodec +import com.gzzn.omms.msgexchange.processing.CodecHolder +import com.gzzn.omms.msgexchange.processing.Handler +import com.gzzn.omms.msgexchange.processing.HandlerHolder +import com.gzzn.omms.msgexchange.processing.HandlerRegistry +import io.micronaut.context.annotation.Bean +import io.micronaut.context.annotation.Factory +import jakarta.inject.Singleton + +/** 主泵解码与 Handler 装配（stub / 实装模式共用真实 XmlCodec + 已注册 Handler）。 */ +@Factory +class PipelineHolderFactory { + + @Bean + @Singleton + fun codecHolder(codec: JacksonXmlCodec): CodecHolder = CodecHolder(codec) + + @Bean + @Singleton + fun handlerHolder(handlers: List): HandlerHolder = HandlerHolder(HandlerRegistry(handlers)) +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubAdapters.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubAdapters.kt index 7f38981..cf03809 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubAdapters.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubAdapters.kt @@ -1,32 +1,13 @@ package com.gzzn.omms.msgexchange.infra.stub -import com.gzzn.omms.msgexchange.codec.DecodeFailure -import com.gzzn.omms.msgexchange.codec.DecodeResult -import com.gzzn.omms.msgexchange.codec.XmlCodec import com.gzzn.omms.msgexchange.delivery.DeliveryPort -import com.gzzn.omms.msgexchange.domain.ErrorClass -import com.gzzn.omms.msgexchange.processing.CodecHolder -import com.gzzn.omms.msgexchange.processing.HandlerHolder -import com.gzzn.omms.msgexchange.processing.HandlerRegistry -import io.micronaut.context.annotation.Bean -import io.micronaut.context.annotation.Factory import io.micronaut.context.annotation.Requires import jakarta.inject.Singleton /** - * U07：stub 适配层——XmlCodec / DeliveryPort / Handler 装配，仅在 msgx.stubs=true 时生效（Redis 客户端已按 ACM2-28 移除）。 - * stub codec 未实装 → 报文 decode 返回 CODEC_ERROR（走 FAILED 可重放路径，链路上可观测）， - * 语义见 ACM2-10 U11：不把“未实装”写成报文非法/终态。 + * U07：stub 适配层——DeliveryPort 内存实现，仅在 msgx.stubs=true 时生效。 + * XmlCodec / Handler 由 [com.gzzn.omms.msgexchange.infra.pipeline.PipelineHolderFactory] 统一装配。 */ -@Requires(property = "msgx.stubs", value = "true") -@Singleton -class StubXmlCodec : XmlCodec { - override fun decode(rawXml: String): DecodeResult = - DecodeResult.Err(DecodeFailure(ErrorClass.CODEC_ERROR, "stub:codec-not-implemented")) - - override fun encodeRqrd(kind: String, rangeJson: String): String = "" -} - @Requires(property = "msgx.stubs", value = "true") @Singleton class StubDeliveryPort : DeliveryPort { @@ -37,15 +18,3 @@ class StubDeliveryPort : DeliveryPort { override fun sendKafka(topic: String, payloadJson: String) { sent += topic to payloadJson } override fun indexFlightHts(payloadJson: String) = Unit } - -/** Holder 工厂：CodecHolder/HandlerHolder 由 Micronaut Bean 提供（取代直连构造占位）。 */ -@Factory -@Requires(property = "msgx.stubs", value = "true") -class StubHolderFactory { - - @Bean - fun codecHolder(codec: StubXmlCodec): CodecHolder = CodecHolder(codec) - - @Bean - fun handlerHolder(): HandlerHolder = HandlerHolder(HandlerRegistry(emptyList())) // 阶段 2 前无 Handler 注册 -} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt index a3c9c96..a51f507 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt @@ -2,6 +2,7 @@ package com.gzzn.omms.msgexchange.processing import com.gzzn.omms.msgexchange.config.PipelineProps import com.gzzn.omms.msgexchange.domain.DecodedMessage +import com.gzzn.omms.msgexchange.domain.flight.FlightStateEngine import com.gzzn.omms.msgexchange.domain.ErrorClass import com.gzzn.omms.msgexchange.domain.MsgEvent import com.gzzn.omms.msgexchange.domain.ProcState @@ -222,9 +223,15 @@ class MessageProcessor( decision.schdPush.forEach { add(MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = it.flid, payloadJson = it.payloadJson)) } } + val messageId = head.cminmsgsId.toString() txManager.inTransaction { if (decision.flightChanges.isNotEmpty()) { - flightSchd.upsertIncremental(decision.flightChanges) + val nextStates = decision.flightChanges.map { change -> + val current = flightSchd.findNextStateByFlid(change.flid) + val commands = FlightStateEngine.commandsFromFields(change.flid, change.fields, snapshotReplace = false) + FlightStateEngine.apply(current, commands, messageId, bumpVersion = true) + } + flightSchd.persistNextStates(null, nextStates, snapshotReplace = false) } if (events.isNotEmpty()) { msgEvents.insertAll(events) @@ -232,8 +239,11 @@ class MessageProcessor( procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED) } - // 提交后：backfill CMINMSGS（共享信箱外部副作用，补偿链路） - inbox.backfillOnSuccess(head.cminmsgsId, decoded.meta.sndr, decoded.meta.type, decoded.meta.styp, decoded.meta.seqn) + try { + inbox.backfillOnSuccess(head.cminmsgsId, decoded.meta.sndr, decoded.meta.type, decoded.meta.styp, decoded.meta.seqn) + } catch (e: Exception) { + log.error("backfill failed after SUCCEEDED id={} (compensation required)", head.cminmsgsId, e) + } log.info("SUCCEEDED id={} events={} flightChanges={}", head.cminmsgsId, events.size, decision.flightChanges.size) } } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlow.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlow.kt index cbe2e6d..22be3a7 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlow.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlow.kt @@ -1,6 +1,7 @@ package com.gzzn.omms.msgexchange.processing import com.gzzn.omms.msgexchange.domain.DecodedMessage +import com.gzzn.omms.msgexchange.domain.flight.FlightStateEngine import com.gzzn.omms.msgexchange.domain.ErrorClass import com.gzzn.omms.msgexchange.domain.ProcState import com.gzzn.omms.msgexchange.domain.ProcStatus @@ -36,18 +37,17 @@ class SnapshotFlow( private val log = org.slf4j.LoggerFactory.getLogger(SnapshotFlow::class.java) fun publishSnapshot(head: ProcState, msg: DecodedMessage) { - // 1) staging：流式解析 + 整包校验（内存瞬态，崩溃从 raw 整包重放） + val messageId = head.cminmsgsId.toString() val staged = StageResult.stagingOf(msg) if (staged is StageResult.Invalid) { log.warn("staging not implemented -> FAILED(UNSUPPORTED) id={} reason={}", head.cminmsgsId, staged.reason) - procFailure.fail(head, ErrorClass.UNSUPPORTED, staged.reason) // U10：未实装 → 可重放，非终态 + procFailure.fail(head, ErrorClass.UNSUPPORTED, staged.reason) return } val ok = staged as StageResult.Ok - val normalized = ok.flights // (flid, fields) + val normalized = ok.flights val day = ok.day - // 内存与超大包熔断防御（ACM2-28 评论 4 项 3） if (normalized.size > MAX_FLIGHTS_PER_SNAPSHOT) { log.error("snapshot flights exceed limit -> DEAD(MALFORMED) id={} size={}", head.cminmsgsId, normalized.size) procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, @@ -55,39 +55,56 @@ class SnapshotFlow( return } - // 2) 准备代元数据与差集（删除集 = gen.flids − 新代，ADFT 自动存活） val gen = flightSchd.getGen(day) + if (gen?.lastMessageId == messageId) { + txManager.inTransaction { + procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED) + } + safeBackfill(head, msg) + log.info("snapshot replay no-op id={} day={}", head.cminmsgsId, day) + return + } + val expected = gen?.version ?: 0L val newFlids = normalized.map { it.first }.toSet() val delFields = gen?.flids?.minus(newFlids) ?: emptySet() val newVersion = expected + 1L - // 3) 自有 PG 单事务原子提交（ACM2-28 定案：覆盖新代 + 域内差删 + SQL CAS + 事件 + SUCCEEDED） + val nextStates = normalized.map { (flid, fields) -> + val current = flightSchd.findNextStateByFlid(flid) + val commands = FlightStateEngine.commandsFromFields(flid, fields, snapshotReplace = true) + FlightStateEngine.apply(current, commands, messageId, bumpVersion = true) + } + try { txManager.inTransaction { - // 覆盖新代全量：强行声明 FDAY 归属（JDBC batch 批处理） - flightSchd.upsertSnapshotBatch(day, normalized) + flightSchd.persistNextStates(day, nextStates, snapshotReplace = true) - // 按代差删域化：仅删除 FDAY = day 且在 delFields 中的记录（ADFT 与跨代已迁移行存活） if (delFields.isNotEmpty()) { flightSchd.deleteDiffByDay(day, delFields) } - // SQL CAS 版本推进（防双写断言） val casSuccess = flightSchd.putGenIfVersion( day = day, expected = expected, - newGen = FlightSchdRepository.GenMeta(fday = day, version = newVersion, flids = newFlids), + newGen = FlightSchdRepository.GenMeta( + fday = day, + version = newVersion, + flids = newFlids, + lastMessageId = messageId, + ), ) if (!casSuccess) { - // 重放路径：version 已是目标值 → no-op 视为成功，否则为 CAS 冲突 val again = flightSchd.getGen(day) - if (again == null || again.version != newVersion) { - throw CasConflictException("gen-cas-conflict: expected=$expected current=${again?.version}") + val replayIdentity = again?.lastMessageId == messageId && again.version == newVersion + if (!replayIdentity) { + throw CasConflictException( + "gen-cas-conflict: expected=$expected current=${again?.version} msg=${again?.lastMessageId}", + ) } + log.info("gen idempotent replay within tx id={} day={}", head.cminmsgsId, day) } - // RESP 匹配 REQ_TRACK -> DONE val schdKind = msg.kind as? MsgKind.Schd if (schdKind?.subtype == MsgKind.SchdSubtype.RESP) { reqTrack.findOpenByKind("SCHD")?.let { req -> @@ -95,20 +112,21 @@ class SnapshotFlow( } } - // 构造并批量写入 schd 投递事件（载荷 = 字段集序列化，KAFKA_SCHD 线格式） - val events = normalized.map { (flid, fields) -> - MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = flid, payloadJson = FlightFieldsJson.toJson(fields)) + val events = nextStates.map { state -> + MsgEvent( + target = Targets.KAFKA_SCHD, + partitionKey = state.flid, + payloadJson = FlightFieldsJson.toJson(state.toFlightFields()), + ) } if (events.isNotEmpty()) { msgEvents.insertAll(events) } - // 终态置 SUCCEEDED procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED) } - // 提交后：backfill CMINMSGS（外部副作用，补偿保障） - inbox.backfillOnSuccess(head.cminmsgsId, msg.meta.sndr, msg.meta.type, msg.meta.styp, msg.meta.seqn) + safeBackfill(head, msg) log.info("snapshot SUCCEEDED id={} day={} flights={}", head.cminmsgsId, day, normalized.size) } catch (e: CasConflictException) { log.warn("gen CAS conflict -> FAILED(INFRA) id={} msg={}", head.cminmsgsId, e.message) @@ -116,6 +134,14 @@ class SnapshotFlow( } } + private fun safeBackfill(head: ProcState, msg: DecodedMessage) { + try { + inbox.backfillOnSuccess(head.cminmsgsId, msg.meta.sndr, msg.meta.type, msg.meta.styp, msg.meta.seqn) + } catch (e: Exception) { + log.error("backfill failed after SUCCEEDED id={} (compensation required)", head.cminmsgsId, e) + } + } + /** staging 结果（骨架）。 */ sealed interface StageResult { data class Ok(val day: String, val flights: List>) : StageResult diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/handlers/GtdtHandler.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/handlers/GtdtHandler.kt new file mode 100644 index 0000000..707ad15 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/handlers/GtdtHandler.kt @@ -0,0 +1,44 @@ +package com.gzzn.omms.msgexchange.processing.handlers + +import com.fasterxml.jackson.databind.ObjectMapper +import com.gzzn.omms.msgexchange.codec.FlopPayload +import com.gzzn.omms.msgexchange.domain.DecodedMessage +import com.gzzn.omms.msgexchange.domain.Decision +import com.gzzn.omms.msgexchange.domain.FlightChange +import com.gzzn.omms.msgexchange.domain.MsgKind +import com.gzzn.omms.msgexchange.domain.SchdPush +import com.gzzn.omms.msgexchange.domain.flight.FlightStateEngine +import com.gzzn.omms.msgexchange.infra.persistence.FlightFieldsJson +import com.gzzn.omms.msgexchange.processing.Handler +import jakarta.inject.Singleton + +/** + * FLOP-GTDT：登机门集合级快照替换（SIS §3.34；GTNO=0 清除）。 + * 产出 FlightChange.fields["GTDT"] 为 JSON 数组，供 FlightStateEngine → persistNextStates 写入明细表。 + */ +@Singleton +class GtdtHandler( + private val mapper: ObjectMapper = ObjectMapper(), +) : Handler { + override val kind: MsgKind = MsgKind.Flop("GTDT") + + override fun decide(flightView: Map>, msg: DecodedMessage): Decision { + val body = msg.body as? FlopPayload + ?: throw IllegalArgumentException("GTDT handler requires FlopPayload body") + val gtdtItems = body.collections["GTDT"] ?: emptyList() + val fields = linkedMapOf() + fields["FLID"] = body.flid + body.scalars.forEach { (k, v) -> fields[k] = v } + fields["GTDT"] = mapper.writeValueAsString(gtdtItems) + + val current = flightView[body.flid]?.let { FlightStateEngine.fromFlightFields(body.flid, it) } + val commands = FlightStateEngine.commandsFromFields(body.flid, fields, snapshotReplace = false) + val preview = FlightStateEngine.apply(current, commands, messageId = "", bumpVersion = false) + val payloadJson = FlightFieldsJson.toJson(preview.toFlightFields(mapper)) + + return Decision( + flightChanges = listOf(FlightChange(body.flid, fields)), + schdPush = listOf(SchdPush(body.flid, payloadJson)), + ) + } +}