package tech.easyflow.ai.mapper;

import com.mybatisflex.core.BaseMapper;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Select;
import org.apache.ibatis.annotations.Update;
import tech.easyflow.ai.entity.DocumentImportBatch;

import java.math.BigInteger;
import java.util.Date;
import java.util.List;

/**
 * 文档批量导入批次映射层。
 *
 * @author Codex
 * @since 2026-07-31
 */
public interface DocumentImportBatchMapper extends BaseMapper<DocumentImportBatch> {

    /**
     * 锁定并读取调用者拥有的批次，保证后续重试基于最新状态。
     *
     * @param batchId 批次 ID
     * @param callerType 调用者类型
     * @param callerId 调用者 ID
     * @return 批次；不存在时返回 null
     */
    @Select("SELECT * FROM tb_document_import_batch "
        + "WHERE id=#{batchId} AND caller_type=#{callerType} "
        + "AND caller_id=#{callerId} FOR UPDATE")
    DocumentImportBatch selectOwnedForUpdate(
        @Param("batchId") BigInteger batchId,
        @Param("callerType") String callerType,
        @Param("callerId") BigInteger callerId
    );

    /**
     * 按文件项状态迁移增量更新批次计数。
     *
     * @param batchId 批次 ID
     * @param completedDelta 完成数增量
     * @param processingDelta 处理中数量增量
     * @param failedDelta 失败数增量
     * @param pendingDelta 等待数增量
     * @param uploadedDelta 已上传数增量
     * @param skippedDelta 跳过数增量
     * @param cancelledDelta 取消数增量
     * @param retryableFailedDelta 可重试失败数增量
     * @param modified 修改时间
     * @return 更新行数
     */
    @Update("UPDATE tb_document_import_batch SET "
        + "completed_count=GREATEST(0, completed_count + #{completedDelta}), "
        + "processing_count=GREATEST(0, processing_count + #{processingDelta}), "
        + "failed_count=GREATEST(0, failed_count + #{failedDelta}), "
        + "pending_count=GREATEST(0, pending_count + #{pendingDelta}), "
        + "uploaded_count=GREATEST(0, uploaded_count + #{uploadedDelta}), "
        + "skipped_count=GREATEST(0, skipped_count + #{skippedDelta}), "
        + "cancelled_count=GREATEST(0, cancelled_count + #{cancelledDelta}), "
        + "retryable_failed_count=GREATEST(0, retryable_failed_count + #{retryableFailedDelta}), "
        + "modified=#{modified} WHERE id=#{batchId}")
    int adjustCounters(
        @Param("batchId") BigInteger batchId,
        @Param("completedDelta") int completedDelta,
        @Param("processingDelta") int processingDelta,
        @Param("failedDelta") int failedDelta,
        @Param("pendingDelta") int pendingDelta,
        @Param("uploadedDelta") int uploadedDelta,
        @Param("skippedDelta") int skippedDelta,
        @Param("cancelledDelta") int cancelledDelta,
        @Param("retryableFailedDelta") int retryableFailedDelta,
        @Param("modified") Date modified
    );

    /**
     * 原子熔断仍在运行的自动导入批次。
     *
     * @param batchId 批次 ID
     * @param interruptCode 稳定中断码
     * @param interruptMessage 用户可见中断原因
     * @param modified 中断时间
     * @return 成功取得熔断权时返回 1
     */
    @Update("UPDATE tb_document_import_batch SET "
        + "status='INTERRUPTED', interrupt_code=#{interruptCode}, "
        + "interrupt_message=#{interruptMessage}, interrupted_at=#{modified}, "
        + "recovery_pending=0, recovery_file_keys_json=NULL, "
        + "recovery_token=NULL, recovery_lease_until=NULL, "
        + "finished_at=NULL, modified=#{modified}, "
        + "version=version + 1 "
        + "WHERE id=#{batchId} AND import_mode='AUTO' AND status='RUNNING'")
    int interruptRunningBatch(
        @Param("batchId") BigInteger batchId,
        @Param("interruptCode") String interruptCode,
        @Param("interruptMessage") String interruptMessage,
        @Param("modified") Date modified
    );

