package com.doumee.biz.system.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.doumee.biz.system.SystemJobTriggerBiz;
import com.doumee.biz.system.dto.TriggerJobDTO;
import com.doumee.core.constants.Constants;
import com.doumee.core.exception.LockedException;
import com.doumee.core.job.BaseDistributer;
import com.doumee.core.job.BaseJob;
import com.doumee.core.job.JobContext;
import com.doumee.core.job.JobParam;
import com.doumee.core.utils.Utils;
import com.doumee.dao.system.SystemJobLogMapper;
import com.doumee.dao.system.SystemJobMapper;
import com.doumee.dao.system.SystemJobSnippetMapper;
import com.doumee.dao.system.model.SystemJob;
import com.doumee.dao.system.model.SystemJobLog;
import com.doumee.dao.system.model.SystemJobSnippet;
import com.doumee.service.system.SystemJobLogService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.List;
import java.util.UUID;

@Slf4j
@Service
public class SystemJobTriggerBizImpl implements SystemJobTriggerBiz {

    @Autowired
    private SystemJobLogMapper systemJobLogMapper;

    @Autowired
    private SystemJobMapper systemJobMapper;

    @Autowired
    private SystemJobSnippetMapper systemJobSnippetMapper;

    private ThreadLocal<String> jobUnique = new ThreadLocal<>();

    @Override
    public void trigger(TriggerJobDTO dto) {
        SystemJob job = null;
        JobParam jobParam = new JobParam();
        boolean locked = false;
        SystemJobLog systemJobLog = new SystemJobLog();
        try {
            Integer jobId = dto.getId();
            job = systemJobMapper.selectById(jobId);
            // 验证JOB
            if (!this.checkJob(job, dto.getBusinessTime())) {
                return;
            }
            // 初始化JOB数据
            jobUnique.set(RandomStringUtils.randomAlphabetic(8));
            log(job, "Job started");
            log(job, "Job thread name: {}", Thread.currentThread().getName());
            log(job, "Job scheduled fire time: {}", Utils.Date.format(dto.getScheduledFireTime()));
            log(job, "Job fire time: {}", Utils.Date.format(dto.getFireTime()));
            log(job, "Job business time: {}", Utils.Date.format(dto.getBusinessTime()));
            jobParam.setId(jobId);
            jobParam.setTriggerType(dto.getTriggerType());
            jobParam.setBusinessTime(dto.getBusinessTime());
            // 记录日志
            if (job.getWithLog()) {
                systemJobLog.setJobId(jobId);
                systemJobLog.setJobName(job.getJobName());
                systemJobLog.setRunTimeStart(new Date());
                systemJobLog.setServerIp(Utils.Server.getIP());
                systemJobLog.setBusinessTime(dto.getBusinessTime());
                systemJobLog.setStatus(Constants.Job.LogStatus.NONE.getCode());
                systemJobLog.setTriggerType(dto.getTriggerType());
                systemJobLog.setRemark("任务正在执行中");
                systemJobLogMapper.insert(systemJobLog);
            }
            // 任务加锁，防止多服务器重复执行
            this.lock(job);
            locked = true;
            // 如果开启了异步执行，则创建一个分片
            if (job.getWithAsync()) {
                String snippetCode = this.createAsyncSnippet(job, jobParam);
                systemJobLog.setStatus(Constants.Job.LogStatus.SUCCESS.getCode());
                systemJobLog.setRemark("异步任务已转为分片任务，分片编码：" + snippetCode);
                return;
            }
            // 如果存在分片
            if (StringUtils.isNotBlank(job.getDistributeHandler())) {
                String distributeGroup = UUID.randomUUID().toString();
                int snippetCount = this.distribute(job, jobParam, distributeGroup);
                systemJobLog.setStatus(Constants.Job.LogStatus.SUCCESS.getCode());
                systemJobLog.setRemark("任务已分发成" + snippetCount + "个分片任务，分发组：" + distributeGroup);
                return;
            }
            // 不存在分片，直接执行任务（根据任务处理器名称获取任务处理器实例并执行）
            BaseJob jobBean = Utils.SpringContext.getBean(job.getHandler(), BaseJob.class);
            JobContext jobContext = jobBean.execute(jobParam);
            // 解锁任务
            this.unlock(job, dto.getNextFireTime());
            locked = false;
            // 修改日志信息
            if (job.getWithLog()) {
                systemJobLog.setRemark("任务执行完成");
                systemJobLog.setHandleTotalSize(jobContext.getHandleTotalSize());
                systemJobLog.setHandleSuccessSize(jobContext.getHandleSuccessSize());
                systemJobLog.setContext(jobContext.getContext());
                systemJobLog.setStatus(Constants.Job.LogStatus.SUCCESS.getCode());
            }
        } catch (LockedException e) {
            log(job, e.getMessage());
            // 修改日志状态
            if (job.getWithLog()) {
                systemJobLog.setStatus(Constants.Job.LogStatus.IGNORE.getCode());
                systemJobLog.setRemark("任务正在被其他服务执行或锁定");
            }
        } catch (Exception e) {
            if (job != null) {
                error(job, e);
                // 修改日志状态
                if (job.getWithLog()) {
                    systemJobLog.setStatus(Constants.Job.LogStatus.FAILED.getCode());
                    systemJobLog.setRemark("任务执行失败，出现异常：" + e.getMessage());
                }
            } else {
                e.printStackTrace();
            }
        } finally {
            // 更新日志
            if (job != null && job.getWithLog() && systemJobLog.getId() != null) {
                systemJobLog.setRunTimeEnd(new Date());
                systemJobLog.setRunSpendTime(Integer.valueOf("" + (systemJobLog.getRunTimeEnd().getTime() - systemJobLog.getRunTimeStart().getTime())));
                systemJobLogMapper.updateById(systemJobLog);
            }
            // 任务解锁
            if (locked) {
                this.unlock(job, dto.getNextFireTime());
                log(job, "Job execute completed.");
            }
        }
    }

