package io.aiven.klaw.clusterapi;

import static org.assertj.core.api.Assertions.assertThat;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.type.MapType;
import com.fasterxml.jackson.databind.type.TypeFactory;
import io.aiven.klaw.clusterapi.models.enums.TopicContentType;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestMethodOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.http.MediaType;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.mock.web.MockHttpServletResponse;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.TestPropertySource;
import org.springframework.test.web.servlet.MockMvc;
import org.springframework.test.web.servlet.request.MockMvcRequestBuilders;

@SpringBootTest(
    webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
    classes = KafkaClusterApiApplication.class)
@AutoConfigureMockMvc
@TestPropertySource(locations = "classpath:application.properties")
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
@DirtiesContext
@EmbeddedKafka(
    brokerProperties = {"listeners=PLAINTEXT://" + TopicContentsControllerIT.BOOTSTRAP_SERVER},
    partitions = 1,
    topics = {TopicContentsControllerIT.TEST_TOPIC_NAME})
public class TopicContentsControllerIT {

  public static final String BOOTSTRAP_SERVER = "localhost:9092";
  public static final String TEST_TOPIC_NAME = "test-topic";
  public static final int TOTAL_TEST_RECORDS = 10;
  public static final String KWCLUSTERAPIUSER = "kwclusterapiuser";
  public static final String AUTHORIZATION = "Authorization";
  public static final String BEARER_PREFIX = "Bearer ";
  public static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

  @Autowired private MockMvc mvc;

  @Value("${klaw.clusterapi.access.base64.secret}")
  private String clusterAccessSecret;

  private static int totalRecordsProduced = 0;
  private final UtilMethods utilMethods = new UtilMethods();

  @Test
  @Order(1)
  void getTopicContentsWhenSelectedNumberOfOffsetsLessThanTotalRecords() throws Exception {
    produceRecords(TOTAL_TEST_RECORDS);

    int totalOffsets = 5;
    String url =
        createUrl(
            String.valueOf(totalOffsets), Integer.MAX_VALUE, Integer.MAX_VALUE, Integer.MAX_VALUE);

    Map<Integer, String> response = getTopicContentsPerformMockRequest(url);

    getTopicContentsVerifyResponse(response, totalOffsets, totalOffsets, TOTAL_TEST_RECORDS - 1);
  }

  @Test
  @Order(2)
  void getTopicContentsWhenSelectedNumberOfOffsetsMoreThanTotalRecords() throws Exception {
    int totalOffsets = TOTAL_TEST_RECORDS + 10;
    String url =
        createUrl(
            String.valueOf(totalOffsets), Integer.MAX_VALUE, Integer.MAX_VALUE, Integer.MAX_VALUE);

    Map<Integer, String> response = getTopicContentsPerformMockRequest(url);

    getTopicContentsVerifyResponse(response, TOTAL_TEST_RECORDS, 0, TOTAL_TEST_RECORDS - 1);
  }

  @Test
  @Order(3)
  void getTopicContentsWhenSelectedNumberOfOffsetsIsNegative() throws Exception {
    String url =
        createUrl(String.valueOf(-1), Integer.MAX_VALUE, Integer.MAX_VALUE, Integer.MAX_VALUE);

    Map<Integer, String> response = getTopicContentsPerformMockRequest(url);

    assertThat(response).isEmpty();
  }

  @Test
  @Order(4)
  void getTopicContentsWhenCustomAndOffsetsLessThanTotalRecords() throws Exception {
    int totalOffsets = 5;
    String url =
        createUrl(
            TopicContentType.CUSTOM.getValue(), totalOffsets, Integer.MAX_VALUE, Integer.MAX_VALUE);

    Map<Integer, String> response = getTopicContentsPerformMockRequest(url);

    getTopicContentsVerifyResponse(response, totalOffsets, totalOffsets, TOTAL_TEST_RECORDS - 1);
  }

  @Test
  @Order(5)
  void getTopicContentsWhenCustomAndOffsetsMoreThanTotalRecords() throws Exception {
    int totalOffsets = TOTAL_TEST_RECORDS + 10;
    String url =
        createUrl(
            TopicContentType.CUSTOM.getValue(), totalOffsets, Integer.MAX_VALUE, Integer.MAX_VALUE);

    Map<Integer, String> response = getTopicContentsPerformMockRequest(url);

    getTopicContentsVerifyResponse(response, TOTAL_TEST_RECORDS, 0, TOTAL_TEST_RECORDS - 1);
  }

  @Test
  @Order(6)
  void getTopicContentsWhenCustomAndOffsetAndOffsetIsNegative() throws Exception {
    String url =
        createUrl(TopicContentType.CUSTOM.getValue(), -1, Integer.MAX_VALUE, Integer.MAX_VALUE);

    Map<Integer, String> response = getTopicContentsPerformMockRequest(url);

    assertThat(response).isEmpty();
  }

