package com.purui.prims.dscheduler.service;

import com.alibaba.fastjson.JSON;
import com.purui.prims.common.config.DolphinSchedulerConfig;
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.result.PageInfo;
import com.purui.prims.dscheduler.utils.ParamUtil;
import com.purui.prims.headless.api.pojo.enums.DolphineSchedulerType;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;

/**
 * 工作流定义
 *
 * @author charlie
 * @date 2024/7/10
 */
@Service
public class ProcessService extends AbstractSchedulerService {

    public ProcessService(DolphinSchedulerConfig config,
                          RestTemplate restTemplate) {
        super(config, restTemplate);
    }

    public List<ProcessDefinition> getProcessDefinition(Long projectCode, String processName) {

        PageInfo data =
            runSchedulerX(projectCode, 0L, DolphineSchedulerType.WORKFLOW_LIST, ParamUtil.getPageParams(processName), PageInfo.class);
        List<ProcessDefinition> lists = new ArrayList<>();
        if (data == null || data.getTotalList() == null) {
            return lists;
        }

        lists = JSON.parseArray(JSON.toJSONString(data.getTotalList()), ProcessDefinition.class);
        return lists;
    }

    public ProcessDefinition getProcessDefinitionInfo(Long projectCode, String processName) {
        List<ProcessDefinition> lists = getProcessDefinition(projectCode, processName);
        Optional<ProcessDefinition> processDefinition = lists.stream()
                                                             .filter(list -> list.getName().equalsIgnoreCase(processName))
                                                             .findFirst();
        return processDefinition.orElse(null);
    }

    public DagData getProcessDefinitionInfo(Long projectCode, Long processCode) {
        return runSchedulerX(projectCode, processCode, DolphineSchedulerType.WORKFLOW_DETAIL, null, DagData.class);
    }

    // 创建或更新流程
    public ProcessDefinition createOrUpdateProcessDefinition(
        Long projectCode,
        Long processCode,
        String processName,
        String taskRelationJson,
        String taskDefinitionJson,
        List<DagNodeLocation> locations,
        boolean isModify) {

        Map<String, Object> params = new HashMap<>();
        params.put("name", processName);
        params.put("description", "系统添加");
        params.put("globalParams", new ArrayList<>());
        params.put("locations", JSON.toJSONString(locations));
        params.put("taskRelationJson", taskRelationJson);
        params.put("taskDefinitionJson", taskDefinitionJson);
        params.put("executionType", "PARALLEL");

        ProcessDefinition processDefinition = null;
        if (isModify) {
            processDefinition =
                runSchedulerX(projectCode, processCode, DolphineSchedulerType.WORKFLOW_UPDATE, params, ProcessDefinition.class);
        } else {
            processDefinition = runSchedulerX(projectCode, 0L, DolphineSchedulerType.WORKFLOW_CREATE, params, ProcessDefinition.class);
        }
        return processDefinition;
    }

    // 发布流程
    public boolean releaseProcessDefinition(ProcessDefinition process) {
        Map<String, Object> params = new HashMap<>();
        params.put("name", process.getName());
        params.put("releaseState", process.getReleaseState().getDescp().toUpperCase(Locale.ROOT));
        return runSchedulerX(process.getProjectCode(), process.getCode(), DolphineSchedulerType.WORKFLOW_RELEASE, params,
            Boolean.class);
    }

    // 删除流程
    public boolean removeProcessDefinition(Long projectCode, Long processCode) {
        return runSchedulerX(projectCode, processCode, DolphineSchedulerType.WORKFLOW_REMOVE, null, Boolean.class);
    }

    // 创建定时任务
    public void scheduleNow(Long projectCode, Long processCode, String cron) {
        if (StringUtils.isEmpty(cron)) {
            // 没有设定任务运行时间，则不创建定时任务
            return;
        }
        Map<String, Object> params = new HashMap<>();
        params.put("processDefinitionCode", processCode);
        params.put("failureStrategy", "CONTINUE");
        params.put("warningType", "NONE");
        params.put("processInstancePriority", "MEDIUM");
        params.put("workerGroup", "default");
        params.put("warningGroupId", "0");
        params.put("tenantCode", "default");
        Map<String, Object> schedule = new HashMap<>();
        schedule.put("crontab", cron);
        schedule.put("startTime", "2024-08-01 00:00:00");
        schedule.put("endTime", "2095-08-01 23:59:59");
        schedule.put("timezoneId", "Asia/Shanghai");

        params.put("schedule", JSON.toJSONString(schedule));
        runSchedulerX(projectCode, processCode, DolphineSchedulerType.SCHEDULE_CREATE, params, Object.class);
    }

    // 停止调度
    public boolean scheduleOffline(Long projectCode, Long id) {
        return runSchedulerX(projectCode, id, DolphineSchedulerType.SCHEDULE_OFFLINE, null, Boolean.class);
    }

    // 启动调度
    public boolean scheduleOnline(Long projectCode, Long id) {
        return runSchedulerX(projectCode, id, DolphineSchedulerType.SCHEDULE_ONLINE, null, Boolean.class);
    }

    // 是否已经有对应的调度器
    public ProcessScheduler getScheduler(Long projectCode, Long processCode) {
        List<ProcessScheduler> list = runSchedulerListX(projectCode, 0L, DolphineSchedulerType.SCHEDULE_LIST, null,
            ProcessScheduler.class);
        if (list == null) {
            return null;
        }
        return list.stream().filter(schedule -> schedule.getProcessDefinitionCode().equals(processCode)).findFirst().orElse(null);
    }
}
