package com.mydreamjob.skill;

import com.mydreamjob.common.exception.BusinessException;
import com.mydreamjob.common.exception.ErrorCode;
import com.mydreamjob.skill.domain.SkillIndustry;
import com.mydreamjob.skill.domain.SkillOccupation;
import com.mydreamjob.skill.dto.SkillCloudItem;
import com.mydreamjob.skill.dto.SkillCloudResponse;
import com.mydreamjob.skill.dto.SkillCloudSkillProjection;
import org.assertj.core.api.ThrowableAssert.ThrowingCallable;
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.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.mock.mockito.MockBean;
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 com.mydreamjob.skill.repository.OccupationSkillRepository;
import com.mydreamjob.skill.repository.SkillCategoryRepository;
import com.mydreamjob.skill.repository.SkillIndustryRepository;
import com.mydreamjob.skill.repository.SkillOccupationRepository;
import org.springframework.transaction.annotation.Transactional;

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

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.catchThrowableOfType;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

@SpringBootTest
@TestPropertySource(properties = {
        "spring.datasource.url=jdbc:h2:mem:integrationdb_skill_query;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.SkillCloudQueryServiceTest$SkillQuerySchemaFilterProvider"
})
@ActiveProfiles("integration-test")
@Transactional
class SkillCloudQueryServiceTest {

    @Autowired
    private SkillCloudQueryService service;

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @MockBean
    private RedisTemplate<String, Object> redisTemplate;

    @Test
    void returnsAllSeedSkillsForIndustryAndOccupationWithContinuousRanks() {
        SkillCloudResponse response = service.query(1L, 1L, null);

        assertThat(response.industry().id()).isEqualTo(1L);
        assertThat(response.industry().code()).isEqualTo("INTERNET");
        assertThat(response.industry().name()).isEqualTo("互联网/信息技术");
        assertThat(response.occupation().id()).isEqualTo(1L);
        assertThat(response.occupation().code()).isEqualTo("FRONTEND_DEVELOPER");
        assertThat(response.occupation().name()).isEqualTo("前端开发工程师");
        assertThat(response.category()).isNull();
        assertThat(response.total()).isEqualTo(15);
        assertThat(response.items()).hasSize(15);
        assertThat(response.items())
                .extracting(SkillCloudItem::rank)
                .containsExactlyElementsOf(IntStream.rangeClosed(1, 15).boxed().toList());
        assertThat(response.items())
                .extracting(SkillCloudItem::skillId)
                .doesNotHaveDuplicates();
    }

    @Test
    void sortsByHeatThenNameAndKeepsNullHeatLast() {
        SkillCloudResponse response = service.query(1L, 1L, 104L);

        assertThat(response.category().id()).isEqualTo(104L);
        assertThat(response.category().code()).isEqualTo("TOOLS");
        assertThat(response.category().name()).isEqualTo("工程化工具");
        assertThat(response.items())
                .extracting(SkillCloudItem::nameZh)
                .containsExactly("Gradle", "Node.js", "Figma");
        assertThat(response.items())
                .extracting(SkillCloudItem::heat)
                .containsExactly(new BigDecimal("4.88"), new BigDecimal("4.88"), null);
        assertThat(response.items().get(response.items().size() - 1).heat()).isNull();

        SkillCloudResponse all = service.query(1L, 1L, null);
        assertThat(all.items().stream()
                .map(SkillCloudItem::heat)
                .filter(java.util.Objects::nonNull)
                .toList())
                .isSortedAccordingTo(java.util.Comparator.reverseOrder());
        assertThat(all.items().get(all.items().size() - 1).heat()).isNull();
        assertThat(all.items().subList(0, all.items().size() - 1))
                .allSatisfy(item -> assertThat(item.heat()).isNotNull());
    }

    @Test
    void comparatorUsesSkillIdWhenEqualKeysArriveInReverseOrder() {
        SkillIndustry industry = new SkillIndustry();
        industry.setId(1L);
        industry.setCode("INTERNET");
        industry.setNameZh("互联网");

        SkillOccupation occupation = new SkillOccupation();
        occupation.setId(1L);
        occupation.setIndustryId(1L);
        occupation.setCode("FRONTEND_DEVELOPER");
        occupation.setNameZh("前端开发工程师");

        SkillIndustryRepository industryRepository = mock(SkillIndustryRepository.class);
        when(industryRepository.findById(1L)).thenReturn(Optional.of(industry));

        SkillOccupationRepository occupationRepository = mock(SkillOccupationRepository.class);
        when(occupationRepository.findById(1L)).thenReturn(Optional.of(occupation));

        OccupationSkillRepository occupationSkillRepository = mock(OccupationSkillRepository.class);
        when(occupationSkillRepository.findSkillProjectionsByOccupationId(1L))
                .thenReturn(List.of(
                        projection(9002L, "TIE_SKILL_HIGH"),
                        projection(9001L, "TIE_SKILL_LOW")));

        SkillCloudQueryService isolatedService = new SkillCloudQueryService(
                industryRepository,
                occupationRepository,
                mock(SkillCategoryRepository.class),
                occupationSkillRepository);

        SkillCloudResponse response = isolatedService.query(1L, 1L, null);

        assertThat(response.items())
                .extracting(SkillCloudItem::skillId)
                .containsExactly(9001L, 9002L);
    }

    private SkillCloudSkillProjection projection(Long skillId, String skillCode) {
        return new SkillCloudSkillProjection(
                skillId,
                skillCode,
                "同名技能",
                "Tie Skill",
                "O*NET",
                "TEST|Tie Skill",
                104L,
                "工程化工具",
                new BigDecimal("4.88"),
                "15-1254.00",
                "4.A.3.b.1",
                true,
                1,
                1
        );
    }

