package com.aizuda.snailjob.server.retry.task.support.schedule;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.lang.Assert;
import com.aizuda.snailjob.common.core.context.SnailSpringContext;
import com.aizuda.snailjob.common.core.enums.RetryStatusEnum;
import com.aizuda.snailjob.common.core.enums.StatusEnum;
import com.aizuda.snailjob.common.core.util.JsonUtil;
import com.aizuda.snailjob.common.core.util.StreamUtils;
import com.aizuda.snailjob.common.log.SnailJobLog;
import com.aizuda.snailjob.server.common.Lifecycle;
import com.aizuda.snailjob.server.common.config.SystemProperties;
import com.aizuda.snailjob.server.common.dto.PartitionTask;
import com.aizuda.snailjob.server.common.enums.SyetemTaskTypeEnum;
import com.aizuda.snailjob.server.common.exception.SnailJobServerException;
import com.aizuda.snailjob.server.common.schedule.AbstractSchedule;
import com.aizuda.snailjob.server.common.util.PartitionTaskUtils;
import com.aizuda.snailjob.server.retry.task.dto.RetryPartitionTask;
import com.aizuda.snailjob.server.retry.task.service.RetryDeadLetterConverter;
import com.aizuda.snailjob.server.retry.task.support.RetryTaskConverter;
import com.aizuda.snailjob.server.retry.task.support.event.RetryTaskFailDeadLetterAlarmEvent;
import com.aizuda.snailjob.template.datasource.access.AccessTemplate;
import com.aizuda.snailjob.template.datasource.access.TaskAccess;
import com.aizuda.snailjob.template.datasource.persistence.mapper.RetryMapper;
import com.aizuda.snailjob.template.datasource.persistence.mapper.RetryTaskMapper;
import com.aizuda.snailjob.template.datasource.persistence.mapper.RetryTaskLogMessageMapper;
import com.aizuda.snailjob.template.datasource.persistence.po.Retry;
import com.aizuda.snailjob.template.datasource.persistence.po.RetryDeadLetter;
import com.aizuda.snailjob.template.datasource.persistence.po.RetryTask;
import com.aizuda.snailjob.template.datasource.persistence.po.RetryTaskLogMessage;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.google.common.collect.Lists;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallbackWithoutResult;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.CollectionUtils;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;

/**
 * Retry清理线程
 * 1. 删除日志信息
 * 2. 删除重试已完成的数据
 * 3. 删除回调任务数据
 * 4. 删除调度日志 sj_retry_task
 * 5. 迁移到达最大重试的数据
 *
 * @author: opensnail
 * @date : 2023-07-21 13:32
 * @since 2.1.0
 */
@Component
@Slf4j
@RequiredArgsConstructor
public class CleanerSchedule extends AbstractSchedule implements Lifecycle {
    private final RetryMapper retryMapper;
    private final RetryTaskMapper retryTaskMapper;
    private final SystemProperties systemProperties;
    private final RetryTaskLogMessageMapper retryTaskLogMessageMapper;
    private final TransactionTemplate transactionTemplate;
    private final AccessTemplate accessTemplate;

    @Override
    public String lockName() {
        return "clearLog";
    }

    @Override
    public String lockAtMost() {
        return "PT4H";
    }

    @Override
    public String lockAtLeast() {
        return "PT1M";
    }

    @Override
    protected void doExecute() {
        try {
            // 清除日志默认保存天数大于零、最少保留最近一天的日志数据
//            if (systemProperties.getLogStorage() <= 1) {
//                SnailJobLog.LOCAL.error("retry clear log storage error", systemProperties.getLogStorage());
//                return;
//            }

            // clean retry log
            LocalDateTime endTime = LocalDateTime.now();
            long total = PartitionTaskUtils.process(startId -> retryTaskBatchList(startId, endTime),
                    this::processRetryLogPartitionTasks, 0);

            SnailJobLog.LOCAL.debug("Retry clear success total:[{}]", total);
        } catch (Exception e) {
            SnailJobLog.LOCAL.error("clear log error", e);
        }
    }

    /**
     * RetryLog List
     *
     * @param startId
     * @param endTime
     * @return
     */
    private List<RetryPartitionTask> retryTaskBatchList(Long startId, LocalDateTime endTime) {

        List<Retry> retryTaskList = retryMapper.selectPage(
                        new Page<>(0, 500),
                        new LambdaUpdateWrapper<Retry>()
                                .ge(Retry::getId, startId)
                                .le(Retry::getCreateDt, endTime)
                                .eq(Retry::getTaskType, SyetemTaskTypeEnum.RETRY.getType())
                                .ne(Retry::getDeleted, StatusEnum.NO.getStatus())
                                .orderByAsc(Retry::getId))
                .getRecords();
        return RetryTaskConverter.INSTANCE.toRetryTaskLogPartitionTasks(retryTaskList);
    }

