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

import java.time.LocalDateTime;
import java.util.Date;
import java.util.Optional;

import org.springblade.core.boot.ctrl.BladeController;
import org.springblade.core.log.annotation.ApiLog;
import org.springblade.core.mp.support.Condition;
import org.springblade.core.mp.support.Query;
import org.springblade.core.secure.utils.AuthUtil;
import org.springblade.core.tool.api.R;
import org.springblade.core.tool.exception.ServiceException;
import org.springblade.etl.dto.etl.JobGraph;
import org.springblade.etl.engine.seatunnel.SeatunnelService;
import org.springblade.etl.entity.Job;
import org.springblade.etl.entity.JobInstance;
import org.springblade.etl.enums.InstanceRunType;
import org.springblade.etl.enums.InstanceStatus;
import org.springblade.etl.enums.JobStatus;
import org.springblade.etl.enums.JobType;
import org.springblade.etl.service.JobInstanceService;
import org.springblade.etl.service.JobService;
import org.springblade.system.client.DataPermClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.xxl.job.core.biz.AdminBiz;
import com.xxl.job.core.biz.model.JobInfo;
import com.xxl.job.core.biz.model.ReturnT;

import cn.hutool.core.util.StrUtil;
import cn.hutool.cron.pattern.parser.PatternParser;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;

/**
 *	任务定义 操作处理
 *
 * @author aokunsang
 * @date   2023-07-28
 */
@Slf4j
@RestController
@RequestMapping("/job")
@Api(value = "任务相关接口", tags = "任务接口")
public class JobController extends BladeController {

	@Autowired
	private JobService jobService;
	
	@Autowired
	private JobInstanceService jobInstanceService;
	
	@Autowired
	private AdminBiz adminBiz;
	
	@Autowired
	private DataPermClient dataPermClient;
	
	@Value("${xxl.job.executor.groupid}")
    private int groupId;
	
	/**
	 * 	列表数据
	 * @param job
	 * @return
	 */
	@GetMapping("/page")
	@ApiOperation(value = "分页查询任务", httpMethod = "GET")
	public R<IPage<Job>> pageList(Job job, Query query){
		Long userId = getUser().getUserId();
		LambdaQueryWrapper<Job> queryWrapper = Condition.getQueryWrapper(job).lambda().orderByDesc(Job::getId);
		Boolean checkSuperPermRet = dataPermClient.checkSuperPermission(getUser().getUserId());
		if(checkSuperPermRet == null || !checkSuperPermRet) {
			queryWrapper.and(t ->{
				t.eq(Job::getCreateUserId, userId)
				.or().inSql(Job::getId, "select a.job_id from etl_job_permission a inner join data_permission_user b on a.data_perm_id = b.data_perm_id where b.user_id = " + userId)
				.or().eq(Job::getPublicStatus, 1);
			});
		}
		return R.data(jobService.page(Condition.getPage(query), queryWrapper));
	}
	
	/**
	 * 单条数据查询接口
	 * @param id
	 * @return
	 */
	@GetMapping("/info")
	@ApiOperation(value = "根据ID获取任务基本信息", httpMethod = "GET")
	public R<Job> info(@RequestParam Integer id) {
		return R.data(jobService.getById(id));
	}
	
	/**
	 *	修改数据接口
	 * @param job
	 * @return
	 */
	@ApiLog("ETL任务流程")
	@PostMapping("/update")
	@ApiOperation(value = "修改任务基本信息", httpMethod = "POST")
	public R<String> doUpdate(@Validated @RequestBody Job job) {
		Job dbJob = jobService.getById(job.getId());
		if(dbJob.getJobStatus() == JobStatus.ONLINE) {
			return R.fail("任务需要先下线后才能修改");
		}
		job.setUpdateBy(AuthUtil.getNickName());
		job.setUpdateUserId(AuthUtil.getUserId());
		job.setUpdateTime(LocalDateTime.now());
		return R.status(jobService.updateById(job));
	}
	
	/**
	 * 	删除数据接口
	 * @param id
	 * @return
	 */
	@ApiLog("ETL任务流程")
	@PostMapping("/delete")
	@ApiOperation(value = "根据ID删除任务", httpMethod = "POST")
	public R<String> doDelete(@RequestParam Integer id) {
		Job dbJob = jobService.getById(id);
		if(dbJob.getJobStatus() == JobStatus.ONLINE) {
			return R.fail("任务需要先下线后才能删除");
		}
		return R.status(jobService.doRemove(id));
	}
	
