package cloud.nstart.terravault.ordinis.app.e2e;

import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEventRepository;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.junit.jupiter.api.Test;
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.test.context.ActiveProfiles;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.springframework.test.web.servlet.MockMvc;

import java.time.Duration;
import java.util.List;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;

/**
 * Этап 2b: end-to-end проверка outbox publishing.
 *
 * <p>POST record → outbox event записан в БД → OutboxPoller подхватывает →
 * KafkaTemplate публикует → Kafka consumer (этого теста) получает.
 *
 * <p>Включаем outbox в этом тесте через {@code ORDINIS_OUTBOX_ENABLED=true}
 * (общий test-profile его выключает чтобы быстрее).
 */
@SpringBootTest(
    webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
    properties = {
        "ordinis.outbox.enabled=true",
        "ordinis.outbox.poll-interval-ms=200",
        "ordinis.outbox.batch-size=20",
    })
@AutoConfigureMockMvc
@ActiveProfiles("test")
class OutboxKafkaE2ETest {

  @DynamicPropertySource
  static void props(DynamicPropertyRegistry registry) {
    E2ESupport.registerProperties(registry);
  }

  @Autowired MockMvc mvc;
  @Autowired ObjectMapper om;
  @Autowired OutboxEventRepository outboxRepo;

  @Value("${spring.kafka.bootstrap-servers}")
  String bootstrapServers;

  @Test
  void postRecord_outboxEventPublishedToKafka() throws Exception {
    String dictName = "out_" + UUID.randomUUID().toString().replace("-", "").substring(0, 8);
    long pendingBefore = outboxRepo.countPending();

    // Создать dictionary (PUBLIC scope → topic ordinis.test.events.public)
    JsonNode schema = om.readTree("""
        {
          "$schema": "http://json-schema.org/draft-07/schema#",
          "type": "object",
          "additionalProperties": false,
          "x-id-source": "code",
          "required": ["code"],
          "properties": { "code": { "type": "string", "x-unique": true } }
        }
        """);
    var dictBody = om.createObjectNode()
        .put("name", dictName)
        .put("scope", "PUBLIC")
        .put("schemaVersion", "1.0.0")
        .put("bundle", "test");
    dictBody.set("schemaJson", schema);
    mvc.perform(post("/api/v1/dictionaries")
            .contentType(MediaType.APPLICATION_JSON)
            .content(om.writeValueAsBytes(dictBody)))
        .andExpect(status().is2xxSuccessful());

    // Pre-create topic чтобы не было race с auto-create при subscribe.
    String topic = "ordinis.test.events.public";
    Properties admin = new Properties();
    admin.put("bootstrap.servers", bootstrapServers);
    try (AdminClient ac = AdminClient.create(admin)) {
      ac.createTopics(List.of(new NewTopic(topic, 1, (short) 1))).all().get(5, TimeUnit.SECONDS);
    } catch (Exception ignored) {
      // Если уже создан предыдущим прогоном — ОК
    }

    // Уникальный consumer-group + auto.offset.reset=earliest → consumer
    // получит все сообщения с начала топика. needleInValue = uuid → найдём
    // именно нашу запись среди возможных от прошлых тестовых прогонов.
    String needle = "EVT_" + UUID.randomUUID().toString().replace("-", "").substring(0, 12).toUpperCase();

    // POST record
    var recBody = om.createObjectNode().put("businessKey", needle);
    recBody.set("data", om.createObjectNode().put("code", needle));
    mvc.perform(post("/api/v1/dictionaries/{n}/records", dictName)
            .contentType(MediaType.APPLICATION_JSON)
            .content(om.writeValueAsBytes(recBody)))
        .andExpect(status().is2xxSuccessful());

    // Дождаться что outbox опустеет (poller подхватил)
    await().atMost(15, TimeUnit.SECONDS)
        .pollInterval(Duration.ofMillis(200))
        .until(() -> outboxRepo.countPending() <= pendingBefore);

    try (KafkaConsumer<String, String> consumer = newConsumer()) {
      consumer.subscribe(List.of(topic));
      // Получить event из Kafka — earliest reset вытащит все сообщения с начала.
      ConsumerRecord<String, String> match = pollUntil(consumer, needle, 15);
      assertThat(match).as("Kafka should receive the event for " + needle).isNotNull();
      assertThat(match.topic()).isEqualTo(topic);

      JsonNode envelope = om.readTree(match.value());
      assertThat(envelope.get("dictionaryName").asText()).isEqualTo(dictName);
      assertThat(envelope.get("dataScope").asText()).isEqualTo("PUBLIC");
      assertThat(envelope.get("bundle").asText()).isEqualTo("test");
      assertThat(envelope.has("payload")).isTrue();
      assertThat(envelope.get("payload").get("businessKey").asText()).isEqualTo(needle);
    }
  }

  private KafkaConsumer<String, String> newConsumer() {
    Properties p = new Properties();
    p.put("bootstrap.servers", bootstrapServers);
    p.put("group.id", "e2e-test-" + UUID.randomUUID());
    p.put("auto.offset.reset", "earliest");
    p.put("key.deserializer", StringDeserializer.class.getName());
    p.put("value.deserializer", StringDeserializer.class.getName());
    return new KafkaConsumer<>(p);
  }

  private ConsumerRecord<String, String> pollUntil(
      KafkaConsumer<String, String> consumer, String needleInValue, int seconds) {
    long deadline = System.currentTimeMillis() + (seconds * 1000L);
    while (System.currentTimeMillis() < deadline) {
      ConsumerRecords<String, String> recs = consumer.poll(Duration.ofMillis(500));
      for (ConsumerRecord<String, String> r : recs) {
        if (r.value() != null && r.value().contains(needleInValue)) return r;
      }
    }
    return null;
  }
}
