package com.purui.prims.headless.server.service.impl;

import com.alibaba.fastjson.JSON;
import com.purui.prims.common.pojo.PersistenceConf;
import com.purui.prims.common.util.ContextUtils;
import com.purui.prims.dscheduler.enums.Flag;
import com.purui.prims.dscheduler.enums.Priority;
import com.purui.prims.dscheduler.enums.ReleaseState;
import com.purui.prims.dscheduler.enums.TimeoutFlag;
import com.purui.prims.dscheduler.model.DSTaskRequest;
import com.purui.prims.dscheduler.model.DagData;
import com.purui.prims.dscheduler.model.DagNodeLocation;
import com.purui.prims.dscheduler.model.ProcessDefinition;
import com.purui.prims.dscheduler.model.ProcessScheduler;
import com.purui.prims.dscheduler.model.ProcessTaskRelation;
import com.purui.prims.dscheduler.model.Project;
import com.purui.prims.dscheduler.model.TaskDefinition;
import com.purui.prims.dscheduler.model.TaskDefinitionVO;
import com.purui.prims.dscheduler.model.TaskInstance;
import com.purui.prims.dscheduler.model.TaskLog;
import com.purui.prims.dscheduler.model.TaskRequest;
import com.purui.prims.dscheduler.service.DatasourceService;
import com.purui.prims.dscheduler.service.ProcessService;
import com.purui.prims.dscheduler.service.ProjectService;
import com.purui.prims.dscheduler.service.TaskService;
import com.purui.prims.headless.api.pojo.response.MetricPersistenceConfigResp;
import com.purui.prims.headless.core.cache.CacheManager;
import com.purui.prims.headless.server.schedule.MetricPersistenceTask;
import com.purui.prims.headless.server.service.SchedulerService;
import com.purui.prims.headless.server.utils.SchedulerConverter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 持久化任务调度服务
 *
 * @author charlie
 * @date 2024/7/17
 */
@Service
@Slf4j
@RequiredArgsConstructor
public class SchedulerServiceImpl implements SchedulerService {

    private static final String PROJECT_CODE_KEY = "PRIMS_PROJECT_CODE";
    private static final String PROJECT_SOURCE_KEY = "PRIMS_PROJECT_SOURCE";
    private static final String PROJECT_TARGET_KEY = "PRIMS_PROJECT_TARGET";
    private final ProcessService processService;
    private final ProjectService projectService;
    private final TaskService taskService;
    private final DatasourceService datasourceService;

    @Override
    public boolean releaseWorkflow(String processName, boolean publishFlag, PersistenceConf persistenceConf) {
        // 查找工作流
        Long projectCode = getProjectCode();
        ProcessDefinition process = processService.getProcessDefinitionInfo(projectCode, processName);
        if (process != null) {
            // 更新工作流
            process.setReleaseState(publishFlag ? ReleaseState.ONLINE : ReleaseState.OFFLINE);
            boolean result = processService.releaseProcessDefinition(process);
            if (result && publishFlag) {
                ProcessScheduler processScheduler = processService.getScheduler(projectCode, process.getCode());
                if (processScheduler == null) {
                    processService.scheduleNow(projectCode, process.getCode(), persistenceConf.getCron());
                } else {
                    processService.scheduleOnline(projectCode, processScheduler.getId());
                }
                return true;
            }
        }
        return false;
    }

    @Override
    public boolean pushWorkflow(String processName, List<DSTaskRequest> taskRequests, PersistenceConf persistenceConf) {
        // 查找工作流
        Long projectCode = getProjectCode();
        ProcessDefinition process = processService.getProcessDefinitionInfo(projectCode, processName);
        if (CollectionUtils.isNotEmpty(taskRequests)) {
            for (DSTaskRequest taskRequest : taskRequests) {
                initDefaultConfig(taskRequest);
            }
        }
        // 更新工作流Location
        List<TaskDefinition> taskDefinitionList = taskRequests.stream().map(task -> {
            TaskDefinition taskDefinition = SchedulerConverter.converterToDefinition(task);
            taskDefinition.setProjectCode(projectCode);
            return taskDefinition;
        }).collect(Collectors.toList());

        // 工作流不存在，需要新建
        if (process == null) {
            createNewProcess(projectCode, processName, taskDefinitionList, persistenceConf);
            return true;
        }
        if (process.getReleaseState() == ReleaseState.ONLINE) {
            log.error("工作流{}已经上线，先下线", processName);
            releaseWorkflow(processName, false, persistenceConf);
        }
        // 更新工作流
        updateProcessDefinition(process, taskDefinitionList, projectCode);
        // 发布并调度
        releaseWorkflow(processName, true, persistenceConf);

        return true;
    }

