package tech.easyflow.ai.documentimport.task;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import tech.easyflow.common.cache.DistributedScheduledLock;

/**
 * 文档导入待处理任务恢复调度器。
 *
 * @author Codex
 * @since 2026-07-31
 */
@Component
public class DocumentImportPendingTaskMonitor {

    private static final Logger LOG =
        LoggerFactory.getLogger(DocumentImportPendingTaskMonitor.class);
    private static final long FAILURE_COOLDOWN_MILLIS = 30_000L;
    private static final long FULL_ERROR_LOG_INTERVAL_MILLIS = 60_000L;

    private final KnowledgeDocumentImportTaskAppService appService;
    private final DocumentImportBatchAppService batchAppService;
    private long retryAfter;
    private long nextFullErrorLogAt;
    private int suppressedFailures;

    /**
     * 创建文档导入恢复调度器。
     *
     * @param appService 文档导入任务服务
     * @param batchAppService 文档导入批次服务
     */
    public DocumentImportPendingTaskMonitor(
        KnowledgeDocumentImportTaskAppService appService,
        DocumentImportBatchAppService batchAppService) {
        this.appService = appService;
        this.batchAppService = batchAppService;
    }

    /**
     * 周期性投递仍处于等待状态的解析与向量化任务。
     */
    @Scheduled(
        fixedDelayString = "${easyflow.ai.document-import.bulk.pending-dispatch-interval:2s}",
        initialDelayString = "${easyflow.ai.document-import.bulk.pending-dispatch-interval:2s}"
    )
    public void dispatchPendingTasks() {
        long now = System.currentTimeMillis();
        if (now < retryAfter) {
            return;
        }
        try {
            batchAppService.recoverPendingBatchRetries();
            appService.dispatchPendingTasks();
            if (suppressedFailures > 0) {
                LOG.info(
                    "文档导入待处理调度已恢复: suppressedFailures={}",
                    suppressedFailures
                );
            }
            retryAfter = 0L;
            nextFullErrorLogAt = 0L;
            suppressedFailures = 0;
        } catch (RuntimeException error) {
            retryAfter = now + FAILURE_COOLDOWN_MILLIS;
            if (now >= nextFullErrorLogAt) {
                LOG.error(
                    "文档导入待处理调度异常，30 秒后重试: "
                        + "suppressedFailures={}",
                    suppressedFailures,
                    error
                );
                nextFullErrorLogAt =
                    now + FULL_ERROR_LOG_INTERVAL_MILLIS;
                suppressedFailures = 0;
            } else {
                suppressedFailures++;
            }
        }
    }

    /**
     * 检测排队超时与长时间无心跳的任务。
     */
    @Scheduled(
        fixedDelayString = "${easyflow.ai.document-import.bulk.interruption-scan-interval:60s}",
        initialDelayString = "${easyflow.ai.document-import.bulk.interruption-scan-interval:60s}"
    )
    @DistributedScheduledLock(
        key = "easyflow:schedule:document-import:interruption-scan",
        leaseSeconds = 30L
    )
    public void recoverTimedOutTasks() {
        appService.expireTimedOutParseSubmissions();
        appService.recoverInterruptedTasks();
        appService.expireTimedOutPendingTasks();
        appService.cleanupCompletedReplacements();
    }
}
