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

import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.springblade.core.secure.utils.AuthUtil;
import org.springblade.core.tool.exception.ServiceException;
import org.springblade.core.tool.utils.CollectionUtil;
import org.springblade.core.tool.utils.SpringUtil;
import org.springblade.datasource.constants.CubeConstants;
import org.springblade.datasource.context.DialectContext;
import org.springblade.etl.dto.dsy.DsyJobDto;
import org.springblade.etl.dto.dsy.SourceTableColumnDto;
import org.springblade.etl.dto.dsy.TargetTableColumnDto;
import org.springblade.etl.engine.datax.event.JobRunEvent;
import org.springblade.etl.engine.datax.event.JobRunInstance;
import org.springblade.etl.engine.datax.event.JobRunTarget;
import org.springblade.etl.entity.DsyJob;
import org.springblade.etl.entity.DsyJobInstance;
import org.springblade.etl.entity.DsyJobPermission;
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.JobStatus;
import org.springblade.etl.mapper.DsyJobMapper;
import org.springblade.etl.service.bi.ReportDatasourceService;
import org.springblade.etl.util.EtlJsonUtil;
import org.springblade.system.client.DatasourceFunctionClient;
import org.springblade.system.client.model.DatasouceFunctionModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;

import java.time.LocalDateTime;
import java.util.*;
import java.util.regex.Matcher;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 *	数据同步任务 服务处理
 *
 * @author aokunsang
 * @date   2023-09-21
 */
@Service
public class DsyJobService extends ServiceImpl<DsyJobMapper, DsyJob>{
	
	@Autowired
	private DsyJobTableService dsyJobTableService;
	
	@Autowired
	private DsyJobPermissionService dsyJobPermissionService;
	
	@Autowired
	private DatasourceFunctionClient dsFunctionClient;
	
	@Autowired
	private DsyJobInstanceService jobInstanceService;
	
	@Autowired
	private DialectContext dialectContext;
	
//	@Autowired
//	private TransformContext transformContext;
//	
//	@Autowired
//	private QueryWrapperService queryWrapperService;
//	
//	@Autowired
//	private DataxJobService dataxJobService;
//	
	@Autowired
	private ReportDatasourceService datasourceService;

	/**
	 * 
	 * @param userId
	 * @return
	 */
	public List<Integer> categoryIdListOfPerm(Long userId){
		return baseMapper.categoryIdListOfPerm(userId);
	}
	
	/**
	 * 
	 * @param userId
	 * @return
	 */
	public List<Integer> jobIdListOfPerm(Long userId){
		return baseMapper.jobIdListOfPerm(userId);
	}
	
	/**
	 * 
	 * @param jobId
	 * @return
	 */
	@Transactional
	public boolean doRemove(Integer jobId) {
		boolean ret = removeById(jobId);
		if(ret) {
			dsyJobTableService.remove(Wrappers.<DsyJobTable>lambdaQuery().eq(DsyJobTable::getJobId, jobId));
		}
		return ret;
	}
	
	/**
	 * 逻辑运行 - 添加任务实例[待执行状态]
	 * @param jobId
	 * @param runType
	 */
	public void jobRunLogic(Integer jobId, InstanceRunType runType) {
		//查询任务记录
		DsyJob dsyJob = getById(jobId);
		//查询任务表记录
		List<DsyJobTable> jobTableList = dsyJobTableService.listByJobId(jobId);
		Assert.notNull(dsyJob, "数据同步任务为空");
		Assert.notEmpty(jobTableList, "数据同步任务的表数据为空");
		List<DsyJobInstance> jobInstanceList = new ArrayList<>();
		for(DsyJobTable jobTable : jobTableList) {
			DsyJobInstance jobInstance = new DsyJobInstance();
			jobInstance.setJobId(jobId);
			jobInstance.setJobTableId(jobTable.getId());
			jobInstance.setServerIp("unknown");
			jobInstance.setInstanceType(dsyJob.getJobType().getInstanceType());
			jobInstance.setInstanceStatus(InstanceStatus.WAITING);
			jobInstance.setInstanceRunType(runType);
			jobInstance.setCreateTime(LocalDateTime.now());
			jobInstance.setTenantId(dsyJob.getTenantId());
			jobInstance.setSourceTableName(jobTable.getSourceTableName());
			jobInstance.setTargetTableName(jobTable.getTargetTableName());
//			jobInstance.setStartTime(new Date());
//			jobInstance.setEndTime(new Date());
			jobInstance.setSyncType(Optional.<DataSyncType>ofNullable(jobTable.getSyncType()).orElse(dsyJob.getSyncType()));
			jobInstanceList.add(jobInstance);
		}
		jobInstanceService.saveBatch(jobInstanceList);
		
		//任务添加成功后，往队列发送数据
		List<JobRunTarget> targets = jobInstanceList.stream().map(item -> new JobRunTarget(item.getId(), runType)).collect(Collectors.toList());
		JobRunInstance.getInstance().add(targets);
		SpringUtil.publishEvent(new JobRunEvent());
	}
	
