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

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.nacos.common.utils.ExceptionUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.springblade.core.tool.exception.ServiceException;
import org.springblade.core.tool.utils.SpringUtil;
import org.springblade.datasource.context.DialectContext;
import org.springblade.datasource.enums.DsTypeEnum;
import org.springblade.datasource.enums.ViewColumnTypeEnum;
import org.springblade.datasource.transform.TransformContext;
import org.springblade.datasource.transform.general.GeneralModel;
import org.springblade.etl.dto.InstanceStatusCount;
import org.springblade.etl.dto.dsy.AutoCreateTableDto;
import org.springblade.etl.dto.dsy.RunningInstanceServerDto;
import org.springblade.etl.dto.dsy.TargetTableColumnDto;
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.datax.DataxJobService;
import org.springblade.etl.engine.datax.dto.DataxReq;
import org.springblade.etl.engine.datax.dto.DataxRes;
import org.springblade.etl.engine.datax.event.JobRunEvent;
import org.springblade.etl.entity.DsyJob;
import org.springblade.etl.entity.DsyJobInstance;
import org.springblade.etl.entity.DsyJobTable;
import org.springblade.etl.entity.bi.ReportDatasource;
import org.springblade.etl.enums.DataSyncType;
import org.springblade.etl.enums.InstanceRunType;
import org.springblade.etl.enums.InstanceStatus;
import org.springblade.etl.enums.InstanceType;
import org.springblade.etl.mapper.DsyJobInstanceMapper;
import org.springblade.etl.properties.DsyProperties.ServerInfo;
import org.springblade.etl.service.bi.ReportDatasourceService;
import org.springblade.etl.task.model.DaemonTaskConfig;
import org.springblade.etl.task.model.DaemonTaskType;
import org.springblade.etl.task.thread.DaemonFactory;
import org.springblade.etl.util.EtlJsonUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import java.math.BigDecimal;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

/**
 *	数据同步任务实例 服务处理
 *
 * @author aokunsang
 * @date 2023-09-21
 */
@Slf4j
@Service
public class DsyJobInstanceService extends ServiceImpl<DsyJobInstanceMapper, DsyJobInstance> {

    @Autowired
    private DsyJobTableService dsyJobTableService;

    @Autowired
    private DataxJobService dataxJobService;

    @Autowired
    private DsyJobService dsyJobService;

    @Autowired
    private ReportDatasourceService datasourceService;

    @Autowired
    private TransformContext transformContext;

    @Autowired
    private QueryWrapperService queryWrapperService;

    @Autowired
    private DialectContext dialectContext;

    /**
     * @param statReq
     * @return
     */
    public List<JobStatRes> stat(JobStatReq statReq) {
        return baseMapper.stat(statReq);
    }

    /**
     * 查询正在运行的所有任务实例
     *
     * @return
     */
    public List<DsyJobInstance> listAllRunningInstanceList() {
        return baseMapper.selectAllRunningInstanceList();
    }

    /**
     * 查询待运行的所有任务实例
     *
     * @return
     */
    public List<DsyJobInstance> listAllWaitingInstanceList() {
        return baseMapper.selectAllWaitingInstanceList();
    }

    /**
     * 统计datax服务器上正在运行的实例数量
     *
     * @return
     */
    public List<RunningInstanceServerDto> countServerByRunningInstance() {
        return baseMapper.countServerByRunningInstance();
    }

    /**
     * @param page
     * @param queryWrapper
     * @return
     */
    public IPage<DsyJobInstance> customPage(IPage<DsyJobInstance> page, LambdaQueryWrapper<DsyJobInstance> queryWrapper) {
        return baseMapper.customPage(page, queryWrapper);
    }

    /**
     * @param queryWrapper
     * @return
     */
    public Long countPage(LambdaQueryWrapper<DsyJobInstance> queryWrapper) {
        return baseMapper.countPage(queryWrapper);
    }

