package com.onsemi.middleware.lite.DbCsv_to_Snowflake;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.StringReader;
import java.sql.Connection;
import java.sql.SQLException;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.zip.GZIPInputStream;
import java.util.zip.ZipEntry;
import java.util.zip.ZipFile;

import javax.annotation.PostConstruct;

import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.component.sql.SqlComponent;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.text.StrSubstitutor;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.util.UriComponents;
import org.springframework.web.util.UriComponentsBuilder;
import org.yaml.snakeyaml.util.UriEncoder;

import com.onsemi.middleware.UECommon.model.Interface;
import com.onsemi.middleware.UECommon.runtime.generics.GenericItemType;
import com.onsemi.middleware.camel.MDWRouteBuilder;
import com.onsemi.middleware.camel.processor.MDWProcessors;
import com.onsemi.middleware.connector.db.DBConnector;
import com.onsemi.middleware.lite.DbCsv_to_Snowflake.model.FileInfo;
import com.onsemi.middleware.lite.DbCsv_to_Snowflake.model.SnowflakeReply;
import com.onsemi.middleware.logging.dispatcher.LogDispatcher;
import com.onsemi.middleware.logging.toolkit.Log;
import com.onsemi.middleware.messaging.InboundMessage;
import com.onsemi.middleware.messaging.Utilities;
import com.onsemi.middleware.ue.execeng.annotation.spring.Connector;
import com.onsemi.middleware.ue.execeng.annotation.spring.InterfaceProperty;
import com.onsemi.middleware.ue.execeng.commons.camel.EndpointInject;

import net.sf.jsqlparser.JSQLParserException;
import net.sf.jsqlparser.expression.Function;
import net.sf.jsqlparser.expression.StringValue;
import net.sf.jsqlparser.expression.TimeKeyExpression;
import net.sf.jsqlparser.parser.CCJSqlParserManager;
import net.sf.jsqlparser.parser.CCJSqlParserUtil;
import net.sf.jsqlparser.schema.Column;
import net.sf.jsqlparser.statement.insert.Insert;
import net.sf.jsqlparser.statement.select.PlainSelect;
import net.sf.jsqlparser.statement.select.Select;
import net.sf.jsqlparser.statement.select.SelectItem;
import net.sf.jsqlparser.statement.select.SetOperationList;
import net.sf.jsqlparser.util.TablesNamesFinder;
import net.snowflake.client.core.SFBaseResultSet;
import net.snowflake.client.core.SFBaseStatement;
import net.snowflake.client.core.SFException;
import net.snowflake.client.jdbc.ErrorCode;
import net.snowflake.client.jdbc.SnowflakeConnectionV1;

public abstract class GenericAnythingToCopyIntoSnowflake extends MDWRouteBuilder {

	protected static final String SOURCE_TARGET_PRE_POST_PROPERTY_NAME = "sourceTargetPrePostPropertyName";
	public static final String FILE_NAME_PLACEHOLDER = "${fileName}";

	public SnowflakeConnectionV1 snowflakeConnection;

	@Connector("Target")
	private DBConnector target;

	@EndpointInject(jndiName = "Target", componentClass = SqlComponent.class , uriParams =  "outputType=SelectOne"
			+ "&query=select 'willBeReplacedInRuntime' FROM DUAL&queryName=dynamicQuery"
	)
	private Endpoint targetPrePostProcessSqls;

	@InterfaceProperty("overwrite")
	private String overwrite = "true";

	protected String fullStageName;
	protected Insert insertStatement;
	protected ProducerTemplate producerTemplate;
	protected CCJSqlParserManager parseSql = new CCJSqlParserManager();

	@InterfaceProperty("InsertSQL2")
	private String insertSql;//schemaName.tableName

	@InterfaceProperty("stageName")
	private String stageName; //"@RAW_PLAN_APS_OWNER.RAW_PLAN_APS_OWNER_STG"

	protected boolean anyRecords = false;
	private Long noOfRecords = 0l;
	private Long passRecords = 0l;
	private Long failedRecords = 0l;

