/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.dolphinscheduler.api.service.impl;

import static java.util.stream.Collectors.toSet;
import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.TASK_TYPE_CONDITIONS;
import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.TASK_TYPE_DEPENDENT;
import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.TASK_TYPE_SUB_PROCESS;

import org.apache.dolphinscheduler.api.dto.taskRelation.TaskRelationCreateRequest;
import org.apache.dolphinscheduler.api.dto.taskRelation.TaskRelationFilterRequest;
import org.apache.dolphinscheduler.api.dto.taskRelation.TaskRelationUpdateUpstreamRequest;
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.exceptions.ServiceException;
import org.apache.dolphinscheduler.api.service.ProcessTaskRelationService;
import org.apache.dolphinscheduler.api.service.ProjectService;
import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.common.enums.ConditionType;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinitionLog;
import org.apache.dolphinscheduler.dao.entity.ProcessTaskRelation;
import org.apache.dolphinscheduler.dao.entity.ProcessTaskRelationLog;
import org.apache.dolphinscheduler.dao.entity.Project;
import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
import org.apache.dolphinscheduler.dao.entity.TaskDefinitionLog;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionLogMapper;
import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.ProcessTaskRelationLogMapper;
import org.apache.dolphinscheduler.dao.mapper.ProcessTaskRelationMapper;
import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionLogMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.service.process.ProcessService;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;

import lombok.extern.slf4j.Slf4j;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.google.common.collect.Lists;

/**
 * process task relation service impl
 */
@Service
@Slf4j
public class ProcessTaskRelationServiceImpl extends BaseServiceImpl implements ProcessTaskRelationService {

    @Autowired
    private ProjectMapper projectMapper;

    @Autowired
    private ProjectService projectService;

    @Autowired
    private ProcessTaskRelationMapper processTaskRelationMapper;

    @Autowired
    private TaskDefinitionLogMapper taskDefinitionLogMapper;

    @Autowired
    private TaskDefinitionMapper taskDefinitionMapper;

    @Autowired
    private ProcessDefinitionMapper processDefinitionMapper;

    @Autowired
    private ProcessDefinitionLogMapper processDefinitionLogMapper;

    @Autowired
    private ProcessService processService;

    @Autowired
    private ProcessTaskRelationLogMapper processTaskRelationLogMapper;

