feat(acm2-92): RQRD 人工登记与 POST /refdata/sync(C-12)

- 新增 POST /refdata/sync:STYP 14 类校验,RSTA 须带 RTYP
- REQ_TRACK 加 styp/rtyp 列(V8 迁移);RQRD/RQFD 各自单开放槽
- OutboundRequestService 加 registerRqrdSync,派发按行目标编码
- Pump 加 REF-RESP 守卫:无在途 RQRD 时 SKIPPED
- 回归测试覆盖 C-12 登记/派发/400/409/独立槽/守卫
This commit is contained in:
windyboy
2026-09-23 16:10:54 +08:00
parent f45a0ac898
commit 4fbb974307
28 changed files with 568 additions and 68 deletions
@@ -10,6 +10,7 @@ import com.gzzn.omms.msgexchange.domain.ErrorClass
import com.gzzn.omms.msgexchange.domain.MetaFields
import com.gzzn.omms.msgexchange.domain.MsgKind
import com.gzzn.omms.msgexchange.domain.OutboundRequestKeys
import com.gzzn.omms.msgexchange.domain.RqfdTimeFilter
import jakarta.inject.Singleton
import javax.xml.stream.XMLInputFactory
@@ -90,10 +91,18 @@ class JacksonXmlCodec : XmlCodec {
}
override fun encodeOutboundRqfd(seqn: Long, dttm: Long): String =
encodeOutboundRqfd(seqn, dttm, RqfdTimeFilter())
override fun encodeOutboundRqfd(seqn: Long, dttm: Long, filter: RqfdTimeFilter): String =
writeOutbound(
SisOutboundMessageXml(
meta = outboundMeta("RQFD", "NONE", seqn, dttm),
rqfd = RqfdXml(),
rqfd = RqfdXml(
stdb = filter.stdb,
stde = filter.stde,
etdb = filter.etdb,
etde = filter.etde,
),
),
)
@@ -1,6 +1,7 @@
package com.gzzn.omms.msgexchange.codec
import com.fasterxml.jackson.annotation.JsonIgnoreProperties
import com.fasterxml.jackson.annotation.JsonInclude
import com.fasterxml.jackson.dataformat.xml.annotation.JacksonXmlElementWrapper
import com.fasterxml.jackson.dataformat.xml.annotation.JacksonXmlProperty
import com.fasterxml.jackson.dataformat.xml.annotation.JacksonXmlRootElement
@@ -83,9 +84,15 @@ data class ErorXml(
@param:JacksonXmlProperty(localName = "ETEX") val etex: String? = null,
)
/** 出站 RQFD 空体占位(C-4 全量不带 STDB/STDE)。 */
/** 出站 RQFD。没传的时间条件保持 null,序列化时不写出(C-4)。 */
@JsonInclude(JsonInclude.Include.NON_NULL)
@JsonIgnoreProperties(ignoreUnknown = true)
class RqfdXml
data class RqfdXml(
@param:JacksonXmlProperty(localName = "STDB") val stdb: String? = null,
@param:JacksonXmlProperty(localName = "STDE") val stde: String? = null,
@param:JacksonXmlProperty(localName = "ETDB") val etdb: String? = null,
@param:JacksonXmlProperty(localName = "ETDE") val etde: String? = null,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class RqrdXml(
@@ -2,6 +2,7 @@ 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.RqfdTimeFilter
/**
* 解码失败的结果:错误分类加上一句原因。
@@ -22,9 +23,13 @@ sealed interface DecodeResult {
interface XmlCodec {
fun decode(rawXml: String): DecodeResult
/** C-4RQFD STYP=NONE,空 RQFD 体。 */
/** C-4RQFD STYP=NONE。不带时间条件时体为空。 */
fun encodeOutboundRqfd(seqn: Long, dttm: Long): String
/** C-4:只写入网页传来的时间条件,没传的标签不出现。 */
fun encodeOutboundRqfd(seqn: Long, dttm: Long, filter: RqfdTimeFilter): String =
encodeOutboundRqfd(seqn, dttm)
/** C-4RQRDSTYP 为参考数据子类型;RSTA 时可带 RTYP。 */
fun encodeOutboundRqrd(styp: String, seqn: Long, dttm: Long, rtyp: String? = null): String
}
@@ -0,0 +1,51 @@
package com.gzzn.omms.msgexchange.domain
/**
* 日计划请求的时间条件(`C-4`)。网页传入,本系统不补、不改。
* 空白视为没传。非空但不符 `DDMONYYHHMM` 则整单拒绝。
*/
data class RqfdTimeFilter(
val stdb: String? = null,
val stde: String? = null,
val etdb: String? = null,
val etde: String? = null,
) {
/** 四个条件都没传:回信是当天完整名单,才删除缺席航班(`US-07` AC2)。 */
fun isCompleteDay(): Boolean = stdb == null && stde == null && etdb == null && etde == null
sealed interface Parse {
data class Ok(val filter: RqfdTimeFilter) : Parse
data object Invalid : Parse
}
companion object {
private val TOKEN = Regex(
"""^(0[1-9]|[12]\d|3[01])(JAN|FEB|MAR|APR|MAY|JUN|JUL|AUG|SEP|OCT|NOV|DEC)\d{2}([01]\d|2[0-3])[0-5]\d$""",
)
fun parse(stdb: String?, stde: String?, etdb: String?, etde: String?): Parse {
val fields = listOf(stdb, stde, etdb, etde).map { one(it) }
if (fields.any { it is Field.Bad }) return Parse.Invalid
return Parse.Ok(
RqfdTimeFilter(
stdb = (fields[0] as Field.Value).text,
stde = (fields[1] as Field.Value).text,
etdb = (fields[2] as Field.Value).text,
etde = (fields[3] as Field.Value).text,
),
)
}
private fun one(raw: String?): Field {
val text = raw?.trim().orEmpty()
if (text.isEmpty()) return Field.Value(null)
if (!TOKEN.matches(text)) return Field.Bad
return Field.Value(text)
}
}
private sealed interface Field {
data class Value(val text: String?) : Field
data object Bad : Field
}
}
@@ -0,0 +1,41 @@
package com.gzzn.omms.msgexchange.domain
/**
* RQRD 参考数据请求的目标类别(`C-12`)。
* 类别码以 SIS 为准(`SIS:4.6.2`);`RSTA` 须带资源类型,其余类别不得带。
*/
data class RqrdTarget(
val styp: String,
val rtyp: String? = null,
) {
sealed interface Parse {
data class Ok(val target: RqrdTarget) : Parse
data object UnknownStyp : Parse
data object MissingRtyp : Parse
data object UnexpectedRtyp : Parse
}
companion object {
/** SIS `4.6.2` 的 14 种 RQRD 子类型。 */
val STYPS = setOf(
"COUL", "ARPT", "AIRL", "AIRC", "REGN", "ORGN", "FLTL",
"TLST", "SLST", "CLST", "GLST", "BLST", "CHLT", "RSTA",
)
/** 资源状态类型,见 implementation.md「静态参考数据」。 */
val RTYPS = setOf("BELT", "CNTR", "GATE", "STND")
fun parse(styp: String?, rtyp: String?): Parse {
val s = styp?.trim().orEmpty().uppercase()
if (s !in STYPS) return Parse.UnknownStyp
val r = rtyp?.trim().orEmpty().uppercase().ifEmpty { null }
if (s == "RSTA") {
if (r == null) return Parse.MissingRtyp
if (r !in RTYPS) return Parse.UnknownStyp
return Parse.Ok(RqrdTarget(s, r))
}
if (r != null) return Parse.UnexpectedRtyp
return Parse.Ok(RqrdTarget(s))
}
}
}
@@ -26,9 +26,8 @@ data class FlightMainRow(
/**
* 日计划报文(SCHD DNLD/RESP)里的一条 FLTR 记录,是解码后的产物。
*
* scalars 和 collections 只装报文里真正出现过的字段:出现就覆盖本地值(标量给空串表示
* 显式清空),没出现就保留库里已有的值;集合一旦出现就按合并后的完整结果整体覆盖写入
* 合并规则见 docs/implementation.md「SCHD 日计划」。
* scalars 和 collections 只装报文里真正出现过的字段。写入当前态时这一班整份替换,
* 没出现的字段清掉(`C-6`)。规则见 docs/implementation.md「SCHD」
*/
data class ScheduleRecord(
val flid: String,
@@ -91,8 +91,8 @@ object FlightStateEngine {
/**
* 把一条日计划记录写成新的当前态:**以 AODB 下发的这份快照为准**——报文带的字段写进去,
* 没带的字段和集合一律清掉(`C-6`、`US-07` AC3)。日计划是覆盖范围内的完整列表,
* 不是增量,所以没有"没出现就保留"这回事(那是 FLOP/ADFT 的语义,见 [mergedState])。
* 没带的字段和集合一律清掉(`C-6`、`US-07` AC3)。这是这一班的整份替换,不是增量
* (增量是 FLOP/ADFT,见 [mergedState])。报文里没有的其他航班是否删除见 `INV-7`。
*
* keepDeleted = true 时即使收到日计划也保持 DELETED:日计划不能把删掉的航班救回来,
* 唯一的恢复入口是 ADFT。调用方负责记一条 SCHD_REVIVE_CONFLICT 告警。
@@ -1,6 +1,8 @@
package com.gzzn.omms.msgexchange.infra.persistence
import com.gzzn.omms.msgexchange.domain.ErrorClass
import com.gzzn.omms.msgexchange.domain.RqfdTimeFilter
import com.gzzn.omms.msgexchange.domain.RqrdTarget
import com.gzzn.omms.msgexchange.domain.MsgEvent
import com.gzzn.omms.msgexchange.domain.ProcState
import com.gzzn.omms.msgexchange.domain.ProcStatus
@@ -321,7 +323,6 @@ interface SnapshotLogRepository {
/**
* 上游请求的跟踪表:记录我们发出去的请求、以及对方回来的应答。
*
* 目前只有表和读写方法,**还没有运行时协调器**(出站写信箱、超时、应答匹配都没实现)。
* 设计意图是:RESP 报文按(运营日、发送方、请求类型)匹配最近一条待应答的请求;
* 同一类请求同时只保留一条有效,发新请求时把旧的置为已过期。
*/
@@ -339,10 +340,25 @@ interface ReqTrackRepository {
val writeUncertain: Boolean = false,
val sentAt: Instant? = null,
val createdAt: Instant? = null,
)
val stdb: String? = null,
val stde: String? = null,
val etdb: String? = null,
val etde: String? = null,
val styp: String? = null,
val rtyp: String? = null,
) {
fun rqfdFilter() = RqfdTimeFilter(stdb, stde, etdb, etde)
fun rqrdTarget() = if (styp != null) RqrdTarget(styp, rtyp) else null
}
/** 登记新请求:同类(类型+运营日+发送方)旧有效请求先置 EXPIRED。 */
fun insert(reqType: String, operationDay: LocalDate, sender: String): Long
/** 登记新请求:同类(类型+运营日+发送方)旧有效请求先置 EXPIRED。日计划时间条件原样留下(C-4)。 */
fun insert(
reqType: String,
operationDay: LocalDate,
sender: String,
filter: RqfdTimeFilter = RqfdTimeFilter(),
target: RqrdTarget? = null,
): Long
fun findLatest(reqType: String, operationDay: LocalDate, sender: String, states: List<ReqState>): Req?
@@ -1091,7 +1091,13 @@ class JdbcReqTrackRepository(
private val ds: DataSource,
private val clock: Clock,
) : ReqTrackRepository {
override fun insert(reqType: String, operationDay: LocalDate, sender: String): Long {
override fun insert(
reqType: String,
operationDay: LocalDate,
sender: String,
filter: com.gzzn.omms.msgexchange.domain.RqfdTimeFilter,
target: com.gzzn.omms.msgexchange.domain.RqrdTarget?,
): Long {
ds.update(
"""
UPDATE req_track SET state = 'EXPIRED'
@@ -1103,12 +1109,19 @@ class JdbcReqTrackRepository(
ps.setString(3, sender)
}
return ds.updateReturningLong(
"INSERT INTO req_track (req_type, operation_day, sender, state, created_at) VALUES (?, ?, ?, 'PENDING', ?) RETURNING req_id",
"INSERT INTO req_track (req_type, operation_day, sender, state, created_at, stdb, stde, etdb, etde, styp, rtyp) " +
"VALUES (?, ?, ?, 'PENDING', ?, ?, ?, ?, ?, ?, ?) RETURNING req_id",
{ ps ->
ps.setString(1, reqType)
ps.setDate(2, java.sql.Date.valueOf(operationDay))
ps.setString(3, sender)
ps.setTimestamp(4, clock.instant().toSqlTimestamp())
ps.setString(5, filter.stdb)
ps.setString(6, filter.stde)
ps.setString(7, filter.etdb)
ps.setString(8, filter.etde)
ps.setString(9, target?.styp)
ps.setString(10, target?.rtyp)
},
)
}
@@ -1245,5 +1258,11 @@ class JdbcReqTrackRepository(
writeUncertain = rs.getBoolean("write_uncertain"),
sentAt = rs.getInstant("sent_at"),
createdAt = rs.getInstant("created_at"),
stdb = rs.getString("stdb"),
stde = rs.getString("stde"),
etdb = rs.getString("etdb"),
etde = rs.getString("etde"),
styp = rs.getString("styp"),
rtyp = rs.getString("rtyp"),
)
}
@@ -496,7 +496,13 @@ class StubReqTrack(private val clock: Clock = Clock.systemUTC()) : ReqTrackRepos
fun clear() = rows.clear()
override fun insert(reqType: String, operationDay: LocalDate, sender: String): Long {
override fun insert(
reqType: String,
operationDay: LocalDate,
sender: String,
filter: com.gzzn.omms.msgexchange.domain.RqfdTimeFilter,
target: com.gzzn.omms.msgexchange.domain.RqrdTarget?,
): Long {
// 同一类请求(类型 + 运营日 + 发送方)只保留一条有效,旧的先置为已过期
rows.values.filter {
it.reqType == reqType && it.operationDay == operationDay && it.sender == sender &&
@@ -506,6 +512,12 @@ class StubReqTrack(private val clock: Clock = Clock.systemUTC()) : ReqTrackRepos
rows[id] = ReqTrackRepository.Req(
id, reqType, operationDay, sender, ReqTrackRepository.ReqState.PENDING,
createdAt = clock.instant(),
stdb = filter.stdb,
stde = filter.stde,
etdb = filter.etdb,
etde = filter.etde,
styp = target?.styp,
rtyp = target?.rtyp,
)
return id
}
@@ -1,8 +1,12 @@
package com.gzzn.omms.msgexchange.ingress
import com.fasterxml.jackson.annotation.JsonIgnoreProperties
import com.fasterxml.jackson.annotation.JsonProperty
import io.micronaut.http.HttpResponse
import io.micronaut.http.HttpStatus
import io.micronaut.http.MediaType
import io.micronaut.http.annotation.Body
import io.micronaut.http.annotation.Consumes
import io.micronaut.http.annotation.Controller
import io.micronaut.http.annotation.Post
import io.micronaut.http.annotation.Produces
@@ -16,6 +20,7 @@ import io.micronaut.http.annotation.Produces
class InboxController(
private val inbox: InboxService,
private val schdSync: SchdSyncService,
private val refdataSync: RefdataSyncService,
) {
@Post("/cminmsgs/send")
@@ -25,14 +30,51 @@ class InboxController(
return HttpResponse.ok(receipt.msgId.toString()) // TODO: 先返回消息 ID 文本,等跟现役响应体逐字对拍过再定稿
}
/** C-8:登记 RQFD 出站请求;开放槽占用时 409,不重复登记。 */
/** C-8:登记 RQFD。请求体是网页选定的时间条件;开放槽占用时 409。 */
@Post("/schd/sync")
@Consumes(MediaType.APPLICATION_JSON)
@Produces(MediaType.TEXT_PLAIN)
fun schdSync(): HttpResponse<String> =
when (val outcome = schdSync.trigger()) {
fun schdSync(@Body body: SchdSyncBody?): HttpResponse<String> =
when (val outcome = schdSync.trigger(body?.stdb, body?.stde, body?.etdb, body?.etde)) {
is SchdSyncService.Outcome.Registered ->
HttpResponse.ok(outcome.reqId.toString())
SchdSyncService.Outcome.OpenExists ->
HttpResponse.status<String>(io.micronaut.http.HttpStatus.CONFLICT).body("open-request-exists")
HttpResponse.status<String>(HttpStatus.CONFLICT).body("open-request-exists")
SchdSyncService.Outcome.InvalidTime ->
HttpResponse.status<String>(HttpStatus.BAD_REQUEST).body("invalid-time-filter")
}
/** C-12:登记 RQRD。请求体是网页选定的类别码;开放槽占用时 409。 */
@Post("/refdata/sync")
@Consumes(MediaType.APPLICATION_JSON)
@Produces(MediaType.TEXT_PLAIN)
fun refdataSync(@Body body: RefdataSyncBody?): HttpResponse<String> =
when (val outcome = refdataSync.trigger(body?.styp, body?.rtyp)) {
is RefdataSyncService.Outcome.Registered ->
HttpResponse.ok(outcome.reqId.toString())
RefdataSyncService.Outcome.OpenExists ->
HttpResponse.status<String>(HttpStatus.CONFLICT).body("open-request-exists")
RefdataSyncService.Outcome.UnknownStyp ->
HttpResponse.status<String>(HttpStatus.BAD_REQUEST).body("unknown-styp")
RefdataSyncService.Outcome.MissingRtyp ->
HttpResponse.status<String>(HttpStatus.BAD_REQUEST).body("missing-rtyp")
RefdataSyncService.Outcome.UnexpectedRtyp ->
HttpResponse.status<String>(HttpStatus.BAD_REQUEST).body("unexpected-rtyp")
}
}
/** 网页提交的参考数据类别。`RSTA` 须带 `RTYP`,其余类别不得带(C-12)。 */
@JsonIgnoreProperties(ignoreUnknown = true)
data class RefdataSyncBody(
@param:JsonProperty("STYP") val styp: String? = null,
@param:JsonProperty("RTYP") val rtyp: String? = null,
)
/** 网页提交的日计划时间条件。字段可缺,缺了表示不按该项筛选(C-4)。 */
@JsonIgnoreProperties(ignoreUnknown = true)
data class SchdSyncBody(
@param:JsonProperty("STDB") val stdb: String? = null,
@param:JsonProperty("STDE") val stde: String? = null,
@param:JsonProperty("ETDB") val etdb: String? = null,
@param:JsonProperty("ETDE") val etde: String? = null,
)
@@ -0,0 +1,32 @@
package com.gzzn.omms.msgexchange.ingress
import com.gzzn.omms.msgexchange.domain.RqrdTarget
import com.gzzn.omms.msgexchange.processing.OutboundRequestService
import jakarta.inject.Singleton
/**
* C-12登记 RQRD请求体是网页选定的类别码RSTA 须带资源类型
*/
@Singleton
class RefdataSyncService(private val outbound: OutboundRequestService) {
sealed interface Outcome {
data class Registered(val reqId: Long) : Outcome
data object OpenExists : Outcome
data object UnknownStyp : Outcome
data object MissingRtyp : Outcome
data object UnexpectedRtyp : Outcome
}
fun trigger(styp: String?, rtyp: String?): Outcome {
val target = when (val parsed = RqrdTarget.parse(styp, rtyp)) {
is RqrdTarget.Parse.Ok -> parsed.target
RqrdTarget.Parse.UnknownStyp -> return Outcome.UnknownStyp
RqrdTarget.Parse.MissingRtyp -> return Outcome.MissingRtyp
RqrdTarget.Parse.UnexpectedRtyp -> return Outcome.UnexpectedRtyp
}
return when (val r = outbound.registerRqrdSync(target)) {
is OutboundRequestService.RegisterOutcome.Registered -> Outcome.Registered(r.reqId)
OutboundRequestService.RegisterOutcome.OpenExists -> Outcome.OpenExists
}
}
}
@@ -1,5 +1,6 @@
package com.gzzn.omms.msgexchange.ingress
import com.gzzn.omms.msgexchange.domain.RqfdTimeFilter
import com.gzzn.omms.msgexchange.processing.OutboundRequestService
import jakarta.inject.Singleton
@@ -8,11 +9,17 @@ class SchdSyncService(private val outbound: OutboundRequestService) {
sealed interface Outcome {
data class Registered(val reqId: Long) : Outcome
data object OpenExists : Outcome
data object InvalidTime : Outcome
}
fun trigger(): Outcome =
when (val r = outbound.registerRqfdSync()) {
fun trigger(stdb: String? = null, stde: String? = null, etdb: String? = null, etde: String? = null): Outcome {
val filter = when (val parsed = RqfdTimeFilter.parse(stdb, stde, etdb, etde)) {
RqfdTimeFilter.Parse.Invalid -> return Outcome.InvalidTime
is RqfdTimeFilter.Parse.Ok -> parsed.filter
}
return when (val r = outbound.registerRqfdSync(filter)) {
is OutboundRequestService.RegisterOutcome.Registered -> Outcome.Registered(r.reqId)
OutboundRequestService.RegisterOutcome.OpenExists -> Outcome.OpenExists
}
}
}
@@ -4,6 +4,8 @@ import com.gzzn.omms.msgexchange.codec.XmlCodec
import com.gzzn.omms.msgexchange.config.OperationDayProps
import com.gzzn.omms.msgexchange.config.PipelineProps
import com.gzzn.omms.msgexchange.domain.OutboundRequestKeys
import com.gzzn.omms.msgexchange.domain.RqfdTimeFilter
import com.gzzn.omms.msgexchange.domain.RqrdTarget
import com.gzzn.omms.msgexchange.infra.persistence.CoutmsgOutboxRepository
import com.gzzn.omms.msgexchange.infra.persistence.ReqTrackRepository
import jakarta.inject.Singleton
@@ -50,20 +52,40 @@ class OutboundRequestService(
listOf(ReqTrackRepository.ReqState.PENDING, ReqTrackRepository.ReqState.SENT),
) != null
fun registerRqfdSync(): RegisterOutcome {
fun registerRqfdSync(filter: RqfdTimeFilter = RqfdTimeFilter()): RegisterOutcome {
val day = currentOperationDay()
if (hasOpenRequest(OutboundRequestKeys.RQFD_REQ_TYPE, day)) return RegisterOutcome.OpenExists
val reqId = reqTrack.insert(OutboundRequestKeys.RQFD_REQ_TYPE, day, OutboundRequestKeys.SENDER)
val reqId = reqTrack.insert(OutboundRequestKeys.RQFD_REQ_TYPE, day, OutboundRequestKeys.SENDER, filter)
dispatchPending(limit = 1)
return RegisterOutcome.Registered(reqId)
}
fun hasOpenSentRqfd(operationDay: LocalDate = currentOperationDay()): Boolean =
/** C-12:登记一次 RQRD。开放槽占用时 OpenExists;落信沿用排队规则。 */
fun registerRqrdSync(target: RqrdTarget): RegisterOutcome {
val day = currentOperationDay()
if (hasOpenRequest(OutboundRequestKeys.RQRD_REQ_TYPE, day)) return RegisterOutcome.OpenExists
val reqId = reqTrack.insert(OutboundRequestKeys.RQRD_REQ_TYPE, day, OutboundRequestKeys.SENDER, target = target)
dispatchPending(limit = 1)
return RegisterOutcome.Registered(reqId)
}
fun openSentRqfd(operationDay: LocalDate = currentOperationDay()): ReqTrackRepository.Req? =
reqTrack.findLatest(
OutboundRequestKeys.RQFD_REQ_TYPE,
operationDay,
OutboundRequestKeys.SENDER,
listOf(ReqTrackRepository.ReqState.SENT),
)
fun hasOpenSentRqfd(operationDay: LocalDate = currentOperationDay()): Boolean =
openSentRqfd(operationDay) != null
fun hasOpenSentRqrd(operationDay: LocalDate = currentOperationDay()): Boolean =
reqTrack.findLatest(
OutboundRequestKeys.RQRD_REQ_TYPE,
operationDay,
OutboundRequestKeys.SENDER,
listOf(ReqTrackRepository.ReqState.SENT),
) != null
fun completeRqfdResponse(operationDay: LocalDate = currentOperationDay()): Boolean =
@@ -102,8 +124,13 @@ class OutboundRequestService(
val seqn = outbox.nextOutboundSeqn()
val dttm = beijingMetaDttm()
val xml = when (req.reqType) {
OutboundRequestKeys.RQFD_REQ_TYPE -> codec.encodeOutboundRqfd(seqn, dttm)
OutboundRequestKeys.RQRD_REQ_TYPE -> codec.encodeOutboundRqrd("AIRL", seqn, dttm)
OutboundRequestKeys.RQFD_REQ_TYPE -> codec.encodeOutboundRqfd(seqn, dttm, req.rqfdFilter())
OutboundRequestKeys.RQRD_REQ_TYPE -> {
val styp = req.styp ?: return false.also {
log.warn("RQRD row without STYP reqId={}", req.reqId)
}
codec.encodeOutboundRqrd(styp, seqn, dttm, req.rtyp)
}
else -> {
log.warn("unknown req_type for dispatch reqId={} type={}", req.reqId, req.reqType)
return false
@@ -203,9 +203,10 @@ class MessageProcessor(
if (body == null) return deadMalformed(head, "missing-schd-body")
when (kind.subtype) {
MsgKind.SchdSubtype.DNLD ->
scheduleProcessor.applyScheduleRecords(head, decoded)
scheduleProcessor.applyScheduleRecords(head, decoded, deleteAbsent = true)
MsgKind.SchdSubtype.RESP -> {
if (!outbound.hasOpenSentRqfd()) {
val open = outbound.openSentRqfd()
if (open == null) {
log.info("SCHD-RESP without open RQFD -> SKIPPED msgId={}", head.msgId)
procState.markTerminal(
head.msgId, ProcStatus.SKIPPED,
@@ -214,7 +215,11 @@ class MessageProcessor(
)
return
}
val respResult = scheduleProcessor.applyScheduleRecords(head, decoded)
val respResult = scheduleProcessor.applyScheduleRecords(
head,
decoded,
deleteAbsent = open.rqfdFilter().isCompleteDay(),
)
if (respResult is ApplyResult.Succeeded || respResult is ApplyResult.ReplaySkipped) {
outbound.completeRqfdResponse()
}
@@ -240,6 +245,15 @@ class MessageProcessor(
is MsgKind.RefData -> {
val body = decoded.body as? com.gzzn.omms.msgexchange.domain.ref.RefDataBody // validated below
if (body == null) return deadMalformed(head, "missing-refdata-body")
if (body.styp.equals("RESP", ignoreCase = true) && !outbound.hasOpenSentRqrd()) {
log.info("REF-RESP without open RQRD -> SKIPPED msgId={}", head.msgId)
procState.markTerminal(
head.msgId, ProcStatus.SKIPPED,
lastError = "resp-guard:no-open-req",
now = clock.instant(),
)
return
}
referenceDataProcessor.apply(head, decoded, kind.type)
}
MsgKind.Eror -> {
@@ -50,7 +50,8 @@ class ProtocolViolation(message: String) : RuntimeException(message)
* 校验或运营日核对不过就整包拒绝一条都不落`INV-4`写入阶段按批分事务
* 整份写完且投影刷完才算处理完成中途失败下轮整包重来`INV-9`
*
* 日计划是覆盖范围内的完整列表范围内缺席航班要标删`INV-7`范围外的不受影响
* 报文里的航班整份替换只有完整名单才删除覆盖范围内缺席航班`INV-7`
* 带了时间条件的应答不是完整名单不删除回信里没有的航班
*/
@Singleton
class ScheduleProcessor(
@@ -70,7 +71,7 @@ class ScheduleProcessor(
cutoffHour = operationDayProps.cutoffHour,
)
fun applyScheduleRecords(head: ProcState, msg: DecodedMessage): ApplyResult {
fun applyScheduleRecords(head: ProcState, msg: DecodedMessage, deleteAbsent: Boolean = true): ApplyResult {
val body = msg.body as? ScheduleBody ?: return ApplyResult.DeadProtocol("missing-schd-body")
val started = System.nanoTime()
@@ -103,7 +104,7 @@ class ScheduleProcessor(
val batchSize = props.schd.snapshotBatch
return try {
// 分批写:每批一个事务(`US-07` AC4),整份写完才刷投影、才记终态(`INV-9`)。
// 覆盖范围内缺席航班在最后清扫`INV-7`)。
// 完整名单才在最后清扫覆盖范围内缺席航班(`INV-7`)。
val upserted = commit.commitBatched(head) { batches ->
// 一次批量查出这些航班现有的运营日,逐个比对(避免逐条查询)。
// 这一比对必须在任何一批写入之前跑完:运营日冲突要整包拒绝、一条都不落(`INV-4`)。
@@ -150,14 +151,16 @@ class ScheduleProcessor(
}
}
sweepAbsent(
head = head,
batches = batches,
coverage = ok.perRecordDay.values.toSet(),
present = ok.perRecordDay.keys,
batchSize = batchSize,
flags = flags,
)
if (deleteAbsent) {
sweepAbsent(
head = head,
batches = batches,
coverage = ok.perRecordDay.values.toSet(),
present = ok.perRecordDay.keys,
batchSize = batchSize,
flags = flags,
)
}
written
}
logSnapshot(head, body, SnapshotResult.COMMITTED, upserted, flags, started)
@@ -169,7 +172,7 @@ class ScheduleProcessor(
}
/**
* 覆盖范围内缺席的航班标删登记删除通知从投影里删掉`INV-7``US-07` AC2/AC5
* 完整名单覆盖范围内缺席的航班标删登记删除通知从投影里删掉`INV-7``US-07` AC2/AC5
*
* [coverage] 是这份报文覆盖的运营日取自报文自身每条记录的 `SODT` 推出的运营日
* 整包校验已保证每条都算得出覆盖范围外的航班一条都不碰前一日延误的航班不在
@@ -287,14 +287,13 @@ CREATE UNIQUE INDEX uq_schd_event
WHERE TARGET = 'KAFKA:schd';
-- ⑥ 请求状态机:只有 RESP 完成 RQFD 请求,按(运营日、发送方、请求类型)匹配最新一条
-- 开放请求。同类请求只留一条有效,新请求置旧请求为 EXPIRED。登记、超时与应答匹配
-- 尚未实现(G-REQ-TRACK、G-REQ-OPEN-UNIQUE)。
-- 开放请求。同类请求只留一条有效,新请求置旧请求为 EXPIRED。
CREATE TABLE REQ_TRACK (
REQ_ID BIGSERIAL PRIMARY KEY,
REQ_TYPE VARCHAR(20) NOT NULL,
OPERATION_DAY DATE NOT NULL, -- 请求覆盖运营日
SENDER VARCHAR(64) NOT NULL, -- 请求发送方(匹配键之一)
STATE VARCHAR(16) NOT NULL, -- PENDING/SENT/DONE/EXPIRED
STATE VARCHAR(16) NOT NULL, -- PENDING/SENT/DONE/EXPIRED/FAILED
COUTMSGS_ID BIGINT,
SENT_AT TIMESTAMP(6) WITH TIME ZONE,
COMPLETED_AT TIMESTAMP(6) WITH TIME ZONE,
@@ -0,0 +1,6 @@
-- 日计划请求的时间条件:网页传入后原样留到落信,空表示没传(C-4)。
ALTER TABLE req_track
ADD COLUMN IF NOT EXISTS stdb VARCHAR(11),
ADD COLUMN IF NOT EXISTS stde VARCHAR(11),
ADD COLUMN IF NOT EXISTS etdb VARCHAR(11),
ADD COLUMN IF NOT EXISTS etde VARCHAR(11);
@@ -0,0 +1,5 @@
-- RQRD 参考数据请求的子类型:网页选定后原样留到落信(C-12)。
-- RQFD 行两列为空。开放槽仍按 REQ_TYPE 单槽,不按 STYP 分槽。
ALTER TABLE req_track
ADD COLUMN IF NOT EXISTS styp VARCHAR(4),
ADD COLUMN IF NOT EXISTS rtyp VARCHAR(4);
@@ -1,6 +1,7 @@
package com.gzzn.omms.msgexchange.codec
import com.gzzn.omms.msgexchange.domain.MsgKind
import com.gzzn.omms.msgexchange.domain.RqfdTimeFilter
import com.gzzn.omms.msgexchange.domain.flight.FlightSnapshot
import com.gzzn.omms.msgexchange.domain.flight.FlightState
import com.gzzn.omms.msgexchange.domain.flight.FlightStateEngine
@@ -17,12 +18,27 @@ class JacksonXmlCodecTest {
private val codec = JacksonXmlCodec()
@Test
fun `encode outbound RQFD uses OMMS meta and empty body per C-4`() {
fun `encode outbound RQFD uses OMMS meta and omits time filters that were not sent`() {
val xml = codec.encodeOutboundRqfd(1243L, 20021010090311L)
assertTrue(xml.contains("<SNDR>OMMS</SNDR>"))
assertTrue(xml.contains("<TYPE>RQFD</TYPE>"))
assertTrue(xml.contains("<STYP>NONE</STYP>"))
assertTrue(xml.contains("<RQFD"))
assertFalse(xml.contains("STDB"))
assertFalse(xml.contains("ETDE"))
}
@Test
fun `encode outbound RQFD writes only the time filters the page sent`() {
val xml = codec.encodeOutboundRqfd(
1243L,
20021010090311L,
RqfdTimeFilter(stdb = "12JAN041730", etde = "13JAN042359"),
)
assertTrue(xml.contains("<STDB>12JAN041730</STDB>"))
assertTrue(xml.contains("<ETDE>13JAN042359</ETDE>"))
assertFalse(xml.contains("STDE"))
assertFalse(xml.contains("ETDB"))
}
@Test
@@ -11,7 +11,7 @@ import java.sql.DriverManager
/**
* 在真实 PostgreSQL 上跑一遍迁移链确认结果符合预期`V1__flight_state_baseline.sql`
* 基线加 V2V6 增量迁移
* 基线加 V2V8 增量迁移
* 依序执行成功该建的表和 PIPELINE_LOCK 单行种子都在回填事实落在基线里`INBOX_CURSOR`
* `BACKFILL_TODO``idx_evt_flid``PROC_STATE` 的处理开始时间列都不复存在FLIGHT_CHUTE
* 的类字段列已由 V2 更名为 CCLS/CTYPSIS 口径
@@ -59,6 +59,8 @@ class FlywayMigrationTest {
"4" to "V4__req_track_outbound_seqn.sql",
"5" to "V5__unmapped_field_srvt_vipf.sql",
"6" to "V6__basicdata_ref_data.sql",
"7" to "V7__req_track_rqfd_window.sql",
"8" to "V8__req_track_rqrd_target.sql",
),
records.map { it.first to it.second },
)
@@ -133,6 +135,14 @@ class FlywayMigrationTest {
assertTrue(indexDef.contains("UNIQUE"), "uq_req_open 必须是唯一索引:$indexDef")
assertTrue(indexDef.contains("WHERE"), "uq_req_open 必须是仅约束开放态的部分索引:$indexDef")
}
stmt.executeQuery(
"SELECT column_name FROM information_schema.columns WHERE table_name = 'req_track' " +
"AND column_name IN ('stdb', 'stde', 'etdb', 'etde', 'styp', 'rtyp')",
).use { rs ->
val cols = mutableSetOf<String>()
while (rs.next()) cols.add(rs.getString("column_name"))
assertEquals(setOf("stdb", "stde", "etdb", "etde", "styp", "rtyp"), cols)
}
stmt.executeUpdate(
"INSERT INTO req_track (req_type, operation_day, sender, state, created_at) " +
"VALUES ('RQFD-NONE', DATE '2026-09-12', 'RMS', 'PENDING', now())",
@@ -5,6 +5,8 @@ import com.gzzn.omms.msgexchange.codec.JacksonXmlCodec
import com.gzzn.omms.msgexchange.config.OperationDayProps
import com.gzzn.omms.msgexchange.config.PipelineProps
import com.gzzn.omms.msgexchange.domain.OutboundRequestKeys
import com.gzzn.omms.msgexchange.domain.RqfdTimeFilter
import com.gzzn.omms.msgexchange.ingress.SchdSyncService
import com.gzzn.omms.msgexchange.infra.persistence.CoutmsgOutboxRepository
import com.gzzn.omms.msgexchange.infra.persistence.ReqTrackRepository
import com.gzzn.omms.msgexchange.infra.stub.StubCoutmsgOutbox
@@ -45,6 +47,28 @@ class OutboundRequestTest {
assertTrue(outbox.messages.values.first().contains("<TYPE>RQFD</TYPE>"))
}
@Test
fun `schd sync writes the page time filter into COUTMSGS and leaves out the rest`() {
val req = StubReqTrack(clock)
val outbox = StubCoutmsgOutbox()
val svc = service(req, outbox)
svc.registerRqfdSync(RqfdTimeFilter(stdb = "12JAN041730"))
val xml = outbox.messages.values.first()
assertTrue(xml.contains("<STDB>12JAN041730</STDB>"))
assertFalse(xml.contains("STDE"))
assertFalse(xml.contains("ETDB"))
assertFalse(xml.contains("ETDE"))
}
@Test
fun `invalid time filter is rejected before registration`() {
val req = StubReqTrack(clock)
val sync = SchdSyncService(service(req, StubCoutmsgOutbox()))
assertEquals(SchdSyncService.Outcome.InvalidTime, sync.trigger(stdb = "today"))
assertTrue(req.rows.isEmpty())
}
@Test
fun `open slot rejects duplicate schd sync registration`() {
val req = StubReqTrack(clock)
@@ -108,4 +132,71 @@ class OutboundRequestTest {
assertTrue(svc.failFromEror(42L, "RQFD", "NONE"))
assertEquals(ReqTrackRepository.ReqState.FAILED, req.rows[reqId]!!.state)
}
@Test
fun `refdata sync registers AIRL and dispatches RQRD into COUTMSGS`() {
val req = StubReqTrack(clock)
val outbox = StubCoutmsgOutbox()
val sync = com.gzzn.omms.msgexchange.ingress.RefdataSyncService(service(req, outbox))
val outcome = sync.trigger("AIRL", null)
assertTrue(outcome is com.gzzn.omms.msgexchange.ingress.RefdataSyncService.Outcome.Registered)
val xml = outbox.messages.values.first()
assertTrue(xml.contains("<TYPE>RQRD</TYPE>"))
assertTrue(xml.contains("<STYP>AIRL</STYP>"))
}
@Test
fun `refdata sync writes RTYP only for RSTA`() {
val req = StubReqTrack(clock)
val outbox = StubCoutmsgOutbox()
val svc = service(req, outbox)
svc.registerRqrdSync(com.gzzn.omms.msgexchange.domain.RqrdTarget("RSTA", "GATE"))
val xml = outbox.messages.values.first()
assertTrue(xml.contains("<STYP>RSTA</STYP>"))
assertTrue(xml.contains("<RTYP>GATE</RTYP>"))
}
@Test
fun `refdata sync rejects unknown styp and rtyp misuse before registration`() {
val req = StubReqTrack(clock)
val sync = com.gzzn.omms.msgexchange.ingress.RefdataSyncService(service(req, StubCoutmsgOutbox()))
assertEquals(
com.gzzn.omms.msgexchange.ingress.RefdataSyncService.Outcome.UnknownStyp,
sync.trigger("NOPE", null),
)
assertEquals(
com.gzzn.omms.msgexchange.ingress.RefdataSyncService.Outcome.MissingRtyp,
sync.trigger("RSTA", null),
)
assertEquals(
com.gzzn.omms.msgexchange.ingress.RefdataSyncService.Outcome.UnexpectedRtyp,
sync.trigger("AIRL", "GATE"),
)
assertTrue(req.rows.isEmpty())
}
@Test
fun `open slot rejects duplicate refdata sync registration`() {
val req = StubReqTrack(clock)
val sync = com.gzzn.omms.msgexchange.ingress.RefdataSyncService(service(req, StubCoutmsgOutbox()))
sync.trigger("AIRL", null)
assertEquals(
com.gzzn.omms.msgexchange.ingress.RefdataSyncService.Outcome.OpenExists,
sync.trigger("AIRL", null),
)
}
@Test
fun `RQRD and RQFD use separate open slots`() {
val req = StubReqTrack(clock)
val rqrd = com.gzzn.omms.msgexchange.ingress.RefdataSyncService(service(req, StubCoutmsgOutbox()))
val rqfd = SchdSyncService(service(req, StubCoutmsgOutbox()))
rqrd.trigger("AIRL", null)
assertTrue(rqfd.trigger() is SchdSyncService.Outcome.Registered)
}
}
@@ -9,6 +9,10 @@ import com.gzzn.omms.msgexchange.domain.DecodedMessage
import com.gzzn.omms.msgexchange.domain.MetaFields
import com.gzzn.omms.msgexchange.domain.MsgKind
import com.gzzn.omms.msgexchange.domain.OutboundRequestKeys
import com.gzzn.omms.msgexchange.domain.RqfdTimeFilter
import com.gzzn.omms.msgexchange.domain.flight.FlightSnapshot
import com.gzzn.omms.msgexchange.domain.flight.FlightState
import com.gzzn.omms.msgexchange.domain.flight.ScheduleRecord
import com.gzzn.omms.msgexchange.domain.ProcState
import com.gzzn.omms.msgexchange.domain.ProcStatus
import com.gzzn.omms.msgexchange.infra.metrics.PipelineCounters
@@ -25,6 +29,7 @@ import com.gzzn.omms.msgexchange.infra.stub.StubReqTrack
import com.gzzn.omms.msgexchange.infra.stub.StubSnapshotLog
import com.gzzn.omms.msgexchange.MutableClock
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assertions.assertTrue
import org.junit.jupiter.api.Test
import java.time.Instant
import java.time.LocalDate
@@ -49,10 +54,10 @@ class RespGuardTest {
inbox: StubInbox,
req: StubReqTrack,
decoded: DecodedMessage,
flights: StubFlightState = StubFlightState(),
): MessageProcessor {
val props = PipelineProps()
val opDay = OperationDayProps().apply { zone = "Asia/Shanghai"; cutoffHour = 0 }
val flights = StubFlightState()
val events = StubMsgEvents()
val commit = testCommit(procState = proc, msgEvents = events, projection = StubFlightProjectionPort(), clock = clock)
return MessageProcessor(
@@ -94,6 +99,41 @@ class RespGuardTest {
assertEquals("resp-guard:no-open-req", proc.find(msgId)!!.lastError)
}
@Test
fun `filtered SCHD-RESP replaces flights in the reply and does not delete the rest`() {
val req = StubReqTrack(clock)
val reqId = req.insert(
OutboundRequestKeys.RQFD_REQ_TYPE,
day,
OutboundRequestKeys.SENDER,
RqfdTimeFilter(stdb = "15DEC261700"),
)
req.linkCoutmsgs(reqId, 1L, 1L)
req.markSent(reqId, clock.instant())
val flights = StubFlightState()
val flightDay = LocalDate.of(2026, 12, 15)
listOf("121", "122").forEach { flid ->
flights.persistFullState(
FlightSnapshot(flid, flightDay, FlightState.ACTIVE, 1, mapOf("SODT" to "15DEC261723"), emptyMap()),
msgId = 1,
now = Instant.EPOCH,
)
}
val body = ScheduleBody(1, listOf(ScheduleRecord("121", mapOf("SODT" to "15DEC261800"))))
val proc = StubProcState()
proc.insertIfAbsent(msgId, null)
val inbox = StubInbox()
inbox.raws[msgId] = "<x/>"
val p = processor(proc, inbox, req, schdMsg(MsgKind.SchdSubtype.RESP, body), flights)
p.processOne(ProcState(msgId, ProcStatus.PENDING, updatedAt = Instant.EPOCH))
assertEquals(ProcStatus.SUCCEEDED, proc.find(msgId)!!.state)
assertEquals(FlightState.ACTIVE, flights.findMainRow("122")!!.state)
assertTrue(flights.findMainRow("121")!!.stateVersion > 1L)
}
@Test
fun `SCHD-DNLD does not require open request`() {
val proc = StubProcState()
@@ -125,4 +165,23 @@ class RespGuardTest {
assertEquals(ProcStatus.SKIPPED, proc.find(msgId)!!.state)
}
@Test
fun `REF-RESP without open RQRD is SKIPPED`() {
val proc = StubProcState()
proc.insertIfAbsent(msgId, null)
val inbox = StubInbox()
inbox.raws[msgId] = "<x/>"
val decoded = DecodedMessage(
meta = MetaFields("AODB", "AIRL", "RESP", 1L, 1L),
kind = MsgKind.RefData("AIRL"),
rawXml = "<MSG/>",
body = com.gzzn.omms.msgexchange.domain.ref.RefDataBody(category = "AIRL", styp = "RESP", records = emptyList()),
)
val p = processor(proc, inbox, StubReqTrack(clock), decoded)
p.processOne(ProcState(msgId, ProcStatus.PENDING, updatedAt = Instant.EPOCH))
assertEquals(ProcStatus.SKIPPED, proc.find(msgId)!!.state)
assertEquals("resp-guard:no-open-req", proc.find(msgId)!!.lastError)
}
}
@@ -233,6 +233,34 @@ class ScheduleProcessorTest {
assertTrue(log.entries.single().flags.contains(SnapshotFlag.SCHD_ABSENT_DELETED))
}
/** 带了时间条件的应答不是完整名单:回信里的航班整份替换,其余航班不删(`US-07` AC2)。 */
@Test
fun `partial reply replaces flights in the message and leaves the others`() {
val flights = StubFlightState()
val events = StubMsgEvents()
val projection = StubFlightProjectionPort()
seed(flights, "122", LocalDate.of(2026, 12, 15))
projection.snapshots["122"] = """{"flid":"122"}"""
flights.persistFullState(
com.gzzn.omms.msgexchange.domain.flight.FlightSnapshot(
"121", LocalDate.of(2026, 12, 15), FlightState.ACTIVE, 1,
mapOf("SODT" to "15DEC261723", "REMC" to "old-note"), emptyMap(),
),
msgId = 1,
now = java.time.Instant.now(),
)
val result = processor(flights = flights, events = events, projection = projection)
.applyScheduleRecords(head(), message(makeBody("121" to "15DEC261800")), deleteAbsent = false)
assertEquals(ApplyResult.Succeeded, result)
assertEquals(FlightState.ACTIVE, flights.findMainRow("122")!!.state)
assertEquals(1L, flights.findMainRow("122")!!.stateVersion)
assertTrue(projection.snapshots.containsKey("122"))
assertFalse(events.rows.values.any { it.partitionKey == "122" })
assertFalse(flights.loadFullSnapshot("121")!!.scalars.containsKey("REMC"))
}
/** 还没被日计划收录的航班(运营日为空)不属于任何覆盖范围,缺席清扫碰不到它。 */
@Test
fun `flights without an operation day are out of every coverage range`() {