package com.mydreamjob.skill;

import org.junit.jupiter.api.AfterEach;
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.dao.DataAccessException;
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 javax.sql.DataSource;
import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.TreeMap;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

@SpringBootTest
@TestPropertySource(properties =
        "spring.datasource.url=jdbc:h2:mem:integrationdb_skill_taxonomy;MODE=MySQL;DB_CLOSE_DELAY=-1;DB_CLOSE_ON_EXIT=FALSE")
@ActiveProfiles("integration-test")
class SkillTaxonomyMigrationTest {

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Autowired
    private DataSource dataSource;

    @MockBean
    private RedisTemplate<String, Object> redisTemplate;

    private Long sourceVersionId;
    private Long industryId;
    private Long occupationId;
    private Long categoryId;
    private Long skillId;
    private Long secondSkillId;
    private Long occupationSkillId;

    @Test
    void migrationCreatesTaxonomyTablesColumnsConstraintsAndIndexes() throws SQLException {
        assertTableAndColumns("skill_source_versions",
                "id", "source_name", "source_version", "license", "attribution_url",
                "imported_at", "created_at", "updated_at");
        assertTableAndColumns("skill_industries",
                "id", "code", "name_zh", "source_code", "sort_order", "created_at", "updated_at");
        assertTableAndColumns("skill_occupations",
                "id", "industry_id", "code", "name_zh", "name_en", "source_code",
                "source_version_id", "sort_order", "created_at", "updated_at");
        assertTableAndColumns("skill_categories",
                "id", "occupation_id", "code", "name_zh", "suggestion_template_code",
                "sort_order", "created_at", "updated_at");
        assertTableAndColumns("skills",
                "id", "code", "name_zh", "name_en", "source_type", "source_code",
                "source_version_id", "sort_order", "created_at", "updated_at");
        assertTableAndColumns("skill_aliases",
                "id", "skill_id", "alias", "normalized_alias", "created_at");
        assertTableAndColumns("occupation_skills",
                "id", "occupation_id", "skill_id", "category_id", "importance", "importance_source_code", "importance_element_id", "importance_is_proxy",
                "relationship_type", "source_version_id", "created_at", "updated_at");

        assertUniqueIndex("skill_source_versions", "source_name", "source_version");
        assertUniqueIndex("skill_industries", "code");
        assertUniqueIndex("skill_occupations", "industry_id", "code");
        assertUniqueIndex("skill_categories", "occupation_id", "code");
        assertUniqueIndex("skills", "code");
        assertUniqueIndex("skills", "source_type", "source_code");
        assertUniqueIndex("skill_aliases", "normalized_alias");
        assertUniqueIndex("skill_aliases", "skill_id", "alias");
        assertUniqueIndex("occupation_skills", "occupation_id", "skill_id");

        assertForeignKey("skill_occupations", "industry_id", "skill_industries", "id");
        assertForeignKey("skill_occupations", "source_version_id", "skill_source_versions", "id");
        assertForeignKey("skill_categories", "occupation_id", "skill_occupations", "id");
        assertForeignKey("skills", "source_version_id", "skill_source_versions", "id");
        assertForeignKey("skill_aliases", "skill_id", "skills", "id");
        assertForeignKey("occupation_skills", "occupation_id", "skill_occupations", "id");
        assertForeignKey("occupation_skills", "skill_id", "skills", "id");
        assertForeignKey("occupation_skills", "category_id", "skill_categories", "id");
        assertForeignKey("occupation_skills", "source_version_id", "skill_source_versions", "id");

        assertIndex("occupation_skills", "occupation_id", "category_id");
        assertIndex("occupation_skills", "occupation_id", "importance");
        assertBooleanColumnMetadata("occupation_skills", "importance_is_proxy");
        assertEquals(1L, jdbcTemplate.queryForObject(
                "SELECT COUNT(*) FROM information_schema.table_constraints " +
                        "WHERE UPPER(table_name) = 'OCCUPATION_SKILLS' " +
                        "AND UPPER(constraint_name) = 'CK_OCCUPATION_SKILLS_IMPORTANCE_PROXY' " +
                        "AND UPPER(constraint_type) = 'CHECK'",
                Long.class));
    }

