/**
 * 
 */
package org.springblade.etl.connector.sink;

import com.alibaba.druid.util.JdbcUtils;
import lombok.extern.slf4j.Slf4j;
import org.springblade.core.tool.exception.ServiceException;
import org.springblade.datasource.enums.DsTypeEnum;
import org.springblade.datasource.transform.TransformContext;
import org.springblade.etl.connector.sink.config.PlatomixJdbcSinkBaseConfig;
import org.springblade.etl.entity.bi.ReportDatasource;
import org.springblade.etl.enums.DataWriteMode;
import org.springblade.etl.service.bi.ReportDatasourceService;
import org.springblade.etl.util.EtlJsonUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;
import java.util.Map;
import java.util.Objects;
import java.util.Properties;

/**
 * @author wookvn
 *
 */
@Slf4j
@Component
public class CommonJdbcSink  {

	@Autowired
	private ReportDatasourceService reportDatasourceService;
	
	@Autowired
	private TransformContext transformContext;
	
	/**
	 * jdbc - sink 前置操作
	 * @param configJson
	 */
	public void prepare(String configJson) {
		log.info("开始执行Sink的前置操作");
		PlatomixJdbcSinkBaseConfig sinkBaseConfig = EtlJsonUtil.parseJsonString(configJson, PlatomixJdbcSinkBaseConfig.class);
		if(sinkBaseConfig.getAutoCreateTable()) {
			log.info("自动建表开始|dsId={},tablename={}", sinkBaseConfig.getDatasourceId(), sinkBaseConfig.getTableName());
			ReportDatasource sinkDs = getDs(sinkBaseConfig.getDatasourceId());
			ReportDatasource sourceDs = null;
			if(sinkBaseConfig.getSourceDatasourceId() == null) { //source没有JDBC数据源，比如：http-source/kafka-source等
				sourceDs = new ReportDatasource();
				sourceDs.setDsType(DsTypeEnum.common.getValue());
			}else {
				sourceDs = getDs(sinkBaseConfig.getSourceDatasourceId());
			}
			if (Objects.equals(sinkDs.getDsType(), DsTypeEnum.mongo.getValue())){
				return;
			}
			long startTime = System.currentTimeMillis();
			transformContext.transformAndCreatTable(sourceDs, sinkDs, 
					sinkBaseConfig.getGeneralColumnList(), sinkBaseConfig.getSourceTableName(), sinkBaseConfig.getTableName(), sinkBaseConfig.getSinkTableComment(),true);
			log.info("自动建表结束|cost:{}ms", (System.currentTimeMillis() - startTime));
		}else if(sinkBaseConfig.getWriteMode() == DataWriteMode.TRUNCATE_INSERT) { //清空数据
			log.info("清空数据开始|dsId={},tablename={}", sinkBaseConfig.getDatasourceId(), sinkBaseConfig.getTableName());
			Connection connection = null;
			Statement statement = null;
			try {
				ReportDatasource sourceModel = getDs(sinkBaseConfig.getDatasourceId());
				Properties props = new Properties();
				props.put("user", sourceModel.getDsUser());
				props.put("password", sourceModel.getDsPassword());
				//设置超时时间
				Map<String, String> timeoutMap = DsTypeEnum.resolve(sourceModel.getDsType()).jdbcConnectionPropTimeout(false);
				if(timeoutMap != null) props.putAll(timeoutMap);
				connection = DriverManager.getConnection(sourceModel.getUrl(), props);
				statement = connection.createStatement();
				String truncateSql = String.format("truncate table %s", sinkBaseConfig.getTableName());
				log.info("truncateSql:{}", truncateSql);
				statement.execute(truncateSql);
			} catch (Exception e) {
				log.error(e.getMessage());
				throw new ServiceException("执行Sink【前置操作:清空表】失败|"+e.getMessage(), e);
			} finally {
				JdbcUtils.close(connection);
				JdbcUtils.close(statement);
			}
		}
	}
	
	/**
	 * 获取数据源
	 * @param configJson
	 * @return
	 */
	private ReportDatasource getDs(Integer databaseId) {
		ReportDatasource sourceModel = reportDatasourceService.getById(databaseId);
		if(sourceModel == null) {
			throw new ServiceException("未查询到数据源|datasourceId="+databaseId);
		}
		return sourceModel;
	}
}
