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

package io.airbyte.integrations.destination.jdbc.copy;

import com.fasterxml.jackson.databind.JsonNode;
import io.airbyte.db.factory.DataSourceFactory;
import io.airbyte.db.jdbc.JdbcDatabase;
import io.airbyte.integrations.BaseConnector;
import io.airbyte.integrations.base.Destination;
import io.airbyte.integrations.destination.ExtendedNameTransformer;
import io.airbyte.integrations.destination.jdbc.AbstractJdbcDestination;
import io.airbyte.integrations.destination.jdbc.SqlOperations;
import io.airbyte.protocol.models.AirbyteConnectionStatus;
import javax.sql.DataSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public abstract class CopyDestination extends BaseConnector implements Destination {

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

  /**
   * The default database schema field in the destination config is "schema". To change it, pass the
   * field name to the constructor.
   */
  private String schemaFieldName = "schema";

  public CopyDestination() {}

  public CopyDestination(final String schemaFieldName) {
    this.schemaFieldName = schemaFieldName;
  }

  /**
   * A self contained method for writing a file to the persistence for testing. This method should try
   * to clean up after itself by deleting the file it creates.
   */
  public abstract void checkPersistence(JsonNode config) throws Exception;

  public abstract ExtendedNameTransformer getNameTransformer();

  public abstract DataSource getDataSource(JsonNode config);

  public abstract JdbcDatabase getDatabase(DataSource dataSource);

  public abstract SqlOperations getSqlOperations();

  @Override
  public AirbyteConnectionStatus check(final JsonNode config) {
    try {
      checkPersistence(config);
    } catch (final Exception e) {
      LOGGER.error("Exception attempting to access the staging persistence: ", e);
      return new AirbyteConnectionStatus()
          .withStatus(AirbyteConnectionStatus.Status.FAILED)
          .withMessage("Could not connect to the staging persistence with the provided configuration. \n" + e.getMessage());
    }

    final DataSource dataSource = getDataSource(config);

    try {
      final JdbcDatabase database = getDatabase(dataSource);
      final var nameTransformer = getNameTransformer();
      final var outputSchema = nameTransformer.convertStreamName(config.get(schemaFieldName).asText());
      AbstractJdbcDestination.attemptSQLCreateAndDropTableOperations(outputSchema, database, nameTransformer, getSqlOperations());

      return new AirbyteConnectionStatus().withStatus(AirbyteConnectionStatus.Status.SUCCEEDED);
    } catch (final Exception e) {
      LOGGER.error("Exception attempting to connect to the warehouse: ", e);
      return new AirbyteConnectionStatus()
          .withStatus(AirbyteConnectionStatus.Status.FAILED)
          .withMessage("Could not connect to the warehouse with the provided configuration. \n" + e.getMessage());
    } finally {
      try {
        DataSourceFactory.close(dataSource);
      } catch (final Exception e) {
        LOGGER.warn("Unable to close data source.", e);
      }
    }
  }

}
