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

package io.airbyte.integrations.destination.s3;

import static io.airbyte.integrations.destination.s3.constant.S3Constants.ACCESS_KEY_ID;
import static io.airbyte.integrations.destination.s3.constant.S3Constants.FILE_NAME_PATTERN;
import static io.airbyte.integrations.destination.s3.constant.S3Constants.SECRET_ACCESS_KEY;
import static io.airbyte.integrations.destination.s3.constant.S3Constants.S_3_BUCKET_NAME;
import static io.airbyte.integrations.destination.s3.constant.S3Constants.S_3_BUCKET_PATH;
import static io.airbyte.integrations.destination.s3.constant.S3Constants.S_3_BUCKET_REGION;
import static io.airbyte.integrations.destination.s3.constant.S3Constants.S_3_ENDPOINT;
import static io.airbyte.integrations.destination.s3.constant.S3Constants.S_3_PATH_FORMAT;

import com.amazonaws.ClientConfiguration;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
import com.fasterxml.jackson.databind.JsonNode;
import io.airbyte.integrations.destination.s3.credential.S3AWSDefaultProfileCredentialConfig;
import io.airbyte.integrations.destination.s3.credential.S3AccessKeyCredentialConfig;
import io.airbyte.integrations.destination.s3.credential.S3CredentialConfig;
import io.airbyte.integrations.destination.s3.credential.S3CredentialType;
import java.util.Objects;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * An S3 configuration. Typical usage sets at most one of {@code bucketPath} (necessary for more
 * delicate data syncing to S3)
 */
public class S3DestinationConfig {

  private static final Logger LOGGER = LoggerFactory.getLogger(S3DestinationConfig.class);

  private final String endpoint;
  private final String bucketName;
  private final String bucketPath;
  private final String bucketRegion;
  private final String pathFormat;
  private final S3CredentialConfig credentialConfig;
  private final S3FormatConfig formatConfig;
  private String fileNamePattern;

  private final Object lock = new Object();
  private AmazonS3 s3Client;

  public S3DestinationConfig(final String endpoint,
                             final String bucketName,
                             final String bucketPath,
                             final String bucketRegion,
                             final String pathFormat,
                             final S3CredentialConfig credentialConfig,
                             final S3FormatConfig formatConfig,
                             final AmazonS3 s3Client) {
    this.endpoint = endpoint;
    this.bucketName = bucketName;
    this.bucketPath = bucketPath;
    this.bucketRegion = bucketRegion;
    this.pathFormat = pathFormat;
    this.credentialConfig = credentialConfig;
    this.formatConfig = formatConfig;
    this.s3Client = s3Client;
  }

  public S3DestinationConfig(final String endpoint,
                             final String bucketName,
                             final String bucketPath,
                             final String bucketRegion,
                             final String pathFormat,
                             final S3CredentialConfig credentialConfig,
                             final S3FormatConfig formatConfig,
                             final AmazonS3 s3Client,
                             final String fileNamePattern) {
    this.endpoint = endpoint;
    this.bucketName = bucketName;
    this.bucketPath = bucketPath;
    this.bucketRegion = bucketRegion;
    this.pathFormat = pathFormat;
    this.credentialConfig = credentialConfig;
    this.formatConfig = formatConfig;
    this.s3Client = s3Client;
    this.fileNamePattern = fileNamePattern;
  }

  public static Builder create(final String bucketName, final String bucketPath, final String bucketRegion) {
    return new Builder(bucketName, bucketPath, bucketRegion);
  }

  public static Builder create(final S3DestinationConfig config) {
    return new Builder(config.getBucketName(), config.getBucketPath(), config.getBucketRegion())
        .withEndpoint(config.getEndpoint())
        .withCredentialConfig(config.getS3CredentialConfig())
        .withFormatConfig(config.getFormatConfig());
  }

  public static S3DestinationConfig getS3DestinationConfig(final JsonNode config) {
    Builder builder = create(
        config.get(S_3_BUCKET_NAME).asText(),
        "",
        config.get(S_3_BUCKET_REGION).asText());

    if (config.has(S_3_BUCKET_PATH)) {
      builder = builder.withBucketPath(config.get(S_3_BUCKET_PATH).asText());
    }

    if (config.has(FILE_NAME_PATTERN)) {
      builder = builder.withFileNamePattern(config.get(FILE_NAME_PATTERN).asText());
    }

    if (config.has(S_3_PATH_FORMAT)) {
      builder = builder.withPathFormat(config.get(S_3_PATH_FORMAT).asText());
    }

    if (config.has(S_3_ENDPOINT)) {
      builder = builder.withEndpoint(config.get(S_3_ENDPOINT).asText());
    }

    final S3CredentialConfig credentialConfig;
    if (config.has(ACCESS_KEY_ID)) {
      credentialConfig = new S3AccessKeyCredentialConfig(config.get(ACCESS_KEY_ID).asText(), config.get(SECRET_ACCESS_KEY).asText());
    } else {
      credentialConfig = new S3AWSDefaultProfileCredentialConfig();
    }
    builder = builder.withCredentialConfig(credentialConfig);

    // In the "normal" S3 destination, this is never null. However, the Redshift and Snowflake copy
    // destinations don't set a Format config.
    if (config.has("format")) {
      builder = builder.withFormatConfig(S3FormatConfigs.getS3FormatConfig(config));
    }

    return builder.get();
  }

