msgexchange-v2/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPoller.kt

package com.gzzn.omms.msgexchange.ingress

import com.gzzn.omms.msgexchange.config.PipelineProps
import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository
import com.gzzn.omms.msgexchange.infra.persistence.InboxCursorRepository
import com.gzzn.omms.msgexchange.infra.persistence.MailboxRow
import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager
import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository
import jakarta.inject.Singleton
import java.time.Duration
import java.time.Instant

/**
 * 收报：轮询共享信箱，把新消息登记到自有 PG 的 PROC_STATE，等着主泵处理。
 *
 * 每一轮只做三件事：
 * 1. 从水位 W 之后按 ID 升序读一批行（只看 ID，不看处理标记）；
 * 2. 把读到的行登记成 PENDING，并把水位推到"连续"的位置；
 * 3. 登记和水位推进写在同一个事务里——中途崩溃时水位没动，下一轮重扫即可补齐。
 *
 * "连续"是这里唯一需要理解的规则。如果 W 后面缺了一个 ID，说明可能有 ID 更小的消息
 * 还没提交上来。此时先停在缺口前，不把缺口后面的消息放进队列：否则那条迟到的消息
 * 会排到它们后面，破坏"先来先处理"的约定，同一航班的报文可能被乱序应用。
 *
 * 缺口等超过 `msgx.pipeline.max-commit-delay` 仍未出现，就认定它永远不会来了
 * （典型情况是自增回滚留下的空位），跳过它继续推进——否则水位会卡在第一个空位上
 * 再也不动。
 *
 * 这一层不解析报文，也不写信箱处理标记：标记由处理完成后的 BackfillService 负责补。
 */
@Singleton
class InboxPoller(
    private val mailbox: CminmsgInboxRepository,
    private val procState: ProcStateRepository,
    private val cursor: InboxCursorRepository,
    private val txManager: PipelineTransactionManager,
    private val props: PipelineProps,
) {
    private val log = org.slf4j.LoggerFactory.getLogger(InboxPoller::class.java)

    @Volatile
    private var running = false

    /** @return 本轮新登记的消息条数（已登记过的行不计入，也不影响水位推进）。 */
    fun pollOnce(now: Instant = Instant.now()): Int {
        val batch = props.pipeline.claimBatch.coerceAtLeast(1)
        val watermark = cursor.load()
        val rows = mailbox.readRange(watermark.committedUpTo, batch)
        if (rows.isEmpty()) return 0

        // 找到连续部分的末尾；如果这批里出现了缺口，缺口后面的行这一轮先不入队
        val contiguous = contiguousUpTo(watermark.committedUpTo, rows) ?: watermark.committedUpTo
        var committedTo = contiguous
        var holeSince: Instant? = null
        if (rows.last().msgId > contiguous) {
            val since = watermark.holeSince ?: now
            if (Duration.between(since, now) < props.pipeline.maxCommitDelay) {
                holeSince = since
            } else {
                // 缺口等太久了：当成永久缺失跳过，让水位继续往前走
                committedTo = rows.first { it.msgId > contiguous }.msgId - 1
                log.warn("hole after W={} aged out, watermark advanced to {}", contiguous, committedTo)
            }
        }

        val enqueued = txManager.inTransaction {
            var n = 0
            rows.takeWhile { it.msgId <= committedTo }.forEach { row ->
                if (procState.insertIfAbsent(row.msgId, row.receivedAt)) n++
            }
            cursor.save(InboxCursorRepository.Cursor(committedTo, holeSince))
            n
        }
        if (enqueued > 0) {
            log.info("polled {} new messages, W {} -> {}", enqueued, watermark.committedUpTo, committedTo)
        }
        return enqueued
    }

    fun loop() {
        running = true
        log.info("inbox poller loop started")
        while (running) {
            try {
                pollOnce()
                sleepQuietly(props.pipeline.pollInterval)
            } catch (_: InterruptedException) {
                Thread.currentThread().interrupt()
                break
            } catch (e: Exception) {
                log.error("inbox poller tick failed", e)
                sleepQuietly(props.pipeline.pollInterval)
            }
        }
        log.info("inbox poller loop stopped")
    }

    fun stop() {
        running = false
    }

    /**
     * 从 W+1 开始数，返回 ID 逐 1 相连的最后一个 ID；遇到第一个缺号就停。
     * 返回 null 表示 W+1 本身就不存在。
     */
    private fun contiguousUpTo(from: Long, rows: List<MailboxRow>): Long? {
        var expected = from + 1
        var last: Long? = null
        for (row in rows) {
            if (row.msgId != expected) break
            last = row.msgId
            expected++
        }
        return last
    }

    private fun sleepQuietly(d: Duration) {
        try {
            Thread.sleep(d.toMillis().coerceAtLeast(1))
        } catch (_: InterruptedException) {
            Thread.currentThread().interrupt()
        }
    }
}