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 com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.time.OffsetDateTime;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

/**
 * Unit tests {@link LineageIndexService}. Покрывает:
 *
 * <ul>
 *   <li>Schema-level reverse FK: not_found, scope hidden, malformed,
 *       self-reference skip, sorting, onClose из x-references-on-close.</li>
 *   <li>Record-level: not_active record → exception, no schema refs → empty,
 *       perSource counts + paginated items, pagination across multiple
 *       sources.</li>
 * </ul>
 *
 * <p>Mockito не может mock'нуть DictionaryDefinition / DictionaryRecord
 * на JDK 25 (subclass mock-maker incompat). Используем real entities.
 */
class LineageIndexServiceTest {

  private static final ObjectMapper OM = new ObjectMapper();

  private DictionaryDefinitionRepository defRepo;
  private DictionaryRecordRepository recRepo;
  private DependentsQuery depsQuery;
  private LineageIndexService service;

  @BeforeEach
  void setUp() {
    defRepo = mock(DictionaryDefinitionRepository.class);
    recRepo = mock(DictionaryRecordRepository.class);
    depsQuery = mock(DependentsQuery.class);
    service = new LineageIndexService(defRepo, recRepo, depsQuery);
  }

  // === Schema-level findSchemaDependents ===

  @Test
  void schemaDeps_targetNotFound_returnsEmpty() {
    when(defRepo.findByName("missing")).thenReturn(Optional.empty());
    var result = service.findSchemaDependents("missing", Set.of(DataScope.PUBLIC));
    assertThat(result).isEmpty();
  }

  @Test
  void schemaDeps_targetScopeHidden_returnsEmpty() {
    var target = mockDef("internal", DataScope.INTERNAL);
    when(defRepo.findByName("internal")).thenReturn(Optional.of(target));
    var result = service.findSchemaDependents("internal", Set.of(DataScope.PUBLIC));
    assertThat(result).isEmpty();
  }

  @Test
  void schemaDeps_oneReferenceFound_returnsIt() throws Exception {
    var target = mockDef("satellite_type", DataScope.PUBLIC);
    var source = mockDefWithSchema("spacecraft", DataScope.PUBLIC, """
        {"properties":{"satTypeCode":{"type":"string","x-references":"satellite_type.code"}}}""");
    when(defRepo.findByName("satellite_type")).thenReturn(Optional.of(target));
    when(defRepo.findAll()).thenReturn(List.of(target, source));
    List<Object[]> rows = new java.util.ArrayList<>();
    rows.add(new Object[]{source.getId(), 7L});
    when(recRepo.countActiveGroupedByDictionary(any(), any())).thenReturn(rows);

    var result = service.findSchemaDependents("satellite_type", Set.of(DataScope.PUBLIC));

    assertThat(result).hasSize(1);
    var dep = result.get(0);
    assertThat(dep.sourceDict()).isEqualTo("spacecraft");
    assertThat(dep.sourceField()).isEqualTo("satTypeCode");
    assertThat(dep.targetField()).isEqualTo("code");
    assertThat(dep.onClose()).isEqualTo(OnCloseAction.BLOCK); // default
    assertThat(dep.activeRecordsInSourceDict()).isEqualTo(7L);
  }

  @Test
  void schemaDeps_onCloseFromSchema_propagated() throws Exception {
    var target = mockDef("ground_station", DataScope.PUBLIC);
    var source = mockDefWithSchema("contact_session", DataScope.PUBLIC, """
        {"properties":{
          "stationCode":{"type":"string",
            "x-references":"ground_station.code",
            "x-references-on-close":"cascade"}
        }}""");
    when(defRepo.findByName("ground_station")).thenReturn(Optional.of(target));
    when(defRepo.findAll()).thenReturn(List.of(target, source));
    when(recRepo.countActiveGroupedByDictionary(any(), any())).thenReturn(List.of());

    var result = service.findSchemaDependents("ground_station", Set.of(DataScope.PUBLIC));

    assertThat(result).hasSize(1);
    assertThat(result.get(0).onClose()).isEqualTo(OnCloseAction.CASCADE);
  }

