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

package com.gzzn.omms.msgexchange.processing

import com.gzzn.omms.msgexchange.codec.FlopPayload
import com.gzzn.omms.msgexchange.codec.ScheduleBody
import com.gzzn.omms.msgexchange.codec.XmlCodec
import com.gzzn.omms.msgexchange.config.PipelineProps
import com.gzzn.omms.msgexchange.domain.DecodedMessage
import com.gzzn.omms.msgexchange.domain.ErrorClass
import com.gzzn.omms.msgexchange.domain.MsgKind
import com.gzzn.omms.msgexchange.domain.ProcState
import com.gzzn.omms.msgexchange.domain.ProcStatus
import com.gzzn.omms.msgexchange.infra.log.TraceLog
import com.gzzn.omms.msgexchange.infra.persistence.BackfillTodoRepository
import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository
import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository
import com.gzzn.omms.msgexchange.infra.retry.ProcFailure
import jakarta.inject.Singleton
import java.time.Duration
import java.time.Instant

/**
 * 处理主泵（docs/flight-state.md §1 严格有序 + §4 处理事务与失败规则）：
 * 单活动主泵严格 FIFO + HOL 阻塞 + 队头滞留转 DEAD；航班状态、事件、处理终态
 * 与回填待办在处理器事务内原子提交。
 */
