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

import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEventRepository;
import cloud.nstart.terravault.ordinis.domain.webhook.WebhookDelivery;
import cloud.nstart.terravault.ordinis.domain.webhook.WebhookDeliveryRepository;
import cloud.nstart.terravault.ordinis.domain.webhook.WebhookSubscription;
import cloud.nstart.terravault.ordinis.domain.webhook.WebhookSubscriptionRepository;
import cloud.nstart.terravault.ordinis.outbox.webhook.HmacSigner;
import cloud.nstart.terravault.ordinis.outbox.webhook.WebhookDispatcher;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.sun.net.httpserver.HttpServer;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.context.TestConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Primary;
import org.springframework.http.MediaType;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.Trigger;
import org.springframework.security.oauth2.jwt.Jwt;
import org.springframework.security.oauth2.server.resource.authentication.JwtAuthenticationToken;
import org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors;
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.io.IOException;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.time.Instant;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

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

/**
 * End-to-end webhook delivery test:
 * <ol>
 *   <li>Стартуем embedded {@link HttpServer} на random localhost port —
 *       imitate consumer's webhook receiver.</li>
 *   <li>Создаём WebhookSubscription указывающий на этот endpoint.</li>
 *   <li>POST a record — попадает в outbox_events.</li>
 *   <li>Тест вручную вызывает {@code dispatcher.matchTick()} и
 *       {@code dispatcher.sendTick()} — никакого фонового scheduler'а.</li>
 *   <li>Проверяем: receiver получил body + valid HMAC signature +
 *       Ordinis headers; delivery row mark'нут как delivered.</li>
 * </ol>
 *
 * <p><b>Почему no-op scheduler:</b> ранее тест полагался на @Scheduled tick'и
 * с интервалом 200ms. На CI postgres testcontainer с reuse=true тащил leftover
 * subscriptions с dead URLs из предыдущих runs — scheduler в фоне забивал
 * thread pool retry'ями ДО того как наш @BeforeEach успевал truncate. Result:
 * matchTick на наш event не доходил, ConditionTimeout 60s. Три итерации фиксов
 * (cleanup leftovers, pool=4, send-timeout=1s) лечили симптомы, не причину.
 *
 * <p>Финальный fix: {@link NoSchedulingConfig} подменяет {@link TaskScheduler}
 * на no-op. @Scheduled методы регистрируются, но никогда не запускаются.
 * Тест вручную дёргает matchTick/sendTick — synchronous, deterministic. Нет
 * race с background threads, нет dependency на 200ms timer'ы.
 */
@SpringBootTest(
    webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
    properties = {
        // Outbox poller off — он единственный требует KafkaTemplate. Webhook
        // dispatcher читает outbox_events напрямую, Kafka не нужна.
        "ordinis.outbox.enabled=false",
        "ordinis.webhook.enabled=true",
        // Send timeout оставляем агрессивным на случай если manual sendTick
        // встретит leftover dead URL (хоть и cleanup в @BeforeEach должен убрать).
        "ordinis.webhook.send-timeout-ms=2000",
        "ordinis.webhook.allowed-hosts=localhost,127.0.0.1",
        // Spring auto-config Kafka не нужно — нет KafkaTemplate потребителей
        // без OutboxPoller. Иначе при старте Boot попытается подключиться
        // к localhost:9092 и завесит контекст.
        "spring.autoconfigure.exclude="
            + "org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration",
    })
@AutoConfigureMockMvc
@ActiveProfiles("test")
class WebhookE2ETest {

  @DynamicPropertySource
  static void props(DynamicPropertyRegistry registry) {
    // Только Postgres — без Kafka container. Экономит ~5-10s cold start
    // и убирает класс flakiness.
    E2ESupport.registerPostgresOnlyProperties(registry);
  }

  /**
   * Подменяет TaskScheduler на no-op. Spring ScheduledAnnotationBeanPostProcessor
   * вызовет {@code scheduleWithFixedDelay} для matchTick/sendTick — наш stub
   * вернёт cancelled future, ничего не запустит.
   *
   * <p>marked @Primary чтобы перебить дефолтный Spring Boot bean.
   */
  @TestConfiguration
  static class NoSchedulingConfig {
    @Bean
    @Primary
    TaskScheduler noopTaskScheduler() {
      return new NoopTaskScheduler();
    }
  }

