package com.ozs.da.service.task.impl;

import java.util.List;
import java.util.UUID;

import com.ozs.da.entity.remotedb.MdmInfoColumn;
import com.ozs.da.service.task.TaskService;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import com.ozs.da.dao.TaskDao;
import com.ozs.da.entity.PageParams;
import com.ozs.da.entity.ReturnData;
import com.ozs.da.entity.remotedb.MdmCreateTableDTO;
import com.ozs.da.entity.remotedb.MdmInfoDatabase;
import com.ozs.da.entity.remotedb.MdmInfoTable;
import com.ozs.da.entity.task.JobOperateCmd;
import com.ozs.da.entity.task.Task;
import com.ozs.da.feign.DaMdmFeign;
import com.ozs.da.job.JobServer;
import com.ozs.da.service.job.impl.SchedulerTimeTriggeJob;
import com.ozs.da.service.job.impl.TriggerJob;
import com.ozs.da.utils.CronExpresUtil;
import com.ozs.da.utils.R;
import com.github.pagehelper.Page;
import com.github.pagehelper.PageHelper;


/**
 * @author 许明
 * on 2019/3/5.
 * @version 2.0.0
 */
@Service
@Transactional(rollbackFor = Exception.class)
public class TaskServiceImpl implements TaskService {

    private static final Logger logger = LogManager.getLogger(TaskServiceImpl.class);

    @Autowired
    private TaskDao taskDao;

    @Autowired
    private DaMdmFeign daMdmFeign;

    @Override
    public void addTask(Task task) throws Exception {
    	String taskCron = task.getTaskCron();
        boolean isValidCron = CronExpresUtil.isValidExpress(taskCron);
        if(!isValidCron){
        	logger.error("Cron表达式[" + taskCron + "]不合法，请检查！");
        	throw new Exception("Cron表达式不合法，请检查！");
        }
        //新增任务，默认新增到临时表
        String taskId = UUID.randomUUID().toString();
        task.setId(taskId);
        task.setTaskStatus("2");
        task.setRunStatus("0");
        R<MdmCreateTableDTO> sourceTableDTO = daMdmFeign.getTableInfoById(task.getSourceTableId());
        R<MdmCreateTableDTO> targetTableDTO = daMdmFeign.getTableInfoById(task.getTargetTableId());
        String sourceTableName = sourceTableDTO.getData().getTableName();
        String targetTableName = targetTableDTO.getData().getTableName();
        logger.info("JOB[" + taskId + "]源表[" + sourceTableName + "][" + task.getSourceTableId() + "]");
        logger.info("JOB[" + taskId + "]目标表[" + targetTableName + "][" + task.getTargetTableId() + "]");
        task.setSourceTableName(sourceTableName);
        task.setTargetTableName(targetTableName);
        taskDao.addTask(task);
        JobServer.MEM_JOB_CACHE.put(taskId, task);
    }

    @Override
    public Task getTaskInfoById(String taskId) throws Exception {
        return taskDao.getTaskInfoById(taskId);
    }

    @Override
    public int updateTaskById(Task task) throws Exception {
        String taskCron = task.getTaskCron();
        boolean isValidCron = CronExpresUtil.isValidExpress(taskCron);
        if(!isValidCron){
            logger.error("Cron表达式[" + taskCron + "]不合法，请检查！");
            throw new Exception("Cron表达式不合法，请检查！");
        }
    	R<MdmCreateTableDTO> sourceTableDTO = daMdmFeign.getTableInfoById(task.getSourceTableId());
        R<MdmCreateTableDTO> targetTableDTO = daMdmFeign.getTableInfoById(task.getTargetTableId());
        String sourceTableName = sourceTableDTO.getData().getTableName();
        String targetTableName = targetTableDTO.getData().getTableName();
        logger.info("JOB[" + task.getId() + "]源表[" + sourceTableName + "][" + task.getSourceTableId() + "]");
        logger.info("JOB[" + task.getId() + "]目标表[" + targetTableName + "][" + task.getTargetTableId() + "]");
        task.setSourceTableName(sourceTableName);
        task.setTargetTableName(targetTableName);
        int result = taskDao.updateTaskById(task);
        Task task1 = taskDao.getTaskInfoById(task.getId());
        
        //根据id从正式表中获取要执行的任务信息，发布到队列，判断是否满足运行条件
        long seq = JobServer.SCHEDULE_CONDITION_JUDGE_BUFFER.next();
        JobServer.SCHEDULE_CONDITION_JUDGE_BUFFER.get(seq).setTask(task1);
        JobServer.SCHEDULE_CONDITION_JUDGE_BUFFER.publish(seq);
        JobServer.MEM_JOB_CACHE.put(task1.getId(), task1);
        return result;
    }