	private AtomicBoolean targetPreProcessSqlExecuted = new AtomicBoolean(false);
	private AtomicBoolean targetPostProcessSqlExecuted = new AtomicBoolean(false);

	@Autowired
	protected LogDispatcher logDispatcher;
	@Autowired
	protected InboundMessage inboundMessage;
	@Autowired
	private Optional<SendTriggerFile> sendTriggerFile;
	@Autowired
	private Optional<ArchiveFileSend> archiveFileSend;

	@InterfaceProperty("SelectSQL")
	protected String selectSql;
	@InterfaceProperty("purge")
	protected String purge = "true";
	@InterfaceProperty("unzip")
	protected boolean unzip = true;

	private Map<String, String> propsAndPayloadReplaceMap = new HashMap<>();
	private Supplier<LocalDateTime> nowSupplier = LocalDateTime::now;

	@PostConstruct
	public void init(){
		propsAndPayloadReplaceMap = new HashMap<>();
		propsAndPayloadReplaceMap.putAll(inboundMessage.getPropertiesMap());
		inboundMessage.getXmlValues()
					  .entrySet().stream()
					  .forEach(e -> propsAndPayloadReplaceMap.put(e.getKey(), e.getValue().toString()));
		snowflakeGlobalColumnNames = snowflakeGlobalColumnNames.concat(",BIW_INS_DTTM,BIW_UPD_DTTM,BIW_BATCH_ID");
	}

	@InterfaceProperty("datePattern")
	private String datePattern = "yyyy-MM-dd HH:mm:ss";
	@InterfaceProperty("notAllowedCharsInSnowflake")
	private String notAllowedCharsInSnowflake = StringUtils.EMPTY;
	@InterfaceProperty("snowflakeGlobalColumnNames")
	private String snowflakeGlobalColumnNames = StringUtils.EMPTY;

	void setNowSupplier(Supplier<LocalDateTime> nowSupplier) {
		this.nowSupplier = nowSupplier;
	}

	void setInboundMessage(InboundMessage inboundMessage) {
		this.inboundMessage = inboundMessage;
	}

	@NotNull
	protected String replacePayloadValuesInQuery() {
		selectSql = replacePlaceholdersInQuery(selectSql);
		return selectSql;
	}

	private String replacePlaceholdersInQuery(String query) {
		AtomicReference<String> queryRef = new AtomicReference<>(query);
		Map<String, Object> xmlValues = Utilities.getXmlValues(inboundMessage.getMessage());
		xmlValues.forEach((key, value) -> queryRef.set(queryRef.get().replace("${" + key + "}", value.toString())));
		replaceDatesInSelectItemsOnly(queryRef);
		if (!placeholderConditionPass(queryRef)) {
			throw new RuntimeException(
					"There are some unresolved ${placeholders} in select query (" + selectSql + "). They need to be provided via XML payload");
		}
		return queryRef.get();
	}

	private AtomicReference<String> replaceDatesInSelectItemsOnly(AtomicReference<String> query) {
		try {
			String formattedDate = getFormatedDate();
			Select select = (Select) parseSql.parse(new StringReader(query.get()));
			select.getPlainSelect()
				  .getSelectItems()
				  .forEach(selectItem -> {
					  if (Column.class.isAssignableFrom(selectItem.getExpression().getClass())) {
						  Column dateColumn = selectItem.getExpression(Column.class);
						  if ("sysdate".equalsIgnoreCase(dateColumn.getColumnName())) {
							  dateColumn.setColumnName("'" + formattedDate + "'");
						  }
					  } else if (TimeKeyExpression.class.isAssignableFrom(selectItem.getExpression().getClass())) {
						  TimeKeyExpression dateColumn = selectItem.getExpression(TimeKeyExpression.class);
						  if ("current_timestamp".equalsIgnoreCase(dateColumn.getStringValue())) {
							  dateColumn.setStringValue("'" + formattedDate + "'");
						  }
					  }
				  });
			query.set(select.toString());
		} catch (JSQLParserException e) {
			throw new RuntimeException(e);
		}
		return query;
	}

