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

package io.airbyte.integrations.source.mysql;

import com.fasterxml.jackson.databind.JsonNode;
import io.airbyte.db.jdbc.JdbcUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

import static io.airbyte.integrations.util.MySqlSslConnectionUtils.checkOrCreatePassword;

public class MySqlCdcProperties {

  static Properties getDebeziumProperties(final JsonNode config) {
    final Properties props = new Properties();

    // debezium engine configuration
    props.setProperty("connector.class", "io.debezium.connector.mysql.MySqlConnector");

    // https://debezium.io/documentation/reference/connectors/mysql.html#mysql-boolean-values
    // https://debezium.io/documentation/reference/1.9/development/converters.html
    /**
     * {@link io.debezium.connector.mysql.converters.TinyIntOneToBooleanConverter}
     * {@link MySQLConverter}
     */
    props.setProperty("converters", "boolean, datetime");
    props.setProperty("boolean.type", "io.debezium.connector.mysql.converters.TinyIntOneToBooleanConverter");
    props.setProperty("datetime.type", "io.airbyte.integrations.debezium.internals.MySQLDateTimeConverter");

    // snapshot config
    if (config.has("snapshot_mode")) {
      //The parameter `snapshot_mode` is passed in test to simulate reading the binlog directly and skip initial snapshot
      props.setProperty("snapshot.mode", config.get("snapshot_mode").asText());
    } else {
      // https://debezium.io/documentation/reference/1.9/connectors/mysql.html#mysql-property-snapshot-mode
      props.setProperty("snapshot.mode", "when_needed");
    }
    // https://debezium.io/documentation/reference/1.9/connectors/mysql.html#mysql-property-snapshot-locking-mode
    // This is to make sure other database clients are allowed to write to a table while Airbyte is
    // taking a snapshot. There is a risk involved that
    // if any database client makes a schema change then the sync might break
    props.setProperty("snapshot.locking.mode", "none");
    // https://debezium.io/documentation/reference/1.9/connectors/mysql.html#mysql-property-include-schema-changes
    props.setProperty("include.schema.changes", "false");
    // This to make sure that binary data represented as a base64-encoded String.
    // https://debezium.io/documentation/reference/1.9/connectors/mysql.html#mysql-property-binary-handling-mode
    props.setProperty("binary.handling.mode", "base64");
    props.setProperty("database.include.list", config.get("database").asText());
    // Check params for SSL connection in config and add properties for CDC SSL connection
    // https://debezium.io/documentation/reference/stable/connectors/mysql.html#mysql-property-database-ssl-mode
    if (!config.has(JdbcUtils.SSL_KEY) || config.get(JdbcUtils.SSL_KEY).asBoolean()) {
      if (config.has(JdbcUtils.SSL_MODE_KEY) && config.get(JdbcUtils.SSL_MODE_KEY).has(JdbcUtils.MODE_KEY)) {
        props.setProperty("database.ssl.mode", config.get(JdbcUtils.SSL_MODE_KEY).get(JdbcUtils.MODE_KEY).asText());
        final var method = config.get(JdbcUtils.SSL_MODE_KEY).get(JdbcUtils.MODE_KEY).asText();
        if (method.equals("verify_ca") || method.equals("verify_identity")) {
          var sslPassword = checkOrCreatePassword(config.get(JdbcUtils.SSL_MODE_KEY));
          props.setProperty("database.history.producer.security.protocol", "SSL");
          props.setProperty("database.history.producer.ssl.truststore.location", "customtruststore.jks");
          props.setProperty("database.history.producer.ssl.truststore.password", sslPassword);
          props.setProperty("database.history.producer.ssl.key.password", sslPassword);

          props.setProperty("database.history.consumer.security.protocol", "SSL");
          props.setProperty("database.history.consumer.ssl.truststore.location", "customtruststore.jks");
          props.setProperty("database.history.consumer.ssl.truststore.password", sslPassword);
          props.setProperty("database.history.consumer.ssl.key.password", sslPassword);
          if (method.equals("verify_identity")) {
            props.setProperty("database.history.producer.ssl.keystore.location", "customkeystore.jks");
            props.setProperty("database.history.producer.ssl.keystore.password", sslPassword);

            props.setProperty("database.history.consumer.ssl.keystore.location", "customkeystore.jks");
            props.setProperty("database.history.consumer.ssl.keystore.password", sslPassword);
          }
        }
      } else {
        props.setProperty("database.ssl.mode", "required");
      }
    }
    return props;
  }

}