  @Autowired MockMvc mvc;
  @Autowired ObjectMapper om;
  @Autowired OutboxEventRepository outboxRepo;
  @Autowired WebhookSubscriptionRepository subRepo;
  @Autowired WebhookDeliveryRepository deliveryRepo;
  @Autowired WebhookDispatcher dispatcher;

  private HttpServer receiver;
  private final ConcurrentLinkedQueue<ReceivedRequest> received = new ConcurrentLinkedQueue<>();

  record ReceivedRequest(String path, String body, String signatureHeader, String eventTypeHeader, String scopeHeader) {}

  @BeforeEach
  void startReceiver() throws IOException {
    // Cleanup leftover state из предыдущего CI run.
    // Postgres testcontainer withReuse(true) сохраняет данные между запусками.
    // Order: deliveries (FK), subscriptions, outbox events.
    deliveryRepo.deleteAllInBatch();
    subRepo.deleteAllInBatch();
    outboxRepo.deleteAllInBatch();

    receiver = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
    receiver.createContext("/hook", exchange -> {
      byte[] body = exchange.getRequestBody().readAllBytes();
      received.add(new ReceivedRequest(
          exchange.getRequestURI().getPath(),
          new String(body, StandardCharsets.UTF_8),
          exchange.getRequestHeaders().getFirst(HmacSigner.HEADER_NAME),
          exchange.getRequestHeaders().getFirst("X-Ordinis-Event-Type"),
          exchange.getRequestHeaders().getFirst("X-Ordinis-Scope")));
      exchange.sendResponseHeaders(200, -1);
      exchange.close();
    });
    receiver.start();
  }

  @AfterEach
  void stopReceiver() {
    if (receiver != null) receiver.stop(0);
  }

  @Test
  void recordCreate_dispatchedToWebhookReceiver_withValidHmac() throws Exception {
    int port = receiver.getAddress().getPort();
    String hookUrl = "http://127.0.0.1:" + port + "/hook";

    // 1. Создаём dict
    String dictName = "wh_" + UUID.randomUUID().toString().replace("-", "").substring(0, 8);
    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());

    // 2. Создаём webhook subscription напрямую через repository (CRUD endpoint
    //    требует RESTRICTED scope от JWT, в тесте без auth — проще через repo)
    String secret = "test-secret-" + UUID.randomUUID();
    WebhookSubscription sub = new WebhookSubscription(
        "e2e-test-receiver", hookUrl, secret, "test-user");
    sub = subRepo.save(sub);
    UUID subId = sub.getId();

