/**
 * 
 */
package org.springblade.etl.engine.sql;

import lombok.extern.slf4j.Slf4j;
import org.springblade.etl.entity.SqlJobInstance;
import org.springblade.etl.entity.SqlJobScript;
import org.springblade.etl.entity.bi.ReportDatasource;
import org.springblade.etl.enums.InstanceStatus;
import org.springblade.etl.service.SqlJobInstanceService;
import org.springblade.etl.service.SqlJobScriptService;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;

/**
 * @author wookvn
 *
 */
@Slf4j
public class SqlJobThread implements Runnable {

	private ReportDatasource ds;
	private SqlJobInstanceService sqlJobInstanceService;
	private SqlJobScriptService sqlJobScriptService;
	private List<SqlJobInstance> jobInstanceList;
	public SqlJobThread(ReportDatasource ds, List<SqlJobInstance> jobInstanceList,
			SqlJobInstanceService sqlJobInstanceService, SqlJobScriptService sqlJobScriptService) {
		this.ds = ds;
		this.sqlJobInstanceService = sqlJobInstanceService;
		this.sqlJobScriptService = sqlJobScriptService;
		this.jobInstanceList = jobInstanceList;
	}
	
	@Override
	public void run() {
		//先排序再执行
		jobInstanceList.sort(Comparator.comparing(SqlJobInstance::getJobScriptSortNum));
		List<SqlJobInstance> errJobInstanceList = new ArrayList<>();
		for(SqlJobInstance jobInstance : jobInstanceList) {
			SqlJobScript sqlJobScript = sqlJobScriptService.getById(jobInstance.getJobScriptId());
			String sqlScript = sqlJobScript.getSqlScript();
			if(errJobInstanceList.isEmpty()) {
				try {
					//1.先更新为运行中
					jobInstance.setStartTime(new Date());
					jobInstance.setInstanceStatus(InstanceStatus.RUNNING);
					sqlJobInstanceService.updateById(jobInstance);
					//2.开始执行SQL脚本
					sqlJobScriptService.runScript(ds, sqlScript, true);
					jobInstance.setEndTime(new Date());
					jobInstance.setInstanceStatus(InstanceStatus.SUCCESS);
					sqlJobInstanceService.updateById(jobInstance);
				} catch (Exception e) {
					log.error(e.getMessage(), e);
					jobInstance.setEndTime(new Date());
					jobInstance.setInstanceStatus(InstanceStatus.FAILURE);
					jobInstance.setId(jobInstance.getId());
					jobInstance.setErrMsg(e.getMessage());
					errJobInstanceList.add(jobInstance);
				}
			}else {
				jobInstance.setInstanceStatus(InstanceStatus.CANCELED);
				errJobInstanceList.add(jobInstance);
			}
		}
		if(!errJobInstanceList.isEmpty()) {
			sqlJobInstanceService.updateBatchById(errJobInstanceList);
		}
	}
}
