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

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
import com.gzzn.omms.msgexchange.domain.ProcStatus
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.MsgEventRepository
import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository
import com.gzzn.omms.msgexchange.infra.persistence.PumpJobRepository
import com.gzzn.omms.msgexchange.infra.persistence.FlightSchdRepository
import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager
import com.gzzn.omms.msgexchange.infra.retry.ProcFailure
import com.gzzn.omms.msgexchange.jobs.JobExecutor
import jakarta.inject.Singleton
import java.time.Duration
import java.time.Instant

/**
 * ACMA-8 流程 2：处理主泵——严格 FIFO + HOL + 毒丸 DEAD 升级（I1）；
 * 阶段 A 运营航班 FLIGHT_SCHD 与待发事件、终态同事务原子提交（I2/I5，ACM2-28 定案）。
 */
@Singleton
class Pump(
    private val procState: ProcStateRepository,
    private val pumpJobs: PumpJobRepository,
    private val inbox: CminmsgInboxRepository,
    private val processor: MessageProcessor,
    private val jobExecutor: JobExecutor,
    private val props: PipelineProps,
) {
    private val log = org.slf4j.LoggerFactory.getLogger(Pump::class.java)

    @Volatile
    private var running = true

    /** 优雅停机：loop 在当前 tick 收尾后退出；线程中断由 Runner 负责。 */
    fun stop() {
        running = false
    }

    fun loop() {
        while (running) {
            try {
                tick()
            } catch (e: InterruptedException) {
                Thread.currentThread().interrupt()
                return
            } catch (e: Exception) {
                // U08：loop 只作最后防线——失败状态迁移已在 processOne/execute 的边界内完成；
                // 致命 Error 不在此捕获（任其终止进程，保证“异常必可见”）。
                sleepQuietly(props.pipeline.pollInterval)
            }
        }
    }

    internal fun tick() {
        val job = pumpJobs.headQueued()
        val head = procState.headUnfinished()

        // 统一 FIFO：JOB 与消息同队列语义（定时任务产物不绕过队头顺序，决策 1）。
        val nextJob = job?.takeIf { jobBefore(job, head) }
        when {
            nextJob != null -> execute(nextJob)
            head == null -> sleepQuietly(props.pipeline.pollInterval)
            head.state == ProcStatus.FAILED && (head.nextAttemptAt ?: Instant.EPOCH) > Instant.now() ->
                if (poisoned(head)) {
                    log.error("poison -> DEAD id={} attempts={} lastError={}", head.cminmsgsId, head.attempts, head.lastError)
                    procState.update(head.cminmsgsId, ProcStatus.DEAD,
                        errorClass = ErrorClass.EXHAUSTED, lastError = head.lastError ?: "head-deadline-exceeded")
                } else {
                    sleepQuietly(Duration.between(Instant.now(), head.nextAttemptAt))
                }
            // PENDING、或 FAILED 退避已到期：交处理入口（内部有边界化失败迁移与 attempts 守卫）
            else -> processor.processOne(head)
        }
    }

    private fun poisoned(head: ProcState): Boolean =
        head.attempts >= props.pipeline.maxAttempts ||
            Duration.between(head.updatedAt, Instant.now()) > props.pipeline.headDeadline

    /** JOB 与队头消息的先后由入队时间近似；实装以统一序号列保证（阶段 1 后续）。 */
    private fun jobBefore(job: PumpJobRepository.Job, head: ProcState?) =
        head == null || head.state == ProcStatus.FAILED

    private fun execute(job: PumpJobRepository.Job) {
        pumpJobs.markRunning(job.jobId)
        try {
            jobExecutor.execute(job)
            pumpJobs.markDone(job.jobId)
        } catch (e: Exception) {
            pumpJobs.markFailed(job.jobId, e.message ?: "unknown")
        }
    }

    private fun sleepQuietly(d: Duration) {
        if (!d.isNegative && !d.isZero) Thread.sleep(d.toMillis().coerceAtLeast(1))
    }
}

