package cloud.nstart.terravault.ordinis.outbox;

import cloud.nstart.terravault.ordinis.domain.DataScope;
import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEvent;
import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEventRepository;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.TopicPartition;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.data.domain.Pageable;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.test.util.ReflectionTestUtils;

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeoutException;

import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

class OutboxPollerTest {

  private OutboxEventRepository repo;
  private KafkaTemplate<String, String> kafka;
  private OutboxPoller poller;

  @BeforeEach
  void setUp() {
    repo = mock(OutboxEventRepository.class);
    kafka = mock(KafkaTemplate.class);
    poller = new OutboxPoller(repo, kafka, new ObjectMapper(), new SimpleMeterRegistry());
    ReflectionTestUtils.setField(poller, "batchSize", 100);
    ReflectionTestUtils.setField(poller, "sendTimeoutMs", 1_000L);
    ReflectionTestUtils.setField(poller, "attemptMax", 3);
  }

  private OutboxEvent newEvent() {
    return new OutboxEvent(
        "DICT_RECORD_CREATED",
        "DictionaryRecord",
        "spacecraft:RESURS-P-3",
        DataScope.PUBLIC,
        JsonNodeFactory.instance.objectNode().put("foo", "bar"),
        "ordinis.cuod.events.public",
        "spacecraft:RESURS-P-3");
  }

  @Test
  void empty_batch_does_nothing() {
    when(repo.findUnpublished(any(Pageable.class))).thenReturn(List.of());

    poller.poll();

    verify(kafka, never()).send(anyString(), anyString(), anyString());
    verify(repo, never()).save(any());
  }

  @Test
  void successful_publish_marks_published() {
    OutboxEvent event = newEvent();
    when(repo.findUnpublished(any(Pageable.class))).thenReturn(List.of(event));

    SendResult<String, String> result = mock(SendResult.class);
    when(result.getRecordMetadata()).thenReturn(
        new RecordMetadata(new TopicPartition("ordinis.cuod.events.public", 0), 0L, 0, 0L, 0, 0));
    when(kafka.send(eq("ordinis.cuod.events.public"), anyString(), anyString()))
        .thenReturn(CompletableFuture.completedFuture(result));

    poller.poll();

    assertThat(event.getPublishedAt()).isNotNull();
    assertThat(event.getDlqAt()).isNull();
    verify(repo).save(event);
  }

  @Test
  void failure_below_cap_marks_failed_not_dlq() {
    OutboxEvent event = newEvent();
    when(repo.findUnpublished(any(Pageable.class))).thenReturn(List.of(event));

    CompletableFuture<SendResult<String, String>> failed = new CompletableFuture<>();
    failed.completeExceptionally(new TimeoutException("kafka down"));
    when(kafka.send(anyString(), anyString(), anyString())).thenReturn(failed);

    poller.poll();

    assertThat(event.getAttemptCount()).isEqualTo(1);
    assertThat(event.getDlqAt()).isNull();
    assertThat(event.getLastError()).contains("kafka down");
    verify(repo).save(event);
  }

  @Test
  void failure_at_cap_marks_dlq() {
    OutboxEvent event = newEvent();
    // Симулируем что 2 предыдущих попытки уже были.
    event.markFailed("prev 1");
    event.markFailed("prev 2");
    assertThat(event.getAttemptCount()).isEqualTo(2);

    when(repo.findUnpublished(any(Pageable.class))).thenReturn(List.of(event));

    CompletableFuture<SendResult<String, String>> failed = new CompletableFuture<>();
    failed.completeExceptionally(new TimeoutException("still down"));
    when(kafka.send(anyString(), anyString(), anyString())).thenReturn(failed);

    poller.poll();

    assertThat(event.getAttemptCount()).isEqualTo(3);
    assertThat(event.getDlqAt()).isNotNull();
    assertThat(event.isDlq()).isTrue();
    assertThat(event.getLastError()).contains("still down");
    verify(repo).save(event);
  }

  @Test
  void failure_above_cap_stays_dlq_not_re_marked() {
    OutboxEvent event = newEvent();
    for (int i = 0; i < 3; i++) event.markFailed("err " + i);
    event.markDlq("first dlq");
    java.time.OffsetDateTime firstDlqAt = event.getDlqAt();

    when(repo.findUnpublished(any(Pageable.class))).thenReturn(List.of());

    poller.poll();

    // Poller не должен подобрать DLQ событие — findUnpublished фильтрует dlq_at IS NULL.
    assertThat(event.getDlqAt()).isEqualTo(firstDlqAt);
  }

  @Test
  void mixed_batch_publishes_some_dlqs_others() {
    OutboxEvent ok = newEvent();
    OutboxEvent doomed = newEvent();
    doomed.markFailed("p1");
    doomed.markFailed("p2");

    when(repo.findUnpublished(any(Pageable.class))).thenReturn(List.of(ok, doomed));

    SendResult<String, String> result = mock(SendResult.class);
    when(result.getRecordMetadata()).thenReturn(
        new RecordMetadata(new TopicPartition("ordinis.cuod.events.public", 0), 0L, 0, 0L, 0, 0));
    CompletableFuture<SendResult<String, String>> okFut = CompletableFuture.completedFuture(result);
    CompletableFuture<SendResult<String, String>> badFut = new CompletableFuture<>();
    badFut.completeExceptionally(new TimeoutException("doomed"));

    // Первый event получает success, второй — failure.
    when(kafka.send(anyString(), anyString(), anyString()))
        .thenReturn(okFut)
        .thenReturn(badFut);

    poller.poll();

    assertThat(ok.getPublishedAt()).isNotNull();
    assertThat(ok.getDlqAt()).isNull();
    assertThat(doomed.getDlqAt()).isNotNull();
    verify(repo, times(2)).save(any());
  }
}
