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

import java.io.Writer;
import java.net.URL;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.ConfigUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
import org.apache.flink.runtime.rest.handler.legacy.messages.ClusterOverviewWithVersion;
import org.apache.flink.runtime.rest.messages.JobExceptionsInfoWithHistory;
import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointingStatistics;
import org.apache.flink.runtime.rest.messages.job.JobDetailsInfo;
import org.springblade.core.tool.utils.CollectionUtil;
import org.springblade.datasource.enums.DsTypeEnum;
import org.springblade.etl.config.EtlImportConfig;
import org.springblade.etl.engine.JobService;
import org.springblade.etl.engine.flink.client.base.JobClient;
import org.springblade.etl.engine.flink.client.base.RestClient;
import org.springblade.etl.engine.flink.client.rest.FlinkRestClient;
import org.springblade.etl.engine.flink.client.session.SessionFactory;
import org.springblade.etl.engine.flink.client.submit.CommandFactory;
import org.springblade.etl.engine.flink.client.submit.SubmitCommand;
import org.springblade.etl.engine.flink.dto.PackageJarJob;
import org.springblade.etl.engine.flink.dto.SubmitJobReq;
import org.springblade.etl.engine.flink.dto.SubmitJobRes;
import org.springblade.etl.engine.flink.util.FlinkKit;
import org.springblade.etl.enums.ConnectorType;
import org.springblade.etl.enums.DeploymentTarget;
import org.springblade.etl.enums.FlinkDeployMode;
import org.springblade.etl.enums.PluginType;
import org.springblade.etl.enums.ResourceProvider;
import org.springblade.etl.properties.EtlProperties;
import org.springblade.etl.util.FileKit;
import org.springblade.etl.util.SeaTunnelKit;
import org.springblade.etl.util.SystemKit;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.FileCopyUtils;

import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.extra.spring.SpringUtil;
import lombok.extern.slf4j.Slf4j;

/**
 * @author wookvn
 *
 */
@Slf4j
@Service
public class FlinkJobService implements JobService {

	@Autowired
	private EtlProperties etlConfig;