    // 查询详细日志
    @Override
    public TaskLog queryTaskLog(int instanceId) {
        return taskService.queryTaskLog(getProjectCode(), instanceId);
    }

    // 查询任务实例
    @Override
    public List<TaskInstance> getTaskInstances(String taskName, String startDate, String endDate) {
        return taskService.getTaskInstances(getProjectCode(), startDate, endDate, taskName);
    }

    @Override
    public Long getProjectCode() {
        CacheManager cacheManager = ContextUtils.getBean(CacheManager.class);
        Object projectCode = cacheManager.get(PROJECT_CODE_KEY);
        if (projectCode != null) {
            return Long.valueOf(projectCode.toString());
        } else {
            Project project = projectService.getProject();
            cacheManager.put(PROJECT_CODE_KEY, project.getCode());
            return project.getCode();
        }
    }

    @Override
    public boolean runTask(Long metricId) {
        MetricPersistenceTask metricPersistenceTask = ContextUtils.getBean(MetricPersistenceTask.class);
        if (metricPersistenceTask == null) {
            return false;
        }
        MetricPersistenceConfigResp configResp = metricPersistenceTask.getConfig(metricId);
        if (configResp == null) {
            log.error("未找到持久化配置{}", metricId);
            return false;
        }
        String taskName = configResp.getTableName();
        String processName = metricPersistenceTask.getProcessName(configResp.getRunCycleCode());
        ProcessDefinition process = processService.getProcessDefinitionInfo(getProjectCode(), processName);
        if (process == null) {
            log.error("未找到工作流{},重新创建", processName);
            // 工作流不存在，需要新建
            Long projectCode = getProjectCode();
            List<DSTaskRequest> taskRequests = metricPersistenceTask.buildTaskList(configResp, metricId);
            List<TaskDefinition> taskDefinitionList = taskRequests.stream().map(task -> {
                TaskDefinition taskDefinition = SchedulerConverter.converterToDefinition(task);
                taskDefinition.setProjectCode(projectCode);
                return taskDefinition;
            }).collect(Collectors.toList());
            createNewProcess(projectCode, processName, taskDefinitionList, configResp.getConf());
            process = processService.getProcessDefinitionInfo(getProjectCode(), processName);
            if (process == null) {
                log.error("工作流{}创建失败", processName);
                return false;
            }
        }
        Long taskCode = taskService.getTaskCode(getProjectCode(), taskName);
        if (taskCode == null || taskCode <= 0L) {
            log.error("未找到任务{},自动重新更新...", taskName);
            metricPersistenceTask.runPersistNow();
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                log.error(e.getMessage());
            }
        }
        return taskService.runTaskNow(getProjectCode(), process, taskCode);
    }

    @Override
    public int getSourceId() {
        CacheManager cacheManager = ContextUtils.getBean(CacheManager.class);
        Object source = cacheManager.get(PROJECT_SOURCE_KEY);
        if (source == null) {
            source = datasourceService.getSourceId();
            cacheManager.put(PROJECT_SOURCE_KEY, source);
        }
        return Integer.valueOf(source.toString());
    }

    @Override
    public int getTargetId() {
        CacheManager cacheManager = ContextUtils.getBean(CacheManager.class);
        Object target = cacheManager.get(PROJECT_TARGET_KEY);
        if (target == null) {
            target = datasourceService.getTargetId();
            cacheManager.put(PROJECT_TARGET_KEY, target);
        }
        return Integer.valueOf(target.toString());
    }

    @Override
    public String getSourceType() {
        return datasourceService.getSourceType();
    }

    @Override
    public String getTargetType() {
        return datasourceService.getTargetType();
    }

    // 创建一个新工作流
    private void createNewProcess(Long projectCode, String processName, List<TaskDefinition> taskDefinitionList, PersistenceConf persistenceConf) {
        List<DagNodeLocation> locations = new ArrayList<>();
        List<ProcessTaskRelation> processTaskRelationList = new ArrayList<>();
        if (CollectionUtils.isEmpty(taskDefinitionList)) {
            taskDefinitionList = new ArrayList<>();
        }
        createTaskDepends(projectCode, processTaskRelationList, taskDefinitionList, locations);

        processService.createOrUpdateProcessDefinition(
            projectCode,
            0L,
            processName,
            JSON.toJSONString(processTaskRelationList),
            JSON.toJSONString(taskDefinitionList),
            locations,
            false);

        // 同时发布
        releaseWorkflow(processName, true, persistenceConf);
    }

    private void createTaskDepends(Long projectCode, List<ProcessTaskRelation> processTaskRelationList,
                                   List<TaskDefinition> taskDefinitionList, List<DagNodeLocation> locations) {
        // 获取新任务code
        List<TaskDefinition> newTaskList = taskDefinitionList.stream().filter(taskDefinition -> taskDefinition.getCode() == null).collect(
            Collectors.toList());
        List<Long> taskCodes = taskService.genTaskCodes(projectCode, newTaskList.size());

        // 按顺序构建依赖关系（需要先排序）
        int i = 0;
        TaskDefinition preTask = null;
        for (TaskDefinition taskDefinition : taskDefinitionList) {
            if (taskDefinition.getCode() == null) {
                taskDefinition.setCode(taskCodes.get(i));
                i++;
                // 不改变旧关系，仅新增
                ProcessTaskRelation processTaskRelation = ProcessTaskRelation.generateProcessTaskRelation(preTask, taskDefinition);
                processTaskRelationList.add(processTaskRelation);
            }
            createLocation(taskDefinition.getCode(), locations);
            preTask = taskDefinition;
        }
    }

    private void createLocation(Long taskCode, List<DagNodeLocation> locations) {
        // 自动计算坐标(多个)
        if (CollectionUtils.isNotEmpty(locations)) {
            boolean matched = locations.stream().anyMatch(location -> location.getTaskCode() == taskCode);
            // 获取最大的 x y 坐标
            long xMax =
                locations.stream().mapToLong(DagNodeLocation::getX).max().getAsLong();

            long yMax =
                locations.stream().mapToLong(DagNodeLocation::getY).max().getAsLong();

            // if not matched, add a new location
            if (!matched) {
                // 随机出一个 x y 坐标
                DagNodeLocation dagNodeLocation = new DagNodeLocation();
                dagNodeLocation.setTaskCode(taskCode);
                dagNodeLocation.setX(RandomUtils.nextLong(xMax - 200, xMax));
                dagNodeLocation.setY(RandomUtils.nextLong(yMax - 200, yMax - 50));
                locations.add(dagNodeLocation);
            }
        } else {
            // 随机出一个 x y 坐标
            DagNodeLocation dagNodeLocation = new DagNodeLocation();
            dagNodeLocation.setTaskCode(taskCode);
            dagNodeLocation.setX(RandomUtils.nextLong(200, 800));
            dagNodeLocation.setY(RandomUtils.nextLong(600, 800));
            locations.add(dagNodeLocation);
        }
    }

    // 更新工作流（多个任务）
    private void updateProcessDefinition(ProcessDefinition process, List<TaskDefinition> taskDefinitionList, long projectCode) {
        DagData dagData = processService.getProcessDefinitionInfo(projectCode, process.getCode());
        if (dagData == null) {
            log.error("工作流不存在！");
            throw new RuntimeException("工作流不存在！");
        }
        List<TaskDefinition> oldTaskList = dagData.getTaskDefinitionList();
        if (CollectionUtils.isEmpty(oldTaskList)) {
            log.error("工作流任务不存在！");
            throw new RuntimeException("工作流任务不存在！");
        }
        // 匹配已有数据
        List<TaskDefinition> newTaskList = new ArrayList<>();
        newTaskList.addAll(oldTaskList);
        for (TaskDefinition taskDefinition : taskDefinitionList) {
            TaskDefinition finalTaskDefinition = taskDefinition;
            TaskDefinition oldTask =
                newTaskList.stream().filter(task -> task.getName().equals(finalTaskDefinition.getName())).findFirst().orElse(null);
            if (oldTask != null) {
                // 仅更新关键数据
                oldTask.setTaskParams(taskDefinition.getTaskParams());
            } else {
                newTaskList.add(taskDefinition);
            }
        }
        List<ProcessTaskRelation> processTaskRelationList = dagData.getProcessTaskRelationList();
        List<DagNodeLocation> locations = JSON.parseArray(process.getLocations(), DagNodeLocation.class);
        // 重新构建依赖关系
        createTaskDepends(process.getProjectCode(), processTaskRelationList, newTaskList, locations);

        processService.createOrUpdateProcessDefinition(
            process.getProjectCode(),
            process.getCode(),
            process.getName(),
            JSON.toJSONString(processTaskRelationList),
            JSON.toJSONString(newTaskList),
            locations,
            true);

        log.info(
            "工作流{}-更新位置{}",
            process.getName(),
            locations);
    }

    private void updateProcessDefinition(ProcessDefinition process, Long taskCode, TaskRequest task, long projectCode) {
        task.setCode(taskCode);
        TaskDefinition newTask = SchedulerConverter.converterToDefinition(task);
        newTask.setProjectCode(projectCode);
        updateProcessDefinition(process, Arrays.asList(newTask), projectCode);
    }

    // 加载默认参数
    private void initDefaultConfig(TaskRequest request) {
        // 默认参数
        if (StringUtils.isEmpty(request.getTaskPriority())) {
            request.setTaskPriority(Priority.MEDIUM.getDescp().toUpperCase());
        }
        if (StringUtils.isEmpty(request.getTimeoutFlag())) {
            request.setTimeoutFlag(TimeoutFlag.OPEN.getDesc().toUpperCase());
        }
        if (StringUtils.isEmpty(request.getTimeoutNotifyStrategy())) {
            request.setTimeoutNotifyStrategy(TimeoutFlag.OPEN.getDesc().toUpperCase());
        }
        if (StringUtils.isEmpty(request.getIsCache())) {
            request.setIsCache(Flag.NO.getDescp().toUpperCase());
        }
        if (StringUtils.isEmpty(request.getFlag())) {
            request.setFlag(Flag.YES.getDescp().toUpperCase());
        }
    }

    /**
     * 新增修改Task任务,暂不使用独立的Task创建
     */
    @Deprecated
    public boolean pushUpdateTask(
        long projectCode, long processCode, long taskCode, DSTaskRequest dsTaskRequest) {
        TaskDefinitionVO taskDefinition = taskService.getTaskDefinition(projectCode, taskCode);
        if (taskDefinition == null) {
            throw new RuntimeException("任务不存在!");
        }

        DagData dagData = processService.getProcessDefinitionInfo(projectCode, processCode);
        if (dagData == null) {
            throw new RuntimeException("工作流定义不存在!");
        }

        ProcessDefinition process = dagData.getProcessDefinition();
        if (process == null) {
            throw new RuntimeException("工作流不存在!");
        }

        if (process.getReleaseState() == ReleaseState.ONLINE) {
            throw new RuntimeException("工作流已经上线，不能修改!");
        }
        /**
         TaskRequest taskRequest = SchedulerConverter.converter(taskDefinition);
         taskRequest.setTaskParams(dsTaskRequest.getTaskParams());
         initDefaultConfig(taskRequest);
         */
        taskDefinition.setProcessDefinitionCode(process.getCode());
        taskDefinition.setProcessDefinitionName(process.getName());
        String taskDefinitionJsonObj = JSON.toJSONString(taskDefinition);
        Long updatedTaskDefinition = taskService.updateTaskDefinition(
            projectCode, taskCode, dsTaskRequest.getUpstreamCodes(), taskDefinitionJsonObj);

        if (updatedTaskDefinition != null && updatedTaskDefinition > 0) {
            log.info("修改成功!");
            return true;
        }
        log.error("修改失败!");
        return false;
    }
}