    /**
     * 仅允许当前恢复令牌持有者中断运行批次。
     *
     * @param batchId 批次 ID
     * @param recoveryToken 当前恢复调度令牌
     * @param interruptCode 稳定中断码
     * @param interruptMessage 用户可见中断原因
     * @param modified 中断时间
     * @return 当前令牌仍持有恢复权且中断成功时返回 1
     */
    @Update("UPDATE tb_document_import_batch SET "
        + "status='INTERRUPTED', interrupt_code=#{interruptCode}, "
        + "interrupt_message=#{interruptMessage}, interrupted_at=#{modified}, "
        + "recovery_pending=0, recovery_file_keys_json=NULL, "
        + "recovery_token=NULL, recovery_lease_until=NULL, "
        + "finished_at=NULL, modified=#{modified}, "
        + "version=version + 1 "
        + "WHERE id=#{batchId} AND import_mode='AUTO' AND status='RUNNING' "
        + "AND recovery_pending=1 AND recovery_token=#{recoveryToken} "
        + "AND recovery_lease_until > #{modified}")
    int interruptOwnedRecoveryBatch(
        @Param("batchId") BigInteger batchId,
        @Param("recoveryToken") String recoveryToken,
        @Param("interruptCode") String interruptCode,
        @Param("interruptMessage") String interruptMessage,
        @Param("modified") Date modified
    );

    /**
     * 仅在触发任务仍是当前活跃任务时中断运行批次。
     *
     * @param batchId 批次 ID
     * @param taskId 触发异常的任务 ID
     * @param interruptCode 稳定中断码
     * @param interruptMessage 用户可见中断原因
     * @param modified 中断时间
     * @return 任务仍活跃且中断成功时返回 1
     */
    @Update("UPDATE tb_document_import_batch batch "
        + "INNER JOIN tb_document_import_task task ON task.batch_id=batch.id SET "
        + "batch.status='INTERRUPTED', batch.interrupt_code=#{interruptCode}, "
        + "batch.interrupt_message=#{interruptMessage}, "
        + "batch.interrupted_at=#{modified}, batch.recovery_pending=0, "
        + "batch.recovery_file_keys_json=NULL, batch.recovery_token=NULL, "
        + "batch.recovery_lease_until=NULL, batch.finished_at=NULL, "
        + "batch.modified=#{modified}, batch.version=batch.version + 1 "
        + "WHERE batch.id=#{batchId} AND batch.import_mode='AUTO' "
        + "AND batch.status='RUNNING' AND task.id=#{taskId} "
        + "AND task.status IN ('PENDING','RUNNING')")
    int interruptRunningBatchForActiveTask(
        @Param("batchId") BigInteger batchId,
        @Param("taskId") BigInteger taskId,
        @Param("interruptCode") String interruptCode,
        @Param("interruptMessage") String interruptMessage,
        @Param("modified") Date modified
    );

    /**
     * 根据批次项真实状态重新汇总计数。
     *
     * @param batchId 批次 ID
     * @param modified 修改时间
     * @return 更新行数
     */
    @Update("UPDATE tb_document_import_batch batch INNER JOIN ("
        + "SELECT batch_id, "
        + "SUM(status='COMPLETED') completed_count, "
        + "SUM(status='RUNNING') processing_count, "
        + "SUM(status='FAILED') failed_count, "
        + "SUM(status='PENDING') pending_count, "
        + "SUM(status='UPLOADED') uploaded_count, "
        + "SUM(status='SKIPPED') skipped_count, "
        + "SUM(status='CANCELLED') cancelled_count, "
        + "SUM(status='FAILED' AND retryable=1) retryable_failed_count "
        + "FROM tb_document_import_batch_item WHERE batch_id=#{batchId} "
        + "GROUP BY batch_id"
        + ") counters ON counters.batch_id=batch.id SET "
        + "batch.completed_count=counters.completed_count, "
        + "batch.processing_count=counters.processing_count, "
        + "batch.failed_count=counters.failed_count, "
        + "batch.pending_count=counters.pending_count, "
        + "batch.uploaded_count=counters.uploaded_count, "
        + "batch.skipped_count=counters.skipped_count, "
        + "batch.cancelled_count=counters.cancelled_count, "
        + "batch.retryable_failed_count=counters.retryable_failed_count, "
        + "batch.modified=#{modified} WHERE batch.id=#{batchId}")
    int refreshCountersFromItems(
        @Param("batchId") BigInteger batchId,
        @Param("modified") Date modified
    );

    /**
     * 查询已经提交但尚未完成恢复调度的运行批次。
     *
     * @param now 当前时间
     * @param limit 最大批次数
     * @return 待恢复批次
     */
    @Select("SELECT * FROM tb_document_import_batch "
        + "WHERE status='RUNNING' AND recovery_pending=1 "
        + "AND (recovery_token IS NULL OR recovery_lease_until <= #{now}) "
        + "ORDER BY modified, id LIMIT #{limit}")
    List<DocumentImportBatch> selectRecoveryPendingBatches(
        @Param("now") Date now,
        @Param("limit") int limit
    );