	/**
	 * 保存或修改
	 * @param dsyJobDto
	 * @return
	 */
	@Transactional
	public Integer doSaveOrUpdate(DsyJobDto dsyJobDto) {
		DsyJob job = dsyJobDto.getJob();
		List<DsyJobTable> tableList = dsyJobDto.getTableList();
		List<Integer> dataPermIdList = dsyJobDto.getDataPermIdList();
		if(job.getId() == null) {
			job.setCreateTime(LocalDateTime.now());
			job.setCreateBy(AuthUtil.getNickName());
			job.setCreateUserId(AuthUtil.getUserId());
		}else {
			//判断是否上线状态
			DsyJob dbJob = this.getById(job.getId());
			Assert.notNull(dbJob, "任务不存在");
			if(dbJob.getJobStatus() == JobStatus.ONLINE){
				throw new ServiceException("任务需要先下线后才能修改");
			}
			job.setUpdateBy(AuthUtil.getNickName());
			job.setUpdateUserId(AuthUtil.getUserId());
			job.setUpdateTime(LocalDateTime.now());
			dsyJobTableService.remove(Wrappers.<DsyJobTable>lambdaQuery().eq(DsyJobTable::getJobId, job.getId()));
		}
		if(job.getJobStatus() == org.springblade.etl.enums.JobStatus.ONLINE_RUN) {
			job.setJobStatus(org.springblade.etl.enums.JobStatus.ONLINE);
		}
		//保存任务
		saveOrUpdate(job);
		Integer jobId = job.getId();
		ReportDatasource sourceDatasource = datasourceService.getById(job.getSourceDsId());
		ReportDatasource targetDatasource = datasourceService.getById(job.getTargetDsId());
		if(sourceDatasource == null) {
			throw new ServiceException("数据源库找不到");
		}
		if(targetDatasource == null) {
			throw new ServiceException("目标数据源找不到");
		}
		//查询数据源函数列表
		List<DatasouceFunctionModel> funcList = dsFunctionClient.list(job.getSourceDsType().getValue(), AuthUtil.getTenantId());
		//开始入库表数据
		for(DsyJobTable jobTable : tableList) {
			//如果带有schema，先清除掉；重新设置查询的表名
			jobTable.setSourceTableName(StrUtil.subSuf(jobTable.getSourceTableName(), jobTable.getSourceTableName().lastIndexOf('.') + 1));
			jobTable.setTargetTableName(StrUtil.subSuf(jobTable.getTargetTableName(), jobTable.getTargetTableName().lastIndexOf('.') + 1));
			DataSyncType syncType = Optional.<DataSyncType>ofNullable(jobTable.getSyncType()).orElse(job.getSyncType());
			//判断增量同步，必须有增量字段
			if(syncType == DataSyncType.INCREMENT && StrUtil.isBlank(jobTable.getIncrementColumnName())) {
				throw new ServiceException("【增量同步】请选择增量字段|目标表："+jobTable.getTargetTableName());
			}
			jobTable.setJobId(jobId);
			jobTable.setCreateTime(LocalDateTime.now());
			//解析preSql
			jobTable.setPreSql(_parsePreSql(jobTable, job.getTruncateTableStatus()));
			//解析querySql
			jobTable.setQuerySql(_parseQuerySql(jobTable, job, funcList, sourceDatasource));
		}
		dsyJobTableService.saveBatch(tableList);
		//开始操作权限组
		if(jobId != null && jobId > 0 && CollectionUtil.isNotEmpty(dataPermIdList)) {
			List<DsyJobPermission> jobPermList = new ArrayList<>();
			dataPermIdList.forEach(item -> {
				DsyJobPermission jobPermission = new DsyJobPermission();
				jobPermission.setCreateBy(AuthUtil.getNickName());
				jobPermission.setCreateTime(new Date());
				jobPermission.setCreateUserId(AuthUtil.getUserId());
				jobPermission.setDataPermId(item);
				jobPermission.setJobId(jobId);
				jobPermList.add(jobPermission);
			});
			dsyJobPermissionService.saveBatch(jobPermList);
		}
		return job.getId();
	}
	
	/**
	 * 解析preSql
	 * @param jobTable
	 * @return
	 */
	private String _parsePreSql(DsyJobTable jobTable, Integer truncateTableStatus) {
		if(Optional.<Integer>ofNullable(jobTable.getTruncateTableStatus()).orElse(truncateTableStatus) == 1) {
//			return String.format("truncate table %s", jobTable.getSourceTableName());
			return "truncate table @table";  //当目标入库表是多个时，该命令可以轮询执行清空操作
		}
		return null;
	}
	