@Singleton
class Pump(
    private val procState: ProcStateRepository,
    private val inbox: CminmsgInboxRepository,
    private val processor: MessageProcessor,
    private val props: PipelineProps,
) {
    private val log = org.slf4j.LoggerFactory.getLogger(Pump::class.java)

    @Volatile
    private var running = true

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

    fun loop() {
        while (running) {
            try {
                tick()
            } catch (e: InterruptedException) {
                Thread.currentThread().interrupt()
                return
            } catch (e: Exception) {
                // 最后防线：失败状态迁移已在 processOne 边界内完成；致命 Error 不捕获
                sleepQuietly(props.pipeline.pollInterval)
            }
        }
    }

    internal fun tick() {
        val head = procState.headUnfinished()
        when {
            head == null -> sleepQuietly(props.pipeline.pollInterval)
            head.state == ProcStatus.FAILED && (head.nextAttemptAt ?: Instant.EPOCH) > Instant.now() ->
                if (poisoned(head)) {
                    log.error("poison -> DEAD msgId={} attempts={} lastError={}", head.msgId, head.attempts, head.lastError)
                    procState.update(
                        head.msgId, 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

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

/**
 * processOne：解码 → 绑定 → 处理器（事务内决策+落库）→ 终态迁移 → 回填。
 * 边界化失败迁移（ProcFailure）：任何意外异常归于本条 head，FAILED(INFRA)+退避，不穿出杀泵；
 * MALFORMED / PROTOCOL 直接 DEAD 不重试（docs/design.md §2.3 错误分类）。
 */
@Singleton
class MessageProcessor(
    private val inbox: CminmsgInboxRepository,
    private val procState: ProcStateRepository,
    private val codec: XmlCodec,
    private val scheduleProcessor: ScheduleProcessor,
    private val flopProcessor: FlopProcessor,
    private val fdelProcessor: FdelProcessor,
    private val adftProcessor: AdftProcessor,
    private val procFailure: ProcFailure,
    private val props: PipelineProps,
    private val backfillTodo: BackfillTodoRepository? = null,
) {
    private val log = org.slf4j.LoggerFactory.getLogger(MessageProcessor::class.java)

    fun processOne(head: ProcState) {
        TraceLog.withTrace(head.msgId) {
            try {
                processInternal(head)
            } catch (e: InterruptedException) {
                Thread.currentThread().interrupt()
                throw e
            } catch (e: Exception) {
                // 边界化：异常归于本条 head，写 FAILED(INFRA)/DEAD，而不是穿出杀 pump
                log.warn("processOne unexpected failure msgId={} ec=INFRA msg={}", head.msgId, 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 msgId={} attempts={}", head.msgId, head.attempts)
            procState.update(head.msgId, ProcStatus.DEAD, errorClass = ErrorClass.EXHAUSTED, lastError = head.lastError ?: "max-attempts")
            return
        }

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

        // I3：identity 仅首次绑定（head.identityKey == null）；FAILED 重试不重绑
        if (head.identityKey == null) {
            val identity = Identity.of(decoded, props.identity)
            if (!procState.tryBindIdentity(head.msgId, identity)) {
                val owner = procState.ownerOfIdentity(identity) ?: -1L
                log.info("duplicate-of:{} -> SKIPPED msgId={}", owner, head.msgId)
                procState.update(head.msgId, ProcStatus.SKIPPED, lastError = "duplicate-of:$owner")
                compensateBackfill(head, decoded)
                return
            }
        }

        // 处理器分派：SCHD 日计划主链路（§3.1）/ FLOP / FDEL / ADFT；缺载荷按 MALFORMED 终态
        val result: ApplyResult = when (val kind = decoded.kind) {
            is MsgKind.Schd -> {
                val body = decoded.body as? ScheduleBody
                if (body == null) {
                    deadMalformed(head, "missing-schd-body")
                    return
                }
                when (kind.subtype) {
                    MsgKind.SchdSubtype.DNLD, MsgKind.SchdSubtype.RESP ->
                        scheduleProcessor.applyScheduleRecords(head, decoded)
                    MsgKind.SchdSubtype.ADFT -> {
                        val record = body.records.singleOrNull()
                        if (record == null) {
                            deadMalformed(head, "adft-needs-single-fltr")
                            return
                        }
                        adftProcessor.apply(head, decoded, record)
                    }
                }
            }
            MsgKind.Fdel -> {
                val payload = decoded.body as? FlopPayload
                if (payload == null) {
                    deadMalformed(head, "missing-fdel-flid")
                    return
                }
                fdelProcessor.apply(head, decoded, payload)
            }
            is MsgKind.Flop -> {
                val payload = decoded.body as? FlopPayload
                if (payload == null) {
                    deadMalformed(head, "missing-flop-body")
                    return
                }
                flopProcessor.apply(head, decoded, payload)
            }
            is MsgKind.Unsupported -> {
                // design.md §2.3：未支持类型 → FAILED(UNSUPPORTED) 退避重试，达阈值转 DEAD；绝不写终态
                log.warn("unsupported type -> FAILED(UNSUPPORTED) msgId={} tag={}", head.msgId, kind.tag)
                procFailure.fail(head, ErrorClass.UNSUPPORTED, "no-handler:${kind.tag}")
                return
            }
        }

        when (result) {
            is ApplyResult.Succeeded, ApplyResult.ReplaySkipped ->
                procState.update(head.msgId, ProcStatus.SUCCEEDED)
            is ApplyResult.DeadProtocol -> {
                // design.md §2.3：整包拒绝 DEAD(PROTOCOL)，立即释放队头，交人工确认
                log.error("DEAD(PROTOCOL) msgId={} reason={} flags={}", head.msgId, result.reason, result.flags)
                procState.update(head.msgId, ProcStatus.DEAD, errorClass = ErrorClass.PROTOCOL, lastError = result.reason.take(1000))
                compensateBackfill(head, decoded)   // 拒绝包同样要回填信箱，防止反复轮询
                return
            }
        }

        backfill(head, decoded)
        log.info("SUCCEEDED msgId={} kind={}", head.msgId, decoded.typeTag)
    }

    /** design.md §3.3/§6.1：提交后回填共享信箱；失败不得把 SUCCEEDED 改回 FAILED，待办已事务内预登记。 */
    private fun backfill(head: ProcState, decoded: DecodedMessage) {
        try {
            inbox.backfillOnSuccess(head.msgId, decoded.meta.sndr, decoded.meta.type, decoded.meta.styp, decoded.meta.seqn)
            backfillTodo?.delete(head.msgId)
        } catch (e: Exception) {
            log.error("backfill failed after SUCCEEDED msgId={} (todo pre-registered, sweep will retry)", head.msgId, e)
        }
    }

    /** SKIPPED/DEAD(PROTOCOL) 包的回填（无预登记待办）：失败落补偿待办。 */
    private fun compensateBackfill(head: ProcState, decoded: DecodedMessage) {
        try {
            inbox.backfillOnSuccess(head.msgId, decoded.meta.sndr, decoded.meta.type, decoded.meta.styp, decoded.meta.seqn)
        } catch (e: Exception) {
            log.error("backfill failed msgId={} (compensation required)", head.msgId, e)
            backfillTodo?.record(
                BackfillTodoRepository.BackfillTask(
                    msgId = head.msgId,
                    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 msgId={}", head.msgId)
        }
    }

    private fun deadMalformed(head: ProcState, detail: String) {
        procState.update(head.msgId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = detail)
    }
}