package com.github.kfcfans.powerjob.server.service.instance;

import com.github.kfcfans.powerjob.common.InstanceStatus;
import com.github.kfcfans.powerjob.common.TimeExpressionType;
import com.github.kfcfans.powerjob.common.request.TaskTrackerReportInstanceStatusReq;
import com.github.kfcfans.powerjob.server.common.utils.SpringUtils;
import com.github.kfcfans.powerjob.server.persistence.core.model.InstanceInfoDO;
import com.github.kfcfans.powerjob.server.persistence.core.model.JobInfoDO;
import com.github.kfcfans.powerjob.server.persistence.core.model.UserInfoDO;
import com.github.kfcfans.powerjob.server.persistence.core.repository.InstanceInfoRepository;
import com.github.kfcfans.powerjob.server.persistence.core.repository.JobInfoRepository;
import com.github.kfcfans.powerjob.server.service.DispatchService;
import com.github.kfcfans.powerjob.server.service.InstanceLogService;
import com.github.kfcfans.powerjob.server.service.UserService;
import com.github.kfcfans.powerjob.server.service.alarm.Alarmable;
import com.github.kfcfans.powerjob.server.service.alarm.JobInstanceAlarmContent;
import com.github.kfcfans.powerjob.server.service.timing.schedule.HashedWheelTimerHolder;
import com.github.kfcfans.powerjob.server.service.workflow.WorkflowInstanceManager;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.TimeUnit;

/**
 * 管理被调度的服务
 *
 * @author tjq
 * @since 2020/4/7
 */
@Slf4j
@Service
public class InstanceManager {

    // 存储 instanceId 对应的 Job 信息，便于重试
    private static Cache<Long, JobInfoDO> instanceId2JobInfo;

    // Spring Bean
    @Resource
    private DispatchService dispatchService;
    @Resource
    private InstanceLogService instanceLogService;
    @Resource(name = "omsCenterAlarmService")
    private Alarmable omsCenterAlarmService;
    @Resource
    private InstanceInfoRepository instanceInfoRepository;
    @Resource
    private JobInfoRepository jobInfoRepository;
    @Resource
    private WorkflowInstanceManager workflowInstanceManager;

    private static final int CACHE_CONCURRENCY_LEVEL = 8;
    private static final int CACHE_MAX_SIZE = 4096;

    static {
        instanceId2JobInfo = CacheBuilder.newBuilder()
                .concurrencyLevel(CACHE_CONCURRENCY_LEVEL)
                .maximumSize(CACHE_MAX_SIZE)
                .build();
    }