    // 3. POST record → outbox_events row created
    String needle = "EVT_" + UUID.randomUUID().toString().replace("-", "").substring(0, 12).toUpperCase();
    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());

    // 4. Manually drive dispatcher — никакого scheduler'а в тесте.
    //    matchTick: outbox_events × subscriptions → INSERT delivery rows.
    //    sendTick: due deliveries → POST на receiver.
    dispatcher.matchTick();
    dispatcher.sendTick();

    // 5. Wait for receiver: HTTP server async, network round-trip <100ms на localhost.
    await().atMost(5, TimeUnit.SECONDS)
        .pollInterval(Duration.ofMillis(50))
        .until(() -> received.stream().anyMatch(r -> r.body().contains(needle)));

    // 6. Verify request properties
    ReceivedRequest req = received.stream()
        .filter(r -> r.body().contains(needle))
        .findFirst()
        .orElseThrow();

    assertThat(req.path()).isEqualTo("/hook");
    assertThat(req.scopeHeader()).isEqualTo("PUBLIC");
    assertThat(req.eventTypeHeader()).isNotBlank();

    // HMAC signature валиден: receiver мог бы verify через свой secret
    String expectedSig = HmacSigner.sign(secret, req.body().getBytes(StandardCharsets.UTF_8));
    assertThat(req.signatureHeader()).isEqualTo(expectedSig);

    // 7. Verify DB delivery row marked delivered. sendTick уже запустился —
    // если success path, status=delivered должен быть синхронно. Маленький
    // poll на случай если sendTick ставит status в отдельной транзакции.
    await().atMost(2, TimeUnit.SECONDS)
        .pollInterval(Duration.ofMillis(50))
        .until(() -> deliveryRepo.findBySubscriptionIdOrderByCreatedAtDesc(
                subId, org.springframework.data.domain.PageRequest.of(0, 10))
            .getContent().stream()
            .anyMatch(d -> d.getStatus() == WebhookDelivery.Status.delivered));

    List<WebhookDelivery> all = deliveryRepo.findBySubscriptionIdOrderByCreatedAtDesc(
        subId, org.springframework.data.domain.PageRequest.of(0, 10)).getContent();
    assertThat(all).isNotEmpty();
    WebhookDelivery delivered = all.stream()
        .filter(d -> d.getStatus() == WebhookDelivery.Status.delivered)
        .findFirst()
        .orElseThrow();
    assertThat(delivered.getLastStatusCode()).isEqualTo(200);
    assertThat(delivered.getDeliveredAt()).isNotNull();
    assertThat(delivered.getLastError()).isNull();
  }

  @Test
  void receiverReturning500_deliveryRetried_notDelivered() throws Exception {
    HttpServer failing = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
    failing.createContext("/fail", exchange -> {
      try {
        exchange.getRequestBody().readAllBytes();
      } catch (Exception ignored) {
      }
      exchange.sendResponseHeaders(500, -1);
      exchange.close();
    });
    failing.start();
    try {
      int port = failing.getAddress().getPort();
      String hookUrl = "http://127.0.0.1:" + port + "/fail";

      String dictName = "whf_" + UUID.randomUUID().toString().replace("-", "").substring(0, 8);
      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());

      WebhookSubscription sub = subRepo.save(new WebhookSubscription(
          "e2e-failing", hookUrl, "secret-fail", "test-user"));

      String needle = "EVT_" + UUID.randomUUID().toString().replace("-", "").substring(0, 12);
      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());

      // Manual ticks — no scheduler.
      dispatcher.matchTick();
      dispatcher.sendTick();

      // sendTick должен mark delivery как retrying (HTTP 500). Маленький poll
      // на случай задержки status update в отдельной транзакции.
      await().atMost(5, TimeUnit.SECONDS)
          .pollInterval(Duration.ofMillis(50))
          .until(() -> deliveryRepo.findBySubscriptionIdOrderByCreatedAtDesc(
                  sub.getId(), org.springframework.data.domain.PageRequest.of(0, 10))
              .getContent().stream()
              .anyMatch(d -> d.getStatus() == WebhookDelivery.Status.retrying
                  && d.getLastStatusCode() != null && d.getLastStatusCode() == 500));
    } finally {
      failing.stop(0);
    }
  }

  /**
   * Mock JWT с INTERNAL role для admin endpoints. spring-security-test
   * authentication() post-processor правильно проходит через filter chain
   * (filter не overwrite'ит auth когда передан таким образом).
   */
  private static org.springframework.test.web.servlet.request.RequestPostProcessor internalJwt() {
    Jwt jwt = Jwt.withTokenValue("test-token")
        .header("alg", "none")
        .claim("sub", "test-user")
        .claim("realm_access", java.util.Map.of("roles", java.util.List.of("ordinis-internal")))
        .build();
    return SecurityMockMvcRequestPostProcessors.authentication(
        new JwtAuthenticationToken(jwt, java.util.Collections.emptyList()));
  }

  @Test
  void statsEndpoint_returnsContinuousTimeline_withCorrectCounts() throws Exception {
    WebhookSubscription sub = subRepo.save(new WebhookSubscription(
        "stats-test", "http://127.0.0.1:1/null", "stats-secret", "test-user"));
    UUID subId = sub.getId();

    Long eventId1 = insertMinimalOutboxEvent("EVT-stats-1");
    Long eventId2 = insertMinimalOutboxEvent("EVT-stats-2");
    Long eventId3 = insertMinimalOutboxEvent("EVT-stats-3");

    WebhookDelivery del1 = deliveryRepo.save(new WebhookDelivery(subId, eventId1));
    del1.markDelivered(200);
    deliveryRepo.save(del1);

    WebhookDelivery del2 = deliveryRepo.save(new WebhookDelivery(subId, eventId2));
    del2.markFailed("connection refused", null);
    deliveryRepo.save(del2);

    deliveryRepo.save(new WebhookDelivery(subId, eventId3));

    mvc.perform(get("/api/v1/admin/webhooks/subscriptions/{id}/stats", subId)
            .with(internalJwt())
            .param("bucketBy", "hour"))
        .andExpect(status().isOk())
        .andExpect(jsonPath("$.bucketBy").value("hour"))
        .andExpect(jsonPath("$.buckets").isArray())
        .andExpect(jsonPath("$.buckets[?(@.total > 0)].pending").value(org.hamcrest.Matchers.hasItem(1)))
        .andExpect(jsonPath("$.buckets[?(@.total > 0)].retrying").value(org.hamcrest.Matchers.hasItem(1)))
        .andExpect(jsonPath("$.buckets[?(@.total > 0)].delivered").value(org.hamcrest.Matchers.hasItem(1)));
  }

  @Test
  void statsEndpoint_rejectsInvalidBucketBy() throws Exception {
    WebhookSubscription sub = subRepo.save(new WebhookSubscription(
        "stats-bad", "http://127.0.0.1:1/null", "secret-bad", "test-user"));
    mvc.perform(get("/api/v1/admin/webhooks/subscriptions/{id}/stats", sub.getId())
            .with(internalJwt())
            .param("bucketBy", "year"))
        .andExpect(status().isBadRequest())
        .andExpect(jsonPath("$.code").value("webhook_stats_bucket_invalid"));
  }

  @Test
  void statsEndpoint_rejectsWindowTooLarge() throws Exception {
    WebhookSubscription sub = subRepo.save(new WebhookSubscription(
        "stats-big", "http://127.0.0.1:1/null", "secret-big", "test-user"));
    String fromIso = java.time.OffsetDateTime.now().minusDays(60).toString();
    String toIso = java.time.OffsetDateTime.now().toString();
    mvc.perform(get("/api/v1/admin/webhooks/subscriptions/{id}/stats", sub.getId())
            .with(internalJwt())
            .param("bucketBy", "hour")
            .param("from", fromIso)
            .param("to", toIso))
        .andExpect(status().isBadRequest())
        .andExpect(jsonPath("$.code").value("webhook_stats_window_too_large"));
  }

  @Autowired org.springframework.jdbc.core.JdbcTemplate jdbcTemplate;

  /**
   * Минимальный outbox_event insert для удовлетворения FK constraint
   * webhook_deliveries.outbox_event_id. NOT NULL columns set к dummy values.
   */
  private Long insertMinimalOutboxEvent(String key) {
    return jdbcTemplate.queryForObject(
        """
        INSERT INTO outbox_events
          (event_type, aggregate_type, aggregate_id, data_scope,
           payload, kafka_topic, kafka_key, created_at, attempt_count)
        VALUES
          ('TestEvent', 'Test', ?, 'PUBLIC',
           '{}'::jsonb, 'test-topic', ?, now(), 0)
        RETURNING id
        """,
        Long.class,
        key, key);
  }

  // ---------- no-op scheduler implementation ----------

  /**
   * No-op {@link TaskScheduler}: {@code schedule*} методы возвращают cancelled
   * {@link ScheduledFuture}. @Scheduled методы регистрируются Spring'ом, но
   * наш stub их не запускает.
   */
  private static final class NoopTaskScheduler implements TaskScheduler {
    @Override public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) { return cancelled(); }
    @Override public ScheduledFuture<?> schedule(Runnable task, Instant startTime) { return cancelled(); }
    @Override public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Instant startTime, Duration period) { return cancelled(); }
    @Override public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Duration period) { return cancelled(); }
    @Override public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Instant startTime, Duration delay) { return cancelled(); }
    @Override public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Duration delay) { return cancelled(); }

    private static ScheduledFuture<?> cancelled() {
      return new ScheduledFuture<>() {
        @Override public long getDelay(TimeUnit unit) { return 0; }
        @Override public int compareTo(Delayed o) { return 0; }
        @Override public boolean cancel(boolean mayInterruptIfRunning) { return true; }
        @Override public boolean isCancelled() { return true; }
        @Override public boolean isDone() { return true; }
        @Override public Object get() { return null; }
        @Override public Object get(long timeout, TimeUnit unit) throws TimeoutException { throw new TimeoutException(); }
      };
    }
  }
}
