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

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.runtime.rest.messages.JobExceptionsInfoWithHistory;
import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointStatistics.CompletedCheckpointStatistics;
import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointingStatistics;
import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointingStatistics.LatestCheckpoints;
import org.apache.flink.runtime.rest.messages.job.JobDetailsInfo;
import org.apache.flink.runtime.rest.messages.job.JobDetailsInfo.JobVertexDetailsInfo;
import org.apache.flink.runtime.rest.messages.job.metrics.IOMetricsInfo;
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.dto.etl.JobGraph;
import org.springblade.etl.dto.etl.JobStatReq;
import org.springblade.etl.dto.etl.JobStatRes;
import org.springblade.etl.dto.vo.InstanceCountVO;
import org.springblade.etl.engine.flink.FlinkJobService;
import org.springblade.etl.engine.flink.dto.SubmitJobRes;
import org.springblade.etl.entity.JobInstance;
import org.springblade.etl.enums.InstanceStatus;
import org.springblade.etl.enums.InstanceType;
import org.springblade.etl.mapper.JobInstanceMapper;
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.util.StopWatch;
import redis.clients.jedis.JedisCluster;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 *	任务实例 服务处理
 *
 * @author aokunsang
 * @date   2023-07-28
 */
@Slf4j
@Service
public class JobInstanceService extends ServiceImpl<JobInstanceMapper, JobInstance>{
	
	@Autowired
	private FlinkJobService flinkJobService;
	
	@Autowired
	private JobService jobService;
    //fixme 上线前打开
//	@Autowired
//	@Qualifier("previewJedisCluster")
	private JedisCluster jedisCluster;

	@Autowired
	private JobInstanceMapper jobInstanceMapper;

	/**
	 * 查询正在运行的所有任务实例
	 * @return
	 */
	public List<JobInstance> listAllRunningInstanceList(){
		return baseMapper.selectAllRunningInstanceList();
	}
	
	/**
	 * 
	 * @param jobId
	 * @return
	 */
	public long countRunningInstanceByJobId(Integer jobId) {
		Long cnt = baseMapper.countRunningInstanceByJobId(jobId);
		return cnt == null ? 0L : cnt.longValue();
	}
	
	/**
	 * 
	 * @param statReq
	 * @return
	 */
	public List<JobStatRes> stat(JobStatReq statReq){
		return baseMapper.stat(statReq);
	}
	
	/**
	 * 取消任务实例
	 * @param jobInstanceId
	 * @throws Exception 
	 */
	public void cancelJobInstance(Integer jobInstanceId) throws Exception {
		JobInstance jobInstance = getById(jobInstanceId);
		flinkJobService.cancel(jobInstance.getJobEngineId(), jobInstance.getWebInterfaceUrl());
	}
	
	/**
	 * 
	 * @param jobInstanceId
	 * @throws Exception
	 */
	public void recoverJobInstance(Integer jobInstanceId) throws Exception{
		JobInstance jobInstance = getById(jobInstanceId);
		if(StrUtil.isBlank(jobInstance.getJobEngineId()) || StrUtil.isBlank(jobInstance.getWebInterfaceUrl())) {
			throw new ServiceException("该任务未正常提交，不能恢复");
		}
		if(jobInstance.getInstanceType() == InstanceType.BATCH) {
			throw new ServiceException("目前只支持流任务的恢复");
		}
		CheckpointingStatistics checkpoints = flinkJobService.checkpoints(jobInstance.getJobEngineId(), jobInstance.getWebInterfaceUrl());
		if(checkpoints == null || checkpoints.getLatestCheckpoints() == null) {
			throw new ServiceException("没有找到有效的快照信息，请确认任务是否开启作业快照功能。");
		}
		LatestCheckpoints latestCheckpoints = checkpoints.getLatestCheckpoints();
//		CompletedCheckpointStatistics savepointStatistics = latestCheckpoints.getSavepointStatistics();
		CompletedCheckpointStatistics completedCheckpoint = latestCheckpoints.getCompletedCheckpointStatistics();
		if(completedCheckpoint == null) {
			throw new ServiceException("没有找到有效的快照信息，请确认任务是否开启作业快照功能。");
		}
		String savepointPath = completedCheckpoint.getExternalPath();
		log.info("任务恢复开始|jobId={},checkpointPath:{}", jobInstance.getJobId(), savepointPath);
		jobService.jobRun(jobInstance.getJobId(), savepointPath);
	}
	
