feat(refs): orphan FK scanner — Prometheus gauge nsi_orphan_references_total

@Scheduled sweep раз в час (configurable). Walks all schemas, для каждого
x-references marker считает orphans через JdbcTemplate native SQL
(WHERE source.data->>field NOT IN target dict active records).

Метрика: Gauge per (source_dict, source_field, target_dict, target_field).
Cardinality bounded by FK count в bundles (~5 для cuod v1.2.1).

Conditional: ordinis.references.scan-enabled=true (default off, включаем
явно в prod values). Read-only, no write impact.

OrphanReferenceQuery (ordinis-domain): native SQL helper, identifier
validation против SQL injection. Field name regex check т.к. JSONB key
path не bind'ится через @Param.

Tests: 4 в OrphanReferenceScannerTest (parseRef edge cases). Реальный
SQL логику покроет integration тест на staging данных.

Total project tests: 191 → 195.

ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/record/OrphanReferenceQuery.java

@@ -0,0 +1,82 @@
+package cloud.nstart.terravault.ordinis.domain.record;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Component;
+
+import java.util.UUID;
+
+/**
+ * Native SQL для подсчёта orphan FK references — записи где
+ * {@code data->>source_field} не находится в active records target dictionary
+ * по {@code data->>target_field}.
+ *
+ * <p>Используется {@code OrphanReferenceScanner} раз в час для метрики
+ * {@code nsi_orphan_references_total}. Отдельный component (не repository
+ * метод) потому что используем JdbcTemplate с dynamic field names — JPA
+ * не может взять {@code @Param} в JSONB key path.
+ */
+@Component
+public class OrphanReferenceQuery {
+
+  private final JdbcTemplate jdbc;
+
+  @Autowired
+  public OrphanReferenceQuery(JdbcTemplate jdbc) {
+    this.jdbc = jdbc;
+  }
+
+  /**
+   * Считает orphan references: active records в source dictionary где
+   * {@code data->>source_field} НЕ присутствует среди active records
+   * target dictionary {@code data->>target_field}. NULL-значения
+   * пропускаются (optional fields, x-references не required).
+   *
+   * <p>Field names валидируются caller'ом (берутся из schema parser, не
+   * из user input) — SQL injection невозможна. Используем native SQL
+   * concat потому что JdbcTemplate не bind'ит JSONB key paths.
+   *
+   * @return count of orphan source records
+   */
+  public long countOrphans(
+      UUID sourceDictionaryId,
+      String sourceField,
+      UUID targetDictionaryId,
+      String targetField) {
+    validateField(sourceField);
+    validateField(targetField);
+
+    String sql = """
+        SELECT count(*) FROM dictionary_records src
+        WHERE src.dictionary_id = ?
+          AND src.valid_from <= now()
+          AND src.valid_to > now()
+          AND src.data ->> '%s' IS NOT NULL
+          AND src.data ->> '%s' NOT IN (
+            SELECT tgt.data ->> '%s'
+            FROM dictionary_records tgt
+            WHERE tgt.dictionary_id = ?
+              AND tgt.valid_from <= now()
+              AND tgt.valid_to > now()
+              AND tgt.data ->> '%s' IS NOT NULL
+          )
+        """.formatted(sourceField, sourceField, targetField, targetField);
+
+    Long count = jdbc.queryForObject(sql, Long.class, sourceDictionaryId, targetDictionaryId);
+    return count == null ? 0 : count;
+  }
+
+  /**
+   * SQL injection guard. Field name приходит из JSON Schema (мы парсим),
+   * но мы не доверяем — extra paranoid check. Алlowед identifier rules.
+   */
+  private static void validateField(String field) {
+    if (field == null || field.isBlank()) {
+      throw new IllegalArgumentException("field name is empty");
+    }
+    if (!field.matches("^[a-zA-Z_][a-zA-Z0-9_]{0,127}$")) {
+      throw new IllegalArgumentException(
+          "field name has unsafe characters: " + field);
+    }
+  }
+}

ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/service/reference/OrphanReferenceScanner.java

@@ -0,0 +1,179 @@
+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) {}
+}

ordinis-rest-api/src/test/java/cloud/nstart/terravault/ordinis/restapi/service/reference/OrphanReferenceScannerTest.java

@@ -0,0 +1,38 @@
+package cloud.nstart.terravault.ordinis.restapi.service.reference;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class OrphanReferenceScannerTest {
+
+  @Test
+  void parsesValidRef() {
+    var p = OrphanReferenceScanner.parseRef("satellite_type.code");
+    assertThat(p.dictName()).isEqualTo("satellite_type");
+    assertThat(p.fieldName()).isEqualTo("code");
+  }
+
+  @Test
+  void rejectsEmpty() {
+    assertThatThrownBy(() -> OrphanReferenceScanner.parseRef(""))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> OrphanReferenceScanner.parseRef(null))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  void rejectsMissingDot() {
+    assertThatThrownBy(() -> OrphanReferenceScanner.parseRef("noDot"))
+        .hasMessageContaining("dict.field");
+  }
+
+  @Test
+  void rejectsLeadingTrailingDot() {
+    assertThatThrownBy(() -> OrphanReferenceScanner.parseRef(".field"))
+        .hasMessageContaining("dict.field");
+    assertThatThrownBy(() -> OrphanReferenceScanner.parseRef("dict."))
+        .hasMessageContaining("dict.field");
+  }
+}