    /**
     * 锁定JOB
     *
     * @param job JOB
     */
    private void lock (SystemJob job) throws LockedException{
        log(job, "Lock job.");
        if (Constants.equalsInteger(job.getStatus(),Constants.Job.JobStatus.RUNNING.getCode())) {
            throw new LockedException("EVA: job is running.");
        }
        SystemJob newJob = new SystemJob();
        newJob.setStatus(Constants.Job.JobStatus.RUNNING.getCode());
        newJob.setLockNumber(job.getLockNumber() + 1);
        // 不修改更新人和更新时间
        newJob.setUpdateUser(Constants.Ignore.IGNORE_USER);
        newJob.setUpdateTime(Constants.Ignore.IGNORE_TIME);
        int result = systemJobMapper.update(newJob, new UpdateWrapper<SystemJob>()
                .lambda()
                .eq(SystemJob::getId, job.getId())
                .eq(SystemJob::getLockNumber, job.getLockNumber())
        );
        if (result == 0) {
            throw new LockedException("EVA: job is locked.");
        }
        log(job, "Lock job completed.");
    }

    /**
     * 解锁JOB
     *
     * @param job JOB
     */
    private void unlock(SystemJob job, Date nextFireTime) {
        log(job, "Unlock job.");
        SystemJob newJob = new SystemJob();
        newJob.setId(job.getId());
        newJob.setStatus(Constants.Job.JobStatus.READY.getCode());
        newJob.setNextFireTime(nextFireTime);
        // 不修改更新人和更新时间
        newJob.setUpdateUser(Constants.Ignore.IGNORE_USER);
        newJob.setUpdateTime(Constants.Ignore.IGNORE_TIME);
        systemJobMapper.updateById(newJob);
        log(job, "Unlock job completed.");
    }

