package tech.powerjob.worker.core.tracker.task;

import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import tech.powerjob.common.enums.ExecuteType;
import tech.powerjob.common.enums.InstanceStatus;
import tech.powerjob.common.model.InstanceDetail;
import tech.powerjob.common.model.JobAdvancedRuntimeConfig;
import tech.powerjob.common.request.ServerQueryInstanceStatusReq;
import tech.powerjob.common.request.ServerScheduleJobReq;
import tech.powerjob.common.request.TaskTrackerReportInstanceStatusReq;
import tech.powerjob.common.serialize.JsonUtils;
import tech.powerjob.worker.common.WorkerRuntime;
import tech.powerjob.worker.common.utils.TransportUtils;
import tech.powerjob.worker.pojo.model.InstanceInfo;

import java.util.Collections;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author Echo009
 * @since 2022/9/19
 */
@Slf4j
public abstract class TaskTracker {

    /**
     * TaskTracker创建时间
     */
    protected final long createTime;
    /**
     * 任务实例ID，使用频率过高，从 InstanceInfo 提取出来单独保存一份
     */
    protected final long instanceId;
    /**
     * 任务实例信息
     */
    protected final InstanceInfo instanceInfo;
    protected final ExecuteType executeType;

    protected final JobAdvancedRuntimeConfig advancedRuntimeConfig;
    /**
     * 追加的工作流上下文数据
     *
     * @since 2021/02/05
     */
    protected final Map<String, String> appendedWfContext;
    /**
     * worker 运行时元数据
     */
    protected final WorkerRuntime workerRuntime;
    /**
     * 是否结束
     */
    protected final AtomicBoolean finished;
    /**
     * 连续上报多次失败后放弃上报，视为结果不可达，TaskTracker down
     */
    protected int reportFailedCnt = 0;

    protected static final int MAX_REPORT_FAILED_THRESHOLD = 5;

    protected TaskTracker(ServerScheduleJobReq req, WorkerRuntime workerRuntime) {
        this.createTime = System.currentTimeMillis();
        this.workerRuntime = workerRuntime;
        this.instanceId = req.getInstanceId();

        this.instanceInfo = new InstanceInfo();

        // PowerJob 值拷贝场景不多，引入三方值拷贝类库可能引入类冲突等问题，综合评估手写 ROI 最高
        instanceInfo.setJobId(req.getJobId());
        instanceInfo.setInstanceId(req.getInstanceId());
        instanceInfo.setWfInstanceId(req.getWfInstanceId());
        instanceInfo.setExecuteType(req.getExecuteType());
        instanceInfo.setProcessorType(req.getProcessorType());
        instanceInfo.setProcessorInfo(req.getProcessorInfo());
        instanceInfo.setJobParams(req.getJobParams());
        instanceInfo.setInstanceParams(req.getInstanceParams());
        instanceInfo.setThreadConcurrency(req.getThreadConcurrency());
        instanceInfo.setTaskRetryNum(req.getTaskRetryNum());
        instanceInfo.setLogConfig(req.getLogConfig());
        instanceInfo.setInstanceTimeoutMS(req.getInstanceTimeoutMS());
        instanceInfo.setAdvancedRuntimeConfig(req.getAdvancedRuntimeConfig());

        // 常用变量初始化
        executeType = ExecuteType.valueOf(req.getExecuteType());
        advancedRuntimeConfig = Optional.ofNullable(req.getAdvancedRuntimeConfig()).map(x -> JsonUtils.parseObjectIgnoreException(x, JobAdvancedRuntimeConfig.class)).orElse(new JobAdvancedRuntimeConfig());

        // 特殊处理超时时间
        if (instanceInfo.getInstanceTimeoutMS() <= 0) {
            instanceInfo.setInstanceTimeoutMS(Integer.MAX_VALUE);
        }
        // 只有工作流中的任务允许向工作流中追加上下文数据
        this.appendedWfContext = req.getWfInstanceId() == null ? Collections.emptyMap() : Maps.newConcurrentMap();
        this.finished = new AtomicBoolean(false);
    }

    /**
     * 销毁
     */
    public abstract void destroy();

    /**
     * 停止任务
     */
    public abstract void stopTask();


    /**
     * 查询任务实例的详细运行状态
     *
     * @return 任务实例的详细运行状态
     */
    public abstract InstanceDetail fetchRunningStatus(ServerQueryInstanceStatusReq req);


    public static void reportCreateErrorToServer(ServerScheduleJobReq req, WorkerRuntime workerRuntime, Exception e) {
        log.warn("[TaskTracker-{}] create TaskTracker from request({}) failed.", req.getInstanceId(), req, e);
        // 直接发送失败请求
        TaskTrackerReportInstanceStatusReq response = new TaskTrackerReportInstanceStatusReq();

        response.setAppId(workerRuntime.getAppId());
        response.setJobId(req.getJobId());
        response.setInstanceId(req.getInstanceId());
        response.setWfInstanceId(req.getWfInstanceId());

        response.setInstanceStatus(InstanceStatus.FAILED.getV());
        response.setResult(String.format("init TaskTracker failed, reason: %s", e.toString()));
        response.setReportTime(System.currentTimeMillis());
        response.setStartTime(System.currentTimeMillis());
        response.setSourceAddress(workerRuntime.getWorkerAddress());

        TransportUtils.ttReportInstanceStatus(response, workerRuntime.getServerDiscoveryService().getCurrentServerAddress(), workerRuntime.getTransporter());
    }

    protected void reportFinalStatusThenDestroy(WorkerRuntime workerRuntime, TaskTrackerReportInstanceStatusReq reportInstanceStatusReq) {
        String currentServerAddress = workerRuntime.getServerDiscoveryService().getCurrentServerAddress();
        // 最终状态需要可靠上报
        boolean serverAccepted = false;
        try {
            serverAccepted = TransportUtils.reliableTtReportInstanceStatus(reportInstanceStatusReq, currentServerAddress, workerRuntime.getTransporter());
        } catch (Exception e) {
            log.warn("[TaskTracker-{}] report finished status failed, req={}.", instanceId, reportInstanceStatusReq, e);
        }
        if (!serverAccepted) {
            if (++reportFailedCnt > MAX_REPORT_FAILED_THRESHOLD) {
                log.error("[TaskTracker-{}] try to report finished status(detail={}) lots of times but all failed, it's time to give up, so the process result will be dropped", instanceId, reportInstanceStatusReq);
                destroy();
            }
            return;
        }
        log.info("[TaskTracker-{}] report finished status(detail={}) success", instanceId, reportInstanceStatusReq);
        destroy();
    }
}