/**
 * 
 */
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.etl.connector.base.SeatunnelConnector;
import org.springblade.etl.connector.sink.config.PlatomixJdbcSinkBaseConfig;
import org.springblade.etl.connector.sink.config.SeatunnelClickhouseSinkConfig;
import org.springblade.etl.entity.bi.ReportDatasource;
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;

/**
 * @author wookvn
 *
 */
@Component
public class PlatomixClickhouseSink implements SeatunnelConnector<SeatunnelClickhouseSinkConfig> {

	@Autowired
	private ReportDatasourceService reportDatasourceService;
	
	@Autowired
	private CommonJdbcSink commonJdbcSink;
	
	@Data
	@EqualsAndHashCode(callSuper = true)
	public static class Config extends PlatomixJdbcSinkBaseConfig {
		/**
		 */
		private static final long serialVersionUID = 1L;
		
		@ApiModelProperty("source的主键列表")
		private List<String> primaryKeys;
	}
	
	@Override
	public SeatunnelClickhouseSinkConfig buildConfig(String configJson) {
		Config ptConfig = (Config)validate(configJson);
		ReportDatasource sourceModel = getDs(ptConfig.getDatasourceId());
		SeatunnelClickhouseSinkConfig seatunnelConfig = new SeatunnelClickhouseSinkConfig();
		seatunnelConfig.setHost(sourceModel.getDsIp() + ":" + sourceModel.getDsPort());
		seatunnelConfig.setUsername(sourceModel.getDsUser());
		seatunnelConfig.setDatabase(sourceModel.getDsDb());
		seatunnelConfig.setPassword(sourceModel.getDsPassword());
		seatunnelConfig.setTable(ptConfig.getTableName());
		seatunnelConfig.setPrimaryKey(CollectionUtil.isNotEmpty(ptConfig.getPrimaryKeys()) ? ptConfig.getPrimaryKeys().get(0) : null);
		seatunnelConfig.setSupportUpsert(CollectionUtil.isNotEmpty(ptConfig.getPrimaryKeys()));
		seatunnelConfig.setLightWeightDelete(CollectionUtil.isNotEmpty(ptConfig.getPrimaryKeys()));
		seatunnelConfig.setSupportUpsert(CollectionUtil.isNotEmpty(ptConfig.getPrimaryKeys()));
		return seatunnelConfig;
	}

	@Override
	public Object validate(String configJson) {
		Config ptConfig = EtlJsonUtil.parseJsonString(configJson, PlatomixClickhouseSink.Config.class);
		ValidateUtil.validate(ptConfig);
		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;
	}
}