    @Test
    void minimumTaxonomyChainIsUsableAndCoversCoreConstraints() {
        sourceVersionId = insertAndReturnId(
                "INSERT INTO skill_source_versions " +
                        "(source_name, source_version, license, attribution_url, imported_at) " +
                        "VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)",
                "ESCO", "test-1.2.1", "Commission Decision 2011/833/EU",
                "https://esco.ec.europa.eu/en/use-esco/download");
        industryId = insertAndReturnId(
                "INSERT INTO skill_industries (code, name_zh, source_code, sort_order) " +
                        "VALUES (?, ?, ?, ?)",
                "TEST_INTERNET", "互联网", "PROJECT-INTERNET", 1);
        occupationId = insertAndReturnId(
                "INSERT INTO skill_occupations " +
                        "(industry_id, code, name_zh, name_en, source_code, source_version_id, sort_order) " +
                        "VALUES (?, ?, ?, ?, ?, ?, ?)",
                industryId, "TEST_FRONTEND", "前端开发工程师", "Frontend Developer",
                "ESCO-2513.2", sourceVersionId, 1);
        categoryId = insertAndReturnId(
                "INSERT INTO skill_categories " +
                        "(occupation_id, code, name_zh, suggestion_template_code, sort_order) " +
                        "VALUES (?, ?, ?, ?, ?)",
                occupationId, "TEST_PROGRAMMING_LANGUAGE", "编程语言",
                "PROGRAMMING_LANGUAGE", 1);
        skillId = insertAndReturnId(
                "INSERT INTO skills " +
                        "(code, name_zh, name_en, source_type, source_code, source_version_id, sort_order) " +
                        "VALUES (?, ?, ?, ?, ?, ?, ?)",
                "TEST_SKILL_JAVA", "Java", "Java", "ESCO", "ESCO-SKILL-JAVA",
                sourceVersionId, 1);
        insertAndReturnId(
                "INSERT INTO skill_aliases (skill_id, alias, normalized_alias) VALUES (?, ?, ?)",
                skillId, "Java SE", "java se");
        occupationSkillId = insertAndReturnId(
                "INSERT INTO occupation_skills " +
                        "(occupation_id, skill_id, category_id, importance, relationship_type, source_version_id) " +
                        "VALUES (?, ?, ?, ?, ?, ?)",
                occupationId, skillId, categoryId, null, "ESSENTIAL", sourceVersionId);

        Map<String, Object> joined = jdbcTemplate.queryForMap(
                "SELECT s.name_zh AS skill_name, c.name_zh AS category_name, os.importance, " +
                        "os.importance_source_code, os.importance_element_id, os.importance_is_proxy " +
                        "FROM occupation_skills os " +
                        "JOIN skills s ON s.id = os.skill_id " +
                        "JOIN skill_categories c ON c.id = os.category_id " +
                        "WHERE os.id = ?", occupationSkillId);
        assertEquals("Java", joined.get("skill_name"));
        assertEquals("编程语言", joined.get("category_name"));
        assertNull(joined.get("importance"));
        assertNull(joined.get("importance_source_code"));
        assertNull(joined.get("importance_element_id"));
        assertEquals(Boolean.FALSE, joined.get("importance_is_proxy"));
        assertEquals(1L, jdbcTemplate.queryForObject(
                "SELECT COUNT(*) FROM occupation_skills WHERE id = ? AND importance_is_proxy = FALSE",
                Long.class, occupationSkillId));

        assertConstraintFailure("UPDATE occupation_skills SET importance = ? WHERE id = ?",
                new BigDecimal("4.25"), occupationSkillId);
        jdbcTemplate.update("UPDATE occupation_skills SET importance = ?, " +
                        "importance_source_code = ?, importance_element_id = ?, importance_is_proxy = ? " +
                        "WHERE id = ?",
                new BigDecimal("4.25"), "15-1252.00", "4.A.3.b.1", Boolean.TRUE, occupationSkillId);
        Map<String, Object> weighted = jdbcTemplate.queryForMap(
                "SELECT importance, importance_source_code, importance_element_id, importance_is_proxy " +
                        "FROM occupation_skills WHERE id = ?",
                occupationSkillId);
        assertEquals(new BigDecimal("4.25"), weighted.get("importance"));
        assertEquals("15-1252.00", weighted.get("importance_source_code"));
        assertEquals("4.A.3.b.1", weighted.get("importance_element_id"));
        assertEquals(Boolean.TRUE, weighted.get("importance_is_proxy"));
        assertEquals(1L, jdbcTemplate.queryForObject(
                "SELECT COUNT(*) FROM occupation_skills WHERE id = ? AND importance_is_proxy = TRUE",
                Long.class, occupationSkillId));
        assertConstraintFailure(
                "INSERT INTO skill_source_versions " +
                        "(source_name, source_version, license, attribution_url) VALUES (?, ?, ?, ?)",
                "ESCO", "test-1.2.1", "duplicate", "https://example.invalid");
        assertConstraintFailure(
                "INSERT INTO skill_categories " +
                        "(occupation_id, code, name_zh, suggestion_template_code, sort_order) " +
                        "VALUES (?, ?, ?, ?, ?)",
                occupationId, "TEST_INVALID_CATEGORY", "非法分类", "INVALID_TEMPLATE", 2);
        assertConstraintFailure(
                "INSERT INTO skills " +
                        "(code, name_zh, source_type, source_code, source_version_id, sort_order) " +
                        "VALUES (?, ?, ?, ?, ?, ?)",
                "TEST_INVALID_SKILL", "非法技能", "INVALID_SOURCE", "INVALID-SOURCE-1",
                sourceVersionId, 2);
        secondSkillId = insertAndReturnId(
                "INSERT INTO skills " +
                        "(code, name_zh, source_type, source_code, source_version_id, sort_order) " +
                        "VALUES (?, ?, ?, ?, ?, ?)",
                "TEST_SKILL_SPRING", "Spring Boot", "ESCO", "ESCO-SKILL-SPRING",
                sourceVersionId, 2);
        assertConstraintFailure(
                "INSERT INTO skills " +
                        "(code, name_zh, source_type, source_code, source_version_id, sort_order) " +
                        "VALUES (?, ?, ?, ?, ?, ?)",
                "TEST_SKILL_NULL_NAME", null, "PROJECT", "PROJECT-NULL-NAME",
                sourceVersionId, 3);
        assertConstraintFailure(
                "INSERT INTO occupation_skills " +
                        "(occupation_id, skill_id, category_id, relationship_type, source_version_id) " +
                        "VALUES (?, ?, ?, ?, ?)",
                occupationId, secondSkillId, categoryId, "INVALID_RELATIONSHIP", sourceVersionId);
        assertConstraintFailure(
                "INSERT INTO skill_aliases (skill_id, alias, normalized_alias) VALUES (?, ?, ?)",
                secondSkillId, "Java SE alternate", "java se");

        assertConstraintFailure(
                "UPDATE occupation_skills SET importance_is_proxy = ? WHERE id = ?",
                (Object) null, occupationSkillId);
        assertTrue(jdbcTemplate.queryForObject(
                "SELECT COUNT(*) FROM skill_aliases WHERE skill_id = ? AND normalized_alias = ?",
                Integer.class, skillId, "java se") == 1);
    }

