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

package io.airbyte.integrations.source.mysql;

import com.fasterxml.jackson.databind.JsonNode;
import com.mysql.cj.MysqlType;
import io.airbyte.db.Database;
import io.airbyte.integrations.standardtest.source.AbstractSourceDatabaseTypeTest;
import io.airbyte.integrations.standardtest.source.TestDataHolder;
import io.airbyte.protocol.models.JsonSchemaType;
import java.io.File;
import java.io.IOException;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.commons.codec.binary.Base64;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.MySQLContainer;

public abstract class AbstractMySqlSourceDatatypeTest extends AbstractSourceDatabaseTypeTest {

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

  protected MySQLContainer<?> container;
  protected JsonNode config;

  @Override
  protected JsonNode getConfig() {
    return config;
  }

  @Override
  protected String getImageName() {
    return "airbyte/source-mysql:dev";
  }

  @Override
  protected abstract Database setupDatabase() throws Exception;

  @Override
  protected String getNameSpace() {
    return container.getDatabaseName();
  }

  @Override
  protected void initTests() {
    // bit defaults to bit(1), which is equivalent to boolean
    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("bit")
            .airbyteType(JsonSchemaType.BOOLEAN)
            .addInsertValues("null", "1", "0")
            .addExpectedValues(null, "true", "false")
            .build());

