package com.mydreamjob.skill;

import com.mydreamjob.profile.StudentProfile;
import com.mydreamjob.profile.StudentProfileRepository;
import com.mydreamjob.skill.domain.SkillAlias;
import com.mydreamjob.skill.dto.SkillCloudItem;
import com.mydreamjob.skill.dto.SkillCloudNode;
import com.mydreamjob.skill.dto.SkillCloudResponse;
import com.mydreamjob.skill.dto.SkillMatchResponse;
import com.mydreamjob.skill.repository.SkillAliasRepository;
import org.hibernate.boot.model.relational.Namespace;
import org.hibernate.boot.model.relational.Sequence;
import org.hibernate.mapping.Table;
import org.hibernate.tool.schema.spi.SchemaFilter;
import org.hibernate.tool.schema.spi.SchemaFilterProvider;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.mock.mockito.MockBean;
import org.springframework.boot.test.mock.mockito.SpyBean;
import org.springframework.boot.test.system.CapturedOutput;
import org.springframework.boot.test.system.OutputCaptureExtension;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.test.context.ActiveProfiles;
import org.springframework.test.context.TestPropertySource;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigDecimal;
import java.util.List;
import java.util.Locale;
import java.util.Optional;
import java.util.Set;
import java.util.stream.IntStream;
import java.util.stream.Stream;

import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyCollection;
import static org.mockito.Mockito.doReturn;

@SpringBootTest
@ExtendWith(OutputCaptureExtension.class)
@TestPropertySource(properties = {
        "spring.datasource.url=jdbc:h2:mem:integrationdb_skill_match;MODE=MySQL;DB_CLOSE_DELAY=-1;DB_CLOSE_ON_EXIT=FALSE",
        "spring.jpa.hibernate.ddl-auto=validate",
        "spring.jpa.properties.hibernate.hbm2ddl.schema_filter_provider=com.mydreamjob.skill.SkillMatchServiceTest$SkillMatchSchemaFilterProvider"
})
@ActiveProfiles("integration-test")
@Transactional
class SkillMatchServiceTest {

    private static final Long USER_ID = 7001L;

    @Autowired
    private SkillMatchService service;

    @SpyBean
    private SkillCloudQueryService queryService;

    @SpyBean
    private StudentProfileRepository profileRepository;

    @SpyBean
    private SkillAliasRepository aliasRepository;

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @MockBean
    private RedisTemplate<String, Object> redisTemplate;

    @Test
    void normalizerAppliesUnicodeRulesAndPreservesProgrammingSymbols() {
        SkillNameNormalizer normalizer = new SkillNameNormalizer();

        assertThat(normalizer.normalize("  Ｓｐｒｉｎｇ‐Ｂｏｏｔ.  ")).isEqualTo("spring boot");
        assertThat(normalizer.normalize("  JAVA\n\tSCRIPT ")).isEqualTo("java script");
        assertThat(normalizer.normalize("Node.js")).isEqualTo("nodejs");
        assertThat(normalizer.normalize("C++")).isEqualTo("c++");
        assertThat(normalizer.normalize("C#")).isEqualTo("c#");
    }

    @Test
    void usesOnlyTopTenWhenMoreThanTenSkillsExist() {
        List<SkillCloudItem> items = IntStream.rangeClosed(1, 11)
                .mapToObj(index -> item((long) index, "Skill " + index, "Skill " + index, "SKILL_" + index))
                .toList();
        stubCloud(cloud(items));
        saveProfile(USER_ID, json(items.stream().map(SkillCloudItem::nameZh).toList()));

        SkillMatchResponse result = service.match(USER_ID, 1L, 1L, null);

        assertThat(result.totalCount()).isEqualTo(10);
        assertThat(result.matchedCount()).isEqualTo(10);
        assertThat(result.matchScore()).isEqualTo(100);
        assertThat(result.matchedSkills())
                .extracting(SkillCloudItem::skillId)
                .containsExactlyElementsOf(IntStream.rangeClosed(1, 10).mapToLong(Long::valueOf).boxed().toList());
        assertThat(result.missingSkills()).isEmpty();
    }

    @Test
    void usesTenAsDenominatorWhenExactlyTenSkillsExist() {
        List<SkillCloudItem> items = IntStream.rangeClosed(1, 10)
                .mapToObj(index -> item((long) index, "Skill " + index, "Skill " + index, "SKILL_" + index))
                .toList();
        stubCloud(cloud(items));
        saveProfile(USER_ID, json(items.stream().limit(5).map(SkillCloudItem::nameZh).toList()));

        SkillMatchResponse result = service.match(USER_ID, 1L, 1L, null);

        assertThat(result.totalCount()).isEqualTo(10);
        assertThat(result.matchedCount()).isEqualTo(5);
        assertThat(result.matchScore()).isEqualTo(50);
        assertThat(result.missingSkills()).hasSize(5);
    }