    @AfterEach
    void cleanUp() {
        if (occupationSkillId != null) {
            jdbcTemplate.update("DELETE FROM occupation_skills WHERE id = ?", occupationSkillId);
        }
        if (secondSkillId != null) {
            jdbcTemplate.update("DELETE FROM skill_aliases WHERE skill_id = ?", secondSkillId);
            jdbcTemplate.update("DELETE FROM skills WHERE id = ?", secondSkillId);
        }
        if (skillId != null) {
            jdbcTemplate.update("DELETE FROM skill_aliases WHERE skill_id = ?", skillId);
            jdbcTemplate.update("DELETE FROM skills WHERE id = ?", skillId);
        }
        if (categoryId != null) {
            jdbcTemplate.update("DELETE FROM skill_categories WHERE id = ?", categoryId);
        }
        if (occupationId != null) {
            jdbcTemplate.update("DELETE FROM skill_occupations WHERE id = ?", occupationId);
        }
        if (industryId != null) {
            jdbcTemplate.update("DELETE FROM skill_industries WHERE id = ?", industryId);
        }
        if (sourceVersionId != null) {
            jdbcTemplate.update("DELETE FROM skill_source_versions WHERE id = ?", sourceVersionId);
        }
    }

    private void assertBooleanColumnMetadata(String table, String column) {
        Map<String, Object> metadata = jdbcTemplate.queryForMap(
                "SELECT data_type, column_default, is_nullable FROM information_schema.columns " +
                        "WHERE UPPER(table_name) = ? AND UPPER(column_name) = ?",
                table.toUpperCase(Locale.ROOT), column.toUpperCase(Locale.ROOT));
        assertEquals("BOOLEAN", String.valueOf(metadata.get("data_type")).toUpperCase(Locale.ROOT));
        assertEquals("FALSE", String.valueOf(metadata.get("column_default")).toUpperCase(Locale.ROOT));
        assertEquals("NO", String.valueOf(metadata.get("is_nullable")).toUpperCase(Locale.ROOT));
    }