  @Test
  @Order(7)
  void getTopicContentsWhenRangeAndValidBounds() throws Exception {
    int start = 4;
    int end = 7;
    String url = createUrl(TopicContentType.RANGE.getValue(), Integer.MAX_VALUE, start, end);

    Map<Integer, String> response = getTopicContentsPerformMockRequest(url);

    getTopicContentsVerifyResponse(response, 4, start, end);
  }

  @Test
  @Order(8)
  void getTopicContentsWhenRangeAndStartOffsetNegative() throws Exception {
    int end = 7;
    String url = createUrl(TopicContentType.RANGE.getValue(), Integer.MAX_VALUE, -1, end);

    Map<Integer, String> response = getTopicContentsPerformMockRequest(url);

    assertThat(response).isEmpty();
  }

  @Test
  @Order(9)
  void getTopicContentsWhenRangeAndEndOffsetNegative() throws Exception {
    int start = 4;
    String url = createUrl(TopicContentType.RANGE.getValue(), Integer.MAX_VALUE, start, -1);

    Map<Integer, String> response = getTopicContentsPerformMockRequest(url);

    assertThat(response).isEmpty();
  }

  @Test
  @Order(10)
  void getTopicContentsWhenRangeAndStartAndEndOffsetNegative() throws Exception {
    String url = createUrl(TopicContentType.RANGE.getValue(), Integer.MAX_VALUE, -1, -1);

    Map<Integer, String> response = getTopicContentsPerformMockRequest(url);

    assertThat(response).isEmpty();
  }

  @Test
  @Order(11)
  void getTopicContentsWhenRangeAndStartOffsetBiggerThanEndOffset() throws Exception {
    String url = createUrl(TopicContentType.RANGE.getValue(), Integer.MAX_VALUE, 6, 3);

    Map<Integer, String> response = getTopicContentsPerformMockRequest(url);

    assertThat(response).isEmpty();
  }

  @Test
  @Order(12)
  void getTopicContentsWhenRangeAndStartAndEndOffsetEqual() throws Exception {
    int start = 5;
    String url = createUrl(TopicContentType.RANGE.getValue(), Integer.MAX_VALUE, start, start);

    Map<Integer, String> response = getTopicContentsPerformMockRequest(url);

    getTopicContentsVerifyResponse(response, 1, start, start);
  }

  @Test
  @Order(13)
  void getTopicContentsWhenRangeAndTotalOffsetsLargerThanMax() throws Exception {
    produceRecords(200);
    String url = createUrl(TopicContentType.RANGE.getValue(), Integer.MAX_VALUE, 20, 180);

    Map<Integer, String> response = getTopicContentsPerformMockRequest(url);

    getTopicContentsVerifyResponse(response, 100, 20, 119);
  }

  private String createUrl(
      String offsetType, int customNumberOfOffsets, int rangeStart, int rangeEnd) {
    return String.format(
        "/topics/getTopicContents/%s/PLAINTEXT/notdefined/test-topic/%s/partitionId/0/selectedNumberOfOffsets/%d/DEV_CLUSTER1/rangeOffsets/%d/%d",
        BOOTSTRAP_SERVER, offsetType, customNumberOfOffsets, rangeStart, rangeEnd);
  }

  private void produceRecords(int number) {
    Properties properties = new Properties();
    properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    properties.setProperty(
        ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.setProperty(
        ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

    ProducerRecord<String, String> producerRecord;
    for (int x = 0; x < number; x++) {
      producerRecord =
          new ProducerRecord<>("test-topic", String.format("value%d", totalRecordsProduced++));
      kafkaProducer.send(producerRecord);
    }

    kafkaProducer.flush();
    kafkaProducer.close();
  }

  private Map<Integer, String> getTopicContentsPerformMockRequest(String url) throws Exception {
    MockHttpServletResponse response =
        mvc.perform(
                MockMvcRequestBuilders.get(url)
                    .contentType(MediaType.APPLICATION_JSON)
                    .header(
                        AUTHORIZATION,
                        BEARER_PREFIX
                            + utilMethods.generateToken(KWCLUSTERAPIUSER, clusterAccessSecret, 3L))
                    .accept(MediaType.APPLICATION_JSON))
            .andExpect(status().isOk())
            .andReturn()
            .getResponse();

    MapType mapType =
        TypeFactory.defaultInstance().constructMapType(HashMap.class, Integer.class, String.class);
    return OBJECT_MAPPER.readValue(response.getContentAsString(), mapType);
  }

  private void getTopicContentsVerifyResponse(
      Map<Integer, String> response, int responseSize, int offsetIdStart, int offsetIdEnd) {
    assertThat(response.size()).isEqualTo(responseSize);
    for (int x = offsetIdStart; x <= offsetIdEnd; x++) {
      assertThat(response.containsKey(x)).isTrue();
      assertThat(response.get(x)).isEqualTo(String.format("value%d", x));
    }
  }
}
