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 同步确认，at-least-once；阶段 B 追加 ES 投影写入）。 */
interface DeliveryPort {
    /** KAFKA_MSG 变化通知（key=FLID）。 */
    fun sendKafka(topic: String, key: String, payloadJson: String)

    /**
     * KAFKA_SCHD 整态（key=FLID）。KAFKA_MSG 与 KAFKA_SCHD 映射同一 topic 语义由适配层定；
     * target→topic：KAFKA:msg→"msg"，KAFKA:schd→"schd"。
     */
    fun sendKafkaSchd(topic: String, key: String, payloadJson: String)

    /** TOMBSTONE：key=FLID、value=null——整态键缺失表示删除旧值（flight-state.md §5）。 */
    fun sendKafkaNull(topic: String, key: String)

    /** 连通性探测（健康检查用）；默认 true，真实 Kafka 实装时覆写为 producer metadata 校验。 */
    fun ping(): Boolean = true
}

/**
 * 投递调度（docs/flight-state.md §5 + design.md §5.1/§5.2）：逐条 KAFKA_MSG 严格 FIFO；
 * KAFKA_SCHD 走 flushSchd 批量——同一 FLID 未发事件按最新 STATE_VERSION 合并输出，
 * TOMBSTONE 发 null 值消息。两主题间不保证顺序（§5）。
 * 批量闭环：队首退避未到期不 claim；发送失败整批 attempts+1 退避，达上限整批 DEAD/DLQ。
 */
@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)
        }
    }

    /** flushSchd：同 FLID 未发事件按最新 STATE_VERSION 合并（§5）；TOMBSTONE 发 null。 */
    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())
        // 被最新版本合并压掉的未发事件同样关闭（§5：同 FLID 只按最新 STATE_VERSION 输出一次）
        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()
    }

    /** 单条事件失败迁移：attempts+1；达上限 DEAD(EXHAUSTED)（DLQ，attempts 落库审计），否则退避重试。 */
    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))
    }
}