    /**
     * 使用租约令牌领取一个批次恢复待办。
     *
     * @param batchId 批次 ID
     * @param recoveryToken 恢复调度令牌
     * @param recoveryLeaseUntil 租约到期时间
     * @param modified 领取时间
     * @return 更新行数
     */
    @Update("UPDATE tb_document_import_batch SET "
        + "recovery_token=#{recoveryToken}, "
        + "recovery_lease_until=#{recoveryLeaseUntil}, "
        + "modified=#{modified}, version=version + 1 "
        + "WHERE id=#{batchId} AND status='RUNNING' "
        + "AND recovery_pending=1 AND (recovery_token IS NULL "
        + "OR recovery_lease_until <= #{modified})")
    int claimRecoveryPending(
        @Param("batchId") BigInteger batchId,
        @Param("recoveryToken") String recoveryToken,
        @Param("recoveryLeaseUntil") Date recoveryLeaseUntil,
        @Param("modified") Date modified
    );

    /**
     * 续期当前持有者的批次恢复租约。
     *
     * @param batchId 批次 ID
     * @param recoveryToken 恢复调度令牌
     * @param recoveryLeaseUntil 新租约到期时间
     * @param modified 续租时间
     * @return 当前令牌仍有效且续租成功时返回 1
     */
    @Update("UPDATE tb_document_import_batch SET "
        + "recovery_lease_until=#{recoveryLeaseUntil}, "
        + "modified=#{modified}, version=version + 1 "
        + "WHERE id=#{batchId} AND status='RUNNING' "
        + "AND recovery_pending=1 AND recovery_token=#{recoveryToken} "
        + "AND recovery_lease_until > #{modified}")
    int renewRecoveryPendingLease(
        @Param("batchId") BigInteger batchId,
        @Param("recoveryToken") String recoveryToken,
        @Param("recoveryLeaseUntil") Date recoveryLeaseUntil,
        @Param("modified") Date modified
    );

    /**
     * 按恢复令牌读取当前持有者领取后的精确恢复参数。
     *
     * @param batchId 批次 ID
     * @param recoveryToken 恢复调度令牌
     * @return 当前令牌持有的批次；令牌失效时返回空
     */
    @Select("SELECT * FROM tb_document_import_batch "
        + "WHERE id=#{batchId} AND status='RUNNING' "
        + "AND recovery_pending=1 AND recovery_token=#{recoveryToken}")
    DocumentImportBatch selectClaimedRecovery(
        @Param("batchId") BigInteger batchId,
        @Param("recoveryToken") String recoveryToken
    );

    /**
     * 原子完成当前恢复调度并按真实计数收口批次状态。
     *
     * <p>终态汇总与令牌清理在同一条 SQL 中完成，避免进程在两次写入
     * 之间退出后留下无恢复待办的运行批次。</p>
     *
     * @param batchId 批次 ID
     * @param recoveryToken 恢复调度令牌
     * @param modified 修改时间
     * @return 当前令牌和租约仍有效且收尾成功时返回 1
     */
    @Update("UPDATE tb_document_import_batch SET "
        + "status=CASE WHEN "
        + "completed_count + failed_count + skipped_count + cancelled_count "
        + ">= total_count AND processing_count=0 AND pending_count=0 "
        + "THEN CASE WHEN failed_count > 0 THEN 'PARTIAL_SUCCEEDED' "
        + "ELSE 'COMPLETED' END ELSE 'RUNNING' END, "
        + "finished_at=CASE WHEN "
        + "completed_count + failed_count + skipped_count + cancelled_count "
        + ">= total_count AND processing_count=0 AND pending_count=0 "
        + "THEN #{modified} ELSE NULL END, "
        + "recovery_pending=0, recovery_file_keys_json=NULL, "
        + "recovery_token=NULL, recovery_lease_until=NULL, "
        + "modified=#{modified}, version=version + 1 "
        + "WHERE id=#{batchId} AND import_mode='AUTO' AND status='RUNNING' "
        + "AND recovery_pending=1 AND recovery_token=#{recoveryToken} "
        + "AND recovery_lease_until > #{modified}")
    int finalizeRecoveryPending(
        @Param("batchId") BigInteger batchId,
        @Param("recoveryToken") String recoveryToken,
        @Param("modified") Date modified
    );

    /**
     * 原子领取管理端人工继续权并清除上一轮中断信息。
     *
     * @param batchId 批次 ID
     * @param modified 修改时间
     * @return 更新行数
     */
    @Update("UPDATE tb_document_import_batch SET status='RUNNING', "
        + "finished_at=NULL, interrupt_code=NULL, interrupt_message=NULL, "
        + "interrupted_at=NULL, recovery_pending=1, "
        + "recovery_file_keys_json=NULL, "
        + "recovery_token=NULL, recovery_lease_until=NULL, "
        + "modified=#{modified}, version=version + 1 "
        + "WHERE id=#{batchId} "
        + "AND status IN ('PARTIAL_SUCCEEDED','INTERRUPTED')")
    int claimContinue(
        @Param("batchId") BigInteger batchId,
        @Param("modified") Date modified
    );