	private boolean placeholderConditionPass(AtomicReference<String> queryRef) {
		String selectQuery = queryRef.get();
		boolean fileNameInSelect = selectQuery.contains(FILE_NAME_PLACEHOLDER);
		if (selectQuery.contains("${") && !fileNameInSelect) {
			return false;
		} else {
			return StringUtils.countMatches(selectQuery, "${") <= 1 || !fileNameInSelect;
		}
	}

	@NotNull
	String getFormatedDate() {
		DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern(datePattern);
		String formattedDate = dateTimeFormatter.format(nowSupplier.get());
		return formattedDate;
	}

	@PostConstruct
	private void initFields() {
		try {
			initInsert();
			Connection connection = target.getMdwBasicDataSource().getConnection();
			snowflakeConnection = connection.unwrap(SnowflakeConnectionV1.class);
			fullStageName = insertStatement.getTable().getSchemaName() + "." + stageName;
		} catch (JSQLParserException | SQLException e) {
			throw new RuntimeException(e);
		}
	}

	void setInsertSql(String insertSql) {
		this.insertSql = insertSql;
	}

	void initInsert() throws JSQLParserException {
		insertStatement = (Insert) parseSql.parse(new StringReader(insertSql.replaceAll(":", "")));
	}

	protected SnowflakeReply getResponseForSnowflakeCall(SFBaseResultSet resultSet){
		try {
			resultSet.next();
			Map<String, String> resultRow = new HashMap<>();
			for (int i = 1; i < resultSet.getMetaData()
										 .getColumnCount(); i++) {
				String columnName = resultSet.getMetaData()
											 .getColumnName(i);
				String columnValue = resultSet.getString(i);
				if("rows_loaded".equals(columnName)){
					setNoOfRecords(Long.parseLong(columnValue) + getNoOfRecords());
				}
				resultRow.put(columnName, columnValue);
			}
			return new SnowflakeReply(resultSet.getQueryId(), resultRow);
		}
		catch (Throwable e) {
			throw new RuntimeException(e);
		}
	}

	protected Exchange sendFileToSnowflake(Exchange exchange) {
		FileInfo body = exchange.getIn().getBody(FileInfo.class); // just one file is in body for all interface implementations
		try {
			SFBaseResultSet resultSet = getSnowflakeConnection().getHandler()
																.getSFStatement()
																.execute("PUT file://" + body.getTempFile().getAbsoluteFile() + " @" + fullStageName
																				 + " parallel=99 OVERWRITE=" + overwrite,
																		 new HashMap<>(),
																		 SFBaseStatement.CallingMethod.EXECUTE);
			double fileSize = body.length() / 1024.0 / 1024.0;
			SnowflakeReply reply = getResponseForSnowflakeCall(resultSet);
			if (SnowflakeReply.SnowflakeResponseStatus.ERROR.getStatus().equalsIgnoreCase(reply.getStatus())) {
				throw new SFException(ErrorCode.IO_ERROR, "File was not correctly uploaded to Snowflake. Response from Snowflake: " + reply);
			}
			logDispatcher.debug(body.getFileName() + " (" + fileSize + " MiB) was sent", "details: " + reply);
		} catch (Exception | SFException e) {
			throw new RuntimeException("Exception during calling PUT command", e);
		}
		exchange.getIn().setBody(body);
		return exchange;
	}