	/**
	 * 预览数据
	 * @param key
	 * @return
	 * @throws Exception 
	 */
	public R<List<JSONObject>> doPreview(String key, JobGraph jobGraph) throws Exception {
		StopWatch watch = new StopWatch("preview-task");
		List<JSONObject> resultList = new ArrayList<>();
		try {
			Boolean exists = jedisCluster.exists(key);
			if(exists == null || !exists) {
				final int maxRetryTimes = 60;
				watch.start();
				R<SubmitJobRes> retData = jobService.previewJobRun(jobGraph);
				if(!retData.isSuccess()) {
					return R.fail(retData.getCode(), retData.getMsg());
				}
				SubmitJobRes jobRes = retData.getData();
				watch.stop();
				log.info("预览任务提交完成|key={},cost={}ms", key, watch.getLastTaskTimeMillis());
				watch.start();
				//轮询检测任务是否已完成或者抛异常
				int i = 0;
				JobStatus lastJobStatus = null;
				while(++i < maxRetryTimes) {
					JobDetailsInfo jobDetailsInfo = flinkJobService.info(jobRes.getJobId(), jobRes.getWebInterfaceUrl());
					log.info("第{}次查询预览任务是否完成|key={},jobStatus={}", i, key, jobDetailsInfo.getJobStatus());
					lastJobStatus = jobDetailsInfo.getJobStatus();
					if(jobDetailsInfo.getJobStatus() == JobStatus.FINISHED) { //真正完成
						break;
					}
					if(jobDetailsInfo.getJobStatus() == JobStatus.FAILED) { 
						JobExceptionsInfoWithHistory exception = flinkJobService.exception(jobRes.getJobId(), jobRes.getWebInterfaceUrl());
						if(exception != null) {
							String errmsg = exception.getRootException();
							//专门的异常结束任务，标识任务已经完成 platomix:etl:preview:shutdown
							if(errmsg.contains("PreviewConnectorException")) { //中断完成
								break;
							}
							//出现真实异常
							return R.fail(201, StrUtil.subAfter(errmsg, "(ForkJoinWorkerThread.java:157)", false));
						}
					}
					TimeUnit.SECONDS.sleep(1);
				}
				watch.stop();
				log.info("预览任务轮询完成|key={},cost={}ms", key, watch.getLastTaskTimeMillis());
				if(i == maxRetryTimes && lastJobStatus != JobStatus.FINISHED && lastJobStatus != JobStatus.FAILED) { //轮询了最大次数才完成，有可能任务还在运行中，已超时
					try {//强制取消任务
						flinkJobService.cancel(jobRes.getJobId(), jobRes.getWebInterfaceUrl());
					} catch (Exception e) {}
					throw new ServiceException("预览数据已超时，请重新提交。");
				}
			}
			watch.start();
			List<String> dataList = jedisCluster.lrange(key, 0, 100);
			if(CollectionUtil.isNotEmpty(dataList)) {
				dataList.stream().map(JSON::parseObject).forEach(resultList::add);
			}
			//查询完设置Key,直接删除掉
			jedisCluster.del(key, key+":num");
		} finally {
//			if(jedis != null) {
//				jedis.close();
//			}
		}
		watch.stop();
		log.info("预览任务redis数据查询完成|key={},cost={}ms", key, watch.getLastTaskTimeMillis());
		log.info("预览任务总耗时|key={},costPretty:{}ms", key, watch.getTotalTimeMillis());
		return R.data(resultList);
	}
	
