msgexchange-v2/src/main/kotlin/com/gzzn/omms/msgexchange/delivery/Dispatcher.kt

package com.gzzn.omms.msgexchange.delivery

import com.gzzn.omms.msgexchange.config.PipelineProps
import com.gzzn.omms.msgexchange.domain.ErrorClass
import com.gzzn.omms.msgexchange.domain.EventStatus
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 jakarta.inject.Singleton
import java.time.Duration
import java.time.Instant

/** 对外投递的出口：同步等 Kafka 确认，语义是至少发一次（可能重复，但不会丢）。以后 ES 投影也从这里加。 */
interface DeliveryPort {
    /** 发一条变化通知，消息 key 是 FLID（航班实例 ID）。 */
    fun sendKafka(topic: String, key: String, payloadJson: String)

    /** 发一条完整状态，消息 key 是 FLID。适配层负责把 target 映射成 topic：KAFKA:msg 对应 "msg"，KAFKA:schd 对应 "schd"。 */
    fun sendKafkaSchd(topic: String, key: String, payloadJson: String)

    /** 发一条删除通知：key 是 FLID、value 为空；下游按"整态里这个键没了"理解成删除。 */
    fun sendKafkaNull(topic: String, key: String)

    /** 给健康检查用的连通性探测；默认返回 true，真实 Kafka 实现要覆写成向 broker 拉一次 metadata 来判断。 */
    fun ping(): Boolean = true
}

/**
 * 投递调度：把 outbox（待发事件表）里的事件发给下游。
 *
 * KAFKA_MSG 一条一条按登记顺序发，不插队。KAFKA_SCHD 走 flushSchd 批量发：同一个 FLID 攒了
 * 多条未发事件时只发版本号最新的那条，旧的自然作废；删除通知发 value 为空的 tombstone。
 * 两个主题之间不保证先后顺序。
 *
 * 失败处理：队首的重试时间没到就不取；一批里有发送失败，整批重试次数加一并推后退避，
 * 次数用尽整批转 DEAD 当死信。见 docs/flight-state.md §5。
 */
@Singleton
class Dispatcher(
    private val msgEvents: MsgEventRepository,
    private val port: DeliveryPort,
    private val props: PipelineProps,
    private val scheduler: FailureScheduler,
) {
    private val log = org.slf4j.LoggerFactory.getLogger(Dispatcher::class.java)

    @Volatile
    private var running = true

    private var lastFlush: Instant? = null

    /** 请求停机：置位后 loop 走完当前一轮就退出；真正中断线程由 PipelineLifecycle 负责。 */
    fun stop() {
        running = false
    }

    fun loop() {
        while (running) {
            try {
                tick()
            } catch (e: InterruptedException) {
                Thread.currentThread().interrupt()
                return
            } catch (e: Exception) {
                sleepQuietly(props.pipeline.pollInterval)
            }
        }
    }

    internal fun tick() {
        val head = msgEvents.headUnsent(Targets.KAFKA_MSG)
        if (head != null && (head.state == EventStatus.SENT || head.nextAttemptAt == null || head.nextAttemptAt <= scheduler.now())) {
            deliver(Targets.KAFKA_MSG, head)
        }
        if (flushDue()) flushSchd()
        if (running) sleepQuietly(props.pipeline.pollInterval)
    }

    private fun flushDue(): Boolean =
        lastFlush?.let { Duration.between(it, scheduler.now()) >= props.schd.flushPeriod } ?: false

    private fun deliver(target: String, e: MsgEvent) {
        try {
            when (target) {
                Targets.KAFKA_MSG -> port.sendKafka("msg", e.partitionKey, e.payloadJson)
            }
            msgEvents.markSent(e.eventId ?: return)
        } catch (ex: Exception) {
            retryOrDead(e, ex.message ?: ex.javaClass.simpleName)
        }
    }

    /** 批量发 KAFKA_SCHD：每个 FLID 只发版本号最新的那条未发事件，删除通知发空 value。 */
    internal fun flushSchd() {
        val batch = try {
            msgEvents.mergePendingSchd(props.schd.flushLimit)
        } catch (e: Exception) {
            log.warn("mergePendingSchd failed: {}", e.message)
            return
        }
        if (batch.isEmpty()) {
            lastFlush = scheduler.now()
            return
        }
        val failures = mutableListOf<MsgEvent>()
        for (e in batch) {
            try {
                when (e.eventType) {
                    EventType.TOMBSTONE -> port.sendKafkaNull("schd", e.partitionKey)
                    EventType.UPSERT -> port.sendKafkaSchd("schd", e.partitionKey, e.payloadJson)
                }
            } catch (ex: Exception) {
                failures.add(e)
            }
        }
        val sentIds = batch.mapNotNull { it.eventId }.toSet() - failures.mapNotNull { it.eventId }.toSet()
        if (sentIds.isNotEmpty()) msgEvents.markAllSent(sentIds.toList())
        // 被更新版本压掉的旧事件也要标成已发，否则它们会一直留在队里：同一个 FLID 只按最新版本输出一次
        val sentVersions = batch.associate { it.partitionKey to it.stateVersion }
        runCatching {
            while (true) {
                val superseded = msgEvents.mergePendingSchd(props.schd.flushLimit)
                    .filter { sentVersions[it.partitionKey]?.let { v -> it.stateVersion < v } == true }
                if (superseded.isEmpty()) break
                msgEvents.markAllSent(superseded.mapNotNull { it.eventId })
            }
        }.onFailure { log.warn("superseded cleanup failed: {}", it.message) }
        failures.forEach { retryOrDead(it, it.lastError ?: "send-failed") }
        lastFlush = scheduler.now()
    }

    /** 一条事件发失败之后怎么走：重试次数加一，到上限就标成 DEAD(EXHAUSTED) 留作死信（次数落库便于追查），否则按退避推到下次再发。 */
    private fun retryOrDead(e: MsgEvent, lastError: String) {
        val eventId = e.eventId ?: return
        val attempts = e.attempts + 1
        if (scheduler.exhausted(attempts)) {
            msgEvents.markDead(eventId, ErrorClass.EXHAUSTED, lastError, attempts)
        } else {
            msgEvents.scheduleRetry(eventId, scheduler.nextAttemptAt(attempts), attempts)
        }
    }

    private fun sleepQuietly(d: Duration) {
        if (!d.isNegative && !d.isZero) Thread.sleep(d.toMillis().coerceAtLeast(1))
    }
}