    /**
     * create process task relation
     *
     * @param loginUser             login user
     * @param projectCode           project code
     * @param processDefinitionCode processDefinitionCode
     * @param preTaskCode           preTaskCode
     * @param postTaskCode          postTaskCode
     * @return create result code
     */
    @Transactional
    @Override
    public Map<String, Object> createProcessTaskRelation(User loginUser, long projectCode, long processDefinitionCode,
                                                         long preTaskCode, long postTaskCode) {
        Project project = projectMapper.queryByCode(projectCode);
        // check user access for project
        Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode, null);
        if (result.get(Constants.STATUS) != Status.SUCCESS) {
            return result;
        }
        ProcessDefinition processDefinition = processDefinitionMapper.queryByCode(processDefinitionCode);
        if (processDefinition == null) {
            log.error("Process definition does not exist, processCode:{}.", processDefinitionCode);
            putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST, String.valueOf(processDefinitionCode));
            return result;
        }
        if (processDefinition.getProjectCode() != projectCode) {
            log.error("Process definition's project does not match project {}.", projectCode);
            putMsg(result, Status.PROJECT_PROCESS_NOT_MATCH);
            return result;
        }
        updateProcessDefiniteVersion(loginUser, result, processDefinition);
        List<ProcessTaskRelation> processTaskRelationList =
                processTaskRelationMapper.queryByProcessCode(processDefinitionCode);
        List<ProcessTaskRelation> processTaskRelations = Lists.newArrayList(processTaskRelationList);
        if (!processTaskRelations.isEmpty()) {
            Map<Long, ProcessTaskRelation> preTaskCodeMap =
                    processTaskRelations.stream().filter(r -> r.getPostTaskCode() == postTaskCode)
                            .collect(Collectors.toMap(ProcessTaskRelation::getPreTaskCode,
                                    processTaskRelation -> processTaskRelation));
            if (!preTaskCodeMap.isEmpty()) {
                if (preTaskCodeMap.containsKey(preTaskCode) || (!preTaskCodeMap.containsKey(0L) && preTaskCode == 0L)) {
                    putMsg(result, Status.PROCESS_TASK_RELATION_EXIST, String.valueOf(processDefinitionCode));
                    return result;
                }
                if (preTaskCodeMap.containsKey(0L) && preTaskCode != 0L) {
                    // delete no upstream
                    processTaskRelations.remove(preTaskCodeMap.get(0L));
                }
            }
        }
        TaskDefinition postTaskDefinition = taskDefinitionMapper.queryByCode(postTaskCode);
        ProcessTaskRelation processTaskRelation = setRelation(processDefinition, postTaskDefinition);
        if (preTaskCode != 0L) {
            TaskDefinition preTaskDefinition = taskDefinitionMapper.queryByCode(preTaskCode);
            List<ProcessTaskRelation> upstreamTaskRelationList = processTaskRelations.stream()
                    .filter(r -> r.getPostTaskCode() == preTaskCode).collect(Collectors.toList());
            // upstream is or not exist
            if (upstreamTaskRelationList.isEmpty()) {
                ProcessTaskRelation preProcessTaskRelation = setRelation(processDefinition, preTaskDefinition);
                preProcessTaskRelation.setPreTaskCode(0L);
                preProcessTaskRelation.setPreTaskVersion(0);
                processTaskRelations.add(preProcessTaskRelation);
            }
            processTaskRelation.setPreTaskCode(preTaskDefinition.getCode());
            processTaskRelation.setPreTaskVersion(preTaskDefinition.getVersion());
        } else {
            processTaskRelation.setPreTaskCode(0L);
            processTaskRelation.setPreTaskVersion(0);
        }
        processTaskRelations.add(processTaskRelation);
        updateRelation(loginUser, result, processDefinition, processTaskRelations);
        return result;
    }

    private ProcessTaskRelationLog persist2ProcessTaskRelationLog(User user, ProcessTaskRelation processTaskRelation) {
        ProcessTaskRelationLog processTaskRelationLog = new ProcessTaskRelationLog(processTaskRelation);
        processTaskRelationLog.setOperator(user.getId());
        processTaskRelationLog.setOperateTime(new Date());
        int result = processTaskRelationLogMapper.insert(processTaskRelationLog);
        if (result <= 0) {
            throw new ServiceException(Status.CREATE_PROCESS_TASK_RELATION_LOG_ERROR,
                    processTaskRelationLog.getPreTaskCode(), processTaskRelationLog.getPostTaskCode());
        }
        return processTaskRelationLog;
    }

    private List<ProcessTaskRelationLog> batchPersist2ProcessTaskRelationLog(User user,
                                                                             List<ProcessTaskRelation> processTaskRelations) {
        Date now = new Date();
        List<ProcessTaskRelationLog> processTaskRelationLogs = new ArrayList<>();

        for (ProcessTaskRelation processTaskRelation : processTaskRelations) {
            ProcessTaskRelationLog processTaskRelationLog = new ProcessTaskRelationLog(processTaskRelation);
            processTaskRelationLog.setOperator(user.getId());
            processTaskRelationLog.setOperateTime(now);
            processTaskRelationLogs.add(processTaskRelationLog);
        }
        int result = processTaskRelationLogMapper.batchInsert(processTaskRelationLogs);
        if (result != processTaskRelationLogs.size()) {
            throw new ServiceException(Status.CREATE_PROCESS_TASK_RELATION_LOG_ERROR);
        }
        return processTaskRelationLogs;
    }

    private void updateVersions(ProcessTaskRelation processTaskRelation) {
        // workflow
        ProcessDefinition processDefinition =
                processDefinitionMapper.queryByCode(processTaskRelation.getProcessDefinitionCode());
        processTaskRelation.setProcessDefinitionVersion(processDefinition.getVersion());

        // tasks
        TaskDefinition preTaskDefinition = taskDefinitionMapper.queryByCode(processTaskRelation.getPreTaskCode());
        processTaskRelation.setPreTaskVersion(preTaskDefinition.getVersion());
        TaskDefinition postTaskDefinition = taskDefinitionMapper.queryByCode(processTaskRelation.getPostTaskCode());
        processTaskRelation.setPostTaskVersion(postTaskDefinition.getVersion());
    }

    /**
     * create resource process task relation
     *
     * @param loginUser login user
     * @param taskRelationCreateRequest project code
     * @return ProcessTaskRelation object
     */
    @Override
    @Transactional
    public ProcessTaskRelation createProcessTaskRelationV2(User loginUser,
                                                           TaskRelationCreateRequest taskRelationCreateRequest) {
        ProcessTaskRelation processTaskRelation = taskRelationCreateRequest.convert2ProcessTaskRelation();
        ProcessDefinition processDefinition =
                processDefinitionMapper.queryByCode(processTaskRelation.getProcessDefinitionCode());
        if (processDefinition == null) {
            throw new ServiceException(Status.PROCESS_DEFINE_NOT_EXIST,
                    String.valueOf(processTaskRelation.getProcessDefinitionCode()));
        }
        if (processTaskRelation.getProjectCode() == 0) {
            processTaskRelation.setProjectCode(processDefinition.getProjectCode());
        }
        Project project = projectMapper.queryByCode(processTaskRelation.getProjectCode());
        projectService.checkProjectAndAuthThrowException(loginUser, project, null);

        // persistence process task relation and process task relation log to database
        this.updateVersions(processTaskRelation);
        int insert = processTaskRelationMapper.insert(processTaskRelation);
        if (insert <= 0) {
            throw new ServiceException(Status.CREATE_PROCESS_TASK_RELATION_ERROR, processTaskRelation.getPreTaskCode(),
                    processTaskRelation.getPostTaskCode());
        }
        this.persist2ProcessTaskRelationLog(loginUser, processTaskRelation);

        return processTaskRelation;
    }

    private ProcessTaskRelation setRelation(ProcessDefinition processDefinition, TaskDefinition taskDefinition) {
        Date now = new Date();
        ProcessTaskRelation processTaskRelation = new ProcessTaskRelation();
        processTaskRelation.setProjectCode(processDefinition.getProjectCode());
        processTaskRelation.setProcessDefinitionCode(processDefinition.getCode());
        processTaskRelation.setProcessDefinitionVersion(processDefinition.getVersion());
        processTaskRelation.setPostTaskCode(taskDefinition.getCode());
        processTaskRelation.setPostTaskVersion(taskDefinition.getVersion());
        processTaskRelation.setConditionType(ConditionType.NONE);
        processTaskRelation.setConditionParams("{}");
        processTaskRelation.setCreateTime(now);
        processTaskRelation.setUpdateTime(now);
        return processTaskRelation;
    }

    private void updateProcessDefiniteVersion(User loginUser, Map<String, Object> result,
                                              ProcessDefinition processDefinition) {
        int insertVersion = processService.saveProcessDefine(loginUser, processDefinition, Boolean.TRUE, Boolean.TRUE);
        if (insertVersion <= 0) {
            log.error("Update process definition error, projectCode:{}, processDefinitionCode:{}.",
                    processDefinition.getProjectCode(), processDefinition.getCode());
            putMsg(result, Status.UPDATE_PROCESS_DEFINITION_ERROR);
            throw new ServiceException(Status.UPDATE_PROCESS_DEFINITION_ERROR);
        } else
            log.info(
                    "Update process definition complete, new version is {}, projectCode:{}, processDefinitionCode:{}.",
                    insertVersion, processDefinition.getProjectCode(), processDefinition.getCode());
        processDefinition.setVersion(insertVersion);
    }

    /**
     * delete process task relation
     *
     * @param loginUser             login user
     * @param projectCode           project code
     * @param processDefinitionCode process definition code
     * @param taskCode              the post task code
     * @return delete result code
     */
    @Transactional
    @Override
    public Map<String, Object> deleteTaskProcessRelation(User loginUser, long projectCode, long processDefinitionCode,
                                                         long taskCode) {
        Project project = projectMapper.queryByCode(projectCode);
        // check user access for project
        Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode, null);
        if (result.get(Constants.STATUS) != Status.SUCCESS) {
            return result;
        }
        if (taskCode == 0) {
            log.error(
                    "Delete task process relation error due to parameter taskCode is 0, projectCode:{}, processDefinitionCode:{}.",
                    projectCode, processDefinitionCode);
            putMsg(result, Status.DELETE_TASK_PROCESS_RELATION_ERROR);
            return result;
        }
        ProcessDefinition processDefinition = processDefinitionMapper.queryByCode(processDefinitionCode);
        if (processDefinition == null) {
            log.error("Process definition does not exist, processDefinitionCode:{}.", processDefinitionCode);
            putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST, String.valueOf(processDefinitionCode));
            return result;
        }
        TaskDefinition taskDefinition = taskDefinitionMapper.queryByCode(taskCode);
        if (null == taskDefinition) {
            log.error("Task definition does not exist, taskDefinitionCode:{}.", taskCode);
            putMsg(result, Status.TASK_DEFINE_NOT_EXIST, String.valueOf(taskCode));
            return result;
        }
        List<ProcessTaskRelation> processTaskRelations =
                processTaskRelationMapper.queryByProcessCode(processDefinitionCode);
        List<ProcessTaskRelation> processTaskRelationList = Lists.newArrayList(processTaskRelations);
        if (CollectionUtils.isEmpty(processTaskRelationList)) {
            log.error("Process task relations are empty, projectCode:{}, processDefinitionCode:{}.", projectCode,
                    processDefinitionCode);
            putMsg(result, Status.DATA_IS_NULL, "processTaskRelationList");
            return result;
        }
        List<Long> downstreamList = Lists.newArrayList();
        for (ProcessTaskRelation processTaskRelation : processTaskRelations) {
            if (processTaskRelation.getPreTaskCode() == taskCode) {
                downstreamList.add(processTaskRelation.getPostTaskCode());
            }
            if (processTaskRelation.getPostTaskCode() == taskCode) {
                processTaskRelationList.remove(processTaskRelation);
            }
        }
        if (CollectionUtils.isNotEmpty(downstreamList)) {
            String downstream = StringUtils.join(downstreamList, ",");
            log.warn(
                    "Relation can not be deleted because task has downstream tasks:[{}], projectCode:{}, processDefinitionCode:{}, taskDefinitionCode:{}.",
                    downstream, projectCode, processDefinitionCode, taskCode);
            putMsg(result, Status.TASK_HAS_DOWNSTREAM, downstream);
            return result;
        }
        updateProcessDefiniteVersion(loginUser, result, processDefinition);
        updateRelation(loginUser, result, processDefinition, processTaskRelationList);
        if (TASK_TYPE_CONDITIONS.equals(taskDefinition.getTaskType())
                || TASK_TYPE_DEPENDENT.equals(taskDefinition.getTaskType())
                || TASK_TYPE_SUB_PROCESS.equals(taskDefinition.getTaskType())) {
            int deleteTaskDefinition = taskDefinitionMapper.deleteByCode(taskCode);
            if (0 == deleteTaskDefinition) {
                log.error("Delete task definition error, taskDefinitionCode:{}.", taskCode);
                putMsg(result, Status.DELETE_TASK_DEFINE_BY_CODE_ERROR);
                throw new ServiceException(Status.DELETE_TASK_DEFINE_BY_CODE_ERROR);
            } else
                log.info("Delete {} type task definition complete, taskDefinitionCode:{}.",
                        taskDefinition.getTaskType(), taskCode);
        }
        putMsg(result, Status.SUCCESS);
        return result;
    }

    /**
     * delete process task relation, will delete exists relation preTaskCode -> postTaskCode, throw error if not exists
     *
     * @param loginUser login user
     * @param preTaskCode relation upstream code
     * @param postTaskCode relation downstream code
     */
    @Override
    @Transactional
    public void deleteTaskProcessRelationV2(User loginUser,
                                            long preTaskCode,
                                            long postTaskCode) {
        ProcessTaskRelation processTaskRelation =
                new TaskRelationFilterRequest(preTaskCode, postTaskCode).convert2TaskDefinition();

        Page<ProcessTaskRelation> page =
                new Page<>(new TaskRelationFilterRequest(preTaskCode, postTaskCode).getPageNo(),
                        new TaskRelationFilterRequest(preTaskCode, postTaskCode).getPageSize());
        IPage<ProcessTaskRelation> processTaskRelationIPage =
                processTaskRelationMapper.filterProcessTaskRelation(page, processTaskRelation);

        List<ProcessTaskRelation> processTaskRelations = processTaskRelationIPage.getRecords();
        if (processTaskRelations.size() != 1) {
            throw new ServiceException(Status.PROCESS_TASK_RELATION_NOT_EXPECT, 1, processTaskRelations.size());
        }

        ProcessTaskRelation processTaskRelationDb = processTaskRelations.get(0);
        Project project = projectMapper.queryByCode(processTaskRelationDb.getProjectCode());
        projectService.checkProjectAndAuthThrowException(loginUser, project, null);
        processTaskRelationMapper.deleteById(processTaskRelationDb.getId());
    }

    /**
     * delete process task relation, will delete exists relation upstream -> downstream, throw error if not exists
     *
     * @param loginUser login user
     * @param taskCode relation upstream code
     * @param needSyncDag needSyncDag
     * @param taskRelationUpdateUpstreamRequest relation downstream code
     */
    @Override
    @Transactional
    public List<ProcessTaskRelation> updateUpstreamTaskDefinitionWithSyncDag(User loginUser,
                                                                             long taskCode,
                                                                             Boolean needSyncDag,
                                                                             TaskRelationUpdateUpstreamRequest taskRelationUpdateUpstreamRequest) {
        TaskDefinition downstreamTask = taskDefinitionMapper.queryByCode(taskCode);
        if (downstreamTask == null) {
            throw new ServiceException(Status.TASK_DEFINE_NOT_EXIST, taskCode);
        }
        List<Long> upstreamTaskCodes = taskRelationUpdateUpstreamRequest.getUpstreams();

        ProcessTaskRelation processTaskRelation = new ProcessTaskRelation();
        processTaskRelation.setPostTaskCode(taskCode);

        Page<ProcessTaskRelation> page = new Page<>(taskRelationUpdateUpstreamRequest.getPageNo(),
                taskRelationUpdateUpstreamRequest.getPageSize());
        IPage<ProcessTaskRelation> processTaskRelationExistsIPage =
                processTaskRelationMapper.filterProcessTaskRelation(page, processTaskRelation);
        List<ProcessTaskRelation> processTaskRelationExists = processTaskRelationExistsIPage.getRecords();

        ProcessDefinition processDefinition = null;
        if (CollectionUtils.isNotEmpty(processTaskRelationExists)) {
            processDefinition =
                    processDefinitionMapper.queryByCode(processTaskRelationExists.get(0).getProcessDefinitionCode());
        } else if (taskRelationUpdateUpstreamRequest.getWorkflowCode() != 0L) {
            processDefinition =
                    processDefinitionMapper.queryByCode(taskRelationUpdateUpstreamRequest.getWorkflowCode());
        }
        if (processDefinition == null) {
            throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR,
                    taskRelationUpdateUpstreamRequest.toString());
        }
        processDefinition.setUpdateTime(new Date());
        int insertVersion = processDefinition.getVersion();
        if (needSyncDag) {
            insertVersion =
                    this.saveProcessDefine(loginUser, processDefinition);
            if (insertVersion <= 0) {
                throw new ServiceException(Status.UPDATE_PROCESS_DEFINITION_ERROR);
            }
        }
        // get new relation to create and out of date relation to delete
        List<Long> taskCodeCreates = upstreamTaskCodes
                .stream()
                .filter(upstreamTaskCode -> processTaskRelationExists.stream().noneMatch(
                        processTaskRelationExist -> processTaskRelationExist.getPreTaskCode() == upstreamTaskCode))
                .collect(Collectors.toList());
        List<Integer> taskCodeDeletes = processTaskRelationExists.stream()
                .filter(ptr -> !upstreamTaskCodes.contains(ptr.getPreTaskCode()))
                .map(ProcessTaskRelation::getId)
                .collect(Collectors.toList());

        // delete relation not exists
        if (CollectionUtils.isNotEmpty(taskCodeDeletes)) {
            int delete = processTaskRelationMapper.deleteBatchIds(taskCodeDeletes);
            if (delete != taskCodeDeletes.size()) {
                throw new ServiceException(Status.PROCESS_TASK_RELATION_BATCH_DELETE_ERROR, taskCodeDeletes);
            }
        }

        // create relation not exists
        List<ProcessTaskRelation> processTaskRelations = new ArrayList<>();
        for (long createCode : taskCodeCreates) {
            long upstreamCode = 0L;
            int version = 0;
            if (createCode != 0L) {
                // 0 for DAG root, should not, it may already exists and skip to create anymore
                TaskDefinition upstreamTask = taskDefinitionMapper.queryByCode(createCode);
                if (upstreamTask == null) {
                    throw new ServiceException(Status.TASK_DEFINE_NOT_EXIST, createCode);
                }
                upstreamCode = upstreamTask.getCode();
                version = upstreamTask.getVersion();
            }
            ProcessTaskRelation processTaskRelationCreate =
                    new ProcessTaskRelation(null, processDefinition.getVersion(), downstreamTask.getProjectCode(),
                            processDefinition.getCode(), upstreamCode, version,
                            downstreamTask.getCode(), downstreamTask.getVersion(), null, null);
            processTaskRelations.add(processTaskRelationCreate);
        }
        int batchInsert = processTaskRelationMapper.batchInsert(processTaskRelations);
        if (batchInsert != processTaskRelations.size()) {
            throw new ServiceException(Status.PROCESS_TASK_RELATION_BATCH_CREATE_ERROR, taskCodeCreates);
        }

        // batch sync to process task relation log
        int saveTaskRelationResult = saveTaskRelation(loginUser, processDefinition, insertVersion);
        if (saveTaskRelationResult != Constants.EXIT_CODE_SUCCESS) {
            log.error("Save process task relations error, projectCode:{}, processCode:{}, processVersion:{}.",
                    processDefinition.getProjectCode(), processDefinition.getCode(), insertVersion);
            throw new ServiceException(Status.CREATE_PROCESS_TASK_RELATION_ERROR);
        }
        log.info("Save process task relations complete, projectCode:{}, processCode:{}, processVersion:{}.",
                processDefinition.getProjectCode(), processDefinition.getCode(), insertVersion);
        processTaskRelations.get(0).setProcessDefinitionVersion(insertVersion);
        return processTaskRelations;
    }

    public int saveTaskRelation(User loginUser, ProcessDefinition processDefinition,
                                int processDefinitionVersion) {
        long projectCode = processDefinition.getProjectCode();
        long processDefinitionCode = processDefinition.getCode();
        List<ProcessTaskRelation> taskRelations =
                processTaskRelationMapper.queryByProcessCode(processDefinitionCode);
        List<ProcessTaskRelationLog> taskRelationList =
                taskRelations.stream().map(ProcessTaskRelationLog::new).collect(Collectors.toList());

        List<Long> taskCodeList =
                taskRelations.stream().map(ProcessTaskRelation::getPostTaskCode).collect(Collectors.toList());
        List<TaskDefinition> taskDefinitions = taskDefinitionMapper.queryByCodeList(taskCodeList);
        List<TaskDefinitionLog> taskDefinitionLogs =
                taskDefinitions.stream().map(TaskDefinitionLog::new).collect(Collectors.toList());

        if (taskRelationList.isEmpty()) {
            return Constants.EXIT_CODE_SUCCESS;
        }
        Map<Long, TaskDefinitionLog> taskDefinitionLogMap = null;
        if (org.apache.commons.collections.CollectionUtils.isNotEmpty(taskDefinitionLogs)) {
            taskDefinitionLogMap = taskDefinitionLogs
                    .stream()
                    .collect(Collectors.toMap(TaskDefinition::getCode, taskDefinitionLog -> taskDefinitionLog));
        }
        Date now = new Date();
        for (ProcessTaskRelationLog processTaskRelationLog : taskRelationList) {
            processTaskRelationLog.setProjectCode(projectCode);
            processTaskRelationLog.setProcessDefinitionCode(processDefinitionCode);
            processTaskRelationLog.setProcessDefinitionVersion(processDefinitionVersion);
            if (taskDefinitionLogMap != null) {
                TaskDefinitionLog preTaskDefinitionLog =
                        taskDefinitionLogMap.get(processTaskRelationLog.getPreTaskCode());
                if (preTaskDefinitionLog != null) {
                    processTaskRelationLog.setPreTaskVersion(preTaskDefinitionLog.getVersion());
                }
                TaskDefinitionLog postTaskDefinitionLog =
                        taskDefinitionLogMap.get(processTaskRelationLog.getPostTaskCode());
                if (postTaskDefinitionLog != null) {
                    processTaskRelationLog.setPostTaskVersion(postTaskDefinitionLog.getVersion());
                }
            }
            processTaskRelationLog.setCreateTime(now);
            processTaskRelationLog.setUpdateTime(now);
            processTaskRelationLog.setOperator(loginUser.getId());
            processTaskRelationLog.setOperateTime(now);
        }
        if (CollectionUtils.isNotEmpty(taskRelations)) {
            Set<Integer> processTaskRelationSet =
                    taskRelations.stream().map(ProcessTaskRelation::hashCode).collect(toSet());
            Set<Integer> taskRelationSet =
                    taskRelationList.stream().map(ProcessTaskRelationLog::hashCode).collect(toSet());
            boolean isSame = org.apache.commons.collections.CollectionUtils.isEqualCollection(processTaskRelationSet,
                    taskRelationSet);
            if (isSame) {
                return Constants.EXIT_CODE_SUCCESS;
            }
            processTaskRelationMapper.deleteByCode(projectCode, processDefinitionCode);
        }
        List<ProcessTaskRelation> processTaskRelations =
                taskRelationList.stream().map(ProcessTaskRelation::new).collect(Collectors.toList());
        int insert = processTaskRelationMapper.batchInsert(processTaskRelations);
        int resultLog = processTaskRelationLogMapper.batchInsert(taskRelationList);
        return (insert & resultLog) > 0 ? Constants.EXIT_CODE_SUCCESS : Constants.EXIT_CODE_FAILURE;
    }

    public int saveProcessDefine(User loginUser, ProcessDefinition processDefinition) {
        ProcessDefinitionLog processDefinitionLog = new ProcessDefinitionLog(processDefinition);
        Integer version = processDefinitionLogMapper.queryMaxVersionForDefinition(processDefinition.getCode());
        int insertVersion = version == null || version == 0 ? Constants.VERSION_FIRST : version + 1;
        processDefinitionLog.setVersion(insertVersion);
        processDefinitionLog.setOperator(loginUser.getId());
        processDefinitionLog.setOperateTime(processDefinition.getUpdateTime());
        processDefinitionLog.setId(null);
        int insertLog = processDefinitionLogMapper.insert(processDefinitionLog);

        processDefinitionLog.setId(processDefinition.getId());
        int result = processDefinitionMapper.updateById(processDefinitionLog);
        return (insertLog & result) > 0 ? insertVersion : 0;
    }

    private void updateRelation(User loginUser, Map<String, Object> result, ProcessDefinition processDefinition,
                                List<ProcessTaskRelation> processTaskRelationList) {
        List<ProcessTaskRelationLog> relationLogs =
                processTaskRelationList.stream().map(ProcessTaskRelationLog::new).collect(Collectors.toList());
        int insertResult = processService.saveTaskRelation(loginUser, processDefinition.getProjectCode(),
                processDefinition.getCode(),
                processDefinition.getVersion(), relationLogs, Lists.newArrayList(), Boolean.TRUE);
        if (insertResult == Constants.EXIT_CODE_SUCCESS) {
            log.info(
                    "Update task relations complete, projectCode:{}, processDefinitionCode:{}, processDefinitionVersion:{}.",
                    processDefinition.getProjectCode(), processDefinition.getCode(), processDefinition.getVersion());
            putMsg(result, Status.SUCCESS);
            result.put(Constants.DATA_LIST, processDefinition);
        } else {
            log.error(
                    "Update task relations error, projectCode:{}, processDefinitionCode:{}, processDefinitionVersion:{}.",
                    processDefinition.getProjectCode(), processDefinition.getCode(), processDefinition.getVersion());
            putMsg(result, Status.UPDATE_PROCESS_DEFINITION_ERROR);
            throw new ServiceException(Status.UPDATE_PROCESS_DEFINITION_ERROR);
        }
    }

    /**
     * delete task upstream relation
     *
     * @param loginUser    login user
     * @param projectCode  project code
     * @param preTaskCodes the pre task codes, sep ','
     * @param taskCode     the post task code
     * @return delete result code
     */
    @Transactional
    @Override
    public Map<String, Object> deleteUpstreamRelation(User loginUser, long projectCode, String preTaskCodes,
                                                      long taskCode) {
        Project project = projectMapper.queryByCode(projectCode);
        // check user access for project
        Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode, null);
        if (result.get(Constants.STATUS) != Status.SUCCESS) {
            return result;
        }
        if (StringUtils.isEmpty(preTaskCodes)) {
            log.warn("Parameter preTaskCodes is empty.");
            putMsg(result, Status.DATA_IS_NULL, "preTaskCodes");
            return result;
        }
        List<ProcessTaskRelation> upstreamList = processTaskRelationMapper.queryUpstreamByCode(projectCode, taskCode);
        if (CollectionUtils.isEmpty(upstreamList)) {
            log.error("Upstream tasks based on the task do not exist, theTaskDefinitionCode:{}.", taskCode);
            putMsg(result, Status.DATA_IS_NULL, "taskCode");
            return result;
        }

        List<Long> preTaskCodeList = Lists.newArrayList(preTaskCodes.split(Constants.COMMA)).stream()
                .map(Long::parseLong).collect(Collectors.toList());
        if (preTaskCodeList.contains(0L)) {
            log.warn("Parameter preTaskCodes contain 0.");
            putMsg(result, Status.DATA_IS_NULL, "preTaskCodes");
            return result;
        }
        List<Long> currentUpstreamList =
                upstreamList.stream().map(ProcessTaskRelation::getPreTaskCode).collect(Collectors.toList());
        if (currentUpstreamList.contains(0L)) {
            log.error("Upstream taskCodes based on the task contain, theTaskDefinitionCode:{}.", taskCode);
            putMsg(result, Status.DATA_IS_NOT_VALID, "currentUpstreamList");
            return result;
        }
        List<Long> tmpCurrent = Lists.newArrayList(currentUpstreamList);
        tmpCurrent.removeAll(preTaskCodeList);
        preTaskCodeList.removeAll(currentUpstreamList);
        if (!preTaskCodeList.isEmpty()) {
            String invalidPreTaskCodes = StringUtils.join(preTaskCodeList, Constants.COMMA);
            log.error("Some upstream taskCodes are invalid, preTaskCodeList:{}.", invalidPreTaskCodes);
            putMsg(result, Status.DATA_IS_NOT_VALID, invalidPreTaskCodes);
            return result;
        }
        ProcessDefinition processDefinition =
                processDefinitionMapper.queryByCode(upstreamList.get(0).getProcessDefinitionCode());
        if (processDefinition == null) {
            log.error("Process definition does not exist, processDefinitionCode:{}.",
                    upstreamList.get(0).getProcessDefinitionCode());
            putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST,
                    String.valueOf(upstreamList.get(0).getProcessDefinitionCode()));
            return result;
        }
        List<ProcessTaskRelation> processTaskRelations =
                processTaskRelationMapper.queryByProcessCode(processDefinition.getCode());
        List<ProcessTaskRelation> processTaskRelationList = Lists.newArrayList(processTaskRelations);
        List<ProcessTaskRelation> processTaskRelationWaitRemove = Lists.newArrayList();
        for (ProcessTaskRelation processTaskRelation : processTaskRelationList) {
            if (currentUpstreamList.size() > 1) {
                if (currentUpstreamList.contains(processTaskRelation.getPreTaskCode())) {
                    currentUpstreamList.remove(processTaskRelation.getPreTaskCode());
                    processTaskRelationWaitRemove.add(processTaskRelation);
                }
            } else {
                if (processTaskRelation.getPostTaskCode() == taskCode
                        && (currentUpstreamList.isEmpty() || tmpCurrent.isEmpty())) {
                    processTaskRelation.setPreTaskVersion(0);
                    processTaskRelation.setPreTaskCode(0L);
                }
            }
        }
        processTaskRelationList.removeAll(processTaskRelationWaitRemove);
        updateProcessDefiniteVersion(loginUser, result, processDefinition);
        updateRelation(loginUser, result, processDefinition, processTaskRelationList);
        return result;
    }

    /**
     * delete task downstream relation
     *
     * @param loginUser     login user
     * @param projectCode   project code
     * @param postTaskCodes the post task codes, sep ','
     * @param taskCode      the pre task code
     * @return delete result code
     */
    @Transactional
    @Override
    public Map<String, Object> deleteDownstreamRelation(User loginUser, long projectCode, String postTaskCodes,
                                                        long taskCode) {
        Project project = projectMapper.queryByCode(projectCode);
        // check user access for project
        Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode, null);
        if (result.get(Constants.STATUS) != Status.SUCCESS) {
            return result;
        }
        if (StringUtils.isEmpty(postTaskCodes)) {
            log.warn("Parameter postTaskCodes is empty.");
            putMsg(result, Status.DATA_IS_NULL, "postTaskCodes");
            return result;
        }
        List<ProcessTaskRelation> downstreamList =
                processTaskRelationMapper.queryDownstreamByCode(projectCode, taskCode);
        if (CollectionUtils.isEmpty(downstreamList)) {
            log.error("Downstream tasks based on the task do not exist, theTaskDefinitionCode:{}.", taskCode);
            putMsg(result, Status.DATA_IS_NULL, "taskCode");
            return result;
        }
        List<Long> postTaskCodeList = Lists.newArrayList(postTaskCodes.split(Constants.COMMA)).stream()
                .map(Long::parseLong).collect(Collectors.toList());
        if (postTaskCodeList.contains(0L)) {
            log.warn("Parameter postTaskCodes contains 0.");
            putMsg(result, Status.DATA_IS_NULL, "postTaskCodes");
            return result;
        }
        ProcessDefinition processDefinition =
                processDefinitionMapper.queryByCode(downstreamList.get(0).getProcessDefinitionCode());
        if (processDefinition == null) {
            log.error("Process definition does not exist, processDefinitionCode:{}.",
                    downstreamList.get(0).getProcessDefinitionCode());
            putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST,
                    String.valueOf(downstreamList.get(0).getProcessDefinitionCode()));
            return result;
        }
        List<ProcessTaskRelation> processTaskRelations =
                processTaskRelationMapper.queryByProcessCode(processDefinition.getCode());
        List<ProcessTaskRelation> processTaskRelationList = Lists.newArrayList(processTaskRelations);
        processTaskRelationList
                .removeIf(processTaskRelation -> postTaskCodeList.contains(processTaskRelation.getPostTaskCode())
                        && processTaskRelation.getPreTaskCode() == taskCode);
        updateProcessDefiniteVersion(loginUser, result, processDefinition);
        updateRelation(loginUser, result, processDefinition, processTaskRelationList);
        return result;
    }

    /**
     * query task upstream relation
     *
     * @param loginUser   login user
     * @param projectCode project code
     * @param taskCode    current task code (post task code)
     * @return the upstream task definitions
     */
    @Override
    public Map<String, Object> queryUpstreamRelation(User loginUser, long projectCode, long taskCode) {
        Project project = projectMapper.queryByCode(projectCode);
        // check user access for project
        Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode, null);
        if (result.get(Constants.STATUS) != Status.SUCCESS) {
            return result;
        }
        List<ProcessTaskRelation> processTaskRelationList =
                processTaskRelationMapper.queryUpstreamByCode(projectCode, taskCode);
        List<TaskDefinitionLog> taskDefinitionLogList = new ArrayList<>();
        if (CollectionUtils.isNotEmpty(processTaskRelationList)) {
            Set<TaskDefinition> taskDefinitions = processTaskRelationList
                    .stream()
                    .map(processTaskRelation -> {
                        TaskDefinition taskDefinition = buildTaskDefinition();
                        taskDefinition.setProjectCode(processTaskRelation.getProjectCode());
                        taskDefinition.setCode(processTaskRelation.getPreTaskCode());
                        taskDefinition.setVersion(processTaskRelation.getPreTaskVersion());
                        return taskDefinition;
                    })
                    .collect(Collectors.toSet());
            taskDefinitionLogList = taskDefinitionLogMapper.queryByTaskDefinitions(taskDefinitions);
        }
        result.put(Constants.DATA_LIST, taskDefinitionLogList);
        putMsg(result, Status.SUCCESS);
        return result;
    }

    /**
     * query task downstream relation
     *
     * @param loginUser   login user
     * @param projectCode project code
     * @param taskCode    pre task code
     * @return the downstream task definitions
     */
    @Override
    public Map<String, Object> queryDownstreamRelation(User loginUser, long projectCode, long taskCode) {
        Project project = projectMapper.queryByCode(projectCode);
        // check user access for project
        Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode, null);
        if (result.get(Constants.STATUS) != Status.SUCCESS) {
            return result;
        }
        List<ProcessTaskRelation> processTaskRelationList =
                processTaskRelationMapper.queryDownstreamByCode(projectCode, taskCode);
        List<TaskDefinitionLog> taskDefinitionLogList = new ArrayList<>();
        if (CollectionUtils.isNotEmpty(processTaskRelationList)) {
            Set<TaskDefinition> taskDefinitions = processTaskRelationList
                    .stream()
                    .map(processTaskRelation -> {
                        TaskDefinition taskDefinition = buildTaskDefinition();
                        taskDefinition.setProjectCode(processTaskRelation.getProjectCode());
                        taskDefinition.setCode(processTaskRelation.getPostTaskCode());
                        taskDefinition.setVersion(processTaskRelation.getPostTaskVersion());
                        return taskDefinition;
                    })
                    .collect(Collectors.toSet());
            taskDefinitionLogList = taskDefinitionLogMapper.queryByTaskDefinitions(taskDefinitions);
        }
        result.put(Constants.DATA_LIST, taskDefinitionLogList);
        putMsg(result, Status.SUCCESS);
        return result;
    }

    /**
     * delete edge
     *
     * @param loginUser             login user
     * @param projectCode           project code
     * @param processDefinitionCode process definition code
     * @param preTaskCode           pre task code
     * @param postTaskCode          post task code
     * @return delete result code
     */
    @Transactional
    @Override
    public Map<String, Object> deleteEdge(User loginUser, long projectCode, long processDefinitionCode,
                                          long preTaskCode, long postTaskCode) {
        Project project = projectMapper.queryByCode(projectCode);
        // check user access for project
        Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode, null);
        if (result.get(Constants.STATUS) != Status.SUCCESS) {
            return result;
        }
        ProcessDefinition processDefinition = processDefinitionMapper.queryByCode(processDefinitionCode);
        if (processDefinition == null) {
            log.error("Process definition does not exist, projectCode：{}， processDefinitionCode:{}.", projectCode,
                    processDefinitionCode);
            putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST, String.valueOf(processDefinitionCode));
            return result;
        }
        List<ProcessTaskRelation> processTaskRelations =
                processTaskRelationMapper.queryByProcessCode(processDefinitionCode);
        List<ProcessTaskRelation> processTaskRelationList = Lists.newArrayList(processTaskRelations);
        if (CollectionUtils.isEmpty(processTaskRelationList)) {
            log.error("Process task relations are empty, projectCode:{}, processDefinitionCode:{}.", projectCode,
                    processDefinitionCode);
            putMsg(result, Status.DATA_IS_NULL, "processTaskRelationList");
            return result;
        }
        Map<Long, List<ProcessTaskRelation>> taskRelationMap = new HashMap<>();
        for (ProcessTaskRelation processTaskRelation : processTaskRelationList) {
            taskRelationMap.compute(processTaskRelation.getPostTaskCode(), (k, v) -> {
                if (v == null) {
                    v = new ArrayList<>();
                }
                v.add(processTaskRelation);
                return v;
            });
        }
        if (!taskRelationMap.containsKey(postTaskCode)) {
            putMsg(result, Status.DATA_IS_NULL, "postTaskCode");
            return result;
        }
        if (taskRelationMap.get(postTaskCode).size() > 1) {
            for (ProcessTaskRelation processTaskRelation : taskRelationMap.get(postTaskCode)) {
                if (processTaskRelation.getPreTaskCode() == preTaskCode) {
                    int delete = processTaskRelationMapper.deleteById(processTaskRelation.getId());
                    if (delete == 0) {
                        log.error(
                                "Delete task relation edge error, processTaskRelationId:{}, preTaskCode:{}, postTaskCode:{}",
                                processTaskRelation.getId(), preTaskCode, postTaskCode);
                        putMsg(result, Status.DELETE_EDGE_ERROR);
                        throw new ServiceException(Status.DELETE_EDGE_ERROR);
                    } else
                        log.info(
                                "Delete task relation edge complete, processTaskRelationId:{}, preTaskCode:{}, postTaskCode:{}",
                                processTaskRelation.getId(), preTaskCode, postTaskCode);
                    processTaskRelationList.remove(processTaskRelation);
                }
            }
        } else {
            ProcessTaskRelation processTaskRelation = taskRelationMap.get(postTaskCode).get(0);
            processTaskRelationList.remove(processTaskRelation);
            processTaskRelation.setPreTaskVersion(0);
            processTaskRelation.setPreTaskCode(0L);
            processTaskRelationList.add(processTaskRelation);
            log.info(
                    "Delete task relation through set invalid value for it: preTaskCode from {} to 0, processTaskRelationId:{}.",
                    preTaskCode, processTaskRelation.getId());
        }
        updateProcessDefiniteVersion(loginUser, result, processDefinition);
        updateRelation(loginUser, result, processDefinition, processTaskRelationList);
        return result;
    }

    @Override
    public List<ProcessTaskRelation> queryByWorkflowDefinitionCode(long workflowDefinitionCode,
                                                                   int workflowDefinitionVersion) {
        return processTaskRelationMapper.queryProcessTaskRelationsByProcessDefinitionCode(workflowDefinitionCode,
                workflowDefinitionVersion);
    }

    @Override
    public void deleteByWorkflowDefinitionCode(long workflowDefinitionCode, int workflowDefinitionVersion) {
        processTaskRelationMapper.deleteByWorkflowDefinitionCode(workflowDefinitionCode, workflowDefinitionVersion);
    }

    /**
     * build task definition
     *
     * @return task definition
     */
    private TaskDefinition buildTaskDefinition() {

        return new TaskDefinition() {

            @Override
            public boolean equals(Object o) {
                if (this == o) {
                    return true;
                }
                if (!(o instanceof TaskDefinition)) {
                    return false;
                }
                TaskDefinition that = (TaskDefinition) o;
                return getCode() == that.getCode()
                        && getVersion() == that.getVersion()
                        && getProjectCode() == that.getProjectCode();
            }

            @Override
            public int hashCode() {
                return Objects.hash(getCode(), getVersion(), getProjectCode());
            }
        };
    }
}