	/**
	 * 解析querySql
	 * @param jobTable
	 * @return
	 */
	private String _parseQuerySql(DsyJobTable jobTable, DsyJob job, List<DatasouceFunctionModel> funcList, ReportDatasource sourceDatasource) {
		List<TargetTableColumnDto> targetTableColumnList = EtlJsonUtil.parseJsonArray(jobTable.getTargetColumnJson(), TargetTableColumnDto.class);
		if(CollectionUtil.isEmpty(targetTableColumnList)) {
			throw new ServiceException("目标数据源字段信息不能为空");
		}
		List<SourceTableColumnDto> sourceTableColumnList = EtlJsonUtil.parseJsonArray(jobTable.getSourceColumnJson(), SourceTableColumnDto.class);
		if(CollectionUtil.isEmpty(sourceTableColumnList)) {
			throw new ServiceException("输入数据源字段信息不能为空");
		}
		if(targetTableColumnList.size() < sourceTableColumnList.size()) {
			throw new ServiceException("目标字段和输入字段不匹配");
		}
		//非虚拟字段的目标列数量
		long notVirualTargetCount = targetTableColumnList.stream().filter(item -> Optional.<Integer>ofNullable(item.getVirtualColumnStatus()).orElse(0) == 0).count();
		if(notVirualTargetCount > sourceTableColumnList.size()) {
			throw new ServiceException("目标字段和输入字段不匹配，请检查");
		}
		List<String> columnList = new ArrayList<>();
		for(int i = 0; i < targetTableColumnList.size(); i++) {
			TargetTableColumnDto targetTableColumn = targetTableColumnList.get(i);
			//是否虚拟计算字段
			if(Optional.<Integer>ofNullable(targetTableColumn.getVirtualColumnStatus()).orElse(0) == 1) {
				if(StrUtil.isBlank(targetTableColumn.getColumnFormula())) {
					throw new ServiceException(String.format("目标字段[%s]的计算公式为空", targetTableColumn.getColumnName()));
				}
				if(StrUtil.containsAny(targetTableColumn.getColumnFormula().toLowerCase(), CubeConstants.AGGREGATION_FUNCTIONS)) {
					throw new ServiceException(String.format("目标字段[%s]的计算公式不能包含聚合函数", targetTableColumn.getColumnName()));
				}
				//格式化虚拟字段
				columnList.add(_formatColumnFormula(targetTableColumn, funcList, sourceTableColumnList));
			}else {
				//非虚拟字段时，获取源字段信息
				SourceTableColumnDto sourceTableColumn = sourceTableColumnList.get(i);
				String columnName = dialectContext.getDialect(sourceDatasource.getDsTypeSyl()).dbEscape(sourceTableColumn.getColumnName());
				columnList.add(columnName);
			}
		}
		String sourceTableName = dialectContext.getDialect(sourceDatasource.getDsTypeSyl()).dbTableName(sourceDatasource, jobTable.getSourceTableName());
		return String.format("select %s from %s", String.join(",", columnList), sourceTableName);
	}
	
	/**
	 * 格式化计算字段
	 * @param columnFormula
	 * @param sourceTableColumnList
	 * @return
	 */
	private String _formatColumnFormula(TargetTableColumnDto targetTableColumn, List<DatasouceFunctionModel> funcList, List<SourceTableColumnDto> sourceTableColumnList) {
		String columnFormula = targetTableColumn.getColumnFormula();
		//1.解析自定义函数
		if (CollectionUtil.isNotEmpty(funcList)) {
			final String _columnFormula = columnFormula;
			DatasouceFunctionModel function = funcList.stream().filter(item -> _columnFormula.contains(item.getMethodName()+"(")).findAny().orElse(null);
			if (function != null) {
				// 获取自定义函数的参数
				String subBetweenStr = StrUtil.removeSuffix(StrUtil.removePrefixIgnoreCase(columnFormula, function.getMethodName() + "("), ")");
				String[] params = subBetweenStr.split(",");
				Map<String, String> paramMap = IntStream.range(0, params.length).boxed()
						.collect(Collectors.toMap(t -> t.toString(), i -> params[i]));
				columnFormula = StrUtil.format(function.getFuncSql(), paramMap);
				if (columnFormula.contains("{") || columnFormula.contains("}")) {
					throw new ServiceException("虚拟字段[" + targetTableColumn.getColumnName() + "]参数不对应");
				}
			}
		}
		//2.替换变量
		Matcher matcher = CubeConstants.COLUMN_FORMULA_PATTERN.matcher(columnFormula);
		while (matcher.find()) {
			String group = matcher.group();
			String key = group.substring(1, group.length() - 1);
			Optional<SourceTableColumnDto> optional = sourceTableColumnList.stream().filter(item -> item.getColumnName().equals(key)).findFirst();
			if (!optional.isPresent()) {
				throw new ServiceException("未找到需要替换的字段:" + key);
			}
			columnFormula = columnFormula.replace(group, optional.get().getColumnName());
		}
		return columnFormula;
	}
}