    private void assertTableAndColumns(String table, String... columns) throws SQLException {
        try (Connection connection = dataSource.getConnection()) {
            DatabaseMetaData metadata = connection.getMetaData();
            String actualTable = findTable(metadata, table);
            assertTrue(actualTable != null, "Missing table: " + table);

            List<String> actualColumns = new ArrayList<>();
            try (ResultSet resultSet = metadata.getColumns(null, null, actualTable, null)) {
                while (resultSet.next()) {
                    actualColumns.add(resultSet.getString("COLUMN_NAME").toLowerCase());
                }
            }
            for (String column : columns) {
                assertTrue(actualColumns.contains(column.toLowerCase()),
                        "Missing column " + table + "." + column + "; actual=" + actualColumns);
            }
        }
    }

    private void assertUniqueIndex(String table, String... columns) throws SQLException {
        assertTrue(hasIndex(table, true, columns),
                "Missing unique index on " + table + "(" + String.join(", ", columns) + ")");
    }

    private void assertIndex(String table, String... columns) throws SQLException {
        assertTrue(hasIndex(table, false, columns),
                "Missing index on " + table + "(" + String.join(", ", columns) + ")");
    }

    private boolean hasIndex(String table, boolean unique, String... expectedColumns) throws SQLException {
        try (Connection connection = dataSource.getConnection()) {
            DatabaseMetaData metadata = connection.getMetaData();
            String actualTable = findTable(metadata, table);
            if (actualTable == null) {
                return false;
            }
            Map<String, TreeMap<Short, String>> indexes = new LinkedHashMap<>();
            try (ResultSet resultSet = metadata.getIndexInfo(null, null, actualTable, false, false)) {
                while (resultSet.next()) {
                    String indexName = resultSet.getString("INDEX_NAME");
                    if (indexName == null || resultSet.getBoolean("NON_UNIQUE") == unique) {
                        continue;
                    }
                    indexes.computeIfAbsent(indexName, ignored -> new TreeMap<>())
                            .put(resultSet.getShort("ORDINAL_POSITION"),
                                    resultSet.getString("COLUMN_NAME").toLowerCase());
                }
            }
            List<String> expected = new ArrayList<>();
            for (String column : expectedColumns) {
                expected.add(column.toLowerCase());
            }
            return indexes.values().stream()
                    .map(TreeMap::values)
                    .map(values -> new ArrayList<>(values))
                    .anyMatch(expected::equals);
        }
    }

    private void assertForeignKey(String childTable, String childColumn,
                                  String parentTable, String parentColumn) throws SQLException {
        try (Connection connection = dataSource.getConnection()) {
            DatabaseMetaData metadata = connection.getMetaData();
            String actualChildTable = findTable(metadata, childTable);
            String actualParentTable = findTable(metadata, parentTable);
            if (actualChildTable == null || actualParentTable == null) {
                throw new AssertionError("Missing table for foreign key " + childTable + "." + childColumn);
            }
            try (ResultSet resultSet = metadata.getImportedKeys(null, null, actualChildTable)) {
                while (resultSet.next()) {
                    if (actualParentTable.equalsIgnoreCase(resultSet.getString("PKTABLE_NAME"))
                            && childColumn.equalsIgnoreCase(resultSet.getString("FKCOLUMN_NAME"))
                            && parentColumn.equalsIgnoreCase(resultSet.getString("PKCOLUMN_NAME"))) {
                        return;
                    }
                }
            }
        }
        throw new AssertionError("Missing foreign key " + childTable + "." + childColumn
                + " -> " + parentTable + "." + parentColumn);
    }

    private String findTable(DatabaseMetaData metadata, String expected) throws SQLException {
        try (ResultSet resultSet = metadata.getTables(null, null, null, new String[]{"TABLE"})) {
            while (resultSet.next()) {
                String actual = resultSet.getString("TABLE_NAME");
                if (expected.equalsIgnoreCase(actual)) {
                    return actual;
                }
            }
        }
        return null;
    }

    private Long insertAndReturnId(String sql, Object... args) {
        jdbcTemplate.update(sql, args);
        String table = extractTable(sql);
        return jdbcTemplate.queryForObject("SELECT MAX(id) FROM " + table, Long.class);
    }

    private String extractTable(String sql) {
        String upper = sql.toUpperCase();
        int fromIndex = upper.indexOf(" INTO ");
        int openIndex = sql.indexOf('(', fromIndex);
        return sql.substring(fromIndex + " INTO ".length(), openIndex).trim();
    }

    private void assertConstraintFailure(String sql, Object... args) {
        assertThrows(DataAccessException.class, () -> jdbcTemplate.update(sql, args));
    }
}