	/**
	 * 刷新作业实例信息
	 * 1.作业状态
	 * 2.作业错误日志
	 * 3.作业统计指标
	 * @param jobInstanceId
	 * @throws Exception 
	 */
	public void refreshJobInstance(Integer jobInstanceId) throws Exception {
		JobInstance jobInstance = getById(jobInstanceId);
		if(jobInstance.getInstanceStatus() != InstanceStatus.RUNNING) {
			return;
		}
		JobInstance updateEnity = new JobInstance();
		updateEnity.setId(jobInstanceId);
		//1.获取作业状态信息
		JobDetailsInfo jobDetailsInfo = null;
		try {
			jobDetailsInfo = flinkJobService.info(jobInstance.getJobEngineId(), jobInstance.getWebInterfaceUrl());
		} catch (InterruptedException | ExecutionException | IOException e) {
			if(e.getMessage().contains("NotFoundException")) {
				log.warn("flink作业未找到|jobId={}", jobInstance.getJobEngineId());
				//运行状态更新为未知
				if(jobInstance.getInstanceStatus() == InstanceStatus.RUNNING) {
					updateEnity.setEndTime(new Date());
					updateEnity.setInstanceStatus(InstanceStatus.UNKNOWN);
					updateById(updateEnity);
				}
			}else {
				log.error(e.getMessage(), e);
			}
			return;
		}
		updateEnity.setEngineJobStatus(jobDetailsInfo.getJobStatus().name());
		updateEnity.setInstanceStatus(InstanceStatus.of(jobDetailsInfo.getJobStatus()));
		//2.获取异常信息
		if(jobDetailsInfo.getJobStatus() == JobStatus.FAILED) {
			JobExceptionsInfoWithHistory jobExceptionsInfo = flinkJobService.exception(jobInstance.getJobEngineId(), jobInstance.getWebInterfaceUrl());
			updateEnity.setErrMsg(StrUtil.subAfter(jobExceptionsInfo.getRootException(), "(ForkJoinWorkerThread.java:157)", false));
		}
//		//状态已改变
//		if(jobDetailsInfo.getJobStatus() != JobStatus.valueOf(jobInstance.getEngineJobStatus())) {
//			updateById(updateEnity);
//		}
		if(InstanceStatus.isDone(jobDetailsInfo.getJobStatus())) {  //已结束生命流程，设置结束时间
			updateEnity.setEndTime(new Date());
		}
		//3.获取作业统计指标
		List<JobVertexDetailsInfo> jobVertexInfos = new ArrayList<>(jobDetailsInfo.getJobVertexInfos());
		int size = jobVertexInfos.size();
		for(JobVertexDetailsInfo metricInfo : jobVertexInfos) {
			long duration = metricInfo.getDuration();
			IOMetricsInfo metricsInfo = metricInfo.getJobVertexMetrics();
			if(size == 1 || (metricsInfo.getRecordsWritten() > 0 && metricsInfo.getRecordsRead() == 0)) {
				updateEnity.setWriteRowCount(metricsInfo.getRecordsWritten());
				updateEnity.setWriteQps(metricsInfo.getRecordsWritten() / (duration / 1000));
			}
			if(size == 1 || (metricsInfo.getRecordsWritten() == 0 && metricsInfo.getRecordsRead() > 0)) {
				updateEnity.setReadRowCount(metricsInfo.getRecordsRead());
				updateEnity.setReadQps(metricsInfo.getRecordsRead() / (duration / 1000));
			}
		}
		updateById(updateEnity);
		//没有结束，扔到线程池中继续轮询
		if(!InstanceStatus.isDone(jobDetailsInfo.getJobStatus())) {
			DaemonFactory.addTask(new DaemonTaskConfig(jobInstanceId.toString(), DaemonTaskType.flink_job));
		}
	}

	public InstanceCountVO getCounts(){ return jobInstanceMapper.getCounts(); }
	public List<InstanceCountVO> getCountByDate(){ return jobInstanceMapper.getCountByDate(); }
	public InstanceCountVO getCountByType(Integer instanceType){ return jobInstanceMapper.getCountByType(instanceType); }
	public List<InstanceCountVO> instanceLineChart(Integer instanceType){ return jobInstanceMapper.instanceLineChart(instanceType); }

}