    @Test
    void usesActualSizeAsDenominatorAndRoundsOneThirdToThirtyThree() {
        List<SkillCloudItem> items = List.of(
                item(1L, "Java", "Java", "JAVA"),
                item(2L, "Python", "Python", "PYTHON"),
                item(3L, "SQL", "SQL", "SQL")
        );
        stubCloud(cloud(items));
        saveProfile(USER_ID, json(List.of("Java")));

        SkillMatchResponse result = service.match(USER_ID, 1L, 1L, null);

        assertThat(result.totalCount()).isEqualTo(3);
        assertThat(result.matchedCount()).isEqualTo(1);
        assertThat(result.matchScore()).isEqualTo(33);
        assertThat(result.matchedSkills()).extracting(SkillCloudItem::skillId).containsExactly(1L);
        assertThat(result.missingSkills()).extracting(SkillCloudItem::skillId).containsExactly(2L, 3L);
    }

    @Test
    void returnsZeroAndEmptyListsWhenCloudIsEmpty() {
        stubCloud(cloud(List.of()));
        saveProfile(USER_ID, json(List.of("Java")));

        SkillMatchResponse result = service.match(USER_ID, 1L, 1L, null);

        assertThat(result.matchScore()).isZero();
        assertThat(result.matchedCount()).isZero();
        assertThat(result.totalCount()).isZero();
        assertThat(result.matchedSkills()).isEmpty();
        assertThat(result.missingSkills()).isEmpty();
    }

    @Test
    void treatsMissingProfileAsNoSkills() {
        stubCloud(cloud(List.of(item(1L, "Java", "Java", "JAVA"))));

        SkillMatchResponse result = service.match(9999L, 1L, 1L, null);

        assertThat(result.matchScore()).isZero();
        assertThat(result.matchedSkills()).isEmpty();
        assertThat(result.missingSkills()).extracting(SkillCloudItem::skillId).containsExactly(1L);
    }

    @Test
    void treatsBlankAndEmptyArrayAsNoSkills() {
        stubCloud(cloud(List.of(item(1L, "Java", "Java", "JAVA"), item(2L, "SQL", "SQL", "SQL"))));
        doReturn(Optional.of(StudentProfile.builder()
                .id(100L)
                .userId(USER_ID)
                .skillsJson("   ")
                .build()))
                .when(profileRepository).findByUserId(USER_ID);

        SkillMatchResponse blank = service.match(USER_ID, 1L, 1L, null);

        assertThat(blank.matchScore()).isZero();
        assertThat(blank.missingSkills()).extracting(SkillCloudItem::skillId).containsExactly(1L, 2L);

        doReturn(Optional.of(StudentProfile.builder()
                .id(101L)
                .userId(USER_ID)
                .skillsJson("[]")
                .build()))
                .when(profileRepository).findByUserId(USER_ID);

        SkillMatchResponse empty = service.match(USER_ID, 1L, 1L, null);

        assertThat(empty.matchScore()).isZero();
        assertThat(empty.matchedSkills()).isEmpty();
    }

    @Test
    void treatsMalformedJsonAsNoSkillsWithoutLeakingRawJson(CapturedOutput output) {
        String malformedJson = "{\"skills\":[\"Java\"]";
        stubCloud(cloud(List.of(item(1L, "Java", "Java", "JAVA"))));
        doReturn(Optional.of(StudentProfile.builder()
                .id(102L)
                .userId(USER_ID)
                .skillsJson(malformedJson)
                .build()))
                .when(profileRepository).findByUserId(USER_ID);

        SkillMatchResponse result = service.match(USER_ID, 1L, 1L, null);

        assertThat(result.matchScore()).isZero();
        assertThat(result.matchedSkills()).isEmpty();
        assertThat(output.getAll()).contains("Failed to parse student skills JSON");
        assertThat(output.getAll()).doesNotContain(malformedJson);
    }

    @Test
    void matchesNameZhNameEnSkillCodeAndAliasAndDeduplicatesStudentSkills() {
        List<SkillCloudItem> items = List.of(
                item(1L, "数据库", "Database Systems", "DATA_STORE_CLASS"),
                item(2L, "中文技能", "database technology", "UNRELATED_CODE"),
                item(3L, "代码技能", "Code English", "DATABASE_ENGINE")
        );
        stubCloud(cloud(items));
        doReturn(List.of(alias(1L, "JVM Language", "jvm language")))
                .when(aliasRepository).findAllBySkillIdInOrderByIdAsc(anyCollection());
        saveProfile(USER_ID, json(List.of(
                "jvm-language",
                "JVM LANGUAGE",
                "数据库",
                "DATABASE TECHNOLOGY",
                "database_engine"
        )));

        SkillMatchResponse result = service.match(USER_ID, 1L, 1L, null);

        assertThat(result.matchScore()).isEqualTo(100);
        assertThat(result.matchedCount()).isEqualTo(3);
        assertThat(result.matchedSkills()).extracting(SkillCloudItem::skillId).containsExactly(1L, 2L, 3L);
        assertThat(result.missingSkills()).isEmpty();
    }

