package cloud.nstart.terravault.ordinis.restapi.service.reference;

import cloud.nstart.terravault.ordinis.domain.definition.DictionaryDefinition;
import cloud.nstart.terravault.ordinis.domain.definition.DictionaryDefinitionRepository;
import cloud.nstart.terravault.ordinis.domain.record.OrphanReferenceQuery;
import com.fasterxml.jackson.databind.JsonNode;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Tags;
import jakarta.annotation.PostConstruct;
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.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @Scheduled сканер orphan FK references. Раз в
 * {@code ordinis.references.scan-interval-sec} (default 3600 = 1h)
 * проходит по всем dictionary schemas, находит x-references markers
 * и считает orphan'ы через {@link OrphanReferenceQuery}.
 *
 * <p>Метрика {@code nsi_orphan_references_total} — Gauge per
 * (source_dict, source_field, target_dict, target_field). Растёт когда
 * referenced record закрывается (close → valid_to=now), а referencing
 * остаётся → dangling FK.
 *
 * <p>Активируется при {@code ordinis.references.scan-enabled=true}
 * (default false; включаем явно на staging/prod). Read-only сканер,
 * нагрузка на replica.
 *
 * <p>Будет триггерить алерт {@code OrdinisOrphanReferences} в
 * prometheus-rules.yaml.
 */
@Component
@ConditionalOnProperty(name = "ordinis.references.scan-enabled", havingValue = "true", matchIfMissing = false)
public class OrphanReferenceScanner {

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

  private final DictionaryDefinitionRepository definitionRepository;
  private final OrphanReferenceQuery orphanQuery;
  private final MeterRegistry meterRegistry;

  /** State: orphan count per (sourceDict, sourceField, targetDict, targetField). */
  private final Map<MetricKey, AtomicLong> orphanCounts = new ConcurrentHashMap<>();

  @Value("${ordinis.references.scan-batch-warn-threshold:100}")
  private long warnThreshold;

  public OrphanReferenceScanner(
      DictionaryDefinitionRepository definitionRepository,
      OrphanReferenceQuery orphanQuery,
      MeterRegistry meterRegistry) {
    this.definitionRepository = definitionRepository;
    this.orphanQuery = orphanQuery;
    this.meterRegistry = meterRegistry;
  }

  @PostConstruct
  void init() {
    log.info("OrphanReferenceScanner initialized (scan-enabled=true). " +
        "Initial sweep on startup, потом периодически.");
  }

  @Scheduled(
      initialDelayString = "${ordinis.references.scan-initial-delay-sec:60}000",
      fixedDelayString = "${ordinis.references.scan-interval-sec:3600}000")
  @Transactional(readOnly = true)
  public void sweep() {
    long start = System.currentTimeMillis();
    int pairsScanned = 0;
    long totalOrphans = 0;

    List<DictionaryDefinition> all = definitionRepository.findAll();
    Map<String, DictionaryDefinition> byName = new java.util.HashMap<>();
    for (DictionaryDefinition d : all) byName.put(d.getName(), d);

    for (DictionaryDefinition source : all) {
      JsonNode schema = source.getSchemaJson();
      if (schema == null || !schema.isObject()) continue;
      JsonNode properties = schema.get("properties");
      if (properties == null || !properties.isObject()) continue;

      Iterator<Map.Entry<String, JsonNode>> it = properties.fields();
      while (it.hasNext()) {
        Map.Entry<String, JsonNode> entry = it.next();
        JsonNode refNode = entry.getValue().get("x-references");
        if (refNode == null || !refNode.isTextual()) continue;

        ParsedRefSpec parsed;
        try {
          parsed = parseRef(refNode.asText());
        } catch (IllegalArgumentException e) {
          log.warn("Schema {} field {} has malformed x-references: {}",
              source.getName(), entry.getKey(), e.getMessage());
          continue;
        }

        DictionaryDefinition target = byName.get(parsed.dictName());
        if (target == null) {
          log.warn("Schema {} field {} references missing dict '{}'",
              source.getName(), entry.getKey(), parsed.dictName());
          continue;
        }

        try {
          long orphans = orphanQuery.countOrphans(
              source.getId(), entry.getKey(),
              target.getId(), parsed.fieldName());

          MetricKey key = new MetricKey(
              source.getName(), entry.getKey(),
              parsed.dictName(), parsed.fieldName());
          orphanCounts.computeIfAbsent(key, k -> registerGauge(k))
              .set(orphans);

          pairsScanned++;
          totalOrphans += orphans;

          if (orphans >= warnThreshold) {
            log.warn("Orphan FK references: {}.{} → {}.{} = {} (threshold {})",
                source.getName(), entry.getKey(),
                parsed.dictName(), parsed.fieldName(),
                orphans, warnThreshold);
          }
        } catch (Exception e) {
          log.error("Failed orphan scan for {}.{} → {}.{}: {}",
              source.getName(), entry.getKey(),
              parsed.dictName(), parsed.fieldName(), e.getMessage());
        }
      }
    }

    long duration = System.currentTimeMillis() - start;
    log.info("Orphan FK sweep: {} pairs scanned, {} total orphans, took {}ms",
        pairsScanned, totalOrphans, duration);
  }

  private AtomicLong registerGauge(MetricKey key) {
    AtomicLong holder = new AtomicLong(0);
    meterRegistry.gauge(
        "nsi_orphan_references_total",
        Tags.of(
            "source_dict", key.sourceDict(),
            "source_field", key.sourceField(),
            "target_dict", key.targetDict(),
            "target_field", key.targetField()),
        holder,
        AtomicLong::doubleValue);
    return holder;
  }

  static ParsedRefSpec parseRef(String spec) {
    if (spec == null || spec.isBlank()) {
      throw new IllegalArgumentException("empty");
    }
    int dot = spec.indexOf('.');
    if (dot <= 0 || dot == spec.length() - 1) {
      throw new IllegalArgumentException("must be 'dict.field': " + spec);
    }
    return new ParsedRefSpec(spec.substring(0, dot), spec.substring(dot + 1));
  }

  /** Map key + Gauge tag holder — keeps cardinality bounded by schema FK count. */
  private record MetricKey(
      String sourceDict, String sourceField, String targetDict, String targetField) {}

  /** Schema {@code x-references} parse result. */
  record ParsedRefSpec(String dictName, String fieldName) {}
}