/**
 * ACMA-8 流程 2 processOne：解码→绑定→纯函数决策→提交。
 * U08/U10/U11：失败状态迁移全部在“持有具体 head”的边界内完成——
 * 任何意外异常 → FAILED(INFRA)+退避（ProcFailure）；attempts 达上限 → DEAD(EXHAUSTED)；
 * MALFORMED 直接 DEAD；InterruptedException 恢复中断位并上抛（不被普通恢复逻辑吞掉）。
 */
@Singleton
class MessageProcessor(
    private val inbox: CminmsgInboxRepository,
    private val procState: ProcStateRepository,
    private val msgEvents: MsgEventRepository,
    private val codecHolder: CodecHolder,
    private val handlers: HandlerHolder,
    private val flightSchd: FlightSchdRepository,
    private val snapshotFlow: SnapshotFlow,
    private val procFailure: ProcFailure,
    private val props: PipelineProps,
    private val txManager: PipelineTransactionManager,
    private val backfillTodo: BackfillTodoRepository? = null,
) {
    private val log = org.slf4j.LoggerFactory.getLogger(MessageProcessor::class.java)
    private val flidRegex = Regex("<FLID>(.*?)</FLID>", RegexOption.IGNORE_CASE)

    private fun extractCandidateFlids(decoded: DecodedMessage): Set<String> {
        val fromXml = flidRegex.findAll(decoded.rawXml).map { it.groupValues[1].trim() }.filter { it.isNotEmpty() }.toSet()
        if (fromXml.isNotEmpty()) return fromXml
        val b = decoded.body
        if (b is Map<*, *>) {
            val flid = b["FLID"] ?: b["flid"]
            if (flid != null) return setOf(flid.toString())
        }
        return emptySet()
    }

    fun processOne(head: ProcState) {
        com.gzzn.omms.msgexchange.infra.log.TraceLog.withTrace(head.cminmsgsId) {
            try {
                processInternal(head)
            } catch (e: InterruptedException) {
                Thread.currentThread().interrupt()
                throw e
            } catch (e: Exception) {
                // U08：边界化——异常归于本条 head，写 FAILED/DEAD，而不是穿出杀 pump
                log.warn("processOne unexpected failure id={} ec=INFRA msg={}", head.cminmsgsId, e.message ?: e.javaClass.simpleName)
                procFailure.fail(head, ErrorClass.INFRA, e.message ?: e.javaClass.simpleName)
            }
        }
    }

    private fun processInternal(head: ProcState) {
        // 守卫：手工/遗留 FAILED 行若 attempts 已达上限，直接终态（防止退避到期后无限重试）
        if (head.state == ProcStatus.FAILED && procFailure.scheduler.exhausted(head.attempts)) {
            log.error("head exhausted at entry -> DEAD id={} attempts={}", head.cminmsgsId, head.attempts)
            procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.EXHAUSTED,
                lastError = head.lastError ?: "max-attempts")
            return
        }

        val raw = inbox.rawOf(head.cminmsgsId) ?: run {
            log.error("raw missing -> DEAD(MALFORMED) id={}", head.cminmsgsId)
            procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = "raw-missing")
            return
        }
        val decoded = when (val r = codecHolder.codec.decode(raw)) {
            is com.gzzn.omms.msgexchange.codec.DecodeResult.Ok -> r.message
            is com.gzzn.omms.msgexchange.codec.DecodeResult.Err -> {
                // T06（U11）：MALFORMED（报文非法）→ DEAD 不重试；CODEC_ERROR（可随 codec 修复重放）→ FAILED 退避
                if (r.failure.errorClass == ErrorClass.MALFORMED) {
                    log.error("decode MALFORMED -> DEAD id={} detail={}", head.cminmsgsId, r.failure.detail)
                    procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED,
                        lastError = r.failure.detail)
                } else {
                    log.warn("decode {} -> FAILED id={} detail={}", r.failure.errorClass, head.cminmsgsId, r.failure.detail)
                    procFailure.fail(head, r.failure.errorClass, r.failure.detail)
                }
                return
            }
        }

        // I3：identity 仅首次绑定（head.identityKey == null）；FAILED 重试不重绑（N25：入参传递 head，不再二次查询）
        if (head.identityKey == null) {
            val identity = Identity.of(decoded, props.identity)
            if (!procState.tryBindIdentity(head.cminmsgsId, identity)) {
                val owner = procState.ownerOfIdentity(identity) ?: -1L
                log.info("duplicate-of:{} -> SKIPPED id={}", owner, head.cminmsgsId)
                procState.update(head.cminmsgsId, ProcStatus.SKIPPED, lastError = "duplicate-of:$owner")
                // 重复报文自身也是一条信箱记录：同样回填共享信箱，防止被反复轮询（失败落补偿）
                compensateBackfill(head, decoded)
                return
            }
        }
        // 快照消息走专属流程（流程 4：staging → SQL 批处理/域内差删/CAS 同事务）
        val handler = handlers.registry.dispatcherFor(decoded)
        if (handler == null) {
            // U10（N21）：未注册 ≠ 报文非法——写 FAILED（可重放），绝不写终态
            log.warn("no-handler:{} -> FAILED(UNSUPPORTED) id={}", decoded.typeTag, head.cminmsgsId)
            procFailure.fail(head, ErrorClass.UNSUPPORTED, "no-handler:${decoded.typeTag}")
            return
        }
        val schdKind = decoded.kind as? com.gzzn.omms.msgexchange.domain.MsgKind.Schd
        if (schdKind != null &&
            schdKind.subtype == com.gzzn.omms.msgexchange.domain.MsgKind.SchdSubtype.DNLD
        ) {
            snapshotFlow.publishSnapshot(head, decoded)
            return
        }

        val candidateFlids = extractCandidateFlids(decoded)
        val flightView = if (candidateFlids.isNotEmpty()) {
            flightSchd.findByFlids(candidateFlids)
        } else {
            emptyMap()
        }

        val decision = handler.decide(flightView, decoded)   // 纯函数

        // 事务 2（自有 PG 单事务原子）：
        // upsert FLIGHT_SCHD(decision.flightChanges) + insert MSG_EVENT + PROC_STATE → SUCCEEDED
        val events = buildList {
            decision.msgNotifies.forEach { add(MsgEvent(target = Targets.KAFKA_MSG, payloadJson = it.payloadJson)) }
            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()) {
                val nextStates = decision.flightChanges.map { change ->
                    val current = flightSchd.findNextStateByFlid(change.flid)
                    val commands = FlightStateEngine.commandsFromFields(change.flid, change.fields)
                    FlightStateEngine.apply(current, commands, messageId, bumpVersion = true)
                }
                flightSchd.persistNextStates(null, nextStates, snapshotReplace = false)
            }
            if (events.isNotEmpty()) {
                msgEvents.insertAll(events)
            }
            procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED)
        }

        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)
            recordBackfillTodo(head.cminmsgsId, decoded, e)
        }
        log.info("SUCCEEDED id={} events={} flightChanges={}", head.cminmsgsId, events.size, decision.flightChanges.size)
    }

    /** v2 §5：回填失败只落补偿待办（成功终态不降级，业务不重放）；由 BackfillSweepJob 到期重试。 */
    private fun compensateBackfill(head: ProcState, decoded: DecodedMessage) {
        try {
            inbox.backfillOnSuccess(head.cminmsgsId, decoded.meta.sndr, decoded.meta.type, decoded.meta.styp, decoded.meta.seqn)
        } catch (e: Exception) {
            log.error("backfill failed for SKIPPED duplicate id={} (compensation required)", head.cminmsgsId, e)
            recordBackfillTodo(head.cminmsgsId, decoded, e)
        }
    }

    private fun recordBackfillTodo(cminmsgsId: Long, decoded: DecodedMessage, e: Exception) {
        backfillTodo?.record(
            BackfillTodoRepository.BackfillTask(
                cminmsgsId = cminmsgsId,
                sndr = decoded.meta.sndr,
                type = decoded.meta.type,
                styp = decoded.meta.styp,
                seqn = decoded.meta.seqn,
            ),
            e.message ?: e.javaClass.simpleName,
        ) ?: log.warn("no backfill-todo repository bound; compensation NOT persisted id={}", cminmsgsId)
    }
}

/** 延迟装配占位（阶段 1 后续以 Micronaut Bean 替换直连构造）。 */
class CodecHolder(val codec: com.gzzn.omms.msgexchange.codec.XmlCodec)
class HandlerHolder(val registry: HandlerRegistry)