	List<String> getColumnsFromSelect(String fileName){
		List<String> correctColumnsOrder = new ArrayList<>();
		try {
			PlainSelect selectBody = getSelectBody(fileName);
			List<String> selectColumnsOrder = selectBody.getSelectItems()
														.stream()
														.map(this::getCorrectColumnNameFromAlias)
														.map(this::replaceNotAllowedChars)
														.collect(Collectors.toList());
			int columnIndex = 1;
			List<String> insertColumnValuesOrdered = getInsertValuesOrder().stream()
																		  .map(Column::getColumnName)
																		  .toList();

			for (String insertColValuePlaceholder : insertColumnValuesOrdered) {
				if (indexOfIgnoreCase(insertColValuePlaceholder, selectColumnsOrder) != -1){
					correctColumnsOrder.add("$"+columnIndex);
					columnIndex++;
					selectColumnsOrder.remove(insertColValuePlaceholder);
				}
				else {
					Optional<String> colFoundByAliasMatching = (Optional<String>) selectColumnsOrder.stream()
																									.filter(it -> it.contains(" ") || it.toUpperCase().contains(" AS "))
																									.map(it -> resolveAliases(insertColValuePlaceholder, it))
																									.filter(Optional::isPresent)
																									.map(Optional::get)
																									.findAny();
					if (colFoundByAliasMatching.isEmpty()){
						throw new RuntimeException("Cannot determine mapping for column "+insertColValuePlaceholder + ". Please check your insert & select (and aliases)");
					}
					else {
						selectColumnsOrder.remove(colFoundByAliasMatching.get());
						correctColumnsOrder.add(colFoundByAliasMatching.get());
					}
				}
			}
		} catch (JSQLParserException e) {
			throw new RuntimeException("Exception during parsing select statement: " + selectSql, e);
		}
		return correctColumnsOrder;
	}

	private PlainSelect getSelectBody(String fileName) throws JSQLParserException {
		Select select = (Select)CCJSqlParserUtil.parse(selectSql.replace(FILE_NAME_PLACEHOLDER, fileName));
		PlainSelect selectBody;
		if (PlainSelect.class.isAssignableFrom(select.getClass())) {
			selectBody = (PlainSelect) select;
		} else {
			SetOperationList operationList = select.getSetOperationList();
			selectBody = (PlainSelect) operationList.getSelect(0);
		}
		return selectBody;
	}

	private @NotNull Optional<?> resolveAliases(String insertColValuePlaceholder, String it) {
		if (it.contains(" ")) {
			String aliasColName = it.split(" ")[1].trim();
			if (aliasColName.equals(insertColValuePlaceholder)) {
				return Optional.of(it);
			}
		}
		if (it.toUpperCase().contains(" AS ")) {
			String aliasColName = it.toUpperCase().split(" AS ")[1].trim();
			if (aliasColName.equalsIgnoreCase(insertColValuePlaceholder)) {
				return Optional.of(it);
			}
		}
		return Optional.empty();
	}

	private String replaceNotAllowedChars(String columnName) {
		if (containsNotAllowedCharsInSnowflake(columnName)) {
			String newColumnName = columnName;
			String[] split = notAllowedCharsInSnowflake.split(",");
			for (String notAllowedChar : split) {
				newColumnName = columnName.replace(notAllowedChar, StringUtils.EMPTY);
			}
			return newColumnName;
		}
		return columnName;
	}

	/**
	 * For global column names which are same for all snowflake tables (currently they are stored in {@link GenericAnythingToCopyIntoSnowflake#snowflakeGlobalColumnNames} interface property) we need to keep the whole alias -> '15' AS BIW_BATCH_ID.
	 * But for other column names we need to get only alias name -> 'Lojza' AS name -> name. Example on UAT is OM_CH_HEADER_Snowflake.
	 * @param selectItem Column/function from select
	 * @return Correct alias name for snowflake copy-into command. If column/function does not have alias it will return column/function name
	 */
	private String getCorrectColumnNameFromAlias(SelectItem<?> selectItem) {
		if (selectItem.getAlias() != null) {
			AtomicReference<String> alias = new AtomicReference<>(selectItem.getAlias().getName());
			if (StringUtils.isBlank(alias.get())) {
				alias.set(selectItem.getExpression(Column.class).getColumnName());
			}
			if (StringValue.class.isAssignableFrom(selectItem.getExpression().getClass())) {
				Optional<String> isGlobalColumnName = Arrays.stream(snowflakeGlobalColumnNames.split(","))
															.filter(globalColumnName -> globalColumnName.equalsIgnoreCase(alias.get()))
															.findAny();
				if (isGlobalColumnName.isEmpty()) {
					return alias.get();
				}
				return selectItem.toString();
			}
			return alias.get();
		} else {
			if (Column.class.isAssignableFrom(selectItem.getExpression().getClass())){
				return selectItem.getExpression(Column.class).getColumnName();
			} else if (Function.class.isAssignableFrom(selectItem.getExpression().getClass())) {
				return selectItem.getExpression(Function.class).getName();
			} else {
				return selectItem.toString();
			}
		}
	}

