package com.gzzn.omms.msgexchange.delivery

import com.gzzn.omms.msgexchange.MutableClock
import com.gzzn.omms.msgexchange.config.PipelineProps
import com.gzzn.omms.msgexchange.domain.EventType
import com.gzzn.omms.msgexchange.domain.MsgEvent
import com.gzzn.omms.msgexchange.domain.Targets
import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository
import com.gzzn.omms.msgexchange.infra.retry.FailureScheduler
import com.gzzn.omms.msgexchange.infra.stub.StubDeliveryPort
import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assertions.assertNull
import org.junit.jupiter.api.Assertions.assertTrue
import org.junit.jupiter.api.Test
import java.time.Instant

/**
 * 投递调度（docs/flight-state.md §7.3）：
 * KAFKA_MSG 逐条 FIFO；KAFKA_SCHD 唯一出口 flushSchd——同 FLID 未发事件按最新
 * STATE_VERSION 合并；TOMBSTONE 发 null 值消息；失败退避重试、达上限 DEAD。
 */
class DispatcherTickTest {

    private val clock = MutableClock(MutableClock.BASE)

    private fun dispatcher(repo: MsgEventRepository, port: DeliveryPort, p: PipelineProps = PipelineProps()): Dispatcher =
        Dispatcher(repo, port, p, FailureScheduler(p, clock))

    private fun ev(id: Long, target: String, key: String, payload: String, version: Long = 0) =
        MsgEvent(eventId = id, target = target, partitionKey = key, stateVersion = version, payloadJson = payload)

    @Test
    fun `per-target tick delivers KAFKA_MSG only and never consumes schd`() {
        val repo = StubMsgEvents()
        val port = StubDeliveryPort()
        val props = PipelineProps().apply { schd.flushPeriod = java.time.Duration.ofHours(1) }
        repo.insertAll(
            listOf(
                ev(1, Targets.KAFKA_MSG, "F1", """{"flid":"F1"}"""),
                ev(2, Targets.KAFKA_SCHD, "F1", """{"flid":"F1"}""", 1),
            ),
        )
        val d = dispatcher(repo, port, props)
        d.tick()

        assertEquals(1, port.sent.count { it.topic == "msg" })
        assertEquals(0, port.sent.count { it.topic == "schd" })   // N03：schd 唯一出口 flushSchd
    }

    @Test
    fun `flushSchd aggregates latest version per flight and marks sent`() {
        val repo = StubMsgEvents()
        val port = StubDeliveryPort()
        repo.insertAll(
            listOf(
                ev(1, Targets.KAFKA_SCHD, "F1", """{"v":"old"}""", 1),
                ev(2, Targets.KAFKA_SCHD, "F1", """{"v":"new"}""", 2),
                ev(3, Targets.KAFKA_SCHD, "F2", """{"v":"f2"}""", 1),
            ),
        )
        val d = dispatcher(repo, port)
        d.flushSchd()

        val schd = port.sent.filter { it.topic == "schd" }
        assertEquals(2, schd.size)
        assertEquals(1, schd.count { it.key == "F1" && it.payload!!.contains("new") && !it.payload.contains("old") })
        assertEquals(0, repo.rows.values.count { it.state.name == "PENDING" })
    }

    @Test
    fun `tombstone is delivered as null value message`() {
        val repo = StubMsgEvents()
        val port = StubDeliveryPort()
        repo.insertAll(
            listOf(
                MsgEvent(
                    eventId = 1, target = Targets.KAFKA_SCHD, partitionKey = "F1",
                    eventType = EventType.TOMBSTONE, stateVersion = 3, payloadJson = """{"flid":"F1","deleted":true}""",
                ),
            ),
        )
        val d = dispatcher(repo, port)
        d.flushSchd()

        assertEquals(1, port.tombstones.size)
        assertEquals("F1", port.tombstones.single().key)   // 整态键缺失表示删除旧值（§7.3）
        assertNull(port.tombstones.single().payload)
    }

    @Test
    fun `schd send failure schedules backoff and goes DEAD at limit`() {
        val repo = StubMsgEvents()
        val p = PipelineProps()
        val d = dispatcher(repo, FailingSchdPort(), p)
        repo.insertAll(listOf(ev(1, Targets.KAFKA_SCHD, "F1", """{"v":"1"}""", 1)))

        d.flushSchd()
        assertEquals(1, repo.rows[1]!!.attempts)

        repeat(p.pipeline.maxAttempts - 1) {
            clock.advance(p.pipeline.backoffFor(repo.rows[1]!!.attempts) + 1)
            d.flushSchd()
        }
        val final = repo.rows[1]!!
        assertTrue(final.attempts >= p.pipeline.maxAttempts || final.state.name == "DEAD")
    }

    @Test
    fun `KAFKA msg delivery unaffected while schd batch is retrying`() {
        val repo = StubMsgEvents()
        val port = StubDeliveryPort()
        val d = dispatcher(repo, port)
        repo.insertAll(listOf(ev(1, Targets.KAFKA_MSG, "F9", """{"flid":"F9"}""")))
        d.tick()
        assertEquals(1, port.sent.count { it.topic == "msg" && it.key == "F9" })
    }
}

/** schd 发送恒失败的端口（退避/终态闭环验证用）。 */
private class FailingSchdPort : DeliveryPort {
    var calls = 0
    override fun sendKafka(topic: String, key: String, payloadJson: String) = Unit
    override fun sendKafkaSchd(topic: String, key: String, payloadJson: String) {
        calls++
        throw IllegalStateException("broker-down")
    }

    override fun sendKafkaNull(topic: String, key: String) {
        throw IllegalStateException("broker-down")
    }
}