msgexchange-v2/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxServiceTest.kt

package com.gzzn.omms.msgexchange.ingress

import com.gzzn.omms.msgexchange.MutableClock
import com.gzzn.omms.msgexchange.config.PipelineProps
import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository
import com.gzzn.omms.msgexchange.infra.persistence.MailboxMarkResult
import com.gzzn.omms.msgexchange.infra.persistence.MailboxRow
import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository
import com.gzzn.omms.msgexchange.infra.stub.StubInbox
import com.gzzn.omms.msgexchange.infra.stub.StubInboxCursor
import com.gzzn.omms.msgexchange.infra.stub.StubPipelineTx
import com.gzzn.omms.msgexchange.infra.stub.StubProcState
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assertions.assertNotNull
import org.junit.jupiter.api.Assertions.assertNull
import org.junit.jupiter.api.Assertions.assertThrows
import org.junit.jupiter.api.Test
import java.time.Instant

/**
 * 兼容入口的接收边界（`C-8`）：`insertRaw` 返回 ID 即"已落信"。
 * 此后的接收时间读取失败或 PG 入队失败都只留痕/计数，不改接收结果——否则客户端重试会在
 * 信箱里多写一行；缺的 `PROC_STATE` 由收报轮询按 `US-01` 补建，且只建一条。
 */
class InboxServiceTest {

    private val t0: Instant = Instant.parse("2026-09-06T02:00:00Z")

    /** 可注入失败的信箱：覆盖"插入失败 / 接收时间读失败 / 接收时间为 NULL"三种情形。 */
    private class FakeInbox(
        private val rawFailure: Boolean = false,
        private val receivedFailure: Boolean = false,
        private val receivedNull: Boolean = false,
    ) : CminmsgInboxRepository {
        override fun insertRaw(rawXml: String): Long =
            if (rawFailure) throw IllegalStateException("mailbox-down") else 42L

        override fun rawOf(msgId: Long): String? = null

        override fun receivedAtOf(msgId: Long): Instant? = when {
            receivedFailure -> throw IllegalStateException("read-time-down")
            receivedNull -> null
            else -> Instant.parse("2026-09-06T01:59:55Z")
        }

        override fun readRange(fromExclusive: Long, limit: Int): List<MailboxRow> = emptyList()

        override fun maxId(): Long? = null

        override fun minId(): Long? = null

        override fun markProcessedIfUnmarked(msgId: Long, value: String): MailboxMarkResult = MailboxMarkResult.MISSING
    }

    /** PG 入队第一次失败，其余委托给内存仓储（模拟轮询随后补建）。 */
    private class FlakyProcState(private val delegate: StubProcState) : ProcStateRepository by delegate {
        private var failNextInsert = true

        override fun insertIfAbsent(msgId: Long, receivedAt: Instant?, enqueuedAt: Instant?): Boolean {
            if (failNextInsert) {
                failNextInsert = false
                throw IllegalStateException("pg-down")
            }
            return delegate.insertIfAbsent(msgId, receivedAt, enqueuedAt)
        }
    }

    @Test
    fun `receipt survives receive-time read failure and pg enqueue failure`() {
        val proc = StubProcState()
        val service = InboxService(FakeInbox(receivedFailure = true), FlakyProcState(proc), MutableClock(t0))

        val receipt = service.accept("<MSG/>")

        assertEquals(42L, receipt.msgId)                   // 已落信：ID 照常返回
        assertEquals(t0, receipt.receivedAt)               // 读失败 → 回退到注入时钟
        assertEquals(1L, service.pgEnqueueFailures.get())  // 入队失败只计数，不冒泡
        assertNull(proc.find(42L))
    }

    @Test
    fun `missing receive time falls back to the injected clock`() {
        val service = InboxService(FakeInbox(receivedNull = true), StubProcState(), MutableClock(t0))

        val receipt = service.accept("<MSG/>")

        assertEquals(t0, receipt.receivedAt)
        assertEquals(0L, service.pgEnqueueFailures.get())
    }

    @Test
    fun `a real mailbox insert failure still fails without a fake id`() {
        val service = InboxService(FakeInbox(rawFailure = true), StubProcState(), MutableClock(t0))

        assertThrows(IllegalStateException::class.java) { service.accept("<MSG/>") }
        assertEquals(0L, service.pgEnqueueFailures.get())
    }

    @Test
    fun `the poller backfills a compat receipt whose pg enqueue failed, exactly once`() {
        val inbox = StubInbox().apply { clear() }
        val proc = StubProcState().apply { clear() }
        val cursor = StubInboxCursor().apply { clear() }
        val clock = MutableClock(t0)
        val service = InboxService(inbox, FlakyProcState(proc), clock)
        val poller = InboxPoller(inbox, proc, cursor, StubPipelineTx(), PipelineProps(), clock)

        val receipt = service.accept("<MSG/>")   // 落信成功；PG 入队失败但不抛
        assertEquals(1L, service.pgEnqueueFailures.get())
        assertNull(proc.find(receipt.msgId))

        assertEquals(1, poller.pollOnce(t0))     // 轮询补建
        assertNotNull(proc.find(receipt.msgId))
        assertEquals(1, proc.rows.size)
        assertEquals(receipt.msgId, cursor.cursor.committedUpTo)

        assertEquals(0, poller.pollOnce(t0))     // 已存在：不重复建行，也不重复计数
        assertEquals(1, proc.rows.size)
    }
}