    @Test
    void matchesOnlyNameZhForV36JavaWhenOtherIdentityFieldsDoNotMatch() {
        jdbcTemplate.update("UPDATE skills SET code = ? WHERE id = ?", "JAVA_NAME_ZH_ISOLATED", 1L);
        saveProfile(USER_ID, json(List.of("Java")));

        SkillMatchResponse result = service.match(USER_ID, 1L, 2L, 201L);

        assertThat(result.totalCount()).isEqualTo(2);
        assertThat(result.matchScore()).isEqualTo(50);
        assertThat(result.matchedSkills())
                .extracting(SkillCloudItem::skillId)
                .containsExactly(1L);
        assertThat(result.missingSkills())
                .extracting(SkillCloudItem::skillId)
                .containsExactly(2L);
    }

    @Test
    void matchesOnlyInsertedAliasForV36JavaWhenAliasIsTheOnlyIdentityMatch() {
        jdbcTemplate.update("""
                INSERT INTO skill_aliases (id, skill_id, alias, normalized_alias)
                VALUES (?, ?, ?, ?)
                """, 10001L, 1L, "ONLY.ALIAS-JVM-XYZ", "only alias jvm xyz");
        saveProfile(USER_ID, json(List.of("ONLY.ALIAS-JVM-XYZ")));

        SkillMatchResponse result = service.match(USER_ID, 1L, 2L, 201L);

        assertThat(result.totalCount()).isEqualTo(2);
        assertThat(result.matchScore()).isEqualTo(50);
        assertThat(result.matchedSkills())
                .extracting(SkillCloudItem::skillId)
                .containsExactly(1L);
        assertThat(result.missingSkills())
                .extracting(SkillCloudItem::skillId)
                .containsExactly(2L);
    }

    @Test
    void doesNotApplyAliasOwnedByAnotherTopSkillToV36Java() {
        jdbcTemplate.update("""
                INSERT INTO skill_aliases (id, skill_id, alias, normalized_alias)
                VALUES (?, ?, ?, ?)
                """, 10002L, 2L, "ONLY.ALIAS-OWNER-SKILL-2", "only alias owner skill 2");
        saveProfile(USER_ID, json(List.of("ONLY.ALIAS-OWNER-SKILL-2")));

        SkillMatchResponse result = service.match(USER_ID, 1L, 2L, 201L);

        assertThat(result.totalCount()).isEqualTo(2);
        assertThat(result.matchScore()).isEqualTo(50);
        assertThat(result.matchedSkills())
                .extracting(SkillCloudItem::skillId)
                .containsExactly(2L);
        assertThat(result.missingSkills())
                .extracting(SkillCloudItem::skillId)
                .containsExactly(1L);
    }

    @Test
    void keepsMatchedAndMissingListsOrderedAndTheirUnionEqualToTopTen() {
        List<SkillCloudItem> items = List.of(
                item(1L, "Java", "Java", "JAVA"),
                item(2L, "Python", "Python", "PYTHON"),
                item(3L, "SQL", "SQL", "SQL")
        );
        stubCloud(cloud(items));
        saveProfile(USER_ID, json(List.of("Python")));

        SkillMatchResponse result = service.match(USER_ID, 1L, 1L, null);
        List<Long> union = Stream.concat(result.matchedSkills().stream(), result.missingSkills().stream())
                .map(SkillCloudItem::skillId)
                .toList();

        assertThat(result.matchScore()).isEqualTo(33);
        assertThat(result.matchedSkills()).extracting(SkillCloudItem::skillId).containsExactly(2L);
        assertThat(result.missingSkills()).extracting(SkillCloudItem::skillId).containsExactly(1L, 3L);
        assertThat(union).containsExactlyInAnyOrder(1L, 2L, 3L);
    }

    @Test
    void returnsZeroWhenNoTopSkillMatches() {
        List<SkillCloudItem> items = List.of(
                item(1L, "Java", "Java", "JAVA"),
                item(2L, "SQL", "SQL", "SQL")
        );
        stubCloud(cloud(items));
        saveProfile(USER_ID, json(List.of("Marketing")));

        SkillMatchResponse result = service.match(USER_ID, 1L, 1L, null);

        assertThat(result.matchScore()).isZero();
        assertThat(result.matchedSkills()).isEmpty();
        assertThat(result.missingSkills()).extracting(SkillCloudItem::skillId).containsExactly(1L, 2L);
    }