    /**
     * @param jobId
     * @return
     */
    public InstanceStatusCount statInstanceStatusByJobId(Integer jobId) {
        return baseMapper.statInstanceStatusByJobId(jobId);
    }

    /**
     * 取消任务实例
     *
     * @param jobInstanceId
     * @throws Exception
     */
    public void cancelJobInstance(Integer jobInstanceId) throws Exception {
        DsyJobInstance jobInstance = getById(jobInstanceId);
        if (jobInstance.getInstanceStatus() == InstanceStatus.RUNNING && !"unknown".equals(jobInstance.getServerIp())) {
            String errorMsg = dataxJobService.cancel(jobInstance.getId(), jobInstance.getDataxConfigName(), jobInstance.getServerIp());
            if (StrUtil.isNotEmpty(errorMsg)) {
                throw new ServiceException("取消任务实例失败，失败原因：" + errorMsg);
            }
        }
        DsyJobInstance updateEntity = new DsyJobInstance();
        updateEntity.setId(jobInstanceId);
        updateEntity.setEndTime(new Date());
        updateEntity.setInstanceStatus(InstanceStatus.CANCELED);
        updateById(updateEntity);
    }

    /**
     * 刷新数据同步任务实例
     *
     * @param jobInstanceId
     */
    public void refreshJobInstance(Integer jobInstanceId) {
        DsyJobInstance dsyJobInstance = getById(jobInstanceId);
        if (dsyJobInstance.getInstanceStatus() != InstanceStatus.RUNNING) {
            return;
        }

        DsyJobInstance updateEntity = new DsyJobInstance();
        updateEntity.setId(dsyJobInstance.getId());

        //读取datax的log日志信息，并且初始化到updateEntity
        dataxJobService.readMetrics(updateEntity, dsyJobInstance.getServerIp(), dsyJobInstance.getDataxLogName());

        //获取当前datax作业执行状态
        InstanceStatus instanceStatus = updateEntity.getInstanceStatus();

        //任务已结束
        boolean isDone = instanceStatus == InstanceStatus.SUCCESS || instanceStatus == InstanceStatus.FAILURE;
        if (isDone && updateEntity.getEndTime() == null) {
            updateEntity.setEndTime(new Date());
        }

        //更新状态数据
        updateById(updateEntity);

        //如果开启增量，任务同步成功后，需要更新最大值和最小值到jobTable
        if (instanceStatus == InstanceStatus.SUCCESS && dsyJobInstance.getSyncType() == DataSyncType.INCREMENT) {
            DsyJobTable tableEntity = new DsyJobTable();
            tableEntity.setId(dsyJobInstance.getJobTableId());
            tableEntity.setStartIncrementData(dsyJobInstance.getStartIncrementData());
            tableEntity.setEndIncrementData(dsyJobInstance.getEndIncrementData());
            dsyJobTableService.updateById(tableEntity);
        }

        //没有结束，扔到线程池中继续轮询
        if (!isDone) {
            DaemonFactory.addTask(new DaemonTaskConfig(jobInstanceId.toString(), DaemonTaskType.datax_job));
        } else { //已结束，发送消息开始执行下一个任务
            SpringUtil.publishEvent(new JobRunEvent());
        }
    }

