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.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 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
        }

        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

        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 {
                flightSchd.persistNextStates(day, nextStates, snapshotReplace = true)

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

                val casSuccess = flightSchd.putGenIfVersion(
                    day = day,
                    expected = expected,
                    newGen = FlightSchdRepository.GenMeta(
                        fday = day,
                        version = newVersion,
                        flids = newFlids,
                        lastMessageId = messageId,
                    ),
                )
                if (!casSuccess) {
                    val again = flightSchd.getGen(day)
                    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)
                }

                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)
            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)
        }
    }

    /** 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