/**
 * 
 */
package org.springblade.etl.service;

import cn.hutool.core.util.StrUtil;
import cn.hutool.extra.spring.SpringUtil;
import com.alibaba.nacos.common.utils.ExceptionUtil;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.JobStatus;
import org.springblade.core.secure.utils.AuthUtil;
import org.springblade.core.tool.api.R;
import org.springblade.core.tool.exception.ServiceException;
import org.springblade.core.tool.utils.CollectionUtil;
import org.springblade.etl.connector.base.SeatunnelConnector;
import org.springblade.etl.dto.etl.JobGraph;
import org.springblade.etl.engine.flink.dto.SubmitJobReq;
import org.springblade.etl.engine.flink.dto.SubmitJobRes;
import org.springblade.etl.engine.seatunnel.SeatunnelService;
import org.springblade.etl.entity.*;
import org.springblade.etl.enums.*;
import org.springblade.etl.mapper.JobMapper;
import org.springblade.etl.task.model.DaemonTaskConfig;
import org.springblade.etl.task.model.DaemonTaskType;
import org.springblade.etl.task.thread.DaemonFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.*;


/**
 *	任务定义 服务处理
 *
 * @author aokunsang
 * @date   2023-07-28
 */
@Slf4j
@Service
public class JobService extends ServiceImpl<JobMapper, Job> {

	@Autowired
	private JobTaskService jobTaskService;
	
	@Autowired
	private JobLinkService jobLinkService;
	
	@Autowired
	private JobInstanceService jobInstanceService;
	
	@Autowired
	private SeatunnelService seatunnelService;
	
	@Autowired
	private JobPermissionService jobPermissionService;
	
	@Autowired
	private org.springblade.etl.engine.JobService jobService;
	
	/**
	 * 
	 * @param userId
	 * @return
	 */
	public List<Integer> categoryIdListOfPerm(Long userId){
		return baseMapper.categoryIdListOfPerm(userId);
	}
	
	/**
	 * 
	 * @param userId
	 * @return
	 */
	public List<Integer> jobIdListOfPerm(Long userId){
		return baseMapper.jobIdListOfPerm(userId);
	}
	
	/**
	 * 删除任务
	 * @param jobId
	 * @return
	 */
	@Transactional
	public boolean doRemove(Integer jobId) {
		boolean ret = removeById(jobId);
		if(ret) {
			jobLinkService.remove(Wrappers.<JobLink>lambdaQuery().eq(JobLink::getJobId, jobId));
			jobTaskService.remove(Wrappers.<JobTask>lambdaQuery().eq(JobTask::getJobId, jobId));
		}
		return ret;
	}
	
	/**
	 * 
	 * JOB定义
	 * @param jobGraph
	 * @return
	 */
	@Transactional
	public Integer jobDefine(JobGraph jobGraph) {
		List<JobTask> nodes = jobGraph.getNodes();
		//1.保存任务基本信息
		Job job = jobGraph.getJob();
		if(job.getId() == null) {
			job.setCreateTime(LocalDateTime.now());
			job.setCreateBy(AuthUtil.getNickName());
			job.setCreateUserId(AuthUtil.getUserId());
		}else {
			Job dbJob = this.getById(job.getId());
			if(dbJob.getJobStatus() == org.springblade.etl.enums.JobStatus.ONLINE) {
				throw new ServiceException("任务需要先下线后才能修改");
			}
			job.setUpdateBy(AuthUtil.getNickName());
			job.setUpdateUserId(AuthUtil.getUserId());
			job.setUpdateTime(LocalDateTime.now());
			//删除连线和节点
			jobLinkService.remove(Wrappers.<JobLink>lambdaQuery().eq(JobLink::getJobId, job.getId()));
			jobTaskService.remove(Wrappers.<JobTask>lambdaQuery().eq(JobTask::getJobId, job.getId()));
		}
		Integer jobId = jobSave(nodes, jobGraph.getEdges(), job);
		//处理权限组-仅新增时
		if(jobId != null && jobId > 0 && CollectionUtil.isNotEmpty(jobGraph.getDataPermIdList())) {
			List<JobPermission> jobPermList = new ArrayList<>();
			jobGraph.getDataPermIdList().forEach(item -> {
				JobPermission jobPermission = new JobPermission();
				jobPermission.setCreateBy(AuthUtil.getNickName());
				jobPermission.setCreateTime(new Date());
				jobPermission.setCreateUserId(AuthUtil.getUserId());
				jobPermission.setDataPermId(item);
				jobPermission.setJobId(jobId);
				jobPermList.add(jobPermission);
			});
			jobPermissionService.saveBatch(jobPermList);
		}
		return jobId;
	}
	