	@Override
	public SubmitJobRes submit(SubmitJobReq submitJobDto) throws Exception {
		// 获取临时文件目录
		final Path tempWorkspace = SystemKit.getRandomWorkspace();
		try {
			Path finkHomePath = Paths.get(etlConfig.getFlinkHome());
			Path seatunnelHomePath = Paths.get(etlConfig.getSeatunnelHome());

			// 获取config.json文件
			Path configPath = FileKit.createFile(tempWorkspace, IdUtil.getSnowflakeNextIdStr() + ".json");
			try (Writer writer = FileKit.getWriter(configPath)) {
				FileCopyUtils.copy(submitJobDto.getJobConfigJson(), writer);
			}

			// 构造jar作业
			PackageJarJob packageJarJob = new PackageJarJob();
			Path starterJarPath = SeaTunnelKit.getStarterJarPath(seatunnelHomePath);
			packageJarJob.setJarFilePath(starterJarPath.toFile().toURI().toString());
			packageJarJob.setEntryPointClass(SeaTunnelKit.SEATUNNEL_MAIN_CLASS);
			packageJarJob.setProgramArgs(new String[] { "--config", configPath.toString() });

			Set<URL> jars = new HashSet<>();
			jars.add(configPath.toUri().toURL());
//			jars.addAll(SeaTunnelKit.getLibJars(seatunnelHomePath));
			if(submitJobDto.getPlugins().stream().anyMatch(item -> item.getPluginType() == PluginType.TRANSFORM)) {
				jars.add(SeaTunnelKit.seatunnelTransform(seatunnelHomePath));
			}
			//文件需要hadoop依赖
			if(submitJobDto.getPlugins().stream().anyMatch(ConnectorType::isFile)) {
				jars.add(SeaTunnelKit.seatunnelHadoop(seatunnelHomePath));
			}
			submitJobDto.getPlugins().stream().filter(item -> StrUtil.isNotBlank(item.getJarPrefix())).map(enu -> SeaTunnelKit.seatunnelConnector(seatunnelHomePath, enu.getJarPrefix(), etlConfig.getSeatunnelVersion()))
					.filter(Objects::nonNull).forEach(jars::add);
			// 获取依赖的jdbc驱动jar
			if (CollectionUtil.isNotEmpty(submitJobDto.getJdbcs())) {
				submitJobDto.getJdbcs().stream()
						.map(enu -> SeaTunnelKit.getPluginJarPath(seatunnelHomePath, "jdbc", enu.getJarName()))
						.filter(Objects::nonNull).forEach(jars::add);
			}
			DeploymentTarget deployTarget = DeploymentTarget.parseDeployTarget(etlConfig.getFlinkResourceProvider(), submitJobDto.getDeployMode());
			// 设置flink的配置
			Configuration configuration = buildConfiguration(deployTarget);
			ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.JARS, jars, Object::toString);
			configuration.setString(PipelineOptions.NAME, submitJobDto.getJobName());
			//设置flink的类加载过程使用双亲委派模型，否则mysql-cdc找不到驱动
			configuration.set(CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL, Stream.of(DsTypeEnum.values()).map(DsTypeEnum::getDriverClassName).filter(StrUtil::isNotBlank).collect(Collectors.toList()));
//			configuration.setString(CoreOptions.FLINK_JVM_OPTIONS, "-Dflink_job_name=" + submitJobDto.getJobName());
			//从checkpoint恢复
			if(StrUtil.isNotBlank(submitJobDto.getSavePointPath())) {
				packageJarJob.setSavepointSettings(SavepointRestoreSettings.forPath(submitJobDto.getSavePointPath()));
//				configuration.set(SavepointConfigOptions.SAVEPOINT_PATH, submitJobDto.getSavePointPath());
			}
			if (submitJobDto.getDeployMode() == FlinkDeployMode.SESSION) {
				if (etlConfig.getFlinkResourceProvider() == ResourceProvider.STANDALONE) {
					URL url = new URL(submitJobDto.getFlinkInterfaceUrl());
					configuration.setString(RestOptions.ADDRESS, url.getHost());
					configuration.setInteger(RestOptions.PORT, url.getPort());
				} 
//				else if (etlConfig.getFlinkResourceProvider() == ResourceProvider.YARN) {
//					configuration.setString(YarnConfigOptions.APPLICATION_ID, submitJobDto.getApplicationId());
//				}
			}
			deployTarget.apply(configuration);
			SubmitCommand submitCommand = CommandFactory.buildSubmitCommand(deployTarget);
			Tuple2<String, ClusterClient> submitTuple = submitCommand.submit(finkHomePath, configuration, packageJarJob);
			return new SubmitJobRes(submitTuple.f0, submitTuple.f0, submitTuple.f1.getWebInterfaceURL());
		} finally {
			FileKit.deleteDir(tempWorkspace);
		}
	}
	
	@Override
	public void cancel(String jobId, String webInterfaceUrl) throws Exception {
		final URL url = new URL(webInterfaceUrl);
//		RestClient client = new FlinkRestClient(url.getHost(), url.getPort(), new Configuration());
		RestClient client = SpringUtil.getBean(EtlImportConfig.flinkClientBeanName(url), FlinkRestClient.class);
		final JobClient jobClient = client.job();
		jobClient.jobCancel(jobId);
	}