  @Test
  void schemaDeps_multipleSourcesSorted() throws Exception {
    var target = mockDef("dict_t", DataScope.PUBLIC);
    var srcA = mockDefWithSchema("zeta_dict", DataScope.PUBLIC,
        "{\"properties\":{\"r\":{\"x-references\":\"dict_t.f\"}}}");
    var srcB = mockDefWithSchema("alpha_dict", DataScope.PUBLIC,
        "{\"properties\":{\"r\":{\"x-references\":\"dict_t.f\"}}}");
    when(defRepo.findByName("dict_t")).thenReturn(Optional.of(target));
    when(defRepo.findAll()).thenReturn(List.of(target, srcA, srcB));
    when(recRepo.countActiveGroupedByDictionary(any(), any())).thenReturn(List.of());

    var result = service.findSchemaDependents("dict_t", Set.of(DataScope.PUBLIC));

    assertThat(result).extracting(LineageIndexService.SchemaDependent::sourceDict)
        .containsExactly("alpha_dict", "zeta_dict");
  }

  @Test
  void schemaDeps_sourceScopeHidden_skipped() throws Exception {
    var target = mockDef("dict_t", DataScope.PUBLIC);
    var visibleSrc = mockDefWithSchema("public_src", DataScope.PUBLIC,
        "{\"properties\":{\"r\":{\"x-references\":\"dict_t.f\"}}}");
    var hiddenSrc = mockDefWithSchema("internal_src", DataScope.INTERNAL,
        "{\"properties\":{\"r\":{\"x-references\":\"dict_t.f\"}}}");
    when(defRepo.findByName("dict_t")).thenReturn(Optional.of(target));
    when(defRepo.findAll()).thenReturn(List.of(target, visibleSrc, hiddenSrc));
    when(recRepo.countActiveGroupedByDictionary(any(), any())).thenReturn(List.of());

    var result = service.findSchemaDependents("dict_t", Set.of(DataScope.PUBLIC));

    assertThat(result).extracting(LineageIndexService.SchemaDependent::sourceDict)
        .containsExactly("public_src");
  }

  @Test
  void schemaDeps_selfReference_skipped() throws Exception {
    // Schema указывает на свой же dict — не должно зацикливать.
    var selfRef = mockDefWithSchema("hierarchy", DataScope.PUBLIC,
        "{\"properties\":{\"parentCode\":{\"x-references\":\"hierarchy.code\"}}}");
    when(defRepo.findByName("hierarchy")).thenReturn(Optional.of(selfRef));
    when(defRepo.findAll()).thenReturn(List.of(selfRef));
    when(recRepo.countActiveGroupedByDictionary(any(), any())).thenReturn(List.of());

    var result = service.findSchemaDependents("hierarchy", Set.of(DataScope.PUBLIC));

    assertThat(result).isEmpty();
  }

  @Test
  void schemaDeps_malformedRef_skippedSilently() throws Exception {
    var target = mockDef("dict_t", DataScope.PUBLIC);
    var srcGood = mockDefWithSchema("good_src", DataScope.PUBLIC,
        "{\"properties\":{\"r\":{\"x-references\":\"dict_t.f\"}}}");
    var srcBad = mockDefWithSchema("bad_src", DataScope.PUBLIC,
        "{\"properties\":{\"r\":{\"x-references\":\"malformed-no-dot\"}}}");
    when(defRepo.findByName("dict_t")).thenReturn(Optional.of(target));
    when(defRepo.findAll()).thenReturn(List.of(target, srcGood, srcBad));
    when(recRepo.countActiveGroupedByDictionary(any(), any())).thenReturn(List.of());

    var result = service.findSchemaDependents("dict_t", Set.of(DataScope.PUBLIC));

    assertThat(result).extracting(LineageIndexService.SchemaDependent::sourceDict)
        .containsExactly("good_src");
  }

  @Test
  void schemaDeps_targetSchemaWithoutProperties_skipped() throws Exception {
    var target = mockDef("dict_t", DataScope.PUBLIC);
    var src = mockDefWithSchema("src", DataScope.PUBLIC, "{\"type\":\"object\"}"); // no properties
    when(defRepo.findByName("dict_t")).thenReturn(Optional.of(target));
    when(defRepo.findAll()).thenReturn(List.of(target, src));
    when(recRepo.countActiveGroupedByDictionary(any(), any())).thenReturn(List.of());

    var result = service.findSchemaDependents("dict_t", Set.of(DataScope.PUBLIC));

    assertThat(result).isEmpty();
  }

  // === Record-level findRecordDependents ===