    @Test
    void categoryFilterReturnsOnlyTheRequestedCategoryAndOmittedCategoryReturnsMore() {
        SkillCloudResponse all = service.query(1L, 1L, null);
        SkillCloudResponse filtered = service.query(1L, 1L, 101L);

        assertThat(filtered.total()).isEqualTo(2);
        assertThat(filtered.items())
                .extracting(SkillCloudItem::categoryId)
                .containsOnly(101L);
        assertThat(filtered.items())
                .extracting(SkillCloudItem::skillId)
                .containsExactlyInAnyOrder(3L, 4L);
        assertThat(all.total()).isGreaterThan(filtered.total());
    }

    @Test
    void returnsEmptySuccessForLegalCategoryWithoutSkills() {
        jdbcTemplate.update("""
                INSERT INTO skill_categories
                    (id, occupation_id, code, name_zh, suggestion_template_code, sort_order)
                VALUES
                    (9101, 1, 'EMPTY_CATEGORY', '空分类', 'GENERAL', 99)
                """);

        SkillCloudResponse response = service.query(1L, 1L, 9101L);

        assertThat(response.category().id()).isEqualTo(9101L);
        assertThat(response.category().name()).isEqualTo("空分类");
        assertThat(response.total()).isZero();
        assertThat(response.items()).isEmpty();
    }

    @Test
    void mapsSkillAndHeatProvenanceFieldsFromTheDatabase() {
        SkillCloudResponse frameworks = service.query(1L, 1L, 103L);
        SkillCloudItem react = frameworks.items().stream()
                .filter(item -> item.skillId().equals(26L))
                .findFirst()
                .orElseThrow();

        Map<String, Object> row = jdbcTemplate.queryForMap("""
                SELECT os.importance, os.importance_source_code, os.importance_element_id,
                       os.importance_is_proxy
                FROM occupation_skills os
                WHERE os.occupation_id = 1 AND os.skill_id = 26
                """);

        assertThat(react.rank()).isEqualTo(1);
        assertThat(react.skillCode()).isEqualTo("REACT");
        assertThat(react.nameZh()).isEqualTo("React");
        assertThat(react.nameEn()).isEqualTo("React");
        assertThat(react.categoryId()).isEqualTo(103L);
        assertThat(react.categoryName()).isEqualTo("前端框架");
        assertThat(react.sourceType()).isEqualTo("O*NET");
        assertThat(react.sourceCode()).isEqualTo("15-1254.00|React|2.E.7.h");
        assertThat(react.heat()).isEqualByComparingTo((BigDecimal) row.get("importance"));
        assertThat(react.heatSourceCode()).isEqualTo(row.get("importance_source_code"));
        assertThat(react.heatElementId()).isEqualTo(row.get("importance_element_id"));
        assertThat(react.heatProxy()).isEqualTo(toBoolean(row.get("importance_is_proxy")));

        SkillCloudResponse languages = service.query(1L, 1L, 101L);
        SkillCloudItem javascript = languages.items().stream()
                .filter(item -> item.skillId().equals(3L))
                .findFirst()
                .orElseThrow();

        assertThat(javascript.sourceType()).isEqualTo("ESCO");
        assertThat(javascript.sourceCode())
                .isEqualTo("http://data.europa.eu/esco/skill/3cd569a2-4f88-4c1e-9995-8dce8c5e51a7");
        assertThat(javascript.heatSourceCode()).isEqualTo("15-1254.00");
    }

    @Test
    void rejectsMissingOrNonPositiveIdentifiersWithBadRequest() {
        assertBadRequest(() -> service.query(null, 1L, null), "industryId");
        assertBadRequest(() -> service.query(0L, 1L, null), "industryId");
        assertBadRequest(() -> service.query(-1L, 1L, null), "industryId");
        assertBadRequest(() -> service.query(1L, null, null), "occupationId");
        assertBadRequest(() -> service.query(1L, 0L, null), "occupationId");
        assertBadRequest(() -> service.query(1L, -1L, null), "occupationId");
        assertBadRequest(() -> service.query(1L, 1L, 0L), "categoryId");
        assertBadRequest(() -> service.query(1L, 1L, -1L), "categoryId");
    }

    @Test
    void rejectsUnknownOrMismatchedTaxonomyNodesWithBadRequest() {
        assertBadRequest(() -> service.query(999L, 1L, null), "industry not found");
        assertBadRequest(() -> service.query(1L, 999L, null), "occupation not found");
        assertBadRequest(() -> service.query(2L, 1L, null), "occupation does not belong");
        assertBadRequest(() -> service.query(1L, 1L, 201L), "category does not belong");
        assertBadRequest(() -> service.query(1L, 1L, 9999L), "category not found");
    }

    private boolean toBoolean(Object value) {
        if (value instanceof Boolean booleanValue) {
            return booleanValue;
        }
        return ((Number) value).intValue() != 0;
    }

    private void assertBadRequest(ThrowingCallable action, String detailFragment) {
        BusinessException exception = catchThrowableOfType(action, BusinessException.class);
        assertThat(exception).isNotNull();
        assertThat(exception.getErrorCodeInt()).isEqualTo(ErrorCode.BAD_REQUEST.getCode());
        assertThat(exception.getErrorMessage()).contains(detailFragment);
    }

    public static class SkillQuerySchemaFilterProvider 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;
                }
            };
        }
    }
}