	@Transactional
	private Integer jobSave(List<JobTask> nodes, List<JobLink> edges, Job job) {
		Set<JobType> jobTypes = new HashSet<>();
		for(JobTask jobTask : nodes) {
			SeatunnelConnector<?> seatunnelConnector = SpringUtil.getBean(jobTask.getConnectorType().getClazz());
			//校验参数的正确性
			seatunnelConnector.validate(jobTask.getConfigJson());
			//设置任务流程类型
			ConnectorType connectorType = jobTask.getConnectorType();
			if(connectorType.getPluginType() == PluginType.SOURCE) {
				jobTypes.add(connectorType.isStreaming() ? JobType.REALTIME : JobType.PERIOD);
			}
		}
		//根据source判断任务类型，如果存在多个不同类型的source输入源，抛出异常
		if(jobTypes.size() > 1) {
			throw new ServiceException("有且只能一种任务类型的SOURCE");
		}
		//设置类型
		job.setJobType(jobTypes.isEmpty() ? JobType.PERIOD : jobTypes.iterator().next());
		if(job.getJobStatus() == org.springblade.etl.enums.JobStatus.ONLINE_RUN) {
			job.setJobStatus(org.springblade.etl.enums.JobStatus.ONLINE);
		}
		//添加或保存任务流程
		saveOrUpdate(job);
		Integer jobId = job.getId();
		Map<String, JobTask> taskMap = new HashMap<>();
		if(CollectionUtil.isNotEmpty(nodes)) {
			//2.保存任务节点信息
			nodes.stream().forEach(item -> {
				item.setId(null);
				item.setJobId(jobId);
				item.setCreateTime(LocalDateTime.now());
				item.setCreateUserId(AuthUtil.getUserId());
				item.setCreateBy(AuthUtil.getNickName());
				taskMap.put(item.getTaskCode(), item);
			});
			if(taskMap.size() != nodes.size()) {
				throw new ServiceException("节点CODE有重复值");
			}
			jobTaskService.saveBatch(nodes);
		}
		if(CollectionUtil.isNotEmpty(edges)) {
			//3.保存节点连线信息
			edges.stream().forEach(item -> {
				item.setId(null);
				item.setCreateTime(LocalDateTime.now());
				item.setJobId(jobId);
			});
			boolean codeNotMatch = edges.stream().anyMatch(item -> !taskMap.containsKey(item.getFromCode()) || !taskMap.containsKey(item.getToCode()));
			if(codeNotMatch) {
				throw new ServiceException("节点连线CODE任务中不存在");
			}
			jobLinkService.saveBatch(edges);
		}
		return jobId;
	}
	
	/**
	 * 
	 * @param jobId
	 * @return
	 */
	public JobGraph jobEdit(Integer jobId) {
		//校验是否可以修改
		long runingCnt = jobInstanceService.count(Wrappers.<JobInstance>lambdaQuery().eq(JobInstance::getJobId, jobId).eq(JobInstance::getInstanceStatus, InstanceStatus.RUNNING));
		if(runingCnt > 0) {
			throw new ServiceException("当前任务正在执行中，不能修改");
		}
		JobGraph jobGraph = new JobGraph();
		Job job = getById(jobId);
		jobGraph.setJob(job);
		jobGraph.setEdges(jobLinkService.list(Wrappers.<JobLink>lambdaQuery().eq(JobLink::getJobId, jobId)));
		jobGraph.setNodes(jobTaskService.list(Wrappers.<JobTask>lambdaQuery().eq(JobTask::getJobId, jobId)));
		return jobGraph;
	}
	
	/**
	 * 从savepoint恢复
	 * @param jobId
	 * @param savepointPath
	 */
	public boolean jobRun(Integer jobId, String savepointPath) {
		return jobRun(jobId, InstanceRunType.MANUAL, savepointPath);
	}
	
