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

import cloud.nstart.terravault.ordinis.domain.DataScope;
import cloud.nstart.terravault.ordinis.domain.definition.DictionaryDefinition;
import cloud.nstart.terravault.ordinis.domain.definition.DictionaryDefinitionRepository;
import cloud.nstart.terravault.ordinis.domain.record.DependentsQuery;
import cloud.nstart.terravault.ordinis.domain.record.DictionaryRecord;
import cloud.nstart.terravault.ordinis.domain.record.DictionaryRecordRepository;
import com.fasterxml.jackson.databind.JsonNode;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Tags;
import io.micrometer.core.instrument.Timer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.OffsetDateTime;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;

/**
 * Reverse FK lineage index: "кто ссылается на этот словарь / запись".
 *
 * <p>Phase 1 read-side для dict-relationships-v2 epic. Два режима:
 *
 * <ul>
 *   <li><b>Schema-level</b>: {@link #findSchemaDependents(String, Set)} —
 *       берёт target dict name, обходит все definitions и собирает все
 *       fields у которых {@code x-references = "<target>.<field>"}. Дёшево
 *       (in-memory traversal, ~40 dicts × ~20 fields), кэш не нужен в v1.</li>
 *   <li><b>Record-level</b>: {@link #findRecordDependents} — для конкретного
 *       business_key в target_dict находит active records в source dicts
 *       что на него ссылаются. Paginated через {@link DependentsQuery} (native
 *       SQL JSONB lookup). Phase 2 переносит на materialized view.</li>
 * </ul>
 *
 * <p>Caching: Phase 1 не кэширует — schema lookup микросекунды, record-level
 * SQL paginated и каждый bbox разный. Phase 2 добавит {@code @Cacheable} для
 * schema-level + invalidation на bundle import (см. eng review A.1).
 *
 * <p>Scope: filter target dict + source dicts через {@code allowedScopes}.
 * Если каллер не имеет access к target dict — возвращаем "not found"
 * (consistent с CRUD endpoint behaviour, чтобы не leak'ало existence).
 */
@Service
public class LineageIndexService {

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

  private final DictionaryDefinitionRepository definitionRepository;
  private final DictionaryRecordRepository recordRepository;
  private final DependentsQuery dependentsQuery;
  /** Optional — присутствует только когда {@code ordinis.lineage.mv.enabled=true}. */
  private final Optional<MaterializedViewRefreshService> mvRefresh;
  /** Optional — Phase 4 SLO instrumentation. Null в test ctor'е (без metrics). */
  private final Optional<MeterRegistry> meterRegistry;

  @Autowired
  public LineageIndexService(
      DictionaryDefinitionRepository definitionRepository,
      DictionaryRecordRepository recordRepository,
      DependentsQuery dependentsQuery,
      Optional<MaterializedViewRefreshService> mvRefresh,
      Optional<MeterRegistry> meterRegistry) {
    this.definitionRepository = definitionRepository;
    this.recordRepository = recordRepository;
    this.dependentsQuery = dependentsQuery;
    this.mvRefresh = mvRefresh;
    this.meterRegistry = meterRegistry;
  }

  /** Test-friendly ctor без MV refresh service и metrics. */
  public LineageIndexService(
      DictionaryDefinitionRepository definitionRepository,
      DictionaryRecordRepository recordRepository,
      DependentsQuery dependentsQuery) {
    this(definitionRepository, recordRepository, dependentsQuery,
        Optional.empty(), Optional.empty());
  }

  /**
   * Schema-level dependents target dict: какие dict.field ссылаются.
   *
   * @param targetDictName  dict name (e.g. "satellite_type")
   * @param allowedScopes   что caller видит (источники с недоступным scope filtered)
   * @return list reverse FK pointers, sorted by sourceDict, sourceField asc.
   *         Empty если target не существует или у caller нет к нему access.
   *
   * @throws SchemaDependentsTarget#NOT_FOUND нет такого dict.
   */
  @Transactional(readOnly = true)
  public List<SchemaDependent> findSchemaDependents(
      String targetDictName, Set<DataScope> allowedScopes) {
    Timer.Sample sample = startTimer();
    try {
      var result = findSchemaDependentsInternal(targetDictName, allowedScopes);
      stopTimer(sample, "schema", targetDictName);
      return result;
    } catch (RuntimeException e) {
      stopTimer(sample, "schema_error", targetDictName);
      throw e;
    }
  }

