/*
 * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
 */

package io.airbyte.integrations.source.sftp;

import com.fasterxml.jackson.databind.JsonNode;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.ChannelSftp.LsEntry;
import com.jcraft.jsch.SftpException;
import io.airbyte.integrations.source.sftp.enums.SupportedFileExtension;
import io.airbyte.integrations.source.sftp.parsers.SftpFileParser;
import io.airbyte.integrations.source.sftp.parsers.SftpFileParserFactory;
import io.airbyte.integrations.source.sftp.util.JsonSchemaGenerator;
import java.io.ByteArrayInputStream;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Vector;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import org.apache.commons.io.FilenameUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SftpCommand {

  protected static final Logger LOGGER = LoggerFactory.getLogger(SftpClient.class);

  private static final String FILE_TYPE_SEPARATOR = ",";
  private final SftpClient client;
  private final Set<SupportedFileExtension> selectedFileExtensions;
  private final Pattern filePattern;
  private final SftpFileParserFactory sftpFileParserFactory;

  public SftpCommand(SftpClient client, JsonNode config) {
    this.client = client;
    sftpFileParserFactory = new SftpFileParserFactory();
    String commaSeparatedFileExtension = config.has("file_types") ? config.get("file_types").asText() : "";
    Set<String> selectedFileExtension = Set.of(commaSeparatedFileExtension.split(FILE_TYPE_SEPARATOR));
    selectedFileExtensions = selectedFileExtension.stream()
        .map(this::transformFileExtension)
        .collect(Collectors.toSet());
    filePattern = Pattern.compile(config.has("file_pattern") ? config.get("file_pattern").asText() : "");
  }

  private SupportedFileExtension transformFileExtension(String fileExtension) {
    try {
      return SupportedFileExtension.valueOf(fileExtension.toUpperCase());
    } catch (Exception e) {
      LOGGER.error("Unsupported file extension : {}", fileExtension);
      throw new RuntimeException();
    }
  }

  public void tryChangeWorkingDirectory(String path) {
    try {
      checkIfConnected();
      client.changeWorkingDirectory(FilenameUtils.normalize(path));
      LOGGER.info("Working directory set to : {}", path);
    } catch (SftpException e) {
      LOGGER.error("Could not find path : {} : ", path, e);
      throw new RuntimeException(e);
    }
  }

  public Map<String, JsonNode> getFilesSchemas() {
    checkIfConnected();
    Map<String, JsonNode> fileSchemas = new HashMap<>();
    Set<String> fileNames = getFileNames();
    LOGGER.info("Found file for sync : {}", fileNames);
    fileNames.forEach(fileName -> {
      ByteArrayInputStream file = client.getFile(fileName);
      JsonNode parsedFile = tryGetFirstNode(file, fileName);
      JsonNode schema = tryGetSchema(parsedFile, fileName);
      fileSchemas.put(fileName, schema);
    });
    return fileSchemas;
  }

  private Set<String> getFileNames() {
    checkIfConnected();
    Vector<LsEntry> entries = new Vector<>();
    for (SupportedFileExtension fileExtension : selectedFileExtensions) {
      entries.addAll(client.lsFile(fileExtension));
    }
    return entries.stream()
        .map(ChannelSftp.LsEntry::getFilename)
        .filter(filePattern.asPredicate())
        .collect(Collectors.toSet());
  }

  private JsonNode tryGetSchema(JsonNode jsonNode, String fileName) {
    try {
      return JsonSchemaGenerator.getJsonSchema(jsonNode);
    } catch (Exception e) {
      LOGGER.error("Exception occurred while trying to parse schema of file {} : ", fileName, e);
      throw new RuntimeException(e);
    }
  }

  public JsonNode tryGetFirstNode(ByteArrayInputStream file, String fileName) {
    try {
      String extension = FilenameUtils.getExtension(fileName);
      SftpFileParser parser = sftpFileParserFactory.create(transformFileExtension(extension));
      return parser.parseFileFirstEntity(file);
    } catch (Exception e) {
      LOGGER.error("Exception occurred while trying to parse file {} : ", fileName, e);
      throw new RuntimeException();
    }
  }

  public List<JsonNode> getFileData(String fileName) {
    ByteArrayInputStream file = client.getFile(fileName);
    try {
      String extension = FilenameUtils.getExtension(fileName);
      SftpFileParser parser = sftpFileParserFactory.create(transformFileExtension(extension));
      return parser.parseFile(file);
    } catch (Exception e) {
      LOGGER.error("Exception occurred while trying to parse file {} : ", fileName, e);
      throw new RuntimeException();
    }
  }

  private void checkIfConnected() {
    if (!client.isConnected()) {
      LOGGER.info("SFTP client is not connected. Attempting to reconnect.");
      client.connect();
    }
  }

}