	/**
	 * 
	 * @param jobGraph
	 * @return
	 */
	@ApiLog("ETL任务流程")
	@PostMapping("/define")
	@ApiOperation(value = "提交任务定义", httpMethod = "POST")
	public R<Integer> define(@Validated @RequestBody JobGraph jobGraph){
		JobStatus jobStatus = jobGraph.getJob().getJobStatus();
		Integer jobId = jobService.jobDefine(jobGraph);
		String suffixMsg = "", errMsg = "";
		if(jobStatus == JobStatus.ONLINE_RUN) { //保存并运行
			boolean isSucc = false;
			try {
				isSucc = jobService.jobRun(jobId, InstanceRunType.MANUAL);
			} catch (Exception e) {
				log.error(e.getMessage(), e);
				errMsg = e.getMessage();
			}
			suffixMsg = isSucc ? ",并且已提交执行" : ",但是提交执行失败";
		}
		R<Integer> ret = R.data(jobId, "任务保存成功"+suffixMsg);
		ret.setExt(errMsg);
		return ret;
	}
	
	/**
	 * 
	 * @param jobId
	 * @return
	 */
	@ApiLog("ETL任务流程")
	@GetMapping("/edit/{jobId}")
	@ApiOperation(value = "修改任务定义【回显】", httpMethod = "GET")
	public R<JobGraph> edit(@PathVariable Integer jobId){
		return R.data(jobService.jobEdit(jobId));
	}
	
	/**
	 * 运行任务
	 * @param jobId
	 * @return
	 */
	@ApiLog("ETL任务流程")
	@PostMapping("/run/{jobId}")
	@ApiOperation(value = "运行任务", httpMethod = "POST")
	public R<Boolean> run(@PathVariable Integer jobId){
		try {
			jobService.jobRun(jobId, InstanceRunType.MANUAL);
			return R.success("任务已提交执行");
		} catch (Exception e) {
			log.error(e.getMessage(), e);
			R<Boolean> ret = R.fail(e.getMessage());
			ret.setExt(e.getMessage());
			return ret;
		}
		//return R.success("任务已提交执行");
	}
	
	/**
	 * 
	 * @param jobId
	 * @param jobStatus
	 * @return
	 */
	@ApiLog("ETL任务流程")
	@PostMapping("/switch/{jobId}")
	@ApiOperation(value = "切换任务状态", httpMethod = "POST")
	public R<String> switchStatus(@PathVariable Integer jobId, @RequestParam JobStatus jobStatus){
		if(jobStatus != JobStatus.ONLINE && jobStatus != JobStatus.OFFLINE) {
			return R.fail("任务状态不正确");
		}
		if(jobStatus == JobStatus.OFFLINE && jobInstanceService.count(Wrappers.<JobInstance>lambdaQuery().eq(JobInstance::getJobId, jobId)
				.eq(JobInstance::getInstanceStatus, InstanceStatus.RUNNING)) > 0) {
			return R.fail("该任务有实例正在运行，中断后才能下线。");
		}
		Job job = jobService.getById(jobId);
		if(job == null) {
			return R.fail("任务流程不存在");
		}
		Job entity = new Job();
		entity.setId(jobId);
		entity.setJobStatus(jobStatus);
		//任务需要下线，但是调度还在开启状态，需要关闭
		if(jobStatus == JobStatus.OFFLINE && job.getTriggerStatus() == 1) {
			entity.setTriggerStatus(0);
			//需要关闭xxljob
			if(job.getXxlJobId() != null && job.getXxlJobId() > 0) {
				ReturnT<String> jobStop = adminBiz.jobStop(job.getXxlJobId());
				log.info("ETL任务流程:暂停任务|jobId={},xxljobId={},ret={}", jobId, job.getXxlJobId(), jobStop);
			}
		}
		entity.setUpdateBy(AuthUtil.getNickName());
		entity.setUpdateUserId(AuthUtil.getUserId());
		entity.setUpdateTime(LocalDateTime.now());
		return R.status(jobService.updateById(entity));
	}
	
