package cloud.nstart.terravault.ordinis.projection;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

/**
 * Подписан на 3 scope-топика. На RecordCreated/Updated делает upsert
 * projection, на Deleted — delete. На Definition events ничего не делает
 * (projection writer не материализует definitions).
 *
 * <p>Группа consumer'а: {@code ordinis-projection-cuod}. ACL Strimzi даёт
 * этому пользователю Read доступ ко всем 3 scope-топикам и Read к группе
 * {@code ordinis-projection-*} (см. ordinis-infra/k8s/kafka/topology.yaml).
 */
@Component
public class RecordEventListener {

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

  private final ProjectionWriter writer;
  private final ObjectMapper objectMapper;
  private final Counter eventsCounter;
  private final Counter errorsCounter;

  public RecordEventListener(
      ProjectionWriter writer, ObjectMapper objectMapper, MeterRegistry meterRegistry) {
    this.writer = writer;
    this.objectMapper = objectMapper;
    this.eventsCounter = Counter.builder("ordinis_projection_events_processed_total")
        .description("Kafka events обработанные projection writer'ом")
        .register(meterRegistry);
    this.errorsCounter = Counter.builder("ordinis_projection_events_errors_total")
        .description("Ошибки обработки event'a в projection writer'е")
        .register(meterRegistry);
  }

  @KafkaListener(
      topics = {
          "ordinis.cuod.events.public",
          "ordinis.cuod.events.internal",
          "ordinis.cuod.events.restricted"
      },
      groupId = "ordinis-projection-cuod",
      containerFactory = "kafkaListenerContainerFactory"
  )
  public void onMessage(
      @Payload String value,
      @Header(name = KafkaHeaders.RECEIVED_KEY, required = false) String key) {
    try {
      JsonNode envelope = objectMapper.readTree(value);
      String dictionary = envelope.path("dictionaryName").asText(null);
      String bundle = envelope.path("bundle").asText("cuod");
      JsonNode payload = envelope.path("payload");
      String type = payload.path("type").asText(null);

      if (type == null || dictionary == null) {
        log.debug("Skip event (no type or dictionary): key={}", key);
        return;
      }

      String businessKey = payload.path("businessKey").asText(null);
      switch (type) {
        case "RecordCreated", "RecordUpdated" -> {
          if (businessKey == null) return;
          JsonNode data = payload.path("data");
          writer.upsert(bundle, dictionary, businessKey, data);
        }
        case "RecordDeleted" -> {
          if (businessKey == null) return;
          writer.delete(bundle, dictionary, businessKey);
        }
        default -> log.debug("Unhandled event type {} for {}", type, dictionary);
      }
      eventsCounter.increment();
    } catch (Exception e) {
      errorsCounter.increment();
      log.error("Failed to process projection event key={}: {}", key, e.getMessage(), e);
    }
  }
}