    /**
     * 更新任务状态
     * @param req TaskTracker上报任务实例状态的请求
     */
    public void updateStatus(TaskTrackerReportInstanceStatusReq req) throws Exception {

        Long jobId = req.getJobId();
        Long instanceId = req.getInstanceId();

        // 获取相关数据
        JobInfoDO jobInfo = instanceId2JobInfo.get(instanceId, () -> {
            Optional<JobInfoDO> jobInfoOpt = jobInfoRepository.findById(jobId);
            return jobInfoOpt.orElseThrow(() -> new IllegalArgumentException("can't find JobIno by jobId: " + jobId));
        });
        InstanceInfoDO instanceInfo = instanceInfoRepository.findByInstanceId(instanceId);

        // 丢弃过期的上报数据
        if (req.getReportTime() <= instanceInfo.getLastReportTime()) {
            log.warn("[InstanceManager-{}] receive the expired status report request: {}, this report will br dropped.", instanceId, req);
            return;
        }

        // 丢弃非目标 TaskTracker 的上报数据（脑裂情况）
        if (!req.getSourceAddress().equals(instanceInfo.getTaskTrackerAddress())) {
            log.warn("[InstanceManager-{}] receive the other TaskTracker's report: {}, but current TaskTracker is {}, this report will br dropped.", instanceId, req, instanceInfo.getTaskTrackerAddress());
            return;
        }

        InstanceStatus newStatus = InstanceStatus.of(req.getInstanceStatus());
        Integer timeExpressionType = jobInfo.getTimeExpressionType();

        instanceInfo.setStatus(newStatus.getV());
        instanceInfo.setLastReportTime(req.getReportTime());
        instanceInfo.setGmtModified(new Date());

        // FREQUENT 任务没有失败重试机制，TaskTracker一直运行即可，只需要将存活信息同步到DB即可
        // FREQUENT 任务的 newStatus 只有2中情况，一种是 RUNNING，一种是 FAILED（表示该机器 overload，需要重新选一台机器执行）
        // 综上，直接把 status 和 runningNum 同步到DB即可
        if (TimeExpressionType.frequentTypes.contains(timeExpressionType)) {

            instanceInfo.setRunningTimes(req.getTotalTaskNum());
            instanceInfoRepository.saveAndFlush(instanceInfo);
            return;
        }

        boolean finished = false;
        if (newStatus == InstanceStatus.SUCCEED) {
            instanceInfo.setResult(req.getResult());
            instanceInfo.setFinishedTime(System.currentTimeMillis());

            finished = true;
        }else if (newStatus == InstanceStatus.FAILED) {

            // 当前重试次数 <= 最大重试次数，进行重试 （第一次运行，runningTimes为1，重试一次，instanceRetryNum也为1，故需要 =）
            if (instanceInfo.getRunningTimes() <= jobInfo.getInstanceRetryNum()) {

                log.info("[InstanceManager-{}] instance execute failed but will take the {}th retry.", instanceId, instanceInfo.getRunningTimes());

                // 延迟10S重试（由于重试不改变 instanceId，如果派发到同一台机器，上一个 TaskTracker 还处于资源释放阶段，无法创建新的TaskTracker，任务失败）
                HashedWheelTimerHolder.TIMER.schedule(() -> {
                    dispatchService.redispatch(jobInfo, instanceId, instanceInfo.getRunningTimes());
                }, 10, TimeUnit.SECONDS);

                // 修改状态为 等待派发，正式开始重试
                // 问题：会丢失以往的调度记录（actualTriggerTime什么的都会被覆盖）
                instanceInfo.setStatus(InstanceStatus.WAITING_DISPATCH.getV());
            }else {
                instanceInfo.setResult(req.getResult());
                instanceInfo.setFinishedTime(System.currentTimeMillis());
                finished = true;
                log.info("[InstanceManager-{}] instance execute failed and have no chance to retry.", instanceId);
            }
        }

        // 同步状态变更信息到数据库
        instanceInfoRepository.saveAndFlush(instanceInfo);

        if (finished) {
            // 这里的 InstanceStatus 只有 成功/失败 两种，手动停止不会由 TaskTracker 上报
            processFinishedInstance(instanceId, req.getWfInstanceId(), newStatus, req.getResult());
        }
    }

    /**
     * 收尾完成的任务实例
     * @param instanceId 任务实例ID
     * @param wfInstanceId 工作流实例ID，非必须
     * @param status 任务状态，有 成功/失败/手动停止
     * @param result 执行结果
     */
    public void processFinishedInstance(Long instanceId, Long wfInstanceId, InstanceStatus status, String result) {

        log.info("[Instance-{}] process finished, final status is {}.", instanceId, status.name());

        // 上报日志数据
        instanceLogService.sync(instanceId);

        // workflow 特殊处理
        if (wfInstanceId != null) {
            // 手动停止在工作流中也认为是失败（理论上不应该发生）
            workflowInstanceManager.move(wfInstanceId, instanceId, status, result);
        }

        // 告警
        if (status == InstanceStatus.FAILED) {
            JobInfoDO jobInfo = fetchJobInfo(instanceId);
            if (jobInfo == null) {
                log.warn("[InstanceManager] can't find jobInfo by instanceId({}), alarm failed.", instanceId);
                return;
            }

            InstanceInfoDO instanceInfo = instanceInfoRepository.findByInstanceId(instanceId);
            JobInstanceAlarmContent content = new JobInstanceAlarmContent();
            BeanUtils.copyProperties(jobInfo, content);
            BeanUtils.copyProperties(instanceInfo, content);

            List<UserInfoDO> userList = SpringUtils.getBean(UserService.class).fetchNotifyUserList(jobInfo.getNotifyUserIds());
            omsCenterAlarmService.onJobInstanceFailed(content, userList);
        }

        // 过期缓存
        instanceId2JobInfo.invalidate(instanceId);
    }

    /**
     * 根据任务实例ID查询任务相关信息
     * @param instanceId 任务实例ID
     * @return 任务元数据
     */
    public JobInfoDO fetchJobInfo(Long instanceId) {
        JobInfoDO jobInfo = instanceId2JobInfo.getIfPresent(instanceId);
        if (jobInfo != null) {
            return jobInfo;
        }
        InstanceInfoDO instanceInfo = instanceInfoRepository.findByInstanceId(instanceId);
        if (instanceInfo != null) {
            return jobInfoRepository.findById(instanceInfo.getJobId()).orElse(null);
        }
        return null;
    }

    /**
     * 释放本地缓存，防止内存泄漏
     */
    public static void releaseCache() {
        instanceId2JobInfo.cleanUp();
    }
}