    /**
     * 原子领取 Public API 重试代次。
     *
     * @param batchId 批次 ID
     * @param callerType 调用者类型
     * @param callerId 调用者 ID
     * @param expectedGeneration 预期重试代次
     * @param recoveryFileKeysJson 本轮选择恢复的文件键 JSON；为空表示全部
     * @param modified 修改时间
     * @return 更新行数
     */
    @Update("UPDATE tb_document_import_batch SET "
        + "status='RUNNING', finished_at=NULL, "
        + "interrupt_code=NULL, interrupt_message=NULL, interrupted_at=NULL, "
        + "recovery_pending=1, "
        + "recovery_file_keys_json=#{recoveryFileKeysJson}, "
        + "recovery_token=NULL, recovery_lease_until=NULL, "
        + "retry_generation=retry_generation + 1, "
        + "version=version + 1, modified=#{modified} "
        + "WHERE id=#{batchId} AND caller_type=#{callerType} AND caller_id=#{callerId} "
        + "AND retry_generation=#{expectedGeneration} "
        + "AND status IN ('PARTIAL_SUCCEEDED','INTERRUPTED')")
    int claimRetry(
        @Param("batchId") BigInteger batchId,
        @Param("callerType") String callerType,
        @Param("callerId") BigInteger callerId,
        @Param("expectedGeneration") int expectedGeneration,
        @Param("recoveryFileKeysJson") String recoveryFileKeysJson,
        @Param("modified") Date modified
    );

    /**
     * 原子领取无进展的未完成批次取消权。
     *
     * @param batchId 批次 ID
     * @param knowledgeId 知识库 ID
     * @param callerType 调用者类型
     * @param callerId 调用者 ID
     * @param incompleteCutoff 最后进展截止时间
     * @param modified 修改时间
     * @return 更新行数
     */
    @Update("UPDATE tb_document_import_batch SET "
        + "status='CANCELLED', finished_at=#{modified}, "
        + "version=version + 1, modified=#{modified} "
        + "WHERE id=#{batchId} AND knowledge_id=#{knowledgeId} "
        + "AND caller_type=#{callerType} AND caller_id=#{callerId} "
        + "AND status IN ('UPLOADING','READY') "
        + "AND ((modified IS NOT NULL AND modified < #{incompleteCutoff}) "
        + "OR (modified IS NULL AND created < #{incompleteCutoff}))")
    int claimStaleCancellation(
        @Param("batchId") BigInteger batchId,
        @Param("knowledgeId") BigInteger knowledgeId,
        @Param("callerType") String callerType,
        @Param("callerId") BigInteger callerId,
        @Param("incompleteCutoff") Date incompleteCutoff,
        @Param("modified") Date modified
    );

    /**
     * 分批查询无进展的未完成批次。
     *
     * @param incompleteCutoff 最后进展截止时间
     * @param limit 最大返回数量
     * @return 待回收批次
     */
    @Select("SELECT * FROM tb_document_import_batch "
        + "WHERE status IN ('UPLOADING','READY') "
        + "AND ((modified IS NOT NULL AND modified < #{incompleteCutoff}) "
        + "OR (modified IS NULL AND created < #{incompleteCutoff})) "
        + "ORDER BY COALESCE(modified, created), id LIMIT #{limit}")
    List<DocumentImportBatch> selectStaleIncompleteBatches(
        @Param("incompleteCutoff") Date incompleteCutoff,
        @Param("limit") int limit
    );

    /**
     * 释放已取消或超过去重窗口的服务端提交指纹。
     *
     * @param batchId 批次 ID
     * @param submissionFingerprint 当前服务端提交指纹
     * @param deduplicationCutoff 去重窗口起点
     * @param modified 修改时间
     * @return 更新行数
     */
    @Update("UPDATE tb_document_import_batch SET "
        + "idempotency_key_hash=NULL, modified=#{modified}, version=version + 1 "
        + "WHERE id=#{batchId} AND idempotency_key_hash=#{submissionFingerprint} "
        + "AND (status='CANCELLED' OR ("
        + "COALESCE(finished_at, modified, created) < #{deduplicationCutoff} "
        + "AND status IN ('INTERRUPTED','PARTIAL_SUCCEEDED','COMPLETED')))")
    int releaseSubmissionFingerprint(
        @Param("batchId") BigInteger batchId,
        @Param("submissionFingerprint") String submissionFingerprint,
        @Param("deduplicationCutoff") Date deduplicationCutoff,
        @Param("modified") Date modified
    );
}