	private int indexOfIgnoreCase(String insertColName, List<String> selectColumnsOrder) {
		int directIndex =  selectColumnsOrder.indexOf(insertColName);
		if (directIndex == -1){
			for (int i = 0;  i < selectColumnsOrder.size() ;i++){
				if (selectColumnsOrder.get(i).equalsIgnoreCase(insertColName)){
					return i;
				}
			}
		}
		return directIndex;
	}

	protected String updateSelectForCopyInto(String fileName) {
		StringBuilder newSelect = new StringBuilder().append("SELECT ");
		List<String> correctColumnsOrder = getColumnsFromSelect(fileName);

		newSelect.append(String.join(",", correctColumnsOrder))
				 .append(" FROM")
				 .append(" @")
				 .append(insertStatement.getTable().getSchemaName())
				 .append(".")
				 .append(stageName);
		return newSelect.toString();
	}

	private boolean containsNotAllowedCharsInSnowflake(String originalColumn) {
		String[] notAllowedChars = notAllowedCharsInSnowflake.split(",");
		for (String notAllowedChar : notAllowedChars) {
			if (originalColumn.contains(notAllowedChar)) {
				return true;
			}
		}
		return false;
	}

	private SnowflakeConnectionV1 getSnowflakeConnection() {
		try {
			if (snowflakeConnection == null) {
				Connection connection = target.getMdwBasicDataSource().getConnection();
				snowflakeConnection = connection.unwrap(SnowflakeConnectionV1.class);
			}
			return snowflakeConnection;
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}

	void definePrePostProcessSqlRoute(GenericItemType preOrPostSql, AtomicBoolean relatedVariableThatChecksIfSqlAlreadyExecuted){
		from("direct:"+preOrPostSql.getPropertyName()).routeId(preOrPostSql.getPropertyName())
			.choice()
			.when(e -> anyRecords || GenericItemType.TARGET_PRE_PROCESS_SQL.getPropertyName().equals(preOrPostSql.getPropertyName()) || GenericItemType.SOURCE_PRE_PROCESS_SQL.getPropertyName().equals(preOrPostSql.getPropertyName()))
				.setHeader(SOURCE_TARGET_PRE_POST_PROPERTY_NAME, preOrPostSql::getPropertyName)
				.setHeader(preOrPostSql.getPropertyName(), () -> getSqls(preOrPostSql)).id("set"+preOrPostSql.getPropertyName()+"sIfAny")
				.choice().id("if"+preOrPostSql.getPropertyName()+"Exist")
					.when(e -> !e.getIn().getHeader(preOrPostSql.getPropertyName(), List.class).isEmpty() && !relatedVariableThatChecksIfSqlAlreadyExecuted.get())
					.process(e -> relatedVariableThatChecksIfSqlAlreadyExecuted.set(true))
					.to("direct:runTargetSourcePrePostProcessSql")
				.endChoice()
			.end();
	}

	@Override
	public void configure() {
		replacePayloadValuesInQuery();
		producerTemplate = getCamelContext().createProducerTemplate();
		onException(Exception.class).handled(false).routeId("exceptionHandler")
									.process(exchange -> logDispatcher.info(Log.PROCESS_STOP, "Processing done",
																			"Total: " + getNoOfRecords() + ", Passed: "
																					+ getPassRecords() + ", Failed: "
																					+ getNoOfRecords())).id("logStats")
									.process(MDWProcessors.intendedShutdown()).id("shutdownCamelContext");
		definePrePostProcessSqlRoute( GenericItemType.TARGET_PRE_PROCESS_SQL, targetPreProcessSqlExecuted);
		definePrePostProcessSqlRoute( GenericItemType.TARGET_POST_PROCESS_SQL, targetPostProcessSqlExecuted);

		from("direct:doSendFile").routeId("sendFile")
				.choice().id("ifAnyRecordsProcessed")
				.when(exchange -> anyRecords)
					.choice()
						.when(this::needUnzip)
							.process(this::unzipInputFile)
					.endChoice()
					.process(this::sendFileToSnowflake).id("sendFileToSnowflake")
					.process(constructAndSendCopyInto()).id("invokeCopyInto")
					.choice().id("shouldSendTriggerFile")
						.when(e -> sendTriggerFile.isPresent())
							.process(exchange -> {
								if(sendTriggerFile.isPresent()){
									logDispatcher.info("Sending trigger file using DE","");
									sendTriggerFile.get().sendFile();
								}
							}).id("doSendTriggerFile")
					.endChoice()
					.choice().id("shouldArchiveViaDE")
						.when(e -> archiveFileSend.isPresent()).id("archiveViaDE")
							.process(archiveUsingDE()).id("sendToExternalArchivalViaDE")
					.endChoice()
					.process(this::cleanupSentFile).id("anythingCleanup")
				.otherwise()
					.process(e -> logDispatcher.warning("No data processed.", "No data selected/picked up."))
				.end();

		from("direct:runTargetSourcePrePostProcessSql").routeId("runPreOrPostProcessSql")
				.setHeader("bodyBackup", body()).id("backupBody")
				.process(e -> logDispatcher.info("Executing "+e.getIn().getHeader(SOURCE_TARGET_PRE_POST_PROPERTY_NAME, String.class)+"(s)", "")).id("logSql")
				.loopDoWhile(e -> !e.getIn().getHeader(e.getIn().getHeader(SOURCE_TARGET_PRE_POST_PROPERTY_NAME, String.class), List.class).isEmpty()).id("loopPreOrPostProcessSql")
					.process(e -> {
						List<String> sqlsToExecute = e.getIn().getHeader(e.getIn().getHeader(SOURCE_TARGET_PRE_POST_PROPERTY_NAME, String.class), List.class);
						String executeNow = sqlsToExecute.get(0);
						e.getIn().setHeader("prePostToExec", executeNow);
						sqlsToExecute.remove(0);
						e.getIn().setHeader(e.getIn().getHeader(SOURCE_TARGET_PRE_POST_PROPERTY_NAME, String.class), sqlsToExecute);
					}).id("preparePrePostSqls")
					.process(e -> this.rewriteDynamicUri(e,
													 "dynamicSql",
													 "prePostToExec",
													 determineEndpointForPrePostSql(GenericItemType.findProperty(e.getIn().getHeader(
																														  SOURCE_TARGET_PRE_POST_PROPERTY_NAME,
																														  String.class))))).id("prepareDynamicUri")
					.toD("${header.dynamicSql}")
					.process(exchange -> logDispatcher.trace("Executed:", exchange.getIn().getHeader("prePostToExec", String.class))).id("traceExecuted")
				.end()
				.setBody(header("bodyBackup")).id("resetBodyFromBackup");

		from("direct:end")
				.choice()
					.when(e -> getNoOfRecords() > 0)
						.to("direct:"+GenericItemType.TARGET_POST_PROCESS_SQL.getPropertyName()).id("targetPostProcessSql")
				.end()
				.process(exchange -> logDispatcher.info(Log.PROCESS_STOP, "Processing done",
														"Total: " + getNoOfRecords() + ", Passed: "
																+ getNoOfRecords() + ", Failed: "
																+ getFailedRecords())).id("logStats")
				.process(MDWProcessors.intendedShutdown());

	}

	@NotNull
	private Processor archiveUsingDE() {
		return e -> {
			logDispatcher.info("Archiving source file using DE",
							   "File -> " + e.getIn().getHeader(Exchange.FILE_NAME, String.class));
			archiveFileSend.get().setOriginalFile(e.getIn().getBody(FileInfo.class).getTempFile());
			archiveFileSend.get().sendFile();
		};
	}

	private boolean needUnzip(Exchange e) {
		if(!unzip){
			return false;
		}
		String fileNameHeader = e.getIn().getHeader(Exchange.FILE_NAME, String.class);
		//GenericCsvToSnowflake do not have file in source. It only creates the FileInfo and since the header are not reflected to this route this should be the solution
		return !GenericCsvToSnowflake.class.isAssignableFrom(this.getClass()) && (fileNameHeader.toLowerCase().endsWith(".zip") || fileNameHeader.toLowerCase().endsWith(".gz"));
	}

	private void unzipInputFile(Exchange exchange) {
		FileInfo fileInfo = exchange.getIn()
								.getBody(FileInfo.class);
		File tempDir = fileInfo.getTempFile().getParentFile();
		String zippedFileName = exchange.getIn().getHeader(Exchange.FILE_NAME, String.class);
		String unzippedWantedPath = tempDir.getAbsolutePath() + File.separator + zippedFileName.replaceAll("\\.zip$", "").replaceAll("\\.gz$", "");
		File unzipped = new File(unzippedWantedPath);
		if (unzipped.exists()){
			logDispatcher.info("Unzipped file exists", "Deleting first. This is likely a rerun of already processed file");
			unzipped.delete();
		}
		try {
			if (!unzipped.createNewFile()){
				throw new RuntimeException("Could not create temp file for unzipped file. Wanted to create a file - "+unzippedWantedPath);
			}
		} catch (IOException e) {
			throw new RuntimeException(e);
		}
		logDispatcher.info("Unzipped filename", "Original filename: "+fileInfo.getFileName() + " =>  Unzipped: "+unzipped.getAbsolutePath());
		try {
			if (zippedFileName.endsWith("zip")){
					try (java.util.zip.ZipFile zipFile = new ZipFile(fileInfo.getTempFile())) {
						Enumeration<? extends ZipEntry> entries = zipFile.entries();
						while (entries.hasMoreElements()) {
							ZipEntry entry = entries.nextElement();
							File entryDestination = new File(tempDir, entry.getName());
							if (entry.isDirectory()) {
								entryDestination.mkdirs();
							} else {
								entryDestination.getParentFile()
												.mkdirs();
								try (InputStream in = new BufferedInputStream(zipFile.getInputStream(entry)); OutputStream out = new BufferedOutputStream(new FileOutputStream(entryDestination))) {
									IOUtils.copy(in, out);
									out.flush();
									exchange.getIn().setBody(new FileInfo(new File(entryDestination.getAbsolutePath()), logDispatcher));
								}
							}
						}
					}
			}
			else if (zippedFileName.endsWith("gz")){
				try (GZIPInputStream is = new GZIPInputStream(new BufferedInputStream(new FileInputStream(fileInfo.getTempFile()))); OutputStream out = new BufferedOutputStream(new FileOutputStream(unzipped))){
					IOUtils.copy(is, out);
					out.flush();
					exchange.getIn().setBody(new FileInfo(new File(unzipped.getAbsolutePath()), logDispatcher));
				}
			}
			else {
				throw new RuntimeException("Unsupported compression format for file -> "+fileInfo.getFileName());
			}
		} catch (Exception e) {
			logDispatcher.error("Error decompressing",
								"Could not decompress file " + fileInfo.getTempFile()
																	   .getAbsolutePath(),
								e);
		}

	}

	private void rewriteDynamicUri(Exchange e, String targetHeaderName, String anotherSqlHeaderName,Endpoint target) {
		String baseUri = target.getEndpointUri().replace(":", "://");
		UriComponentsBuilder builder = UriComponentsBuilder.fromUriString(UriEncoder.decode(baseUri));
		String anotherSql = e.getIn().getHeader(anotherSqlHeaderName, String.class);
		UriComponents newSql = builder.host(anotherSql)
									  .build();
		String newUri = UriEncoder.encode(newSql.toUriString());
		e.getIn().setHeader(targetHeaderName, newUri.replace("://", ":"));
	}

	private void cleanupSentFile(Exchange exchange) throws IOException {
		if (!GenericFileToSnowflake.class.isAssignableFrom(this.getClass())) {
			FileInfo body = exchange.getIn().getBody(FileInfo.class);
			body.close();
		}
	}

	@NotNull
	protected Processor anyRecords() {
		return exchange -> anyRecords = true;
	}

	void setNoOfRecords(Long noOfRecords) {
		this.noOfRecords = noOfRecords;
	}

	public Long getNoOfRecords() {
		return noOfRecords;
	}

	public Long getPassRecords() {
		return passRecords;
	}

	public Long getFailedRecords() {
		return failedRecords;
	}

	public void setPassRecords(Long passRecords) {
		this.passRecords = passRecords;
	}

	public void setFailedRecords(Long failedRecords) {
		this.failedRecords = failedRecords;
	}

	public abstract Processor constructAndSendCopyInto();

	protected String executeCopyIntoCommand(String copyInto) throws SQLException {
		logDispatcher.debug("COPY INTO command", copyInto);
		try {
			SFBaseResultSet resultSet = snowflakeConnection.getHandler()
														 .getSFStatement()
														 .execute(copyInto, new HashMap<>(), SFBaseStatement.CallingMethod.EXECUTE);
			SnowflakeReply reply = getResponseForSnowflakeCall(resultSet);
			if (SnowflakeReply.SnowflakeResponseStatus.ERROR.getStatus().equalsIgnoreCase(reply.getStatus())) {
				throw new SFException(ErrorCode.BAD_RESPONSE, "Response from Snowflake: " + reply);
			}
			logDispatcher.debug("Result of the COPY INTO command", "Details: "+ reply);
			return reply.toString();
		} catch (SFException e) {
			throw new RuntimeException("Exception during calling COPY INTO command", e);
		}
	}

	protected List<String> getSqls(GenericItemType property){
		return inboundMessage.getPropertiesMap().entrySet()
				.stream()
				.filter(e -> e.getKey().startsWith(property.getPropertyName()))
				.map(e -> replaceFromPropertiesOrPayload(e.getValue()))
				.sorted()
				.collect(Collectors.toList());
	}


	protected List<Column> getInsertValuesOrder() {
		if (insertStatement.getColumns().size() == 1) {
			return List.of(insertStatement.getColumns().get(0));
		}
		List<Column> columns = insertStatement.getValues()
											  .getExpressions()
											  .stream()
											  .filter(it -> Column.class.isAssignableFrom(it.getClass()))
											  .map(Column.class::cast)
											  .toList();
		return new LinkedList<>(columns);
	}


	private String replaceFromPropertiesOrPayload(String replaceInMe){
		return StrSubstitutor.replace(replaceInMe, propsAndPayloadReplaceMap);
	}

	protected Endpoint determineEndpointForPrePostSql(GenericItemType itemType){
		return targetPrePostProcessSqls;
	}

	protected String getTableName() {
		TablesNamesFinder tablesNamesFinder = new TablesNamesFinder();
		List<String> tableList = tablesNamesFinder.getTableList(insertStatement);
		if (tableList.size() > 1) {
			throw new RuntimeException("Currently interface is not implemented to handle more that one table. Please update insertSql2 interface property to have only one table name");
		}
		return tableList.get(0);
	}

	void setSelectSql(String selectSql) {
		this.selectSql = selectSql;
	}

	public String getHeaderLine() {
		return insertStatement.getColumns()
							  .stream().map(Column::getColumnName).collect(Collectors.joining(", "));
	}

	public void setSnowflakeGlobalColumnNames(String snowflakeGlobalColumnNames) {
		this.snowflakeGlobalColumnNames = snowflakeGlobalColumnNames;
	}
}