    /**
     * 运行实例
     *
     * @param jobInstanceId
     * @param runType
     */
    public void runJobInstance(Integer jobInstanceId, InstanceRunType runType, ServerInfo serverInfo) {
        DsyJobInstance dsyJobInstance = getById(jobInstanceId);
        if (dsyJobInstance.getInstanceStatus() != InstanceStatus.WAITING) {
            return;
        }
        //更新状态为运行中
        DsyJobInstance uptJobInstance = new DsyJobInstance();
        uptJobInstance.setId(jobInstanceId);
        uptJobInstance.setInstanceStatus(InstanceStatus.RUNNING);
        updateById(uptJobInstance);

        DataxRes dataxRes = null;
        String errMsg = null;
        Date startTime = null;
        DataSyncType dataSyncType = null;
        try {
            DsyJob dsyJob = dsyJobService.getById(dsyJobInstance.getJobId());
            DsyJobTable jobTable = dsyJobTableService.getById(dsyJobInstance.getJobTableId());

            Assert.notNull(dsyJob, "任务已删除");
            Assert.notNull(jobTable, "任务表已删除");
            dataSyncType = Optional.<DataSyncType>ofNullable(jobTable.getSyncType()).orElse(dsyJob.getSyncType());

            //查询source和target数据源信息
            List<ReportDatasource> dsList = datasourceService.listByIds(Arrays.asList(dsyJob.getSourceDsId(), dsyJob.getTargetDsId()));
            Map<Integer, ReportDatasource> dsMap = dsList.stream().collect(Collectors.toMap(ReportDatasource::getId, t -> t));
            ReportDatasource sourceDs = dsMap.get(dsyJob.getSourceDsId());
            ReportDatasource targetDs = dsMap.get(dsyJob.getTargetDsId());
            Assert.notNull(sourceDs, "输入数据源不存在");
            Assert.notNull(targetDs, "目标数据源不存在");

            DataxReq dataxReq = new DataxReq();

            dataxReq.setSyncType(dataSyncType);
            dataxReq.setJobTableId(jobTable.getId());

            String sourcePluginPrefix = sourceDs.getDsTypeSyl() == DsTypeEnum.dameng ? "rdbms" : sourceDs.getDsTypeSyl().name().toLowerCase();
            String targetPluginPrefix = targetDs.getDsTypeSyl() == DsTypeEnum.dameng ? "rdbms" : targetDs.getDsTypeSyl().name().toLowerCase();

            dataxReq.setSourcePlugin(sourcePluginPrefix + "reader");
            dataxReq.setSourceJdbcUrl(sourceDs.getUrl().replaceAll("CONVERT_TO_NULL", "convertToNull"));
            dataxReq.setSourceUser(sourceDs.getDsUser());
            dataxReq.setSourcePassword(sourceDs.getDsPassword());
            dataxReq.setQuerySql(jobTable.getQuerySql());

            dataxReq.setPreSql(jobTable.getPreSql());
            dataxReq.setTargetPlugin(targetPluginPrefix + "writer");
            dataxReq.setTargetJdbcUrl(targetDs.getUrl().replaceAll("CONVERT_TO_NULL", "convertToNull"));
            dataxReq.setTargetUser(targetDs.getDsUser());
            dataxReq.setTargetPassword(targetDs.getDsPassword());
            String targetTableName = dialectContext.getDialect(targetDs.getDsTypeSyl()).dbTableName(targetDs, jobTable.getTargetTableName());
            dataxReq.setTargetTableName(targetTableName);

            List<TargetTableColumnDto> targetColumnList = EtlJsonUtil.parseJsonArray(jobTable.getTargetColumnJson(), TargetTableColumnDto.class);
            dataxReq.setTargetColumns(targetColumnList.stream().map(item -> "\"" + dialectContext.getDialect(targetDs.getDsTypeSyl()).dbEscape(item.getColumnName()) + "\"").collect(Collectors.joining(",")));

            //构造自动建表model
            AutoCreateTableDto autoCreateTable = null;
            //是否自动建表
            if (Optional.<Integer>ofNullable(jobTable.getCreateTableStatus()).orElse(0) == 1) {
                autoCreateTable = new AutoCreateTableDto();
                autoCreateTable.setTableComment(jobTable.getTargetTableComment());
                autoCreateTable.setJobTableId(jobTable.getId());
                autoCreateTable.setSyncType(dataSyncType);
                autoCreateTable.setTableName(jobTable.getTargetTableName());
                autoCreateTable.setModelList(_buildCreateTableModel(targetColumnList));
            }

            //是否增量同步
            if (dataSyncType == DataSyncType.INCREMENT) {
                dataxReq.setStartIncrementData(jobTable.getEndIncrementData());
                //查询最大值
                dataxReq.setEndIncrementData(_queryMaxValue(sourceDs, jobTable.getIncrementColumnName(), jobTable.getSourceTableName()));
                String whereSql = "";
                if (StrUtil.isNotBlank(dataxReq.getEndIncrementData())) {
                    String incrConditionSql = incrementConditionSql(jobTable, dataxReq.getStartIncrementData(), dataxReq.getEndIncrementData());
                    whereSql = StrUtil.containsIgnoreCase(dataxReq.getQuerySql(), " where ") ? (" and " + incrConditionSql) : (" where " + incrConditionSql);
                }
                dataxReq.setQuerySql(dataxReq.getQuerySql() + whereSql);
            }

            //设置datax mysqlwriter的特殊扩展参数[Mysql支持upsert语义设置，且该参数仅支持mysql]
            if (targetDs.getDsTypeSyl() == DsTypeEnum.mysql) {
                dataxReq.setTargetCustomParamJson("\"writeMode\": \"update\",");
            }

            startTime = new Date();
            //开始自动建表操作
            if (autoCreateTable != null) {
                long oneTime = System.currentTimeMillis();
                ReportDatasource _sourceDs = new ReportDatasource();
                _sourceDs.setDsType(DsTypeEnum.common.getValue());
                transformContext.transformAndCreatTable(_sourceDs, targetDs,
                        autoCreateTable.getModelList(), null, autoCreateTable.getTableName(), autoCreateTable.getTableComment(), false);
                log.info("数据同步:自动建表结束|cost:{}ms", (System.currentTimeMillis() - oneTime));
            }

            //非同步结构开始任务执行
            if (dataSyncType != DataSyncType.STRUCTURE) {
                dataxRes = dataxJobService.submit(dataxReq, serverInfo);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            errMsg = ExceptionUtil.getStackTrace(e);
        } finally {
            DsyJobInstance jobInstance = new DsyJobInstance();
            jobInstance.setId(jobInstanceId);
            jobInstance.setServerIp(serverInfo.getIp());
            jobInstance.setInstanceType(InstanceType.BATCH);
            jobInstance.setCreateTime(LocalDateTime.now());
            jobInstance.setTenantId(dsyJobInstance.getTenantId());
            jobInstance.setSyncType(dataSyncType);
            jobInstance.setInstanceRunType(runType);
            //任务包含两种结果形式：1.同步表结构；2.增量或全量数据同步
            if (dataxRes != null) {
                jobInstance.setStartTime(dataxRes.getStartTime());
                jobInstance.setStartIncrementData(dataxRes.getStartIncrementData());
                jobInstance.setEndIncrementData(dataxRes.getEndIncrementData());
                jobInstance.setErrMsg(dataxRes.getErrorMsg());
                jobInstance.setDataxConfigName(dataxRes.getConfigName());
                jobInstance.setDataxLogName(dataxRes.getLogName());
                jobInstance.setInstanceStatus(StrUtil.isEmpty(dataxRes.getErrorMsg()) ? InstanceStatus.RUNNING : InstanceStatus.FAILURE);
            } else {
                jobInstance.setStartTime(startTime);
                jobInstance.setErrMsg(errMsg);
                jobInstance.setInstanceStatus(StrUtil.isEmpty(errMsg) ? InstanceStatus.SUCCESS : InstanceStatus.FAILURE);
            }
            if (dataxRes == null || StrUtil.isNotEmpty(dataxRes.getErrorMsg())) {
                jobInstance.setEndTime(new Date());
            }
            updateById(jobInstance);
            //运行成功，添加到任务轮训调度中
            if (dataxRes != null && StrUtil.isEmpty(dataxRes.getErrorMsg())) {
                DaemonFactory.addTask(new DaemonTaskConfig(jobInstance.getId().toString(), DaemonTaskType.datax_job));
            }
        }
    }


    /**
     * 构造自动建表model
     *
     * @param targetColumnList
     * @return
     */
    private List<GeneralModel> _buildCreateTableModel(List<TargetTableColumnDto> targetColumnList) {
        List<GeneralModel> dataList = new ArrayList<>();
        for (TargetTableColumnDto targetTableColumn : targetColumnList) {
            GeneralModel createTableModel = new GeneralModel();
            createTableModel.setSourceColumn(false);
            createTableModel.setColumnComments(targetTableColumn.getColumnComment());
            createTableModel.setColumnName(targetTableColumn.getColumnName());
            createTableModel.setPrimaryKey(targetTableColumn.getPrimaryKey());
            createTableModel.setColumnSize(targetTableColumn.getColumnSize());
            createTableModel.setDecimalDigits(targetTableColumn.getPrecisionSize());
            createTableModel.setType(targetTableColumn.getColumnType().name());
            createTableModel.setDestColumnName(targetTableColumn.getColumnName());
            dataList.add(createTableModel);
        }
        return dataList;
    }

    /**
     * @param sourceDs
     * @param incrementColumnName
     * @param sourceTableName
     * @return
     */
    private String _queryMaxValue(ReportDatasource sourceDs, String incrementColumnName, String sourceTableName) {
        long startTime = System.currentTimeMillis();
        //重设表名
        sourceTableName = dialectContext.getDialect(sourceDs.getDsTypeSyl()).dbTableName(sourceDs, sourceTableName);
        String queryMaxValueSql = String.format("select max(%s) as max_value from %s", incrementColumnName, sourceTableName);
        Map<String, Object> maxValueMap = queryWrapperService.tempSelectOne(sourceDs, queryMaxValueSql);
        log.info("数据同步:查询增量字段最大值结束|columnName:{},value:{},cost:{}ms", incrementColumnName, maxValueMap, (System.currentTimeMillis() - startTime));
        String maxValueStr = "";
        if (maxValueMap == null) {
            return maxValueStr;
        }
        Object maxValue = maxValueMap.get("max_value");
        if (maxValue instanceof Timestamp) {
            Timestamp value = (Timestamp) maxValue;
            maxValueStr = DateUtil.format(new Date(value.getTime()), "yyyy-MM-dd HH:mm:ss");
        } else if (maxValue instanceof BigDecimal) {
            BigDecimal value = (BigDecimal) maxValue;
            maxValueStr = value.toPlainString();
        } else {
            maxValueStr = maxValue.toString();
        }
        return maxValueStr;
    }

    /**
     * 构造datax增量条件查询语句
     *
     * @param jobTable
     * @return
     */
    private String incrementConditionSql(DsyJobTable jobTable, String startData, String endData) {
        //构造增量同步where条件
        boolean isStr = jobTable.getIncrementColumnType() == ViewColumnTypeEnum.STRING || jobTable.getIncrementColumnType() == ViewColumnTypeEnum.DATETIME;
        String startExpression = isStr ? String.format("%s > '%s'", jobTable.getIncrementColumnName(), startData)
                : String.format("%s > %s", jobTable.getIncrementColumnName(), startData);
        String endExpression = isStr ? String.format("%s <= '%s'", jobTable.getIncrementColumnName(), endData)
                : String.format("%s <= %s", jobTable.getIncrementColumnName(), endData);
        String whereSql = "";
        if (StrUtil.isEmpty(startData)) {
            whereSql = endExpression;
        } else {
            whereSql = startExpression + " and " + endExpression;
        }
        return whereSql;
    }

    public InstanceCountVO getCounts() {

        return baseMapper.getCounts();
    }

    public List<InstanceCountVO> getCountByDays(int day) {

        return baseMapper.getCountByDays(day);
    }

    public InstanceCountVO getCountToday() {

        return baseMapper.getCountToday();
    }

    public List<InstanceCountVO> instanceLineChart() {

        return baseMapper.instanceLineChart();
    }

    public IPage<DsyJobInstance> pageDo(IPage<DsyJobInstance> page) {

        return baseMapper.pageDo(page);
    }
}