    // bit(1) is equivalent to boolean
    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("bit")
            .fullSourceDataType("bit(1)")
            .airbyteType(JsonSchemaType.BOOLEAN)
            .addInsertValues("null", "1", "0")
            .addExpectedValues(null, "true", "false")
            .build());

    // bit(>1) is binary
    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("bit")
            .fullSourceDataType("bit(7)")
            .airbyteType(JsonSchemaType.STRING_BASE_64)
            // 1000001 is binary for A
            .addInsertValues("null", "b'1000001'")
            // QQo= is base64 encoding in charset UTF-8 for A
            .addExpectedValues(null, "QQ==")
            .build());

    // tinyint without width
    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("tinyint")
            .airbyteType(JsonSchemaType.INTEGER)
            .addInsertValues("null", "-128", "127")
            .addExpectedValues(null, "-128", "127")
            .build());

    // tinyint(1) is equivalent to boolean
    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("tinyint")
            .fullSourceDataType("tinyint(1)")
            .airbyteType(JsonSchemaType.BOOLEAN)
            .addInsertValues("null", "1", "0")
            .addExpectedValues(null, "true", "false")
            .build());

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("tinyint")
            .fullSourceDataType("tinyint(2)")
            .airbyteType(JsonSchemaType.INTEGER)
            .addInsertValues("null", "-128", "127")
            .addExpectedValues(null, "-128", "127")
            .build());

    final Set<String> booleanTypes = Set.of("BOOLEAN", "BOOL");
    for (final String booleanType : booleanTypes) {
      addDataTypeTestData(
          TestDataHolder.builder()
              .sourceType(booleanType)
              .airbyteType(JsonSchemaType.BOOLEAN)
              // MySql booleans are tinyint(1), and only 1 is true
              .addInsertValues("null", "1", "0")
              .addExpectedValues(null, "true", "false")
              .build());
    }

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("smallint")
            .airbyteType(JsonSchemaType.INTEGER)
            .addInsertValues("null", "-32768", "32767")
            .addExpectedValues(null, "-32768", "32767")
            .build());

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("smallint")
            .airbyteType(JsonSchemaType.INTEGER)
            .fullSourceDataType("smallint zerofill")
            .addInsertValues("1")
            .addExpectedValues("1")
            .build());

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("smallint")
            .airbyteType(JsonSchemaType.INTEGER)
            .fullSourceDataType("smallint unsigned")
            .addInsertValues("null", "0", "65535")
            .addExpectedValues(null, "0", "65535")
            .build());

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("mediumint")
            .airbyteType(JsonSchemaType.INTEGER)
            .addInsertValues("null", "-8388608", "8388607")
            .addExpectedValues(null, "-8388608", "8388607")
            .build());

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("mediumint")
            .airbyteType(JsonSchemaType.INTEGER)
            .fullSourceDataType("mediumint zerofill")
            .addInsertValues("1")
            .addExpectedValues("1")
            .build());

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("int")
            .airbyteType(JsonSchemaType.INTEGER)
            .addInsertValues("null", "-2147483648", "2147483647")
            .addExpectedValues(null, "-2147483648", "2147483647")
            .build());

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("int")
            .airbyteType(JsonSchemaType.INTEGER)
            .fullSourceDataType("int unsigned")
            .addInsertValues("3428724653")
            .addExpectedValues("3428724653")
            .build());

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("int")
            .airbyteType(JsonSchemaType.INTEGER)
            .fullSourceDataType("int zerofill")
            .addInsertValues("1")
            .addExpectedValues("1")
            .build());

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("bigint")
            .airbyteType(JsonSchemaType.INTEGER)
            .addInsertValues("null", "9223372036854775807")
            .addExpectedValues(null, "9223372036854775807")
            .build());

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("float")
            .airbyteType(JsonSchemaType.NUMBER)
            .addInsertValues("null", "10.5")
            .addExpectedValues(null, "10.5")
            .build());

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("double")
            .airbyteType(JsonSchemaType.NUMBER)
            .addInsertValues("null", "power(10, 308)", "1/power(10, 45)", "10.5")
            .addExpectedValues(null, String.valueOf(Math.pow(10, 308)), String.valueOf(1 / Math.pow(10, 45)), "10.5")
            .build());

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("decimal")
            .airbyteType(JsonSchemaType.NUMBER)
            .fullSourceDataType("decimal(10,3)")
            .addInsertValues("0.188", "null")
            .addExpectedValues("0.188", null)
            .build());

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("decimal")
            .airbyteType(JsonSchemaType.NUMBER)
            .fullSourceDataType("decimal(19,2)")
            .addInsertValues("1700000.01")
            .addExpectedValues("1700000.01")
            .build());

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("date")
            .airbyteType(JsonSchemaType.STRING_DATE)
            .addInsertValues("null", "'2021-01-01'")
            .addExpectedValues(null, "2021-01-01")
            .build());

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("datetime")
            .airbyteType(JsonSchemaType.STRING_TIMESTAMP_WITHOUT_TIMEZONE)
            .addInsertValues("null", "'2005-10-10 23:22:21'", "'2013-09-05T10:10:02'", "'2013-09-06T10:10:02'")
            .addExpectedValues(null, "2005-10-10T23:22:21.000000", "2013-09-05T10:10:02.000000", "2013-09-06T10:10:02.000000")
            .build());

    addTimestampDataTypeTest();

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("time")
            .airbyteType(JsonSchemaType.STRING_TIME_WITHOUT_TIMEZONE)
            // JDBC driver can process only "clock"(00:00:00-23:59:59) values.
            .addInsertValues("null", "'-22:59:59'", "'23:59:59'", "'00:00:00'")
            .addExpectedValues(null, "22:59:59.000000", "23:59:59.000000", "00:00:00.000000")
            .build());

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("year")
            .airbyteType(JsonSchemaType.STRING)
            // MySQL converts values in the ranges '0' - '69' to YEAR value in the range 2000 - 2069
            // and '70' - '99' to 1970 - 1999.
            .addInsertValues("null", "'1997'", "'0'", "'50'", "'70'", "'80'", "'99'")
            .addExpectedValues(null, "1997", "2000", "2050", "1970", "1980", "1999")
            .build());

    // char types can be string or binary, so they are tested separately
    final Set<String> charTypes = Stream.of(MysqlType.CHAR, MysqlType.VARCHAR)
        .map(Enum::name)
        .collect(Collectors.toSet());
    for (final String charType : charTypes) {
      addDataTypeTestData(
          TestDataHolder.builder()
              .sourceType(charType)
              .airbyteType(JsonSchemaType.STRING)
              .fullSourceDataType(charType + "(63)")
              .addInsertValues("null", "'Airbyte'", "'!\"#$%&\\'()*+,-./:;<=>?\\@[\\]^_\\`{|}~'")
              .addExpectedValues(null, "Airbyte", "!\"#$%&'()*+,-./:;<=>?@[]^_`{|}~")
              .build());

      addDataTypeTestData(
          TestDataHolder.builder()
              .sourceType(charType)
              .airbyteType(JsonSchemaType.STRING)
              .fullSourceDataType(charType + "(63) character set utf16")
              .addInsertValues("0xfffd")
              .addExpectedValues("�")
              .build());

      addDataTypeTestData(
          TestDataHolder.builder()
              .sourceType(charType)
              .airbyteType(JsonSchemaType.STRING)
              .fullSourceDataType(charType + "(63) character set cp1251")
              .addInsertValues("'тест'")
              .addExpectedValues("тест")
              .build());

      // when charset is binary, return binary in base64 encoding in charset UTF-8
      addDataTypeTestData(
          TestDataHolder.builder()
              .sourceType(charType)
              .airbyteType(JsonSchemaType.STRING_BASE_64)
              .fullSourceDataType(charType + "(7) character set binary")
              .addInsertValues("null", "'Airbyte'")
              .addExpectedValues(null, "QWlyYnl0ZQ==")
              .build());
    }

    final Set<String> blobTypes = Stream
        .of(MysqlType.TINYBLOB, MysqlType.BLOB, MysqlType.MEDIUMBLOB, MysqlType.LONGBLOB)
        .map(Enum::name)
        .collect(Collectors.toSet());
    for (final String blobType : blobTypes) {
      addDataTypeTestData(
          TestDataHolder.builder()
              .sourceType(blobType)
              .airbyteType(JsonSchemaType.STRING_BASE_64)
              .addInsertValues("null", "'Airbyte'")
              .addExpectedValues(null, "QWlyYnl0ZQ==")
              .build());
    }

    // binary appends '\0' to the end of the string
    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType(MysqlType.BINARY.name())
            .fullSourceDataType(MysqlType.BINARY.name() + "(10)")
            .airbyteType(JsonSchemaType.STRING_BASE_64)
            .addInsertValues("null", "'Airbyte'")
            .addExpectedValues(null, "QWlyYnl0ZQAAAA==")
            .build());

    // varbinary does not append '\0' to the end of the string
    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType(MysqlType.VARBINARY.name())
            .fullSourceDataType(MysqlType.VARBINARY.name() + "(10)")
            .airbyteType(JsonSchemaType.STRING_BASE_64)
            .addInsertValues("null", "'Airbyte'")
            .addExpectedValues(null, "QWlyYnl0ZQ==")
            .build());

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType(MysqlType.VARBINARY.name())
            .airbyteType(JsonSchemaType.STRING_BASE_64)
            .fullSourceDataType(MysqlType.VARBINARY.name() + "(20000)") // size should be enough to save test.png
            .addInsertValues("null", "'test'", "'тест'", String.format("FROM_BASE64('%s')", getFileDataInBase64()))
            .addExpectedValues(null, "dGVzdA==", "0YLQtdGB0YI=", getFileDataInBase64())
            .build());

    final Set<String> textTypes = Stream
        .of(MysqlType.TINYTEXT, MysqlType.TEXT, MysqlType.MEDIUMTEXT, MysqlType.LONGTEXT)
        .map(Enum::name)
        .collect(Collectors.toSet());
    final String randomText = RandomStringUtils.random(50, true, true);
    for (final String textType : textTypes) {
      addDataTypeTestData(
          TestDataHolder.builder()
              .sourceType(textType)
              .airbyteType(JsonSchemaType.STRING)
              .addInsertValues("null", "'Airbyte'", String.format("'%s'", randomText))
              .addExpectedValues(null, "Airbyte", randomText)
              .build());
    }

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("mediumtext")
            .airbyteType(JsonSchemaType.STRING)
            .addInsertValues(getLogString(1048000), "'test'")
            .addExpectedValues(StringUtils.leftPad("0", 1048000, "0"), "test")
            .build());

    addJsonDataTypeTest();

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("enum")
            .fullSourceDataType("ENUM('xs', 's', 'm', 'l', 'xl')")
            .airbyteType(JsonSchemaType.STRING)
            .addInsertValues("null", "'xs'", "'m'")
            .addExpectedValues(null, "xs", "m")
            .build());

    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("set")
            .fullSourceDataType("SET('xs', 's', 'm', 'l', 'xl')")
            .airbyteType(JsonSchemaType.STRING)
            .addInsertValues("null", "'xs,s'", "'m,xl'")
            .addExpectedValues(null, "xs,s", "m,xl")
            .build());

  }

  protected void addJsonDataTypeTest() {
    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("json")
            .airbyteType(JsonSchemaType.STRING)
            .addInsertValues("null", "'{\"a\": 10, \"b\": 15}'", "'{\"fóo\": \"bär\"}'", "'{\"春江潮水连海平\": \"海上明月共潮生\"}'")
            .addExpectedValues(null, "{\"a\": 10, \"b\": 15}", "{\"fóo\": \"bär\"}", "{\"春江潮水连海平\": \"海上明月共潮生\"}")
            .build());
  }

  protected void addTimestampDataTypeTest() {
    addDataTypeTestData(
        TestDataHolder.builder()
            .sourceType("timestamp")
            .airbyteType(JsonSchemaType.STRING_TIMESTAMP_WITH_TIMEZONE)
            .addInsertValues("null", "'2021-01-00'", "'2021-00-00'", "'0000-00-00'", "'2022-08-09T10:17:16.161342Z'")
            .addExpectedValues(null, null, null, null, "2022-08-09T10:17:16.000000Z")
            .build());
  }

  private String getLogString(final int length) {
    final int maxLpadLength = 262144;
    final StringBuilder stringBuilder = new StringBuilder("concat(");
    final int fullChunks = length / maxLpadLength;
    stringBuilder.append("lpad('0', 262144, '0'),".repeat(fullChunks));
    stringBuilder.append("lpad('0', ").append(length % maxLpadLength).append(", '0'))");
    return stringBuilder.toString();
  }

  private String getFileDataInBase64() {
    final File file = new File(getClass().getClassLoader().getResource("test.png").getFile());
    try {
      return Base64.encodeBase64String(FileUtils.readFileToByteArray(file));
    } catch (final IOException e) {
      LOGGER.error(String.format("Fail to read the file: %s. Error: %s", file.getAbsoluteFile(), e.getMessage()));
    }
    return null;
  }

}