	/**
	 * 
	 * @param job
	 * @return
	 */
	@ApiLog("ETL任务流程")
	@PostMapping("/submitcron")
	@ApiOperation(value = "新增修改任务调度", httpMethod = "POST")
	public R<String> submitCron(@RequestBody Job job){
		if(job.getId() == null || StrUtil.isBlank(job.getJobCron())) {
			return R.fail("请求参数不正确");
		}
		Integer jobId = job.getId();
		Job dbJob = jobService.getById(jobId);
		if(dbJob.getJobType() == JobType.REALTIME) {
			return R.fail("流任务不支持任务调度");
		}
		if(job.getTriggerStatus() == 1 && dbJob.getJobStatus() == JobStatus.OFFLINE) {
			return R.fail("任务必须上线后才能开启任务调度");
		}
		dbJob.setJobCron(job.getJobCron());
		int xxlJobId = Optional.<Integer>ofNullable(dbJob.getXxlJobId()).orElse(0);
		int triggerStatus = Optional.<Integer>ofNullable(job.getTriggerStatus()).orElse(0);
		ReturnT<String> returnT = _handleJob(dbJob, AuthUtil.getUserId(), xxlJobId, triggerStatus);
		log.info("ETL任务流程|新增/修改xxljob任务|jobId={},symbol={},result={}", jobId, (xxlJobId>0?"update":"add"), returnT);
		if (ReturnT.SUCCESS_CODE == returnT.getCode()){
			//首次添加任务，需要把任务ID更新进去
			Job entity = new Job();
			//修改的时候content=null，新增时候content=xxljobId
			int newXxlJobId = xxlJobId > 0 ? xxlJobId : Integer.valueOf(returnT.getContent());
			entity.setXxlJobId(newXxlJobId);
			entity.setTriggerStatus(triggerStatus);
			entity.setId(jobId);
			entity.setJobCron(job.getJobCron());
			jobService.updateById(entity);
			//修改任务，并且之前为暂停状态：需要启动任务
			if(xxlJobId > 0 && triggerStatus == 1 && dbJob.getTriggerStatus() == 0) {
				ReturnT<String> jobStart = adminBiz.jobStart(newXxlJobId);
				log.info("ETL任务流程:启动任务|jobId={},xxljobId={},ret={}", jobId, newXxlJobId, jobStart);
			}
			if(xxlJobId > 0 && triggerStatus == 0 && dbJob.getTriggerStatus() == 1) {
				ReturnT<String> jobStop = adminBiz.jobStop(newXxlJobId);
				log.info("ETL任务流程:暂停任务|jobId={},xxljobId={},ret={}", jobId, newXxlJobId, jobStop);
			}
		}
		return R.data("操作成功");
	}
	
	@Autowired
	private SeatunnelService seatunnelService;
	
	@GetMapping("/config/{jobId}")
	@ApiOperation(value = "[测试]查看seatunnel配置文件", httpMethod = "GET")
	public R<?> config(@PathVariable Integer jobId){
		return R.data(JSON.parseObject(seatunnelService.buildJobRequest(jobId).getJobConfigJson()));
	}
	
	/**
	 * 处理xxljob任务
	 * @return
	 */
	private ReturnT<String> _handleJob(Job jobInfo, Long userId, int jobId, int triggerStatus) {
		try {//校验cron的正确性
			PatternParser.parse(jobInfo.getJobCron());
		} catch (Exception e) {
			throw new ServiceException("请检查Cron表达式不正确");
		}
		JobInfo xxlJobInfo = new JobInfo();
		xxlJobInfo.setJobGroup(groupId);
		xxlJobInfo.setJobDesc("ETL任务:"+jobInfo.getJobName() + ":" + jobInfo.getId());
		xxlJobInfo.setAuthor(userId.toString());
		xxlJobInfo.setScheduleType("CRON"); //调度类型：固定速度【以固定速度，触发任务调度；按照固定的间隔时间，周期性触发】
		xxlJobInfo.setScheduleConf(jobInfo.getJobCron());
		xxlJobInfo.setMisfireStrategy("DO_NOTHING");  //调度过期策略：忽略【调度过期后，忽略过期的任务，从当前时间开始重新计算下次触发时间】
		xxlJobInfo.setExecutorRouteStrategy("ROUND"); //路由策略：轮询
		xxlJobInfo.setExecutorBlockStrategy("SERIAL_EXECUTION"); //阻塞处理策略：单机串行【调度请求进入单机执行器后，调度请求进入FIFO队列并以串行方式运行】
		xxlJobInfo.setExecutorHandler("jobHandler");
		xxlJobInfo.setExecutorParam(jobInfo.getId().toString());
		xxlJobInfo.setExecutorTimeout(60); //超时时间1分钟
		xxlJobInfo.setExecutorFailRetryCount(0);
		xxlJobInfo.setGlueUpdatetime(new Date());
		xxlJobInfo.setGlueType("BEAN");
		xxlJobInfo.setTriggerLastTime(0);
		xxlJobInfo.setTriggerNextTime(0);
		xxlJobInfo.setTriggerStatus(triggerStatus);//1.运行 0.暂停
		if(jobId == 0) { //新增
			xxlJobInfo.setAddTime(new Date());
			return adminBiz.jobAdd(xxlJobInfo);
		}else {
			xxlJobInfo.setId(jobId);
			return adminBiz.jobUpdate(xxlJobInfo);
		}
	}
}