  public String getEndpoint() {
    return endpoint;
  }

  public String getBucketName() {
    return bucketName;
  }

  public String getBucketPath() {
    return bucketPath;
  }

  public String getPathFormat() {
    return pathFormat;
  }

  public String getBucketRegion() {
    return bucketRegion;
  }

  public String getFileNamePattern() {
    return fileNamePattern;
  }

  public S3CredentialConfig getS3CredentialConfig() {
    return credentialConfig;
  }

  public S3FormatConfig getFormatConfig() {
    return formatConfig;
  }

  public AmazonS3 getS3Client() {
    synchronized (lock) {
      if (s3Client == null) {
        return resetS3Client();
      }
      return s3Client;
    }
  }

  AmazonS3 resetS3Client() {
    synchronized (lock) {
      if (s3Client != null) {
        s3Client.shutdown();
      }
      s3Client = createS3Client();
      return s3Client;
    }
  }

  protected AmazonS3 createS3Client() {
    LOGGER.info("Creating S3 client...");

    final AWSCredentialsProvider credentialsProvider = credentialConfig.getS3CredentialsProvider();
    final S3CredentialType credentialType = credentialConfig.getCredentialType();

    if (S3CredentialType.DEFAULT_PROFILE == credentialType) {
      return AmazonS3ClientBuilder.standard()
          .withRegion(bucketRegion)
          .withCredentials(credentialsProvider)
          .build();
    }

    if (null == endpoint || endpoint.isEmpty()) {
      return AmazonS3ClientBuilder.standard()
          .withCredentials(credentialsProvider)
          .withRegion(bucketRegion)
          .build();
    }

    final ClientConfiguration clientConfiguration = new ClientConfiguration();
    clientConfiguration.setSignerOverride("AWSS3V4SignerType");

    return AmazonS3ClientBuilder.standard()
        .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(endpoint, bucketRegion))
        .withPathStyleAccessEnabled(true)
        .withClientConfiguration(clientConfiguration)
        .withCredentials(credentialsProvider)
        .build();
  }

  @Override
  public boolean equals(final Object o) {
    if (this == o) {
      return true;
    }
    if (o == null || getClass() != o.getClass()) {
      return false;
    }
    final S3DestinationConfig that = (S3DestinationConfig) o;
    return Objects.equals(endpoint, that.endpoint) && Objects.equals(bucketName, that.bucketName) && Objects.equals(
        bucketPath, that.bucketPath) && Objects.equals(bucketRegion, that.bucketRegion)
        && Objects.equals(credentialConfig, that.credentialConfig)
        && Objects.equals(formatConfig, that.formatConfig);
  }

  @Override
  public int hashCode() {
    return Objects.hash(endpoint, bucketName, bucketPath, bucketRegion, credentialConfig, formatConfig);
  }

  public static class Builder {

    private String endpoint = "";
    private String pathFormat = S3DestinationConstants.DEFAULT_PATH_FORMAT;

    private String bucketName;
    private String bucketPath;
    private String bucketRegion;
    private S3CredentialConfig credentialConfig;
    private S3FormatConfig formatConfig;
    private AmazonS3 s3Client;
    private String fileNamePattern;

    private Builder(final String bucketName, final String bucketPath, final String bucketRegion) {
      this.bucketName = bucketName;
      this.bucketPath = bucketPath;
      this.bucketRegion = bucketRegion;
    }

    public Builder withBucketName(final String bucketName) {
      this.bucketName = bucketName;
      return this;
    }

    public Builder withFileNamePattern(final String fileNamePattern) {
      this.fileNamePattern = fileNamePattern;
      return this;
    }

    public Builder withBucketPath(final String bucketPath) {
      this.bucketPath = bucketPath;
      return this;
    }

    public Builder withBucketRegion(final String bucketRegion) {
      this.bucketRegion = bucketRegion;
      return this;
    }

    public Builder withPathFormat(final String pathFormat) {
      this.pathFormat = pathFormat;
      return this;
    }

    public Builder withEndpoint(final String endpoint) {
      this.endpoint = endpoint;
      return this;
    }

    public Builder withFormatConfig(final S3FormatConfig formatConfig) {
      this.formatConfig = formatConfig;
      return this;
    }

    public Builder withAccessKeyCredential(final String accessKeyId, final String secretAccessKey) {
      this.credentialConfig = new S3AccessKeyCredentialConfig(accessKeyId, secretAccessKey);
      return this;
    }

    public Builder withCredentialConfig(final S3CredentialConfig credentialConfig) {
      this.credentialConfig = credentialConfig;
      return this;
    }

    public Builder withS3Client(final AmazonS3 s3Client) {
      this.s3Client = s3Client;
      return this;
    }

    public S3DestinationConfig get() {
      return new S3DestinationConfig(
          endpoint,
          bucketName,
          bucketPath,
          bucketRegion,
          pathFormat,
          credentialConfig,
          formatConfig,
          s3Client,
          fileNamePattern);
    }

  }

}