    /**
     * 创建异步分片
     *
     * @param job JOB
     * @param jobParam JOB参数
     */
    private String createAsyncSnippet (SystemJob job, JobParam jobParam) {
        log(job, "Begin create async snippet.");
        String distributeGroup = UUID.randomUUID().toString();
        SystemJobSnippet snippet = new SystemJobSnippet();
        snippet.setJobId(job.getId());
        snippet.setJobName(job.getJobName());
        snippet.setJobBusinessTime(jobParam.getBusinessTime());
        snippet.setJobDistributeGroup(distributeGroup);
        snippet.setDistributeHandler(job.getDistributeHandler());
        snippet.setDistributeLimit(job.getDistributeLimit());
        snippet.setHandler(job.getHandler());
        snippet.setSnippetIndex(0);
        snippet.setSnippetCode(UUID.randomUUID().toString());
        snippet.setWithLog(job.getWithLog());
        snippet.setAllowServerIps(job.getAllowServerIps());
        snippet.setStatus(Constants.Job.JobStatus.READY.getCode());
        systemJobSnippetMapper.insert(snippet);
        log(job, "Create async snippet completed.");
        return snippet.getSnippetCode();
    }

    /**
     * 分发
     *
     * @param job JOB
     * @param jobParam JOB参数
     * @param distributeGroup 分发组
     * @return int
     */
    private int distribute (SystemJob job, JobParam jobParam, String distributeGroup) {
        log(job, "Begin distribute.");
        BaseDistributer distributer = Utils.SpringContext.getBean(job.getDistributeHandler(), BaseDistributer.class);
        List<List<?>> snippetDatas = distributer.getSnippetData(new BaseDistributer.DistributeDTO(job.getDistributeLimit()), jobParam);
        for (int i = 0; i < snippetDatas.size(); i++) {
            List snippetData = snippetDatas.get(i);
            SystemJobSnippet snippet = new SystemJobSnippet();
            snippet.setJobId(job.getId());
            snippet.setJobName(job.getJobName());
            snippet.setJobBusinessTime(jobParam.getBusinessTime());
            snippet.setJobDistributeGroup(distributeGroup);
            snippet.setHandler(job.getHandler());
            snippet.setSnippetIndex(i);
            snippet.setSnippetCode(UUID.randomUUID().toString());
            snippet.setSnippetData(JSON.toJSONString(snippetData));
            snippet.setSnippetDataSize(snippetData.size());
            snippet.setWithLog(job.getWithLog());
            snippet.setAllowServerIps(job.getAllowServerIps());
            snippet.setStatus(Constants.Job.JobStatus.READY.getCode());
            systemJobSnippetMapper.insert(snippet);
        }
        log(job, "distribute completed.");
        return snippetDatas.size();
    }

    /**
     * 验证Job是否可执行
     *
     * @param job JOB
     */
    private boolean checkJob (SystemJob job, Date businessTime) {
        // 执行时间验证（任务预计下一次执行<=当前执行时间即可执行)
        if (businessTime != null && job.getNextFireTime() != null && job.getNextFireTime().compareTo(businessTime) == 1) {
            return Boolean.FALSE;
        }
        // 服务器验证
        if (StringUtils.isNotBlank(job.getAllowServerIps())) {
            return job.getAllowServerIps().contains(Utils.Server.getIP());
        }
        return Boolean.TRUE;
    }

    /**
     * 打印日志
     *
     * @param job JOB
     * @param text 日志模版内容
     * @param params 日志参数
     */
    private void log (SystemJob job, String text, Object... params) {
        log.trace("{}: " + text, String.format("%s_%s", job.getJobName(), jobUnique.get()), params);
    }

    /**
     * 打印错误日志
     *
     * @param job JOB
     * @param e 异常对象
     */
    private void error (SystemJob job, Exception e) {
        log.error(String.format("EVA: %s_%s", job.getJobName(), jobUnique.get()) + ": Handle job throw an exception.", e);
    }
}