  @Test
  void recordDeps_invalidPagination_throws() {
    assertThatThrownBy(() -> service.findRecordDependents(
        "d", "k", Set.of(DataScope.PUBLIC), -1, 50))
        .isInstanceOf(IllegalArgumentException.class)
        .hasMessageContaining("page");
    assertThatThrownBy(() -> service.findRecordDependents(
        "d", "k", Set.of(DataScope.PUBLIC), 0, 0))
        .isInstanceOf(IllegalArgumentException.class)
        .hasMessageContaining("size");
    assertThatThrownBy(() -> service.findRecordDependents(
        "d", "k", Set.of(DataScope.PUBLIC), 0, 501))
        .isInstanceOf(IllegalArgumentException.class)
        .hasMessageContaining("size");
  }

  @Test
  void recordDeps_dictNotFound_throws() {
    when(defRepo.findByName("missing")).thenReturn(Optional.empty());
    assertThatThrownBy(() -> service.findRecordDependents(
        "missing", "k", Set.of(DataScope.PUBLIC), 0, 50))
        .isInstanceOf(IllegalArgumentException.class)
        .hasMessageContaining("dictionary not found");
  }

  @Test
  void recordDeps_dictScopeHidden_throwsAsNotFound() {
    var target = mockDef("internal", DataScope.INTERNAL);
    when(defRepo.findByName("internal")).thenReturn(Optional.of(target));
    assertThatThrownBy(() -> service.findRecordDependents(
        "internal", "k", Set.of(DataScope.PUBLIC), 0, 50))
        .isInstanceOf(IllegalArgumentException.class)
        .hasMessageContaining("dictionary not found");
  }

  @Test
  void recordDeps_recordNotActive_throws() {
    var target = mockDef("dict_t", DataScope.PUBLIC);
    when(defRepo.findByName("dict_t")).thenReturn(Optional.of(target));
    when(recRepo.findActiveAt(eq(target.getId()), eq("missing-key"), any()))
        .thenReturn(Optional.empty());

    assertThatThrownBy(() -> service.findRecordDependents(
        "dict_t", "missing-key", Set.of(DataScope.PUBLIC), 0, 50))
        .isInstanceOf(IllegalArgumentException.class)
        .hasMessageContaining("active record not found");
  }

  @Test
  void recordDeps_noSchemaRefs_returnsEmptyPage() throws Exception {
    var target = mockDef("dict_t", DataScope.PUBLIC);
    when(defRepo.findByName("dict_t")).thenReturn(Optional.of(target));
    when(defRepo.findAll()).thenReturn(List.of(target));
    when(recRepo.countActiveGroupedByDictionary(any(), any())).thenReturn(List.of());

    JsonNode data = OM.readTree("{\"code\":\"X-001\"}");
    DictionaryRecord rec = mockRecord(target.getId(), "X-001", data);
    when(recRepo.findActiveAt(eq(target.getId()), eq("X-001"), any()))
        .thenReturn(Optional.of(rec));

    var page = service.findRecordDependents(
        "dict_t", "X-001", Set.of(DataScope.PUBLIC), 0, 50);

    assertThat(page.items()).isEmpty();
    assertThat(page.perSource()).isEmpty();
    assertThat(page.total()).isZero();
  }

