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

import cn.hutool.core.collection.CollectionUtil;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
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.SeatunnelJdbcSinkConfig;
import org.springblade.etl.connector.sink.config.SeatunnelMongoSinkConfig;
import org.springblade.etl.connector.sink.config.SeatunnelPreviewSinkConfig;
import org.springblade.etl.entity.bi.ReportDatasource;
import org.springblade.etl.enums.JdbcType;
import org.springblade.etl.properties.PreviewRedisProperties;
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 javax.validation.constraints.Max;
import javax.validation.constraints.Min;
import javax.validation.constraints.NotEmpty;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * @author wookvn
 *
 */
@Component
public class PlatomixMongoSink implements SeatunnelConnector<SeatunnelMongoSinkConfig> {


	@Autowired
	private ReportDatasourceService reportDatasourceService;

	@Data
	public static class Config  extends PlatomixJdbcSinkBaseConfig {

	}
	
	@Override
	public SeatunnelMongoSinkConfig buildConfig(String configJson) {
		return (SeatunnelMongoSinkConfig)validate(configJson);
	}

	@Override
	public Object validate(String configJson) {
		PlatomixMongoSink.Config ptConfig = EtlJsonUtil.parseJsonString(configJson, PlatomixMongoSink.Config.class);
		ReportDatasource sourceModel = getDs(ptConfig.getDatasourceId());
		SeatunnelMongoSinkConfig seatunnelConfig = new SeatunnelMongoSinkConfig();
		seatunnelConfig.setUri(sourceModel.getUrl());
		seatunnelConfig.setDatabase(sourceModel.getDbSchema());
		seatunnelConfig.setCollection(ptConfig.getTableName());
//		seatunnelConfig.setSchema();
//		seatunnelConfig.setJdbcType(JdbcType.ofDsType(sourceModel.getDsType()));

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