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

package com.gzzn.omms.msgexchange.processing

import com.gzzn.omms.msgexchange.config.MailboxProps
import com.gzzn.omms.msgexchange.config.PipelineProps
import com.gzzn.omms.msgexchange.domain.ProcStatus
import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository
import com.gzzn.omms.msgexchange.infra.persistence.MailboxMarkResult
import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository
import jakarta.inject.Singleton
import java.time.Clock
import java.time.Duration
import java.time.Instant

/**
 * 把"已处理"标记写回共享信箱。
 *
 * 一条消息处理完要做两件事：记下终态、把标记写回信箱。第一件在业务事务里完成，
 * 同时留下"还欠一次回填"的意图（PROC_STATE 上的 BACKFILL_NEXT_AT）；
 * 这个类负责第二件：
 *
 * - [attempt]：处理刚结束时马上试一次，让标记尽快落到信箱。失败也不影响处理结果，
 *   留给扫描重试即可。
 * - [sweep]：定时把还欠回填的记录挑出来重试。失败就按 30 秒起步、最长 15 分钟的
 *   退避往后推；如果一条消息从收到现在已经超过超期期限，则无视退避强制补写——
 *   否则退避可能一直失败下去，这些行永远打不上标记，库方就没法清理信箱。
 *
 * 两条底线：回填失败不会把终态改回去，也不会重新执行业务逻辑；写标记只写还是空标记的
 * 行，重复执行没有副作用。
 */
@Singleton
class BackfillService(
    private val procState: ProcStateRepository,
    private val mailbox: CminmsgInboxRepository,
    private val mailboxProps: MailboxProps,
    private val props: PipelineProps,
    private val clock: Clock,
    /** 与人工重放共用的互斥门；`internal` 以便装配测试断言两者拿到同一实例。 */
    internal val lifecycleGate: MessageLifecycleGate,
) {
    private val log = org.slf4j.LoggerFactory.getLogger(BackfillService::class.java)

    companion object {
        private val INITIAL_BACKOFF: Duration = Duration.ofSeconds(30)
        private val MAX_BACKOFF: Duration = Duration.ofMinutes(15)
        private val TERMINAL_STATES = setOf(ProcStatus.SUCCEEDED, ProcStatus.SKIPPED, ProcStatus.DEAD)

        /** 放弃原因：运行时查询确认信箱行不存在（确定性结论，重试不会改变结果）。 */
        const val ABANDON_MISSING_ROW = "MISSING_ROW"

        /** 放弃原因：暂时性故障达到尝试上限；停止自动重试，但保留人工恢复能力。 */
        const val ABANDON_MAX_ATTEMPTS = "MAX_ATTEMPTS"

        fun backoffDelayFor(attempts: Int): Duration {
            val shift = (attempts - 1).coerceIn(0, 20)
            return INITIAL_BACKOFF.multipliedBy(1L shl shift).coerceAtMost(MAX_BACKOFF)
        }
    }

    /**
     * 处理完立刻试一次。失败只记一笔退避信息就返回，不抛异常——
     * 调用方是主泵的处理路径，不能被回填问题拖住。
     */
    fun attempt(msgId: Long, now: Instant = clock.instant()) {
        lifecycleGate.exclusive {
            val row = procState.find(msgId) ?: return@exclusive
            if (row.state !in TERMINAL_STATES) return@exclusive
            if (row.backfillAt != null) return@exclusive
            if (row.backfillAbandonedAt != null) return@exclusive
            record(msgId, attempts = row.backfillAttempts, now = now)?.let {
                log.warn("backfill failed msgId={} error={} (sweep will retry)", msgId, it)
            }
        }
    }

    /**
     * 人工恢复入口：清除放弃标记并重排一次回填。
     * 用于暂时性故障恢复之后，或人工对账确认该行仍应打标之后。
     */
    fun reopen(msgId: Long, now: Instant = clock.instant()): Boolean =
        lifecycleGate.exclusive { procState.reopenBackfill(msgId, now) }

    /**
     * 批量补写，由 JobRunner 每 30 秒调用一次。待办状态都在数据库里，
     * 进程重启后接着跑，不需要额外恢复步骤。
     *
     * 注意：**调度周期不是完成时限**——扫描与历史作业串行，批次积压、单行调用超时与
     * 历史作业耗时都会延长实际回填延迟。
     *
     * @return 本批处理的条数
     */
    fun sweep(now: Instant = clock.instant()): Int {
        val due = procState.findBackfillDue(now, now.minus(props.pipeline.overdueBackfill), props.pipeline.backfillBatch)
        due.forEach { attempt(it.msgId, now) }
        return due.size
    }

    /** 回填一条。@return 失败原因；返回 null 表示已处理完（写成功／早已标记／已放弃）。 */
    private fun record(msgId: Long, attempts: Int, now: Instant): String? =
        try {
            when (mailbox.markProcessedIfUnmarked(msgId, mailboxProps.processedValue)) {
                MailboxMarkResult.MARKED, MailboxMarkResult.ALREADY_MARKED -> {
                    // 没有真正写进去说明库里已经有标记了，同样算成功（不覆盖已有值）
                    procState.markBackfilled(msgId, now)
                    null
                }
                MailboxMarkResult.MISSING -> {
                    // 确定性结论：行不存在，重试不会改变结果。停止自动补偿并把事实留痕。
                    // 这**不等于**标记已确认（BACKFILL_AT 仍为空），清除前提因此仍然不成立。
                    procState.markBackfillAbandoned(msgId, ABANDON_MISSING_ROW, now)
                    log.error("backfill abandoned: mailbox row missing msgId={}", msgId)
                    null
                }
            }
        } catch (e: Exception) {
            // 超时/连接失败是暂时性的，**不能**当作缺行证据：按退避重试。
            // 达到上限后停止自动重试（保留人工恢复能力），避免永久占满扫描批次造成饥饿。
            val reason = e.message ?: e.javaClass.simpleName
            val nextAttempts = attempts + 1
            if (nextAttempts >= props.pipeline.backfillMaxAttempts) {
                runCatching { procState.markBackfillAbandoned(msgId, ABANDON_MAX_ATTEMPTS, now) }
                    .onFailure { log.error("abandon backfill failed msgId={}", msgId, it) }
                log.error("backfill abandoned after {} attempts msgId={} error={}", nextAttempts, msgId, reason)
            } else {
                runCatching {
                    procState.recordBackfillFailure(msgId, reason, nextAttempts, now.plus(backoffDelayFor(nextAttempts)), now)
                }.onFailure { log.error("record backfill failure failed msgId={}", msgId, it) }
            }
            reason
        }

}