msgexchange-v2/src/main/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlow.kt

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
import com.gzzn.omms.msgexchange.domain.MsgEvent
import com.gzzn.omms.msgexchange.domain.MsgKind
import com.gzzn.omms.msgexchange.domain.Targets
import com.gzzn.omms.msgexchange.infra.persistence.BackfillTodoRepository
import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository
import com.gzzn.omms.msgexchange.infra.persistence.FlightFields
import com.gzzn.omms.msgexchange.infra.persistence.FlightFieldsJson
import com.gzzn.omms.msgexchange.infra.persistence.FlightSchdRepository
import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository
import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager
import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository
import com.gzzn.omms.msgexchange.infra.persistence.ReqTrackRepository
import com.gzzn.omms.msgexchange.infra.retry.ProcFailure
import jakarta.inject.Singleton

/**
 * staging（内存瞬态，崩溃从 raw 整包重放）→ PG 单事务原子“覆盖+按代差删+SQL CAS 推进+事件入队+SUCCEEDED”（I4/I5，ACM2-28 定案）。
 * U10/T07 占位安全化 + U08 统一失败迁移（ProcFailure）：staging 未实装 → FAILED(UNSUPPORTED)+退避
 * （可重放，绝不写终态）；CAS 冲突 → FAILED(INFRA)+退避；达上限统一 DEAD(EXHAUSTED)。
 */
@Singleton
class SnapshotFlow(
    private val procState: ProcStateRepository,
    private val flightSchd: FlightSchdRepository,
    private val msgEvents: MsgEventRepository,
    private val reqTrack: ReqTrackRepository,
    private val procFailure: ProcFailure,
    private val txManager: PipelineTransactionManager,
    private val inbox: CminmsgInboxRepository,
    private val backfillTodo: BackfillTodoRepository? = null,
) {
    private val log = org.slf4j.LoggerFactory.getLogger(SnapshotFlow::class.java)

    fun publishSnapshot(head: ProcState, msg: DecodedMessage) {
        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)
            return
        }
        val ok = staged as StageResult.Ok
        val normalized = ok.flights
        val day = ok.day

        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,
                lastError = "flights-exceed-limit:${normalized.size}")
            return
        }

        // v2 §5 事务流程：锁外只做解析与整包校验；取得 PIPELINE_LOCK 后在锁内
        // 复核快照身份 → 读取当前代与当前航班态 → 计算 nextState → 写入 → 提交。
        var replayNoOp = false
        try {
            txManager.inTransaction {
                // 锁内复核快照身份：同一消息已成功提交 → 重放短路（不加版本、不重复发事件）
                val gen = flightSchd.getGen(day)
                if (gen?.lastMessageId == messageId) {
                    procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED)
                    replayNoOp = true
                    return@inTransaction
                }

                val expected = gen?.version ?: 0L
                val newFlids = normalized.map { it.first }.toSet()
                val delFields = gen?.flids?.minus(newFlids) ?: emptySet()
                val newVersion = expected + 1L

                // 锁内读取当前航班态，计算 nextState（单写者互斥下读到的一定是已提交最新态）
                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)
                }

                flightSchd.persistNextStates(day, nextStates, snapshotReplace = true)

                if (delFields.isNotEmpty()) {
                    flightSchd.deleteDiffByDay(day, delFields)
                }

                // 锁内 CAS：expected 即锁内刚读到的最新版本，仍失败属数据异常——
                // 一律回滚进入可重试 FAILED(INFRA)，绝不凭版本号推断“已经是我的提交”（v2 §5）
                val casSuccess = flightSchd.putGenIfVersion(
                    day = day,
                    expected = expected,
                    newGen = FlightSchdRepository.GenMeta(
                        fday = day,
                        version = newVersion,
                        flids = newFlids,
                        lastMessageId = messageId,
                    ),
                )
                if (!casSuccess) {
                    throw CasConflictException(
                        "gen-cas-conflict: expected=$expected msg=$messageId (lock-held CAS must not fail)",
                    )
                }

                val schdKind = msg.kind as? MsgKind.Schd
                if (schdKind?.subtype == MsgKind.SchdSubtype.RESP) {
                    reqTrack.findOpenByKind("SCHD")?.let { req ->
                        reqTrack.markDone(req.reqId)
                    }
                }

                val events = nextStates.map { state ->
                    MsgEvent(
                        target = Targets.KAFKA_SCHD,
                        partitionKey = state.flid,
                        payloadJson = FlightFieldsJson.toJson(state.toFlightFields()),
                    )
                }
                if (events.isNotEmpty()) {
                    msgEvents.insertAll(events)
                }

                procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED)
            }

            // 提交后补偿：重放短路同样补做回填（其待办可能仍在重试中）
            safeBackfill(head, msg)
            if (replayNoOp) {
                log.info("snapshot replay no-op id={} day={}", head.cminmsgsId, day)
            } else {
                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)
            procFailure.fail(head, ErrorClass.INFRA, e.message ?: "gen-cas-conflict")
        }
    }

    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)
            backfillTodo?.record(
                BackfillTodoRepository.BackfillTask(
                    cminmsgsId = head.cminmsgsId,
                    sndr = msg.meta.sndr,
                    type = msg.meta.type,
                    styp = msg.meta.styp,
                    seqn = msg.meta.seqn,
                ),
                e.message ?: e.javaClass.simpleName,
            ) ?: log.warn("no backfill-todo repository bound; compensation NOT persisted id={}", head.cminmsgsId)
        }
    }

    /** staging 结果（骨架）。 */
    sealed interface StageResult {
        data class Ok(val day: String, val flights: List<Pair<String, FlightFields>>) : StageResult
        data class Invalid(val reason: String) : StageResult

        companion object {
            var parser: ((DecodedMessage) -> StageResult)? = null

            fun stagingOf(msg: DecodedMessage): StageResult =
                parser?.invoke(msg) ?: Invalid("staging-not-implemented(${msg.typeTag})")   // TODO(阶段2)
        }
    }
}
class CasConflictException(message: String) : RuntimeException(message)
const val MAX_FLIGHTS_PER_SNAPSHOT = 10000