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

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.source.config.SeatunnelMysqlCdcSourceConfig;
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 javax.validation.constraints.NotNull;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author wookvn
 *
 */
@Component
public class PlatomixMysqlCdcSource implements SeatunnelConnector<SeatunnelMysqlCdcSourceConfig> {
	
	@Autowired
	private ReportDatasourceService reportDatasourceService;

	@Data
	public static class Config {
		@NotNull(message = "mysqlCdc数据源不能为空")
		@ApiModelProperty("数据源ID")
		private Integer datasourceId;
		@NotNull(message = "mysqlCdc数据表不能为空")
		@ApiModelProperty("表名列表")
		private List<String> tableNames;
	}
	
	@Override
	public SeatunnelMysqlCdcSourceConfig buildConfig(String configJson) {
		Config ptConfig = (Config)validate(configJson);
		ReportDatasource sourceModel = reportDatasourceService.getById(ptConfig.getDatasourceId());
		if(sourceModel == null) {
			throw new ServiceException("未查询到数据源|datasourceId="+ptConfig.getDatasourceId());
		}
		SeatunnelMysqlCdcSourceConfig seatunnelConfig = new SeatunnelMysqlCdcSourceConfig();
		seatunnelConfig.setJdbcType(JdbcType.ofDsType(sourceModel.getDsType()));
		seatunnelConfig.setJdbcUrl(sourceModel.getUrl());
		seatunnelConfig.setUsername(sourceModel.getDsUser());
		seatunnelConfig.setPassword(sourceModel.getDsPassword());
		seatunnelConfig.setTableNames(ptConfig.getTableNames().stream().map(item -> sourceModel.getDsDb() + "." + item).collect(Collectors.toList()));
		seatunnelConfig.setDatabaseNames(Arrays.asList(sourceModel.getDsDb()));
		return seatunnelConfig;
	}
	
	@Override
	public Object validate(String configJson) {
		Config ptConfig = EtlJsonUtil.parseJsonString(configJson, PlatomixMysqlCdcSource.Config.class);
		ValidateUtil.validate(ptConfig);
		return ptConfig;
	}
}