	/**
	 * 直接启动
	 * @param jobId
	 * @param runType
	 */
	public boolean jobRun(Integer jobId, InstanceRunType runType) {
		return jobRun(jobId, runType, null);
	}
	
	/**
	 * 运行JOB
	 * @param jobId
	 * @return
	 */
	public boolean jobRun(Integer jobId, InstanceRunType runType, String savepointPath) {
		Job job = getById(jobId);
		if(job.getJobStatus() != org.springblade.etl.enums.JobStatus.ONLINE) {
			throw new ServiceException("任务不在线，不能运行");
		}
		//校验是否有任务实例在运行中
		long runningCount = jobInstanceService.countRunningInstanceByJobId(jobId);
		if(runningCount >= 1) {
			throw new ServiceException("该任务有实例正在运行中，请检查");
		}
		SubmitJobRes submitJobRes = null;
		SubmitJobReq submitJobReq = null;
		try {
			if(StrUtil.isBlank(savepointPath)) { //从快照恢复，不能执行前置操作
				List<JobTask> taskList = jobTaskService.listByJobId(jobId);
				//执行准备工作
				for(JobTask task : taskList) {
					SpringUtil.getBean(task.getConnectorType().getClazz()).prepare(task.getConfigJson());
				}
			}
			submitJobReq = seatunnelService.buildJobRequest(jobId);
			submitJobReq.setSavePointPath(savepointPath);
		} catch (Exception ex) {
			log.error(ex.getMessage(), ex);
			throw ex;
		}
		String errMsg = null;
		if(submitJobReq != null) {
			try {
				submitJobRes = jobService.submit(submitJobReq);
			} catch (Exception e) {
				log.error(e.getMessage(), e);
				errMsg = ExceptionUtil.getAllExceptionMsg(e);
			} finally {
				//创建运行实例
				JobInstance jobInstance = new JobInstance();
				jobInstance.setJobId(jobId);
				jobInstance.setStartTime(new Date());
				jobInstance.setApplicationId(submitJobRes != null ? submitJobRes.getClusterId() : null);
				jobInstance.setJobEngineId(submitJobRes != null ? submitJobRes.getJobId() : null);
				jobInstance.setWebInterfaceUrl(submitJobRes != null ? submitJobRes.getWebInterfaceUrl() : null);
				jobInstance.setInstanceType(job.getJobType().getInstanceType());
				jobInstance.setInstanceStatus(submitJobRes != null ? InstanceStatus.RUNNING : InstanceStatus.FAILURE);
				jobInstance.setEngineJobStatus(submitJobRes != null ? JobStatus.INITIALIZING.name() : null);
				jobInstance.setCreateTime(LocalDateTime.now());
				jobInstance.setTenantId(job.getTenantId());
				if(submitJobRes == null) {
					jobInstance.setEndTime(new Date());
				}
				jobInstance.setInstanceRunType(runType);
				jobInstance.setErrMsg(errMsg);
				jobInstanceService.save(jobInstance);
				
				//如果实例运行正常，循环监控
				if(submitJobRes != null && StrUtil.isNotEmpty(submitJobRes.getJobId())) {
					DaemonFactory.addTask(new DaemonTaskConfig(jobInstance.getId().toString(), DaemonTaskType.flink_job));
				}
			}
		}
		return errMsg == null;
	}
	
	/**
	 * @param jobGraph
	 * @return
	 * @throws Exception 
	 */
	public R<SubmitJobRes> previewJobRun(JobGraph jobGraph) throws Exception {
		List<JobTask> taskList = jobGraph.getNodes();
		List<JobLink> taskLinkList = jobGraph.getEdges();
		Job job = jobGraph.getJob();
		//执行准备工作 - 预览暂不执行准备工作
//		for(JobTask task : taskList) {
//			SpringUtil.getBean(task.getConnectorType().getClazz()).prepare(task.getConfigJson());
//		}
		SubmitJobRes submitJobRes = null;
		try {
			SubmitJobReq submitJobReq = seatunnelService.buildJobRequest(job, taskList, taskLinkList);
			submitJobRes = jobService.submit(submitJobReq);
		} catch (Exception e) {
			log.error(e.getMessage(), e);
			return R.fail(201, ExceptionUtil.getAllExceptionMsg(e));
		}
		return R.data(submitJobRes);
	}

}