diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt index 9d81ec9..2f5c417 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt @@ -97,6 +97,7 @@ interface FlightSchdRepository { val fday: String, val version: Long, val flids: Set, + val lastMessageId: String? = null, val updatedAt: Instant = Instant.now(), ) @@ -112,6 +113,9 @@ interface FlightSchdRepository { /** 点查单航班字段集；无字段行（含航班不存在）返回 null。 */ fun findByFlid(flid: String): FlightFields? + /** v2：点查单航班 nextState（含 state_version / last_message_id 追踪字段）。 */ + fun findNextStateByFlid(flid: String): com.gzzn.omms.msgexchange.domain.flight.FlightNextState? + /** 点查多航班字段集：FLID → 字段集，仅含实际存在的航班。 */ fun findByFlids(flids: Collection): Map @@ -140,6 +144,17 @@ interface FlightSchdRepository { * 历史代清理：清理 cutoffDay 之前的历史代记录（与 FLIGHT_SCHD 历史清场生命周期对齐）。 */ fun deleteGenBefore(cutoffDay: String): Int + + /** + * v2 无损明细：将 nextState 写入宽表 + 明细表，并更新追踪字段。 + * 必须在 PipelineTransactionManager 事务内调用。 + */ + fun persistNextStates( + day: String?, + states: List, + snapshotReplace: Boolean, + now: Instant = Instant.now(), + ) } /** 事务管理器抽象：自有 PG 单事务原子保障。 */ diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlightDetailTables.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlightDetailTables.kt new file mode 100644 index 0000000..ab3bb71 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlightDetailTables.kt @@ -0,0 +1,179 @@ +package com.gzzn.omms.msgexchange.infra.persistence.jdbc + +import com.gzzn.omms.msgexchange.domain.flight.FlightNextState +import java.sql.Timestamp +import java.time.Instant +import javax.sql.DataSource + +/** + * v2 明细表读写（flight-state-design-v2 §3.2）。 + * 与宽表双写；读路径优先明细表（完整顺序/源序号保真）。 + */ +internal object FlightDetailTables { + /** 由明细表承载的集合键（v2 读写的权威来源）。 */ + val DETAIL_COLLECTION_KEYS: Set = setOf( + "GTDT", "CKDT", "CLDT", "PSDT", "CHDT", "DELY", "ABTM", "CHOT", "ROUT", "ERUT", + ) + + private data class TableSpec( + val table: String, + val collectionKey: String, + val seqAttr: String, + val columns: List>, // json attr -> db column + ) + + private val SPECS = listOf( + TableSpec("flight_gate", "GTDT", "GTNO", listOf( + "GATE" to "gate", "PGOT" to "pgot", "PGCT" to "pgct", "GOTM" to "gotm", "GCTM" to "gctm", "GTYP" to "gtyp", + )), + TableSpec("flight_checkin", "CKDT", "CKNO", listOf( + "CHKC" to "chkc", "CCLS" to "ccls", "PCOT" to "pcot", "PCCT" to "pcct", "COTM" to "cotm", "CCTM" to "cctm", "CTYP" to "ctyp", + )), + TableSpec("flight_belt", "CLDT", "CLNO", listOf( + "BELT" to "belt", "BCLS" to "bcls", "PCOT" to "pcot", "PCCT" to "pcct", "FBAG" to "fbag", "LBAG" to "lbag", "BTYP" to "btyp", + )), + TableSpec("flight_stand_plan", "PSDT", "PSNO", listOf( + "PSST" to "psst", "STST" to "stst", "STET" to "stet", + )), + TableSpec("flight_chute", "CHDT", "CHNO", listOf( + "CHUT" to "chut", "CHCLS" to "chcls", "PCBT" to "pcbt", "PCET" to "pcet", "CBTM" to "cbtm", "CETM" to "cetm", "CHTYP" to "chtyp", + )), + TableSpec("flight_delay", "DELY", "DLNO", listOf( + "CODE" to "code", "STRT" to "strt", "DURA" to "dura", "REMC" to "remc", + )), + TableSpec("flight_bridge_op", "ABTM", "ASNO", listOf( + "ABDG" to "abdg", "ABOP" to "abop", "AOTM" to "aotm", + )), + TableSpec("flight_chock_op", "CHOT", "CSNO", listOf( + "CHID" to "chid", "CHST" to "chst", "CHTM" to "chtm", + )), + ) + + fun replaceAll(ds: DataSource, state: FlightNextState, now: Instant) { + val ts = now.toSqlTimestamp() + val conn = ds.obtainConnection() + try { + for (spec in SPECS) { + ds.update("DELETE FROM ${spec.table} WHERE flid = ?") { ps -> ps.setString(1, state.flid) } + val items = state.collections[spec.collectionKey] ?: continue + if (items.isEmpty()) continue + val colNames = listOf("flid", "ordinal", "source_seq", "record_version") + + spec.columns.map { it.second } + listOf("created_at", "updated_at") + val sql = "INSERT INTO ${spec.table} (${colNames.joinToString(", ")}) VALUES (${ + colNames.joinToString(", ") { "?" } + })" + conn.prepareStatement(sql).use { ps -> + items.forEachIndexed { index, item -> + var i = 1 + ps.setString(i++, state.flid) + ps.setInt(i++, index + 1) + ps.setString(i++, item[spec.seqAttr]) + ps.setLong(i++, state.stateVersion) + spec.columns.forEach { (attr, _) -> ps.setString(i++, item[attr]) } + ps.setTimestamp(i++, ts) + ps.setTimestamp(i++, ts) + ps.addBatch() + } + ps.executeBatch() + } + } + replaceRoutes(ds, state, now) + } finally { + conn.releaseIfNotInTransaction() + } + } + + private fun replaceRoutes(ds: DataSource, state: FlightNextState, now: Instant) { + val ts = now.toSqlTimestamp() + ds.update("DELETE FROM flight_route_point WHERE flid = ?") { ps -> ps.setString(1, state.flid) } + listOf("ROUT" to "ROUT", "ERUT" to "ERUT").forEach { (key, kind) -> + val items = state.collections[key] ?: return@forEach + if (items.isEmpty()) return@forEach + val conn = ds.obtainConnection() + try { + conn.prepareStatement( + "INSERT INTO flight_route_point (flid, ordinal, source_seq, record_version, route_kind, apcd, scat, scdt, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ).use { ps -> + items.forEachIndexed { index, item -> + ps.setString(1, state.flid) + ps.setInt(2, index + 1) + ps.setString(3, item["RTNO"]) + ps.setLong(4, state.stateVersion) + ps.setString(5, kind) + ps.setString(6, item["APCD"]) + ps.setString(7, item["SCAT"]) + ps.setString(8, item["SCDT"]) + ps.setTimestamp(9, ts) + ps.setTimestamp(10, ts) + ps.addBatch() + } + ps.executeBatch() + } + } finally { + conn.releaseIfNotInTransaction() + } + } + } + + fun deleteForFlids(ds: DataSource, flids: Collection) { + if (flids.isEmpty()) return + val tables = SPECS.map { it.table } + "flight_route_point" + for (table in tables) { + for (chunk in flids.chunked(200)) { + val placeholders = chunk.joinToString(",") { "?" } + ds.update("DELETE FROM $table WHERE flid IN ($placeholders)") { ps -> + chunk.forEachIndexed { i, flid -> ps.setString(i + 1, flid) } + } + } + } + } + + fun loadCollections(ds: DataSource, flid: String): Map { + val out = linkedMapOf() + val mapper = com.fasterxml.jackson.databind.ObjectMapper() + for (spec in SPECS) { + val rows = ds.query( + "SELECT * FROM ${spec.table} WHERE flid = ? ORDER BY ordinal ASC", + { ps -> ps.setString(1, flid) }, + ) { rs -> + spec.columns.associate { (attr, col) -> attr to rs.getString(col) }.toMutableMap().apply { + put(spec.seqAttr, rs.getString("source_seq")) + } + } + if (rows.isNotEmpty()) { + out[spec.collectionKey] = mapper.writeValueAsString(rows) + } + } + listOf("ROUT" to "ROUT", "ERUT" to "ERUT").forEach { (key, kind) -> + val rows = ds.query( + "SELECT source_seq, apcd, scat, scdt FROM flight_route_point WHERE flid = ? AND route_kind = ? ORDER BY ordinal ASC", + { ps -> ps.setString(1, flid); ps.setString(2, kind) }, + ) { rs -> + mapOf( + "RTNO" to rs.getString("source_seq"), + "APCD" to rs.getString("apcd"), + "SCAT" to rs.getString("scat"), + "SCDT" to rs.getString("scdt"), + ) + } + if (rows.isNotEmpty()) { + out[key] = mapper.writeValueAsString(rows) + } + } + return out + } + + fun hasAnyDetailRows(ds: DataSource, flid: String): Boolean { + for (spec in SPECS) { + val count = ds.queryOne( + "SELECT 1 FROM ${spec.table} WHERE flid = ? LIMIT 1", + { ps -> ps.setString(1, flid) }, + ) { 1 } + if (count != null) return true + } + return ds.queryOne( + "SELECT 1 FROM flight_route_point WHERE flid = ? LIMIT 1", + { ps -> ps.setString(1, flid) }, + ) { 1 } != null + } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt index 1536ba6..d85eac7 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt @@ -394,6 +394,9 @@ class JdbcFlightSchdRepository( /** 无界集合紧凑 JSON 数组字串列（键名去 _TEXT 后缀即 legacy 视图键，直存直取）。 */ private val TEXT_COLUMNS: List = listOf("SRVT_TEXT", "VIPF_TEXT", "MAFL_TEXT") + /** v2 主表写入列：标量 + 异常 + 无界文本；集合由明细表承载，不经槽位/里程碑列。 */ + private val SCALAR_WRITE_COLUMNS: List = SCALAR_COLUMNS + EXCEPTION_COLUMNS + TEXT_COLUMNS + /** 库列名全集（读侧 SELECT 与视图重建的稳定顺序，与 V1.2.0 迁移一致）。 */ private val ALL_COLUMNS: List = SCALAR_COLUMNS + EXCEPTION_COLUMNS + OCCURRENCE_COLLECTIONS.flatMap { c -> c.slots.flatMap(Map::values) } + @@ -442,7 +445,22 @@ class JdbcFlightSchdRepository( return out } - /** FDIV/FRET/FLAB：1:0..1 单值异常对象 → 前缀标量列（自由文本取首个命中的文本键）。 */ + /** v2：仅标量/异常/文本列 flatten；集合键由明细表读写，不经过槽位列。 */ + private fun flattenScalarsOnly(scalars: Map): Map { + val fields = linkedMapOf() + scalars.forEach { (key, value) -> fields[key] = value } + val out = mutableMapOf() + scalars.forEach { (key, value) -> + when { + key in SCALAR_KEY_SET -> out[key] = value + key in setOf("SRVT", "VIPF", "MAFL") -> out["${key}_TEXT"] = value + } + } + flattenExceptions(fields, out) + return out + } + + /** FDIV/FRET/FLAB：1:0..1 单值异常对象 → 前缀标量列（自由文本取首个命中的文本键）。 */ private fun flattenExceptions(fields: FlightFields, out: MutableMap) { fun flatten(key: String, mappings: Map, textTarget: String) { val raw = fields[key] ?: return @@ -627,6 +645,7 @@ class JdbcFlightSchdRepository( override fun deleteDiffByDay(day: String, delFlids: Collection): Int { if (delFlids.isEmpty()) return 0 + FlightDetailTables.deleteForFlids(ds, delFlids) var totalDeleted = 0 val sqlDate = toSqlDate(day) for (chunk in delFlids.chunked(200)) { @@ -660,12 +679,17 @@ class JdbcFlightSchdRepository( ?.let { fields["FRET"] = it } exceptionView("FLAB", listOf("ARES" to "FLAB_ARES", "RSN" to "FLAB_RSN"), row) ?.let { fields["FLAB"] = it } - delayView(row)?.let { fields["DELY"] = it } - airbridgeView(row)?.let { fields["ABTM"] = it } - chocksView(row)?.let { fields["CHOT"] = it } - row["ROUT_PATH"]?.let { fields["ROUT"] = routeView(it) } - row["ERUT_PATH"]?.let { fields["ERUT"] = routeView(it) } - OCCURRENCE_COLLECTIONS.forEach { c -> slotView(c, row)?.let { fields[c.key] = it } } + val detailCollections = FlightDetailTables.loadCollections(ds, flid) + val detailKeys = detailCollections.keys + if ("DELY" !in detailKeys) delayView(row)?.let { fields["DELY"] = it } + if ("ABTM" !in detailKeys) airbridgeView(row)?.let { fields["ABTM"] = it } + if ("CHOT" !in detailKeys) chocksView(row)?.let { fields["CHOT"] = it } + if ("ROUT" !in detailKeys) row["ROUT_PATH"]?.let { fields["ROUT"] = routeView(it) } + if ("ERUT" !in detailKeys) row["ERUT_PATH"]?.let { fields["ERUT"] = routeView(it) } + OCCURRENCE_COLLECTIONS.forEach { c -> + if (c.key !in detailKeys) slotView(c, row)?.let { fields[c.key] = it } + } + detailCollections.forEach { (key, value) -> fields[key] = value } return flid to fields } @@ -740,6 +764,20 @@ class JdbcFlightSchdRepository( ::mapFlightRow, )?.second + override fun findNextStateByFlid(flid: String): com.gzzn.omms.msgexchange.domain.flight.FlightNextState? = + ds.queryOne( + "SELECT flid, state_version, last_message_id, ${ALL_COLUMNS.joinToString(", ")} FROM flight_schd WHERE flid = ?", + { ps -> ps.setString(1, flid) }, + ) { rs -> + val (id, fields) = mapFlightRow(rs) + com.gzzn.omms.msgexchange.domain.flight.FlightStateEngine.fromFlightFields( + id, + fields, + rs.getLong("state_version"), + rs.getString("last_message_id") ?: "", + ) + } + override fun findByFlids(flids: Collection): Map { if (flids.isEmpty()) return emptyMap() val result = linkedMapOf() @@ -769,6 +807,7 @@ class JdbcFlightSchdRepository( override fun deleteByFlids(flids: Set): Int { if (flids.isEmpty()) return 0 + FlightDetailTables.deleteForFlids(ds, flids) var totalDeleted = 0 for (chunk in flids.chunked(200)) { val placeholders = chunk.joinToString(",") { "?" } @@ -782,9 +821,11 @@ class JdbcFlightSchdRepository( override fun getGen(day: String): FlightSchdRepository.GenMeta? { val meta = ds.queryOne( - "SELECT version, updated_at FROM schd_gen WHERE fday = ?", + "SELECT version, last_message_id, updated_at FROM schd_gen WHERE fday = ?", { ps -> ps.setDate(1, toSqlDate(day)) }, - ) { rs -> rs.getLong("version") to (rs.getInstant("updated_at") ?: Instant.now()) } + ) { rs -> + Triple(rs.getLong("version"), rs.getString("last_message_id"), rs.getInstant("updated_at") ?: Instant.now()) + } ?: return null val flids = ds.query( "SELECT flid FROM schd_gen_flid WHERE fday = ? ORDER BY flid ASC", @@ -794,7 +835,8 @@ class JdbcFlightSchdRepository( fday = day, version = meta.first, flids = flids, - updatedAt = meta.second, + lastMessageId = meta.second, + updatedAt = meta.third, ) } @@ -832,14 +874,15 @@ class JdbcFlightSchdRepository( if (expected == 0L) { val inserted = ds.update( """ - INSERT INTO schd_gen (fday, version, updated_at) - VALUES (?, ?, ?) + INSERT INTO schd_gen (fday, version, last_message_id, updated_at) + VALUES (?, ?, ?, ?) ON CONFLICT (fday) DO NOTHING """.trimIndent(), ) { ps -> ps.setDate(1, sqlDate) ps.setLong(2, newGen.version) - ps.setTimestamp(3, sqlTimestamp) + ps.setString(3, newGen.lastMessageId) + ps.setTimestamp(4, sqlTimestamp) } if (inserted == 1) { replaceGenFlids(day, newGen.flids) @@ -850,14 +893,15 @@ class JdbcFlightSchdRepository( val updated = ds.update( """ UPDATE schd_gen - SET version = ?, updated_at = ? + SET version = ?, last_message_id = ?, updated_at = ? WHERE fday = ? AND version = ? """.trimIndent(), ) { ps -> ps.setLong(1, newGen.version) - ps.setTimestamp(2, sqlTimestamp) - ps.setDate(3, sqlDate) - ps.setLong(4, expected) + ps.setString(2, newGen.lastMessageId) + ps.setTimestamp(3, sqlTimestamp) + ps.setDate(4, sqlDate) + ps.setLong(5, expected) } if (updated == 1) { replaceGenFlids(day, newGen.flids) @@ -866,6 +910,98 @@ class JdbcFlightSchdRepository( return false } + + /** v2 主表行存在性保障（增量路径新插 FDAY=NULL）。 */ + private fun ensureFlightRow(flid: String, now: Instant) { + val ts = now.toSqlTimestamp() + ds.update( + "INSERT INTO flight_schd (flid, created_at, updated_at) VALUES (?, ?, ?) " + + "ON CONFLICT (flid) DO UPDATE SET updated_at = EXCLUDED.updated_at", + ) { ps -> + ps.setString(1, flid) + ps.setTimestamp(2, ts) + ps.setTimestamp(3, ts) + } + } + + /** v2 快照：标量列全量替换（集合不写槽位/里程碑列）。 */ + private fun upsertSnapshotScalars(day: String, state: com.gzzn.omms.msgexchange.domain.flight.FlightNextState, now: Instant) { + val stored = flattenScalarsOnly(state.scalars) + val sqlDate = toSqlDate(day) + val sqlTimestamp = now.toSqlTimestamp() + val sql = buildString { + append("INSERT INTO flight_schd (flid, fday, ") + append(SCALAR_WRITE_COLUMNS.joinToString(", ")) + append(", last_message_id, state_version, created_at, updated_at) VALUES (?, ?, ") + append(SCALAR_WRITE_COLUMNS.joinToString(", ") { "?" }) + append(", ?, ?, ?, ?) ON CONFLICT (flid) DO UPDATE SET fday = EXCLUDED.fday, ") + append(SCALAR_WRITE_COLUMNS.joinToString(", ") { "$it = EXCLUDED.$it" }) + append(", last_message_id = EXCLUDED.last_message_id, state_version = EXCLUDED.state_version, updated_at = EXCLUDED.updated_at") + } + ds.update(sql) { ps -> + var i = 1 + ps.setString(i++, state.flid) + ps.setDate(i++, sqlDate) + for (column in SCALAR_WRITE_COLUMNS) { + ps.setString(i++, stored[column]) + } + ps.setString(i++, state.lastMessageId) + ps.setLong(i++, state.stateVersion) + ps.setTimestamp(i++, sqlTimestamp) + ps.setTimestamp(i++, sqlTimestamp) + } + } + + /** v2 增量：仅更新出现的标量列 + 追踪字段。 */ + private fun upsertIncrementalScalars(state: com.gzzn.omms.msgexchange.domain.flight.FlightNextState, now: Instant) { + val stored = flattenScalarsOnly(state.scalars) + if (stored.isEmpty()) { + updateTrackingFields(state, now) + return + } + val assignments = stored.keys.map { "$it = ?" }.toMutableList() + assignments += listOf("last_message_id = ?", "state_version = ?", "updated_at = ?") + val sql = "UPDATE flight_schd SET ${assignments.joinToString(", ")} WHERE flid = ?" + val sqlTimestamp = now.toSqlTimestamp() + ds.update(sql) { ps -> + var i = 1 + stored.values.forEach { ps.setString(i++, it) } + ps.setString(i++, state.lastMessageId) + ps.setLong(i++, state.stateVersion) + ps.setTimestamp(i++, sqlTimestamp) + ps.setString(i, state.flid) + } + } + + private fun updateTrackingFields(state: com.gzzn.omms.msgexchange.domain.flight.FlightNextState, now: Instant) { + ds.update( + "UPDATE flight_schd SET last_message_id = ?, state_version = ?, updated_at = ? WHERE flid = ?", + ) { ps -> + ps.setString(1, state.lastMessageId) + ps.setLong(2, state.stateVersion) + ps.setTimestamp(3, now.toSqlTimestamp()) + ps.setString(4, state.flid) + } + } + + override fun persistNextStates( + day: String?, + states: List, + snapshotReplace: Boolean, + now: Instant, + ) { + if (states.isEmpty()) return + for (state in states) { + if (snapshotReplace && day != null) { + upsertSnapshotScalars(day, state, now) + } else { + ensureFlightRow(state.flid, now) + upsertIncrementalScalars(state, now) + } + FlightDetailTables.replaceAll(ds, state, now) + } + } + override fun deleteGenBefore(cutoffDay: String): Int = ds.update( "DELETE FROM schd_gen WHERE fday < ?", diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt index 2dfde4c..5333c83 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt @@ -203,15 +203,19 @@ class StubFlightSchd : FlightSchdRepository { val flid: String, val fday: String?, val fields: FlightFields, + val stateVersion: Long = 0L, + val lastMessageId: String = "", val createdAt: Instant, val updatedAt: Instant, ) private val records = linkedMapOf() + private val detailCollections = mutableMapOf>>>() private val gens = mutableMapOf() fun clear() { records.clear() + detailCollections.clear() gens.clear() } @@ -245,28 +249,56 @@ class StubFlightSchd : FlightSchdRepository { if (delFlids.isEmpty()) return 0 val delSet = delFlids.toSet() val toRemove = records.filter { (flid, rec) -> rec.fday == day && flid in delSet }.keys - toRemove.forEach { records.remove(it) } + toRemove.forEach { flid -> + records.remove(flid) + detailCollections.remove(flid) + } return toRemove.size } - override fun findByFlid(flid: String): FlightFields? = records[flid]?.fields + override fun findByFlid(flid: String): FlightFields? { + val rec = records[flid] ?: return null + return mergeFields(rec) + } + + private fun mergeFields(rec: Record): FlightFields { + val out = linkedMapOf() + out.putAll(rec.fields) + val mapper = com.fasterxml.jackson.databind.ObjectMapper() + detailCollections[rec.flid]?.forEach { (key, items) -> + out[key] = mapper.writeValueAsString(items) + } + return out + } + + override fun findNextStateByFlid(flid: String): com.gzzn.omms.msgexchange.domain.flight.FlightNextState? { + val rec = records[flid] ?: return null + return com.gzzn.omms.msgexchange.domain.flight.FlightNextState( + flid = rec.flid, + scalars = rec.fields.filterKeys { it != "FLID" }, + collections = detailCollections[flid] ?: emptyMap(), + stateVersion = rec.stateVersion, + lastMessageId = rec.lastMessageId, + ) + } override fun findByFlids(flids: Collection): Map = - flids.mapNotNull { flid -> records[flid]?.let { flid to it.fields } }.toMap() + flids.mapNotNull { flid -> records[flid]?.let { flid to mergeFields(it) } }.toMap() override fun findByDay(day: String): List> = records.values.filter { it.fday == day } .sortedBy { it.flid } - .map { it.flid to it.fields } + .map { it.flid to mergeFields(it) } override fun findAll(): Map = - records.mapValues { it.value.fields } + records.mapValues { mergeFields(it.value) } override fun deleteByFlids(flids: Set): Int { if (flids.isEmpty()) return 0 var count = 0 for (flid in flids) { if (records.remove(flid) != null) count++ + detailCollections.remove(flid) } return count } @@ -280,7 +312,48 @@ class StubFlightSchd : FlightSchdRepository { return true } - override fun deleteGenBefore(cutoffDay: String): Int { + override fun persistNextStates( + day: String?, + states: List, + snapshotReplace: Boolean, + now: Instant, + ) { + for (state in states) { + val scalarFields = linkedMapOf("FLID" to state.flid) + state.scalars.forEach { (k, v) -> scalarFields[k] = v } + val existing = records[state.flid] + if (snapshotReplace && day != null) { + records[state.flid] = Record( + flid = state.flid, + fday = day, + fields = scalarFields, + stateVersion = state.stateVersion, + lastMessageId = state.lastMessageId, + createdAt = existing?.createdAt ?: now, + updatedAt = now, + ) + detailCollections[state.flid] = state.collections + } else { + val mergedScalars = (existing?.fields?.filterKeys { + it == "FLID" || it !in com.gzzn.omms.msgexchange.infra.persistence.jdbc.FlightDetailTables.DETAIL_COLLECTION_KEYS + } ?: emptyMap()) + scalarFields + records[state.flid] = Record( + flid = state.flid, + fday = existing?.fday, + fields = mergedScalars, + stateVersion = state.stateVersion, + lastMessageId = state.lastMessageId, + createdAt = existing?.createdAt ?: now, + updatedAt = now, + ) + val mergedCollections = (detailCollections[state.flid] ?: emptyMap()).toMutableMap() + state.collections.forEach { (key, items) -> mergedCollections[key] = items } + detailCollections[state.flid] = mergedCollections + } + } + } + + override fun deleteGenBefore(cutoffDay: String): Int { val toRemove = gens.keys.filter { it < cutoffDay } toRemove.forEach { gens.remove(it) } return toRemove.size