package tech.easyflow.datacenter.schedule;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import tech.easyflow.common.cache.DistributedScheduledLock;

import java.sql.Timestamp;
import java.time.Instant;
import java.time.temporal.ChronoUnit;

/**
 * 定期分批清理过期的数据集写入幂等回执。
 */
@Component
public class DatacenterWriteReceiptCleanupJob {

    private static final Logger log =
            LoggerFactory.getLogger(DatacenterWriteReceiptCleanupJob.class);
    private static final String DELETE_SQL =
            "DELETE FROM tb_datacenter_write_receipt "
                    + "WHERE created < ? ORDER BY created LIMIT ?";
    private static final long MIN_RETENTION_DAYS = 7L;

    private final JdbcTemplate jdbcTemplate;
    private final long retentionDays;
    private final int batchSize;
    private final int maxBatches;

    /**
     * 创建回执清理任务。
     *
     * @param jdbcTemplate JDBC 操作模板
     * @param retentionDays 回执保留天数，最低七天
     * @param batchSize 单批删除行数
     * @param maxBatches 单次调度最多删除批次
     */
    public DatacenterWriteReceiptCleanupJob(
            JdbcTemplate jdbcTemplate,
            @Value("${easyflow.workflow.data-write-receipt-retention-days:14}")
            long retentionDays,
            @Value("${easyflow.workflow.data-write-receipt-cleanup-batch-size:1000}")
            int batchSize,
            @Value("${easyflow.workflow.data-write-receipt-cleanup-max-batches:20}")
            int maxBatches) {
        this.jdbcTemplate = jdbcTemplate;
        this.retentionDays = Math.max(MIN_RETENTION_DAYS, retentionDays);
        this.batchSize = Math.max(1, batchSize);
        this.maxBatches = Math.max(1, maxBatches);
    }

    /**
     * 在单个集群节点上删除一批超过保留期的回执。
     */
    @Scheduled(
            fixedDelayString =
                    "${easyflow.workflow.data-write-receipt-cleanup-interval:1h}",
            initialDelayString =
                    "${easyflow.workflow.data-write-receipt-cleanup-initial-delay:10m}")
    @DistributedScheduledLock(
            key = "easyflow:schedule:datacenter-write-receipt-cleanup",
            leaseSeconds = 300L)
    public void cleanup() {
        Timestamp cutoff = Timestamp.from(
                Instant.now().minus(retentionDays, ChronoUnit.DAYS));
        int totalDeleted = 0;
        try {
            for (int batch = 0; batch < maxBatches; batch++) {
                int deleted = jdbcTemplate.update(
                        DELETE_SQL, cutoff, batchSize);
                totalDeleted += deleted;
                if (deleted < batchSize) {
                    break;
                }
            }
            if (totalDeleted > 0) {
                log.info(
                        "Cleaned {} expired datacenter write receipts",
                        totalDeleted);
            }
        } catch (RuntimeException error) {
            log.error("Datacenter write receipt cleanup failed", error);
        }
    }
}
