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

import cn.hutool.core.collection.CollectionUtil;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import lombok.EqualsAndHashCode;
import org.springblade.core.tool.exception.ServiceException;
import org.springblade.datasource.context.DialectContext;
import org.springblade.etl.connector.base.SeatunnelConnector;
import org.springblade.etl.connector.sink.config.PlatomixJdbcSinkBaseConfig;
import org.springblade.etl.connector.sink.config.SeatunnelJdbcSinkConfig;
import org.springblade.etl.entity.bi.ReportDatasource;
import org.springblade.etl.enums.JdbcType;
import org.springblade.etl.service.bi.ReportDatasourceService;
import org.springblade.etl.util.EtlJsonUtil;
import org.springblade.etl.util.ValidateUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * @author wookvn
 *
 */
@Component
public class PlatomixJdbcSink implements SeatunnelConnector<SeatunnelJdbcSinkConfig> {

	@Autowired
	private ReportDatasourceService reportDatasourceService;
	
	@Autowired
	private CommonJdbcSink commonJdbcSink;
	
	@Autowired
	private DialectContext dialectContext;
	
	@Data
	@EqualsAndHashCode(callSuper = true)
	public static class Config extends PlatomixJdbcSinkBaseConfig {
		/**
		 */
		private static final long serialVersionUID = 1L;
		
		@ApiModelProperty("source的主键列表")
		private List<String> primaryKeys;
		@ApiModelProperty("表字段列表【columnFields和primaryKeys必须有一个】")
		private List<String> columnFields;
	}

	@Override
	public SeatunnelJdbcSinkConfig buildConfig(String configJson) {
		Config ptConfig = (Config)validate(configJson);
		ReportDatasource sourceModel = getDs(ptConfig.getDatasourceId());
		SeatunnelJdbcSinkConfig seatunnelConfig = new SeatunnelJdbcSinkConfig();
		seatunnelConfig.setPrimary_keys(ptConfig.getPrimaryKeys());
		seatunnelConfig.setSupport_upsert_by_query_primary_key_exist(CollectionUtil.isNotEmpty(ptConfig.getPrimaryKeys()));
		seatunnelConfig.setGenerate_sink_sql(CollectionUtil.isNotEmpty(ptConfig.getPrimaryKeys()));
		seatunnelConfig.setJdbcType(JdbcType.ofDsType(sourceModel.getDsType()));
		seatunnelConfig.setUrl(sourceModel.getUrl());
		seatunnelConfig.setUsername(sourceModel.getDsUser());
		seatunnelConfig.setPassword(sourceModel.getDsPassword());
		seatunnelConfig.setDriver(sourceModel.getDriverClassName());
		String database = dialectContext.getDialect(sourceModel.getDsTypeSyl()).dbSchema(sourceModel);
		seatunnelConfig.setDatabase(database);
		seatunnelConfig.setTable(ptConfig.getTableName());
		if(CollectionUtil.isEmpty(ptConfig.getPrimaryKeys())) {
			String insertSql = String.format("insert into %s(%s) values(%s)", 
//					sourceModel.getDsDb(), 
					ptConfig.getTableName(), String.join(",", ptConfig.getColumnFields()),
					IntStream.range(0, ptConfig.getColumnFields().size()).mapToObj(t -> "?").collect(Collectors.joining(",")));
			seatunnelConfig.setQuery(insertSql);
		}
		return seatunnelConfig;
	}
	
	@Override
	public Object validate(String configJson) {
		Config ptConfig = EtlJsonUtil.parseJsonString(configJson, PlatomixJdbcSink.Config.class);
		ValidateUtil.validate(ptConfig);
		if(CollectionUtil.isEmpty(ptConfig.getPrimaryKeys()) && CollectionUtil.isEmpty(ptConfig.getColumnFields())) {
			throw new ServiceException("Sink主键或字段列表必须有一个非空");
		}
		return ptConfig;
	}

	@Override
	public void prepare(String configJson) {
		commonJdbcSink.prepare(configJson);
	}
	
	/**
	 * 获取数据源
	 * @param configJson
	 * @return
	 */
	private ReportDatasource getDs(Integer databaseId) {
		ReportDatasource sourceModel = reportDatasourceService.getById(databaseId);
		if(sourceModel == null) {
			throw new ServiceException("未查询到数据源|datasourceId="+databaseId);
		}
		return sourceModel;
	}
}