  @Test
  void recordDeps_oneSourceWithDependents_returnsItemsAndPerSource() throws Exception {
    var target = mockDef("satellite_type", DataScope.PUBLIC);
    var source = mockDefWithSchema("spacecraft", DataScope.PUBLIC, """
        {"properties":{"satTypeCode":{"type":"string","x-references":"satellite_type.code"}}}""");
    when(defRepo.findByName("satellite_type")).thenReturn(Optional.of(target));
    when(defRepo.findByName("spacecraft")).thenReturn(Optional.of(source));
    when(defRepo.findAll()).thenReturn(List.of(target, source));
    when(recRepo.countActiveGroupedByDictionary(any(), any())).thenReturn(List.of());

    JsonNode targetData = OM.readTree("{\"code\":\"SAR-X\"}");
    DictionaryRecord targetRec = mockRecord(target.getId(), "SAR-X", targetData);
    when(recRepo.findActiveAt(eq(target.getId()), eq("SAR-X"), any()))
        .thenReturn(Optional.of(targetRec));

    when(depsQuery.countActive(eq(source.getId()), eq("satTypeCode"), eq("SAR-X"), any()))
        .thenReturn(2L);
    OffsetDateTime t1 = OffsetDateTime.parse("2026-01-01T00:00:00Z");
    OffsetDateTime t2 = OffsetDateTime.parse("2026-02-01T00:00:00Z");
    when(depsQuery.findActive(eq(source.getId()), eq("satTypeCode"), eq("SAR-X"),
        any(), anyInt(), anyInt())).thenReturn(List.of(
            new DependentsQuery.DependentRow(
                UUID.randomUUID(), source.getId(), "SC-001", t1, OffsetDateTime.MAX, t2),
            new DependentsQuery.DependentRow(
                UUID.randomUUID(), source.getId(), "SC-002", t1, OffsetDateTime.MAX, t1)));

    var page = service.findRecordDependents(
        "satellite_type", "SAR-X", Set.of(DataScope.PUBLIC), 0, 50);

    assertThat(page.total()).isEqualTo(2L);
    assertThat(page.perSource()).hasSize(1);
    assertThat(page.perSource().get(0).count()).isEqualTo(2L);
    assertThat(page.perSource().get(0).sourceDict()).isEqualTo("spacecraft");
    assertThat(page.items()).hasSize(2);
    // sorted by createdAt DESC: SC-001 (t2) first
    assertThat(page.items().get(0).businessKey()).isEqualTo("SC-001");
    assertThat(page.items().get(1).businessKey()).isEqualTo("SC-002");
  }

  @Test
  void recordDeps_pagination_slicesCorrectly() throws Exception {
    var target = mockDef("dict_t", DataScope.PUBLIC);
    var source = mockDefWithSchema("src", DataScope.PUBLIC,
        "{\"properties\":{\"ref\":{\"x-references\":\"dict_t.code\"}}}");
    when(defRepo.findByName("dict_t")).thenReturn(Optional.of(target));
    when(defRepo.findByName("src")).thenReturn(Optional.of(source));
    when(defRepo.findAll()).thenReturn(List.of(target, source));
    when(recRepo.countActiveGroupedByDictionary(any(), any())).thenReturn(List.of());

    JsonNode targetData = OM.readTree("{\"code\":\"K\"}");
    when(recRepo.findActiveAt(eq(target.getId()), eq("K"), any()))
        .thenReturn(Optional.of(mockRecord(target.getId(), "K", targetData)));

    when(depsQuery.countActive(any(), anyString(), anyString(), any())).thenReturn(5L);
    // 5 rows, sorted later by createdAt DESC.
    var rows = new java.util.ArrayList<DependentsQuery.DependentRow>();
    for (int i = 0; i < 5; i++) {
      rows.add(new DependentsQuery.DependentRow(
          UUID.randomUUID(), source.getId(), "K-" + i,
          OffsetDateTime.parse("2026-01-01T00:00:00Z"), OffsetDateTime.MAX,
          OffsetDateTime.parse("2026-01-0" + (i + 1) + "T00:00:00Z")));
    }
    when(depsQuery.findActive(any(), anyString(), anyString(), any(), anyInt(), anyInt()))
        .thenReturn(rows);

    var page1 = service.findRecordDependents("dict_t", "K", Set.of(DataScope.PUBLIC), 0, 2);
    assertThat(page1.items()).hasSize(2);
    assertThat(page1.items().get(0).businessKey()).isEqualTo("K-4"); // newest
    assertThat(page1.total()).isEqualTo(5L);

    var page2 = service.findRecordDependents("dict_t", "K", Set.of(DataScope.PUBLIC), 1, 2);
    assertThat(page2.items()).hasSize(2);
    assertThat(page2.items().get(0).businessKey()).isEqualTo("K-2");

    var page3 = service.findRecordDependents("dict_t", "K", Set.of(DataScope.PUBLIC), 2, 2);
    assertThat(page3.items()).hasSize(1);
    assertThat(page3.items().get(0).businessKey()).isEqualTo("K-0"); // oldest
  }

  // === Batch outgoing FK summary ===

  @Test
  void outgoingFk_emptyDicts_returnsEmpty() {
    when(defRepo.findAll()).thenReturn(List.of());
    var result = service.findOutgoingFkSummary(Set.of(DataScope.PUBLIC), null);
    assertThat(result).isEmpty();
  }