    @Override
    public ReturnData operateTask(JobOperateCmd jobOperateCmd) throws Exception {
        ReturnData returnData = new ReturnData();
        String operateType = jobOperateCmd.getOperateType();
        String result = "";
        switch (operateType) {
            case "1":
                logger.info("收到job[" + jobOperateCmd.getJobId() + "]开始运行命令");
                result = startTask(jobOperateCmd.getJobId());
                break;
            case "2":
                logger.info("收到job[" + jobOperateCmd.getJobId() + "]结束运行命令");
                result = stopTask(jobOperateCmd.getJobId());
                break;
            case "3":
                logger.info("收到job[" + jobOperateCmd.getJobId() + "]立即运行一次的命令");
                result = startTaskNow(jobOperateCmd.getJobId());
                break;
            default:
                logger.info("收到job[" + jobOperateCmd.getJobId() + "]开始运行命令");
                result = startTask(jobOperateCmd.getJobId());
                break;
        }
        if ("0".equals(result)) {
            returnData.setMsg("操作成功");
        }
        if ("-1".equals(result)) {
        	returnData.setCode("-1");
            returnData.setMsg("操作失败");
        }
        if ("2".equals(result)) {
        	returnData.setCode("-1");
        	returnData.setMsg("调度未启用，不允许执行该操作！");
        }
        return returnData;
    }

    @Override
    public Page<Task> listTask(PageParams<Task> pageParams) throws Exception {
        String sortStr = pageParams.getSort() == null ? "createTime" : pageParams.getSort();
        String orderByStr = sortStr + " " + pageParams.getOrder();
        PageHelper.startPage(pageParams.getPage(), pageParams.getPageSize()).setOrderBy(orderByStr);
        List<Task> resultList = taskDao.listTask(pageParams);
        return (Page<Task>) resultList;
    }

    /**
     * 根据id启动任务
     *
     * @param jobId
     */
    private String startTask(String jobId) throws Exception {
        String result = "0";
        //根据id从正式表中获取要执行的任务信息，发布到队列，判断是否满足运行条件
        Task task = JobServer.MEM_JOB_CACHE.get(jobId);
        task.setTaskStatus("1");
        taskDao.updateTaskStatusById(jobId, "1", "");
        long seq = JobServer.SCHEDULE_CONDITION_JUDGE_BUFFER.next();
        JobServer.SCHEDULE_CONDITION_JUDGE_BUFFER.get(seq).setTask(task);
        JobServer.SCHEDULE_CONDITION_JUDGE_BUFFER.publish(seq);
        JobServer.MEM_JOB_CACHE.put(jobId, task);
        return result;
    }

    /**
     * 根据id停止任务
     *
     * @param jobId
     */
    private String stopTask(String jobId) throws Exception {
    	Task task = JobServer.MEM_JOB_CACHE.get(jobId);
    	task.setTaskStatus("2");
        String result = "0";
        taskDao.updateTaskStatusById(jobId, "2", "");
        SchedulerTimeTriggeJob.deleteJobByJobId(task);
        logger.info("停用job[" + jobId + "]成功");
        JobServer.MEM_JOB_CACHE.put(jobId, task);
        return result;
    }

    /**
     * 根据id立即运行任务
     *
     * @param jobId
     */
    private String startTaskNow(String jobId) throws Exception {
        String result = "";
        TriggerJob triggerJob = new TriggerJob();
        Task thisJob = JobServer.MEM_JOB_CACHE.get(jobId);
        if("2".equals(thisJob.getTaskStatus())){
        	result = "2";
        	return result;
    	}
    	try {
            result = triggerJob.execute(jobId, null);
        } catch (Exception e){
            result = "-1";
        }
        if ("0".equals(result)) {
            logger.info("JOBID[" + jobId + "]进行数据采集任务成功");
        }
        if ("-1".equals(result)) {
            logger.info("JOBID[" + jobId + "]进行数据采集任务失败");
        }
        return result;
    }

    @Override
    public MdmInfoDatabase database(String id) throws Exception {
        return daMdmFeign.database(id);
    }

    @Override
    public List<MdmInfoTable> listTableByDatabaseId(String databaseId) {
        return daMdmFeign.listTableByDatabaseId(databaseId);
    }

    @Override
    public List<MdmInfoDatabase> listOriginal(String resourceName) throws Exception {
        return daMdmFeign.listOriginal(resourceName);
    }

    @Override
    public List<MdmInfoDatabase> listDatabase(String resourceName) throws Exception {
        return daMdmFeign.listDatabase(resourceName);
    }

    @Override
    public List<MdmInfoColumn> listColumnByTableId(String tableId) throws Exception {
        return daMdmFeign.listColumnByTableId(tableId);
    }

	@Override
	public void deleteTaskById(String taskId) throws Exception {
		Task task = JobServer.MEM_JOB_CACHE.get(taskId);
		if("1".equals(task.getTaskStatus())){
			throw new Exception("启用状态的调度不允许删除！");
		}
		taskDao.deleteTaskById(taskId);
		
	}
}