    /**
     * clean table RetryTaskLog & RetryTaskLogMessage
     *
     * @param partitionTasks
     */
    public void processRetryLogPartitionTasks(List<? extends PartitionTask> partitionTasks) {

        List<Long> retryIds = StreamUtils.toList(partitionTasks, PartitionTask::getId);
        if (CollectionUtils.isEmpty(retryIds)) {
            return;
        }

        // 查询回调数据
        List<Retry> cbRetries = retryMapper.selectList(new LambdaQueryWrapper<Retry>()
                .select(Retry::getId).in(Retry::getParentId, retryIds));

        List<Long> totalWaitRetryIds = Lists.newArrayList(retryIds);
        List<Long> cbRetryIds = Lists.newArrayList();
        if (!CollectionUtils.isEmpty(cbRetries)) {
            cbRetryIds = StreamUtils.toList(cbRetries, Retry::getId);
            totalWaitRetryIds.addAll(cbRetryIds);
        }

        List<RetryPartitionTask> retryPartitionTasks = (List<RetryPartitionTask>) partitionTasks;

        List<Long> finishRetryIds = retryPartitionTasks.stream().filter(retryPartitionTask ->
                        RetryStatusEnum.FINISH.getStatus().equals(retryPartitionTask.getRetryStatus()))
                .map(PartitionTask::getId).toList();

        // 删除重试任务
        List<RetryTask> retryTaskList = retryTaskMapper.selectList(new LambdaQueryWrapper<RetryTask>()
                .in(RetryTask::getRetryId, totalWaitRetryIds));

        // 删除重试日志信息
        List<RetryTaskLogMessage> retryTaskLogMessageList = retryTaskLogMessageMapper.selectList(
                new LambdaQueryWrapper<RetryTaskLogMessage>()
                        .in(RetryTaskLogMessage::getRetryId, totalWaitRetryIds));

        List<Long> finalCbRetryIds = cbRetryIds;
        transactionTemplate.execute(new TransactionCallbackWithoutResult() {
            @Override
            protected void doInTransactionWithoutResult(final TransactionStatus status) {

                // 删除回调数据
                retryMapper.deleteByIds(finalCbRetryIds);

                // 删除重试完成的数据
                retryMapper.deleteByIds(finishRetryIds);

                // 删除重试任务
                if (!CollectionUtils.isEmpty(retryTaskList)) {
                    List<Long> retryTaskIds = StreamUtils.toList(retryTaskList, RetryTask::getId);
                    Lists.partition(retryTaskIds, 500).forEach(retryTaskMapper::deleteByIds);
                }

                if (!CollectionUtils.isEmpty(retryTaskLogMessageList)) {
                    List<Long> retryTaskLogMessageIds = StreamUtils.toList(retryTaskLogMessageList, RetryTaskLogMessage::getId);
                    Lists.partition(retryTaskLogMessageIds, 500).forEach(retryTaskLogMessageMapper::deleteByIds);
                }

            }
        });

        // 重试最大次数迁移死信表
        List<RetryPartitionTask> maxCountRetries = retryPartitionTasks.stream()
                .filter(retryPartitionTask ->
                        RetryStatusEnum.MAX_COUNT.getStatus().equals(retryPartitionTask.getRetryStatus()))
                .toList();
        moveDeadLetters(maxCountRetries);
    }

    /**
     * 迁移死信队列数据
     *
     * @param retries 待迁移数据
     */
    private void moveDeadLetters(List<RetryPartitionTask> retries) {
        if (CollUtil.isEmpty(retries)) {
            return;
        }

        List<RetryDeadLetter> retryDeadLetters = RetryDeadLetterConverter.INSTANCE.toRetryDeadLetter(retries);
        LocalDateTime now = LocalDateTime.now();
        for (RetryDeadLetter retryDeadLetter : retryDeadLetters) {
            retryDeadLetter.setCreateDt(now);
        }

        Assert.isTrue(retryDeadLetters.size() == accessTemplate
                        .getRetryDeadLetterAccess().insertBatch(retryDeadLetters),
                () -> new SnailJobServerException("插入死信队列失败 [{}]", JsonUtil.toJsonString(retryDeadLetters)));

        TaskAccess<Retry> retryTaskAccess = accessTemplate.getRetryAccess();
        Assert.isTrue(retries.size() == retryTaskAccess.delete(new LambdaQueryWrapper<Retry>()
                        .in(Retry::getId, StreamUtils.toList(retries, RetryPartitionTask::getId))),
                () -> new SnailJobServerException("删除重试数据失败 [{}]", JsonUtil.toJsonString(retries)));

        SnailSpringContext.getContext().publishEvent(new RetryTaskFailDeadLetterAlarmEvent(
                RetryTaskConverter.INSTANCE.toRetryTaskFailDeadLetterAlarmEventDTO(retryDeadLetters)
        ));
    }


    @Override
    public void start() {
        taskScheduler.scheduleAtFixedRate(this::execute, Duration.parse("PT4H"));
    }

    @Override
    public void close() {

    }
}