//	@Override
//	public void stop(String jobId, Integer id) throws Exception {
//		final URL url = new URL(etlConfig.getFlinkWebInterfaceUrl());
//		RestClient client = new FlinkRestClient(url.getHost(), url.getPort(), new Configuration());
//		final JobClient jobClient = client.job();
//
//		String jobCode = "123";
//		String savePointPath = String.join("/", String.valueOf(jobCode), String.valueOf(id));
//		Path targetDirectory = SystemKit.getSavepointDir(savePointPath);
//		StopWithSavepointRequestBody requestBody = new StopWithSavepointRequestBody(targetDirectory.toString(), true, SavepointFormatType.DEFAULT, new TriggerId());
//		jobClient.jobStop(jobId, requestBody);
//	}

	@Override
	public JobDetailsInfo info(String jobId, String webInterfaceUrl) throws Exception {
		final URL url = new URL(webInterfaceUrl);
//		RestClient client = new FlinkRestClient(url.getHost(), url.getPort(), new Configuration());
		RestClient client = SpringUtil.getBean(EtlImportConfig.flinkClientBeanName(url), FlinkRestClient.class);
		return client.job().jobDetail(jobId).get();
	}

	@Override
	public JobExceptionsInfoWithHistory exception(String jobId, String webInterfaceUrl) throws Exception {
		final URL url = new URL(webInterfaceUrl);
//		RestClient client = new FlinkRestClient(url.getHost(), url.getPort(), new Configuration());
		RestClient client = SpringUtil.getBean(EtlImportConfig.flinkClientBeanName(url), FlinkRestClient.class);
		return client.job().jobException(jobId, Optional.<String>of("30")).get();
	}
	
	@Override
	public CheckpointingStatistics checkpoints(String jobId, String webInterfaceUrl) throws Exception {
		final URL url = new URL(webInterfaceUrl);
//		RestClient client = new FlinkRestClient(url.getHost(), url.getPort(), new Configuration());
		RestClient client = SpringUtil.getBean(EtlImportConfig.flinkClientBeanName(url), FlinkRestClient.class);
		return client.job().jobCheckpoints(jobId).get();
	}

	@Override
	public String createSessionCluster(ResourceProvider resourceProvider) {
		try {
			DeploymentTarget deployTarget = DeploymentTarget.parseDeployTarget(resourceProvider, FlinkDeployMode.SESSION);
			Path finkHomePath = Paths.get(etlConfig.getFlinkHome());
			Configuration configuration = buildConfiguration(deployTarget);
			deployTarget.apply(configuration);
			Tuple2<String, ClusterClient> tuple2 = SessionFactory.buildSessionCommand(deployTarget).create(finkHomePath, configuration);
			return tuple2.f0;
		} catch (Exception e) {
			log.error(e.getMessage(), e);
		}
		return null;
	}

	/**
	 * flink配置信息
	 * @param deployTarget
	 * @return
	 */
	private Configuration buildConfiguration(DeploymentTarget deployTarget) {
		Configuration dynamicProperties = null;
		if (etlConfig.getFlinkConfigOptions() != null) {
			dynamicProperties = Configuration.fromMap(etlConfig.getFlinkConfigOptions());
		} else {
			dynamicProperties = new Configuration();
		}
		if (deployTarget.getResourceProvider() == ResourceProvider.STANDALONE) {
			return FlinkKit.loadConfiguration(Paths.get(etlConfig.getFlinkHome()), dynamicProperties);
		} else if (deployTarget.getResourceProvider() == ResourceProvider.YARN) {
			dynamicProperties.set(CoreOptions.FLINK_HADOOP_CONF_DIR,
					Paths.get(etlConfig.getClusterCredentialPath()).toAbsolutePath().toString());
			if (!dynamicProperties.contains(JobManagerOptions.TOTAL_PROCESS_MEMORY)) {
				dynamicProperties.setLong(JobManagerOptions.TOTAL_PROCESS_MEMORY.key(), MemorySize.ofMebiBytes(2048).getBytes());
			}
			if (!dynamicProperties.contains(TaskManagerOptions.TOTAL_PROCESS_MEMORY)) {
				dynamicProperties.setLong(TaskManagerOptions.TOTAL_PROCESS_MEMORY.key(), MemorySize.ofMebiBytes(2048).getBytes());
			}
//			dynamicProperties.setLong(TaskManagerOptions.NUM_TASK_SLOTS.key(), 1);
		} 
		//Kubernetes
//		else {
//			List<Path> childs = null;
//			try {
//				childs = FileKit.listFiles(Paths.get(etlConfig.getClusterCredentialPath()));
//			} catch (IOException e) {
//				log.error(e.getMessage(), e);
//			}
//			if (CollectionUtils.isEmpty(childs)) {
//				throw new IllegalStateException("Kubernetes kubeconfig can't be null");
//			}
//			final Path kubeConfigFile = childs.get(0);
//			dynamicProperties.set(KubernetesConfigOptions.KUBE_CONFIG_FILE, kubeConfigFile.toAbsolutePath().toString());
//			if (!dynamicProperties.contains(JobManagerOptions.TOTAL_PROCESS_MEMORY)) {
//				dynamicProperties.setLong(JobManagerOptions.TOTAL_PROCESS_MEMORY.key(), MemorySize.ofMebiBytes(1024).getBytes());
//			}
//			if (!dynamicProperties.contains(TaskManagerOptions.TOTAL_PROCESS_MEMORY)) {
//				dynamicProperties.setLong(TaskManagerOptions.TOTAL_PROCESS_MEMORY.key(), MemorySize.ofMebiBytes(1024).getBytes());
//			}
//		}
		return dynamicProperties;
	}

	@Override
	public String lighterLoad(List<String> webInterfaceUrls) {
		Map<String, Integer> urlMap = new HashMap<>();
		for(String webInterfaceUrl : webInterfaceUrls) {
			try {
				final URL url = new URL(webInterfaceUrl);
//				RestClient client = new FlinkRestClient(url.getHost(), url.getPort(), new Configuration());
				RestClient client = SpringUtil.getBean(EtlImportConfig.flinkClientBeanName(url), FlinkRestClient.class);
				ClusterOverviewWithVersion clusterOverview = client.cluster().overview().get();
				int runCnt = clusterOverview.getNumJobsRunningOrPending();
				urlMap.put(webInterfaceUrl, runCnt);
			} catch (Exception e) {
				log.error(e.getMessage(), e);
			}
		}
		Optional<Entry<String, Integer>> optional = urlMap.entrySet().stream().sorted(Map.Entry.<String, Integer>comparingByValue().reversed()).findFirst();
		return optional.isPresent() ? optional.get().getKey() : null;
	}
}
