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

/**
 * 收报（docs/message-lifecycle.md §5.1）：共享 MySQL 信箱按 **ID 区间**升序有界读取，
 * 在自有 PG 建 `PROC_STATE(PENDING)`，并把消费水位推进到连续上界；入队与水位推进在同一
 * PG 事务内提交，中断即重扫补建（§4 第一行）。收报层不解析业务载荷、不写处理标记。
 *
 * 三条红线：
 * - **扫描谓词不含处理标记**：标记只用于回填与库方清除，不参与消息发现；否则已入队而未
 *   回填的行会永久占据批次，死信累积到批大小时收报整体停摆；
 * - **水位遇空洞即停**：不越过空洞入队（越过后较小 ID 迟到即 FIFO 越序，architecture §5）；
 * - **空洞老化**：超过最大提交时延（Q2 承诺）的空洞判定为永久并放行，否则水位会永久停摆于
 *   一次自增回滚留下的空位。
 *
 * 空转代价为每轮一次区间 SELECT。
 */
@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

        // 连续上界；读取区间内出现空洞时，只推进到连续部分，空洞之后的行暂不入队（防较小 ID 迟到被越过）
        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 {
                // 空洞老化：超过最大提交时延仍缺席即判永久（Q2），放行水位，否则永久停摆
                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 相邻的最后一个；首个空位即停（rows 为升序且覆盖该区间）。 */
    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()
        }
    }
}