package cloud.nstart.terravault.ordinis.outbox;

import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEvent;
import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEventRepository;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.domain.PageRequest;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @Scheduled poller для outbox. Sweep каждые {@code ordinis.outbox.poll-interval-ms},
 * batch size {@code ordinis.outbox.batch-size}. Активируется только если
 * {@code ordinis.outbox.enabled=true} (отключаем в read-api / projection-writer).
 *
 * <p>После {@code ordinis.outbox.attempt-max} (default 1000) неудачных попыток
 * событие маркируется {@code dlq_at} и исключается из дальнейшего polling. Без
 * cap poller бесконечно ретраит мёртвые сообщения (orphan topic, ACL deny),
 * забивая лог и lag-метрику.
 */
@Component
@ConditionalOnProperty(name = "ordinis.outbox.enabled", havingValue = "true", matchIfMissing = false)
public class OutboxPoller {

  private static final Logger log = LoggerFactory.getLogger(OutboxPoller.class);

  private final OutboxEventRepository repository;
  private final KafkaTemplate<String, String> kafkaTemplate;
  private final com.fasterxml.jackson.databind.ObjectMapper objectMapper;
  private final Counter publishedCounter;
  private final Counter failedCounter;
  private final Counter dlqCounter;
  private final Timer publishTimer;

  @Value("${ordinis.outbox.batch-size:100}")
  private int batchSize;

  @Value("${ordinis.outbox.send-timeout-ms:10000}")
  private long sendTimeoutMs;

  @Value("${ordinis.outbox.attempt-max:1000}")
  private int attemptMax;

  public OutboxPoller(
      OutboxEventRepository repository,
      KafkaTemplate<String, String> kafkaTemplate,
      com.fasterxml.jackson.databind.ObjectMapper objectMapper,
      MeterRegistry meterRegistry) {
    this.repository = repository;
    this.kafkaTemplate = kafkaTemplate;
    this.objectMapper = objectMapper;
    this.publishedCounter = Counter.builder("ordinis_outbox_published_total")
        .description("Количество успешно опубликованных outbox events")
        .register(meterRegistry);
    this.failedCounter = Counter.builder("ordinis_outbox_publish_errors_total")
        .description("Количество ошибок публикации в Kafka")
        .register(meterRegistry);
    this.dlqCounter = Counter.builder("ordinis_outbox_dlq_total")
        .description("Количество событий перенесённых в DLQ (attempt cap)")
        .register(meterRegistry);
    this.publishTimer = Timer.builder("ordinis_kafka_publish_duration_seconds")
        .description("Время публикации одного события в Kafka")
        .register(meterRegistry);
  }

  @Scheduled(fixedDelayString = "${ordinis.outbox.poll-interval-ms:1000}")
  @Transactional
  public void poll() {
    List<OutboxEvent> batch = repository.findUnpublished(PageRequest.of(0, batchSize));
    if (batch.isEmpty()) return;

    log.debug("Outbox poll: {} unpublished events", batch.size());

    for (OutboxEvent event : batch) {
      Timer.Sample sample = Timer.start();
      try {
        String value = objectMapper.writeValueAsString(event.getPayload());
        SendResult<String, String> result = kafkaTemplate
            .send(event.getKafkaTopic(), event.getKafkaKey(), value)
            .get(sendTimeoutMs, TimeUnit.MILLISECONDS);

        event.markPublished();
        repository.save(event);
        publishedCounter.increment();
        sample.stop(publishTimer);

        log.debug("Published outbox event id={} topic={} partition={} offset={}",
            event.getId(),
            event.getKafkaTopic(),
            result.getRecordMetadata().partition(),
            result.getRecordMetadata().offset());
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        markFailed(event, "Interrupted: " + e.getMessage());
        return;
      } catch (ExecutionException | TimeoutException e) {
        markFailed(event, e.getClass().getSimpleName() + ": " + e.getMessage());
      } catch (Exception e) {
        markFailed(event, e.getClass().getSimpleName() + ": " + e.getMessage());
      }
    }
  }

  private void markFailed(OutboxEvent event, String error) {
    event.markFailed(error);
    failedCounter.increment();

    if (event.getAttemptCount() != null && event.getAttemptCount() >= attemptMax) {
      event.markDlq(error);
      dlqCounter.increment();
      log.error(
          "Outbox event id={} topic={} → DLQ after {} attempts. Last error: {}",
          event.getId(), event.getKafkaTopic(), event.getAttemptCount(), error);
    } else {
      log.warn(
          "Failed to publish outbox event id={} topic={} attempt={}/{}: {}",
          event.getId(), event.getKafkaTopic(),
          event.getAttemptCount(), attemptMax, error);
    }
    repository.save(event);
  }
}
