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

package io.airbyte.integrations.destination.cassandra;

import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;

import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.collect.ImmutableMap;
import io.airbyte.commons.json.Jsons;
import io.airbyte.protocol.models.AirbyteMessage;
import io.airbyte.protocol.models.DestinationSyncMode;
import java.util.function.Function;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.MethodOrderer.OrderAnnotation;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.TestMethodOrder;

@TestMethodOrder(OrderAnnotation.class)
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
class CassandraMessageConsumerIT {

  private static final String AIRBYTE_NAMESPACE_1 = "airbyte_namespace_1";
  private static final String AIRBYTE_NAMESPACE_2 = "airbyte_namespace_2";

  private static final String AIRBYTE_STREAM_1 = "airbyte_stream_1";
  private static final String AIRBYTE_STREAM_2 = "airbyte_stream_2";

  private CassandraMessageConsumer cassandraMessageConsumer;

  private CassandraCqlProvider cassandraCqlProvider;

  private CassandraNameTransformer nameTransformer;

  @BeforeAll
  void setup() {
    var cassandraContainer = CassandraContainerInitializr.initContainer();
    var cassandraConfig = TestDataFactory.createCassandraConfig(
        cassandraContainer.getUsername(),
        cassandraContainer.getPassword(),
        cassandraContainer.getHost(),
        cassandraContainer.getFirstMappedPort());

    var stream1 = TestDataFactory.createAirbyteStream(AIRBYTE_STREAM_1, AIRBYTE_NAMESPACE_1);
    var stream2 = TestDataFactory.createAirbyteStream(AIRBYTE_STREAM_2, AIRBYTE_NAMESPACE_2);

    var cStream1 = TestDataFactory.createConfiguredAirbyteStream(DestinationSyncMode.APPEND, stream1);
    var cStream2 = TestDataFactory.createConfiguredAirbyteStream(DestinationSyncMode.OVERWRITE, stream2);

    var catalog = TestDataFactory.createConfiguredAirbyteCatalog(cStream1, cStream2);

    cassandraCqlProvider = new CassandraCqlProvider(cassandraConfig);
    cassandraMessageConsumer = new CassandraMessageConsumer(cassandraConfig, catalog, cassandraCqlProvider, message -> {});
    nameTransformer = new CassandraNameTransformer(cassandraConfig);
  }

  @Test
  @Order(1)
  void testStartTracked() {

    assertDoesNotThrow(() -> cassandraMessageConsumer.startTracked());

  }

  @Test
  @Order(2)
  void testAcceptTracked() {

    Function<String, JsonNode> function =
        data -> Jsons.jsonNode(ImmutableMap.builder().put("property", data).build());

    assertDoesNotThrow(() -> {
      cassandraMessageConsumer.acceptTracked(
          TestDataFactory.createAirbyteMessage(AirbyteMessage.Type.RECORD, AIRBYTE_STREAM_1, AIRBYTE_NAMESPACE_1,
              function.apply("data1")));
      cassandraMessageConsumer.acceptTracked(
          TestDataFactory.createAirbyteMessage(AirbyteMessage.Type.RECORD, AIRBYTE_STREAM_1, AIRBYTE_NAMESPACE_1,
              function.apply("data2")));
      cassandraMessageConsumer.acceptTracked(
          TestDataFactory.createAirbyteMessage(AirbyteMessage.Type.RECORD, AIRBYTE_STREAM_2, AIRBYTE_NAMESPACE_2,
              function.apply("data3")));
      cassandraMessageConsumer.acceptTracked(
          TestDataFactory.createAirbyteMessage(AirbyteMessage.Type.RECORD, AIRBYTE_STREAM_2, AIRBYTE_NAMESPACE_2,
              function.apply("data4")));
      cassandraMessageConsumer.acceptTracked(
          TestDataFactory.createAirbyteMessage(AirbyteMessage.Type.STATE, AIRBYTE_STREAM_2, AIRBYTE_NAMESPACE_2,
              function.apply("data5")));
    });

  }

  @Test
  @Order(3)
  void testClose() {

    assertDoesNotThrow(() -> cassandraMessageConsumer.close(false));

  }

  @Test
  @Order(4)
  void testFinalState() {

    var keyspace1 = nameTransformer.outputKeyspace(AIRBYTE_NAMESPACE_1);
    var keyspace2 = nameTransformer.outputKeyspace(AIRBYTE_NAMESPACE_2);
    var table1 = nameTransformer.outputTable(AIRBYTE_STREAM_1);
    var table2 = nameTransformer.outputTable(AIRBYTE_STREAM_2);

    var resultSet1 = cassandraCqlProvider.select(keyspace1, table1);
    var resultSet2 = cassandraCqlProvider.select(keyspace2, table2);

    assertThat(resultSet1)
        .isNotNull()
        .hasSize(2)
        .anyMatch(r -> r.getData().equals("{\"property\":\"data1\"}"))
        .anyMatch(r -> r.getData().equals("{\"property\":\"data2\"}"));

    assertThat(resultSet2)
        .isNotNull()
        .hasSize(2)
        .anyMatch(r -> r.getData().equals("{\"property\":\"data3\"}"))
        .anyMatch(r -> r.getData().equals("{\"property\":\"data4\"}"));

  }

}