  private List<SchemaDependent> findSchemaDependentsInternal(
      String targetDictName, Set<DataScope> allowedScopes) {
    Optional<DictionaryDefinition> targetOpt = definitionRepository.findByName(targetDictName);
    if (targetOpt.isEmpty()) return List.of();
    DictionaryDefinition target = targetOpt.get();
    if (!allowedScopes.contains(target.getScope())) {
      // Scope hide — target invisible. Return empty (same shape as not_found).
      return List.of();
    }

    List<DictionaryDefinition> all = definitionRepository.findAll();
    // Pre-compute counts active records per source dict (для UI badge).
    Map<java.util.UUID, Long> counts = new HashMap<>();
    OffsetDateTime now = OffsetDateTime.now();
    for (Object[] row : recordRepository.countActiveGroupedByDictionary(allowedScopes, now)) {
      counts.put((java.util.UUID) row[0], (Long) row[1]);
    }

    List<SchemaDependent> out = new ArrayList<>();
    for (DictionaryDefinition source : all) {
      if (source.getId().equals(target.getId())) continue; // self-reference skipped
      if (!allowedScopes.contains(source.getScope())) continue;
      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();
        String fieldName = entry.getKey();
        JsonNode propSchema = entry.getValue();
        JsonNode refNode = propSchema.get("x-references");
        if (refNode == null || !refNode.isTextual()) continue;

        ReferenceValidator.ParsedRef parsed;
        try {
          parsed = ReferenceValidator.parseRef(refNode.asText(), propSchema);
        } catch (IllegalArgumentException e) {
          log.warn("Schema {} field {} malformed x-references: {}",
              source.getName(), fieldName, e.getMessage());
          continue;
        }
        if (!parsed.dictionaryName().equals(target.getName())) continue;

        out.add(new SchemaDependent(
            source.getName(),
            source.getDisplayName(),
            fieldName,
            parsed.fieldName(),
            parsed.onClose(),
            counts.getOrDefault(source.getId(), 0L)));
      }
    }
    out.sort((a, b) -> {
      int c = a.sourceDict().compareTo(b.sourceDict());
      return c != 0 ? c : a.sourceField().compareTo(b.sourceField());
    });
    return out;
  }

  /**
   * Record-level dependents: для target record (dict + business_key) ищет
   * все active records в source dicts что на него ссылаются.
   *
   * <p>Логика: для каждого source dict у которого schema имеет
   * {@code x-references: "<target_dict>.<target_field>"} — берём value
   * {@code target_record.data->>target_field} и считаем/перечисляем active
   * source records где {@code data->>source_field == that_value}.
   *
   * @param page page index 0..N
   * @param size page size 1..200
   * @return paged dependents + per-source totals
   */
  @Transactional(readOnly = true)
  public RecordDependentsPage findRecordDependents(
      String targetDictName,
      String targetBusinessKey,
      Set<DataScope> allowedScopes,
      int page,
      int size) {
    Timer.Sample sample = startTimer();
    try {
      var result = findRecordDependentsInternal(
          targetDictName, targetBusinessKey, allowedScopes, page, size);
      stopTimer(sample, "record", targetDictName);
      return result;
    } catch (RuntimeException e) {
      stopTimer(sample, "record_error", targetDictName);
      throw e;
    }
  }

  private RecordDependentsPage findRecordDependentsInternal(
      String targetDictName,
      String targetBusinessKey,
      Set<DataScope> allowedScopes,
      int page,
      int size) {
    if (page < 0) throw new IllegalArgumentException("page must be >= 0");
    // Service-level cap = 500 (CascadeCloseService запрашивает up to
    // CASCADE_MAX_PER_REQUEST). Controller separately enforces 200 cap для
    // public endpoint — UI не нуждается в больше.
    if (size < 1 || size > 500) {
      throw new IllegalArgumentException("size must be 1..500, got " + size);
    }

    Optional<DictionaryDefinition> targetOpt = definitionRepository.findByName(targetDictName);
    if (targetOpt.isEmpty()) {
      throw new IllegalArgumentException("dictionary not found: " + targetDictName);
    }
    DictionaryDefinition target = targetOpt.get();
    if (!allowedScopes.contains(target.getScope())) {
      throw new IllegalArgumentException("dictionary not found: " + targetDictName);
    }

    OffsetDateTime now = OffsetDateTime.now();
    Optional<DictionaryRecord> recOpt =
        recordRepository.findActiveAt(target.getId(), targetBusinessKey, now);
    if (recOpt.isEmpty()) {
      throw new IllegalArgumentException(
          "active record not found: " + targetDictName + "/" + targetBusinessKey);
    }
    DictionaryRecord targetRecord = recOpt.get();

    // Schema-level reverse index: какие (source_dict, source_field, target_field, onClose).
    List<SchemaDependent> schemaRefs = findSchemaDependents(targetDictName, allowedScopes);
    if (schemaRefs.isEmpty()) {
      return new RecordDependentsPage(
          targetDictName, targetBusinessKey, List.of(), List.of(), 0, page, size,
          buildLineageMeta());
    }

    // Per-source: pull target value, count + paginate.
    // Phase 1: linear scan по source schemas — ОК для <40 dicts.
    // Phase 2: MV сворачивает в один query.
    List<RecordDependent> all = new ArrayList<>();
    List<PerSourceSummary> summary = new ArrayList<>();
    long total = 0;

    for (SchemaDependent ref : schemaRefs) {
      JsonNode targetData = targetRecord.getData();
      if (targetData == null) continue;
      JsonNode valueNode = targetData.get(ref.targetField());
      if (valueNode == null || !valueNode.isTextual()) continue;
      String targetValue = valueNode.asText();

      Optional<DictionaryDefinition> sourceOpt =
          definitionRepository.findByName(ref.sourceDict());
      if (sourceOpt.isEmpty()) continue;
      DictionaryDefinition sourceDef = sourceOpt.get();

      long perSourceCount = dependentsQuery.countActive(
          sourceDef.getId(), ref.sourceField(), targetValue, now);
      summary.add(new PerSourceSummary(
          ref.sourceDict(),
          ref.sourceDisplayName(),
          ref.sourceField(),
          ref.onClose(),
          perSourceCount));
      total += perSourceCount;

      if (perSourceCount == 0) continue;

      // Paginate by aggregated total: каждый source pulled полностью если
      // нужен (proper offset с union в SQL отложен Phase 2 — здесь simpler).
      List<DependentsQuery.DependentRow> rows = dependentsQuery.findActive(
          sourceDef.getId(), ref.sourceField(), targetValue, now,
          // Защитный bound: <= 200 на source per request.
          Math.min(size + page * size + 1, 200), 0);
      for (DependentsQuery.DependentRow r : rows) {
        all.add(new RecordDependent(
            ref.sourceDict(),
            ref.sourceDisplayName(),
            ref.sourceField(),
            ref.onClose(),
            r.id(),
            r.businessKey(),
            r.validFrom(),
            r.validTo(),
            r.createdAt()));
      }
    }

    // Sort all dependents by createdAt DESC, then page slice.
    all.sort((a, b) -> b.createdAt().compareTo(a.createdAt()));
    int from = Math.min(page * size, all.size());
    int to = Math.min(from + size, all.size());
    List<RecordDependent> slice = all.subList(from, to);

    return new RecordDependentsPage(
        targetDictName, targetBusinessKey, slice, summary, total, page, size,
        buildLineageMeta());
  }

  /**
   * Метаданные lineage index source: помечает MV-backed vs JSONB direct
   * + last refresh timestamp (если MV active). Используется UI'ем для
   * staleness badge "обновлено N min назад".
   */
  private LineageMeta buildLineageMeta() {
    if (mvRefresh.isEmpty()) {
      // JSONB direct path — данные always fresh, нет stale window.
      return new LineageMeta(false, null, null, null);
    }
    var meta = mvRefresh.get().lastRefresh();
    if (meta.isEmpty() || meta.get().lastRefreshedAt() == null) {
      return new LineageMeta(true, null, "never", null);
    }
    var m = meta.get();
    return new LineageMeta(true, m.lastRefreshedAt(), m.lastStatus(), m.lastDurationMs());
  }

  /**
   * Batch outgoing FK summary: для каждого visible dict — сколько outgoing
   * FK refs в schema + список target dict names.
   *
   * <p>Используется admin-UI catalog list (`→` column) — раньше frontend
   * парсил schemaJson каждого dict для подсчёта, что требовало включения
   * schemaJson в list response. Этот endpoint позволяет catalog list
   * вернуть только metadata + один graph batch.
   *
   * @param allowedScopes scopes caller видит. Source dicts вне scope skip.
   * @param dictFilter    optional whitelist names (null = все visible).
   * @return map (dictName → OutgoingFkInfo). Empty dicts тоже включены
   *         (fkCount=0) чтобы frontend мог дисплейнуть "—" единообразно.
   */
  @Transactional(readOnly = true)
  public Map<String, OutgoingFkInfo> findOutgoingFkSummary(
      Set<DataScope> allowedScopes, Set<String> dictFilter) {
    List<DictionaryDefinition> all = definitionRepository.findAll();
    Map<String, OutgoingFkInfo> out = new java.util.LinkedHashMap<>();
    for (DictionaryDefinition d : all) {
      if (!allowedScopes.contains(d.getScope())) continue;
      if (dictFilter != null && !dictFilter.contains(d.getName())) continue;

      JsonNode schema = d.getSchemaJson();
      List<String> targets = new ArrayList<>();
      if (schema != null && schema.isObject()) {
        JsonNode properties = schema.get("properties");
        if (properties != null && properties.isObject()) {
          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;
            try {
              ReferenceValidator.ParsedRef parsed =
                  ReferenceValidator.parseRef(refNode.asText(), entry.getValue());
              // Dedup target dict names (один target может быть указан несколько
              // раз через разные fields — UI показывает unique targets list).
              if (!targets.contains(parsed.dictionaryName())) {
                targets.add(parsed.dictionaryName());
              }
            } catch (IllegalArgumentException e) {
              log.warn("Schema {} field {} malformed x-references: {}",
                  d.getName(), entry.getKey(), e.getMessage());
            }
          }
        }
      }
      // Sort target list для stable response (test stability + cache keys).
      targets.sort(String::compareTo);
      out.put(d.getName(), new OutgoingFkInfo(targets.size(), targets));
    }
    return out;
  }

  // === DTOs ===

  /** Per-dict outgoing FK summary. */
  public record OutgoingFkInfo(int fkCount, List<String> targets) {}

  /** Schema-level reverse FK pointer: source dict → target dict через field. */
  public record SchemaDependent(
      String sourceDict,
      String sourceDisplayName,
      String sourceField,
      String targetField,
      OnCloseAction onClose,
      long activeRecordsInSourceDict) {}

  /** Record-level dependent — конкретная active record что ссылается. */
  public record RecordDependent(
      String sourceDict,
      String sourceDisplayName,
      String sourceField,
      OnCloseAction onClose,
      java.util.UUID id,
      String businessKey,
      OffsetDateTime validFrom,
      OffsetDateTime validTo,
      OffsetDateTime createdAt) {}

  /** Per-source summary — сколько dependents в каждом source dict. */
  public record PerSourceSummary(
      String sourceDict,
      String sourceDisplayName,
      String sourceField,
      OnCloseAction onClose,
      long count) {}

  // === Metrics helpers ===

  private Timer.Sample startTimer() {
    if (meterRegistry.isEmpty()) return null;
    return Timer.start(meterRegistry.get());
  }

  private void stopTimer(Timer.Sample sample, String mode, String targetDict) {
    if (sample == null || meterRegistry.isEmpty()) return;
    // Cardinality bound: target_dict tag — bounded by # dictionaries (~40).
    Timer t = Timer.builder("nsi_dependents_query_duration_seconds")
        .description("Dependents lookup latency (Phase 4 SLO instrumentation)")
        .tags(Tags.of("mode", mode, "target_dict", targetDict))
        .publishPercentiles(0.5, 0.95, 0.99)
        .register(meterRegistry.get());
    sample.stop(t);
  }

  /** Paginated page + per-source aggregates + lineage source metadata. */
  public record RecordDependentsPage(
      String targetDict,
      String targetBusinessKey,
      List<RecordDependent> items,
      List<PerSourceSummary> perSource,
      long total,
      int page,
      int size,
      LineageMeta lineageMeta) {}

  /**
   * Metadata о backing data source (Phase 2):
   * @param mvEnabled MV-backed query path active. False — JSONB direct (always fresh).
   * @param lastRefreshedAt последний successful MV refresh, null если MV выключен или never.
   * @param lastStatus "ok" / "error" / "never". null если MV выключен.
   * @param lastDurationMs длительность последнего refresh'а в ms.
   */
  public record LineageMeta(
      boolean mvEnabled,
      OffsetDateTime lastRefreshedAt,
      String lastStatus,
      Long lastDurationMs) {}
}
