msgexchange-v2/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt

package com.gzzn.omms.msgexchange.jobs

import com.gzzn.omms.msgexchange.processing.BackfillService
import jakarta.inject.Singleton
import java.time.Clock
import java.time.Duration
import java.time.Instant
import java.time.LocalDate
import java.time.ZoneId

/**
 * 维护作业线程：定时做那些不需要跟消息一起排队的事。
 *
 * - 每 30 秒扫一次还欠回填的记录，把处理标记补写回共享信箱；
 * - 每天机场时间 3:30 之后跑一次航班历史归档与留痕清理。
 *
 * 作业跑在自己的线程上，不占用消息处理循环，也不会让到期的消息饿死在这里。
 * 回填意图在处理完成时就已经写进数据库了，本线程只负责到期重试。
 */
@Singleton
class JobRunner(
    private val backfill: BackfillService,
    private val historySweep: HistorySweepJob,
    private val clock: Clock,
) {
    private val log = org.slf4j.LoggerFactory.getLogger(JobRunner::class.java)
    private val zone: ZoneId = ZoneId.of("Asia/Shanghai")

    @Volatile
    private var running = true

    @Volatile
    private var lastHistoryDay: LocalDate? = null

    private var thread: Thread? = null

    fun start() {
        if (thread != null) return
        running = true
        thread = Thread.ofPlatform().name("msgx-jobs").daemon(true).start { loop() }
        log.info("job runner started (backfill 30s, history daily 03:30 {})", zone)
    }

    fun stop() {
        running = false
        thread?.interrupt()
        thread = null
    }

    internal fun loop() {
        while (running) {
            try {
                backfill.sweep()
                maybeHistorySweep()
            } catch (e: InterruptedException) {
                Thread.currentThread().interrupt()
                return
            } catch (e: Exception) {
                log.warn("job tick failed: {}", e.message ?: e.javaClass.simpleName)
            }
            if (running) runCatching { Thread.sleep(Duration.ofSeconds(30).toMillis()) }
        }
    }

    private fun maybeHistorySweep() {
        // 时区换算后的机场时钟：切日与 03:30 门槛都以机场时区为准，且走注入 Clock（测试可确定）。
        val airportClock = clock.withZone(zone)
        val today = LocalDate.now(airportClock)
        if (lastHistoryDay == today) return
        val now = java.time.LocalTime.now(airportClock)
        if ((now.hour == 3 && now.minute >= 30) || now.hour > 3) {
            val outcome = historySweep.run(clock.instant())
            lastHistoryDay = today
            if (outcome.selected > 0 || outcome.snapLogPurged > 0) {
                log.info("history sweep: {}", outcome)
            }
        }
    }
}