  @Test
  void outgoingFk_basicCounts_returnsCountsAndTargets() {
    var noRefs = mockDefWithSchema("operators", DataScope.PUBLIC,
        "{\"properties\":{\"name\":{\"type\":\"string\"}}}");
    var threeRefs = mockDefWithSchema("satellites", DataScope.PUBLIC, """
        {"properties":{
          "operator":{"type":"string","x-references":"operators.code"},
          "launchSite":{"type":"string","x-references":"launch_sites.code"},
          "mission":{"type":"string","x-references":"missions.code"}
        }}""");
    when(defRepo.findAll()).thenReturn(List.of(noRefs, threeRefs));

    var result = service.findOutgoingFkSummary(Set.of(DataScope.PUBLIC), null);

    assertThat(result).hasSize(2);
    assertThat(result.get("operators").fkCount()).isEqualTo(0);
    assertThat(result.get("operators").targets()).isEmpty();
    assertThat(result.get("satellites").fkCount()).isEqualTo(3);
    // Sorted alphabetically для stable test + cacheability.
    assertThat(result.get("satellites").targets())
        .containsExactly("launch_sites", "missions", "operators");
  }

  @Test
  void outgoingFk_scopeHidesSource_skipped() {
    var publicDict = mockDef("pub", DataScope.PUBLIC);
    var internalDict = mockDef("internal", DataScope.INTERNAL);
    when(defRepo.findAll()).thenReturn(List.of(publicDict, internalDict));

    var result = service.findOutgoingFkSummary(Set.of(DataScope.PUBLIC), null);

    assertThat(result).containsOnlyKeys("pub");
  }

  @Test
  void outgoingFk_filterWhitelist_respectsFilter() {
    var a = mockDef("a", DataScope.PUBLIC);
    var b = mockDef("b", DataScope.PUBLIC);
    var c = mockDef("c", DataScope.PUBLIC);
    when(defRepo.findAll()).thenReturn(List.of(a, b, c));

    var result = service.findOutgoingFkSummary(
        Set.of(DataScope.PUBLIC), Set.of("a", "c"));

    assertThat(result).containsOnlyKeys("a", "c");
  }

  @Test
  void outgoingFk_dedupRepeatedTarget_returnsUnique() {
    // Source с двумя fields ссылающимися на same target dict — target должен
    // появиться в targets list один раз, но fkCount считает unique targets.
    var src = mockDefWithSchema("doc", DataScope.PUBLIC, """
        {"properties":{
          "primaryAuthor":{"type":"string","x-references":"users.id"},
          "reviewerAuthor":{"type":"string","x-references":"users.id"}
        }}""");
    when(defRepo.findAll()).thenReturn(List.of(src));

    var result = service.findOutgoingFkSummary(Set.of(DataScope.PUBLIC), null);

    assertThat(result.get("doc").fkCount()).isEqualTo(1);
    assertThat(result.get("doc").targets()).containsExactly("users");
  }

  @Test
  void outgoingFk_malformedRef_skippedNotFatal() {
    var src = mockDefWithSchema("buggy", DataScope.PUBLIC, """
        {"properties":{
          "good":{"type":"string","x-references":"ok.code"},
          "bad":{"type":"string","x-references":"no_dot_format"}
        }}""");
    when(defRepo.findAll()).thenReturn(List.of(src));

    var result = service.findOutgoingFkSummary(Set.of(DataScope.PUBLIC), null);

    assertThat(result.get("buggy").fkCount()).isEqualTo(1);
    assertThat(result.get("buggy").targets()).containsExactly("ok");
  }

  // === Helpers ===

  private static DictionaryDefinition mockDef(String name, DataScope scope) {
    return mockDefWithSchema(name, scope, "{\"type\":\"object\"}");
  }

  private static DictionaryDefinition mockDefWithSchema(String name, DataScope scope, String schemaJson) {
    JsonNode schema;
    try {
      schema = OM.readTree(schemaJson);
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
    var def = new DictionaryDefinition(UUID.randomUUID(), name, scope, schema);
    def.setDisplayName(name);
    return def;
  }

  private static DictionaryRecord mockRecord(UUID dictId, String businessKey, JsonNode data) {
    return new DictionaryRecord(
        UUID.randomUUID(), dictId, businessKey, data, DataScope.PUBLIC,
        OffsetDateTime.parse("2026-01-01T00:00:00Z"));
  }
}