    @Test
    void returnsOneHundredAndNoMissingSkillsWhenAllTopSkillsMatch() {
        List<SkillCloudItem> items = List.of(
                item(1L, "Java", "Java", "JAVA"),
                item(2L, "SQL", "SQL", "SQL")
        );
        stubCloud(cloud(items));
        saveProfile(USER_ID, json(List.of("Java", "SQL")));

        SkillMatchResponse result = service.match(USER_ID, 1L, 1L, null);

        assertThat(result.matchScore()).isEqualTo(100);
        assertThat(result.matchedSkills()).extracting(SkillCloudItem::skillId).containsExactly(1L, 2L);
        assertThat(result.missingSkills()).isEmpty();
    }

    @Test
    void excludesCrossOccupationCategoryMismatchFromMatchSet() {
        jdbcTemplate.update("""
                INSERT INTO occupation_skills (
                    id, occupation_id, skill_id, category_id, importance,
                    importance_source_code, importance_element_id, importance_is_proxy,
                    relationship_type, source_version_id
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                10000L, 1L, 1L, 201L, new BigDecimal("5.00"),
                "15-1254.00", "2.E.7.h", true, "ESSENTIAL", 2L);
        saveProfile(USER_ID, json(List.of("Java")));

        SkillCloudResponse cloud = queryService.query(1L, 1L, null);
        SkillMatchResponse result = service.match(USER_ID, 1L, 1L, null);
        List<Long> expectedTop = cloud.items().stream().limit(10).map(SkillCloudItem::skillId).toList();
        List<Long> actualUnion = Stream.concat(result.matchedSkills().stream(), result.missingSkills().stream())
                .map(SkillCloudItem::skillId)
                .toList();

        assertThat(result.matchedCount()).isZero();
        assertThat(actualUnion).containsExactlyElementsOf(expectedTop);
        assertThat(actualUnion).doesNotContain(1L);
    }

    private void stubCloud(SkillCloudResponse response) {
        doReturn(response).when(queryService).query(any(), any(), any());
    }

    private StudentProfile saveProfile(Long userId, String skillsJson) {
        jdbcTemplate.update("""
                INSERT INTO users (id, phone, role, is_active, is_verified, created_at, updated_at)
                VALUES (?, ?, 'STUDENT', TRUE, FALSE, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
                """, userId, String.valueOf(userId));
        return profileRepository.saveAndFlush(StudentProfile.builder()
                .userId(userId)
                .skillsJson(skillsJson)
                .build());
    }

    private SkillCloudResponse cloud(List<SkillCloudItem> items) {
        return new SkillCloudResponse(
                new SkillCloudNode(1L, "INTERNET", "互联网"),
                new SkillCloudNode(1L, "FRONTEND_DEVELOPER", "前端开发工程师"),
                null,
                items.size(),
                items
        );
    }

    private SkillCloudItem item(Long id, String nameZh, String nameEn, String code) {
        return new SkillCloudItem(
                id.intValue(),
                id,
                code,
                nameZh,
                nameEn,
                "O*NET",
                "test-source",
                1L,
                "测试分类",
                BigDecimal.TEN,
                null,
                null,
                false
        );
    }

    private SkillAlias alias(Long skillId, String alias, String normalizedAlias) {
        return SkillAlias.builder()
                .id(1L)
                .skillId(skillId)
                .alias(alias)
                .normalizedAlias(normalizedAlias)
                .build();
    }

    private String json(List<String> values) {
        return values.stream()
                .map(value -> "\"" + value.replace("\\", "\\\\").replace("\"", "\\\"") + "\"")
                .collect(java.util.stream.Collectors.joining(",", "[", "]"));
    }

    public static class SkillMatchSchemaFilterProvider implements SchemaFilterProvider {

        private static final Set<String> VALIDATED_TABLES = Set.of(
                "skill_source_versions",
                "skill_industries",
                "skill_occupations",
                "skill_categories",
                "skills",
                "skill_aliases",
                "occupation_skills"
        );

        @Override
        public SchemaFilter getCreateFilter() {
            return SchemaFilter.ALL;
        }

        @Override
        public SchemaFilter getDropFilter() {
            return SchemaFilter.ALL;
        }

        @Override
        public SchemaFilter getTruncatorFilter() {
            return SchemaFilter.ALL;
        }

        @Override
        public SchemaFilter getMigrateFilter() {
            return SchemaFilter.ALL;
        }

        @Override
        public SchemaFilter getValidateFilter() {
            return new SchemaFilter() {
                @Override
                public boolean includeNamespace(Namespace namespace) {
                    return true;
                }

                @Override
                public boolean includeTable(Table table) {
                    return VALIDATED_TABLES.contains(table.getName().toLowerCase(Locale.ROOT));
                }

                @Override
                public boolean includeSequence(Sequence sequence) {
                    return false;
                }
            };
        }
    }
}