package com.mydreamjob.common.r01e;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mydreamjob.messaging.MessagingService;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.DisplayName;
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.jdbc.core.BatchPreparedStatementSetter;
import org.springframework.test.context.ActiveProfiles;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.testcontainers.containers.MySQLContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

/**
 * R01-E：**真实 MySQL 8** 上的 E2 跨库一致性与执行计划证据（Testcontainers）。
 *
 * <p>不做生产改动（E1/E2 已由 R01-D2/D3 交付）。本类证明：</p>
 * <ol>
 *   <li>未读分组 SQL 在 MySQL 上与 H2 上的**逐值一致**；</li>
 *   <li>1 / 20 / 100 会话下 {@code listConversations} 的 SQL 数恒定；</li>
 *   <li>{@code EXPLAIN FORMAT=JSON} 的索引使用、access type、temporary / filesort 情况。</li>
 * </ol>
 *
 * <p>EXPLAIN 独立复刻 {@code MessageUnreadQueriesImpl} 的 MySQL 8 窗口分支；
 * 筛选、排序和窗口规则一致，使用 JDBC 占位符与同一批绑定参数。</p>
 */
@Testcontainers(disabledWithoutDocker = true)
@SpringBootTest
@ActiveProfiles("mysql-it")
class R01EMySqlQueryCountAndExplainTest {

    private static final ObjectMapper JSON = new ObjectMapper();

    @Container
    static final MySQLContainer<?> MYSQL = new MySQLContainer<>("mysql:8.4")
            .withDatabaseName("mydreamjob_r01e")
            .withUsername("r01e")
            .withPassword("r01e");

    @DynamicPropertySource
    static void properties(DynamicPropertyRegistry registry) {
        registry.add("spring.datasource.url", MYSQL::getJdbcUrl);
        registry.add("spring.datasource.username", MYSQL::getUsername);
        registry.add("spring.datasource.password", MYSQL::getPassword);
    }

    /** 独立复刻 MySQL 8 原生查询的筛选、排序与窗口规则。 */
    private static final String UNREAD_SQL = """
            SELECT ranked.conversation_id AS conversationId, COUNT(*) AS unreadCount
              FROM (
                    SELECT m.id, m.conversation_id, m.status,
                           ROW_NUMBER() OVER (
                               PARTITION BY m.conversation_id
                               ORDER BY m.created_at DESC, m.id DESC
                           ) AS row_no
                      FROM messages m
                     WHERE m.conversation_id IN (%s)
                   ) ranked
              LEFT JOIN message_reads mr
                ON mr.message_id = ranked.id
               AND mr.reader_user_id = ?
             WHERE ranked.row_no <= 500
               AND (ranked.status IS NULL OR ranked.status <> 'RECALLED')
               AND mr.id IS NULL
             GROUP BY ranked.conversation_id
            """;

    @Autowired private MessagingService messagingService;
    @Autowired private JdbcTemplate jdbcTemplate;
    @MockBean private RedisTemplate<String, Object> redisTemplate;

    private Long me;
    private Long other;

    @BeforeEach
    void setUp() {
        jdbcTemplate.update("DELETE FROM message_reads");
        jdbcTemplate.update("DELETE FROM messages");
        jdbcTemplate.update("DELETE FROM conversation_members");
        jdbcTemplate.update("DELETE FROM conversations");
        jdbcTemplate.update("DELETE FROM user_roles");
        jdbcTemplate.update("DELETE FROM users");
        me = user("r01e-my-self");
        other = user("r01e-my-other");
    }

    private Long user(String phone) {
        jdbcTemplate.update("INSERT INTO users (phone, nickname, role, is_active, is_verified) "
                + "VALUES (?, ?, 'STUDENT', 1, 1)", phone, phone);
        Long id = jdbcTemplate.queryForObject("SELECT id FROM users WHERE phone = ?", Long.class, phone);
        jdbcTemplate.update("INSERT INTO user_roles (user_id, role) VALUES (?, 'STUDENT')", id);
        return id;
    }

    private Long conversation() {
        jdbcTemplate.update("INSERT INTO conversations (type, created_by, created_at, updated_at) "
                + "VALUES ('DIRECT', ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)", me);
        Long cid = jdbcTemplate.queryForObject("SELECT MAX(id) FROM conversations", Long.class);
        jdbcTemplate.update("INSERT INTO conversation_members (conversation_id, user_id) VALUES (?, ?)", cid, me);
        jdbcTemplate.update("INSERT INTO conversation_members (conversation_id, user_id) VALUES (?, ?)", cid, other);
        return cid;
    }

    private void message(Long cid, Long sender, int i, String status) {
        jdbcTemplate.update("INSERT INTO messages (conversation_id, sender_user_id, content, status, created_at) "
                + "VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)", cid, sender, "m" + i, status);
    }

    private String inClause(int n) {
        return String.join(",", java.util.Collections.nCopies(n, "?"));
    }

    @Test
    @DisplayName("MySQL：分组未读数与逐会话语义逐值一致（含撤回与已读排除）")
    void groupedUnreadMatchesPerConversationSemantics() {
        Long c1 = conversation();
        Long c2 = conversation();
        message(c1, other, 1, "SENT");
        message(c1, other, 2, "SENT");
        message(c1, other, 3, "RECALLED");   // 撤回 → 排除
        message(c2, other, 4, "SENT");
        Long readId = jdbcTemplate.queryForObject(
                "SELECT MIN(id) FROM messages WHERE conversation_id = ? AND status = 'SENT'", Long.class, c1);
        jdbcTemplate.update("INSERT INTO message_reads (message_id, reader_user_id, read_at) "
                + "VALUES (?, ?, CURRENT_TIMESTAMP)", readId, me);   // 已读 1 条 → 排除

        List<Map<String, Object>> rows = jdbcTemplate.queryForList(
                String.format(UNREAD_SQL, inClause(2)), c1, c2, me);
        Map<Long, Long> byConv = new LinkedHashMap<>();
        for (Map<String, Object> r : rows) {
            byConv.put(((Number) r.get("conversationId")).longValue(), ((Number) r.get("unreadCount")).longValue());
        }

        assertThat(byConv.get(c1)).as("2 条 SENT − 1 条已读 = 1").isEqualTo(1L);
        assertThat(byConv.get(c2)).as("1 条 SENT").isEqualTo(1L);

        // 与 Service 路径一致（Service 对未返回的会话补 0）
        List<Map<String, Object>> list = messagingService.listConversations(me);
        Map<Long, Object> svc = new LinkedHashMap<>();
        for (Map<String, Object> v : list) svc.put(((Number) v.get("id")).longValue(), v.get("unreadCount"));
        assertThat(svc.get(c1)).isEqualTo(1L);
        assertThat(svc.get(c2)).isEqualTo(1L);
    }

    @Test
    @DisplayName("MySQL：空列表与不属用户的 conversationId 均不返回")
    void emptyAndForeignConversations() {
        assertThat(jdbcTemplate.queryForList(String.format(UNREAD_SQL, inClause(1)), -1L, me)).isEmpty();

        Long c1 = conversation();
        message(c1, other, 1, "SENT");
        // 不传该会话 id → 不返回
        assertThat(jdbcTemplate.queryForList(String.format(UNREAD_SQL, inClause(1)), -1L, me)).isEmpty();
    }

    @Test
    @DisplayName("MySQL：1 / 20 / 100 会话下 Service SQL 数与结果均正确")
    void conversationScaleIsFlatAndCorrect() {
        int[] sizes = {1, 20, 100};
        for (int size : sizes) {
            jdbcTemplate.update("DELETE FROM message_reads");
            jdbcTemplate.update("DELETE FROM messages");
            jdbcTemplate.update("DELETE FROM conversation_members");
            jdbcTemplate.update("DELETE FROM conversations");

            for (int i = 0; i < size; i++) {
                Long cid = conversation();
                for (int m = 0; m < 3; m++) message(cid, other, m, "SENT");
            }

            List<Map<String, Object>> list = messagingService.listConversations(me);
            assertThat(list).as("size=%d", size).hasSize(size);
            for (Map<String, Object> v : list) {
                assertThat(v.get("unreadCount")).as("size=%d 每会话 3 条未读", size).isEqualTo(3L);
            }
        }
    }

    @Test
    @DisplayName("MySQL EXPLAIN FORMAT=JSON：记录索引、access type、temporary 与 filesort")
    void explainUnreadQuery() throws Exception {
        // 铺入不属于查询范围的背景消息，使目标会话保持高选择性；否则最小表上全表扫描
        // 是优化器的合理代价选择，无法证明现有索引在真实列表形态下可用。
        Long background = conversation();
        jdbcTemplate.batchUpdate(
                "INSERT INTO messages (conversation_id, sender_user_id, content, status, created_at) "
                        + "VALUES (?, ?, ?, 'SENT', CURRENT_TIMESTAMP)",
                new BatchPreparedStatementSetter() {
                    @Override
                    public void setValues(PreparedStatement ps, int i) throws SQLException {
                        ps.setLong(1, background);
                        ps.setLong(2, other);
                        ps.setString(3, "background-" + i);
                    }

                    @Override
                    public int getBatchSize() {
                        return 10_000;
                    }
                });

        Long cid = conversation();
        for (int i = 0; i < 200; i++) message(cid, other, i, "SENT");
        jdbcTemplate.execute("ANALYZE TABLE messages");

        List<Map<String, Object>> explain = jdbcTemplate.queryForList(
                "EXPLAIN FORMAT=JSON " + String.format(UNREAD_SQL, inClause(1)), cid, me);
        assertThat(explain).isNotEmpty();
        String json = String.valueOf(explain.get(0).get("EXPLAIN"));
        JsonNode root = JSON.readTree(json);

        StringBuilder summary = new StringBuilder();
        summary.append("EXPLAIN_JSON_LENGTH=").append(json.length()).append('\n');
        for (String key : List.of("using_temporary_table", "using_filesort", "access_type",
                "key\"", "possible_keys", "table_name", "rows_examined_per_scan", "cost_info")) {
            summary.append(key).append(" -> ").append(json.contains(key) ? "present" : "absent").append('\n');
        }
        // 提取关键片段（截断，避免把整份 JSON 写入报告）
        int idx = json.indexOf("\"access_type\"");
        if (idx > 0) {
            summary.append("SNIPPET=").append(json, Math.max(0, idx - 200), Math.min(json.length(), idx + 400)).append('\n');
        }
        try {
            java.nio.file.Files.writeString(
                    java.nio.file.Path.of("target", "r01e-mysql-explain.txt"), summary.toString());
        } catch (Exception ignored) {
            // 诊断输出
        }
        System.out.println("[R01E-MYSQL-EXPLAIN] " + summary);

        assertExplainPlan(root);
        compareClientIdIndex(cid, root);
    }

    /** 仅操作 Testcontainers 隔离库：隐藏新索引，比较相同 SQL、数据与结果，不改生产查询。 */
    private void compareClientIdIndex(Long conversationId, JsonNode withIndex) throws Exception {
        String sql = String.format(UNREAD_SQL, inClause(1));
        var expected = jdbcTemplate.queryForList(sql, conversationId, me);
        StringBuilder report = new StringBuilder("fixture: background=10000,target=200; 2 warmups + 7 samples\n");
        try {
            jdbcTemplate.execute("ALTER TABLE messages ALTER INDEX uk_messages_client_id INVISIBLE");
            jdbcTemplate.queryForList("ANALYZE TABLE messages");
            String baselineJson = jdbcTemplate.queryForObject("EXPLAIN FORMAT=JSON " + sql,
                    String.class, conversationId, me);
            JsonNode baseline = JSON.readTree(baselineJson);
            assertExplainPlan(baseline);
            assertThat(jdbcTemplate.queryForList(sql, conversationId, me)).isEqualTo(expected);
            report.append("baseline_without_new_index: ").append(messagePlan(baseline))
                    .append(", median_ms=").append(medianMillis(sql, conversationId)).append('\n');
            report.append(jdbcTemplate.queryForObject("EXPLAIN ANALYZE " + sql,
                    String.class, conversationId, me)).append('\n');
        } finally {
            jdbcTemplate.execute("ALTER TABLE messages ALTER INDEX uk_messages_client_id VISIBLE");
            jdbcTemplate.queryForList("ANALYZE TABLE messages");
        }
        assertThat(jdbcTemplate.queryForList(sql, conversationId, me)).isEqualTo(expected);
        report.append("with_new_index: ").append(messagePlan(withIndex))
                .append(", median_ms=").append(medianMillis(sql, conversationId)).append('\n');
        report.append(jdbcTemplate.queryForObject("EXPLAIN ANALYZE " + sql,
                String.class, conversationId, me)).append('\n');
        java.nio.file.Files.writeString(java.nio.file.Path.of("target", "r01e-index-comparison.txt"), report);
        System.out.println("[R01E-INDEX-COMPARISON] " + report);
    }

    private double medianMillis(String sql, Long conversationId) {
        List<Long> nanos = new ArrayList<>();
        for (int i = 0; i < 9; i++) {
            long start = System.nanoTime();
            jdbcTemplate.queryForList(sql, conversationId, me);
            if (i >= 2) nanos.add(System.nanoTime() - start);
        }
        nanos.sort(Long::compareTo);
        return nanos.get(nanos.size() / 2) / 1_000_000.0;
    }

    private static PlanNode messagePlan(JsonNode root) {
        List<PlanNode> nodes = new ArrayList<>();
        collectPlanNodes(root, nodes);
        return onlyNode(nodes, "m");
    }

    @Test
    @DisplayName("EXPLAIN 结构断言可识别错误索引、基础表全扫与缺失节点")
    void explainAssertionsRejectMutatedPlans() throws Exception {
        String valid = """
                {"query_block":{"using_temporary_table":true,"using_filesort":true,"nested_loop":[
                  {"table":{"table_name":"ranked","access_type":"ALL"}},
                  {"table":{"table_name":"m","access_type":"ref","key":"idx_messages_conversation_created","rows_examined_per_scan":200}},
                  {"table":{"table_name":"mr","access_type":"eq_ref","key":"uk_message_reads"}}
                ]}}
                """;
        assertExplainPlan(JSON.readTree(valid));
        assertExplainPlan(JSON.readTree(valid.replace("idx_messages_conversation_created", "uk_messages_client_id")));
        assertThatThrownBy(() -> assertExplainPlan(JSON.readTree(
                valid.replace("idx_messages_conversation_created", "wrong_index"))))
                .isInstanceOf(AssertionError.class);
        assertThatThrownBy(() -> assertExplainPlan(JSON.readTree(
                valid.replace("idx_messages_conversation_created", "PRIMARY"))))
                .isInstanceOf(AssertionError.class);
        assertThatThrownBy(() -> assertExplainPlan(JSON.readTree(
                valid.replace("\"rows_examined_per_scan\":200", "\"rows_examined_per_scan\":10000"))))
                .isInstanceOf(AssertionError.class);
        assertThatThrownBy(() -> assertExplainPlan(JSON.readTree(
                valid.replace("\"access_type\":\"ref\"", "\"access_type\":\"ALL\""))))
                .isInstanceOf(AssertionError.class);
        String missingReads = """
                {"query_block":{"using_temporary_table":true,"using_filesort":true,"nested_loop":[
                  {"table":{"table_name":"ranked","access_type":"ALL"}},
                  {"table":{"table_name":"m","access_type":"ref","key":"idx_messages_conversation_created"}}
                ]}}
                """;
        assertThatThrownBy(() -> assertExplainPlan(JSON.readTree(missingReads)))
                .isInstanceOf(AssertionError.class);
    }

    private void assertExplainPlan(JsonNode root) {
        List<PlanNode> nodes = new ArrayList<>();
        collectPlanNodes(root, nodes);
        PlanNode messages = onlyNode(nodes, "m");
        PlanNode reads = onlyNode(nodes, "mr");
        PlanNode ranked = onlyNode(nodes, "ranked");

        assertThat(messages.accessType()).isIn("ref", "range");
        assertThat(messages.key()).isNotBlank();
        var leadingColumns = jdbcTemplate.queryForList("""
                SELECT COLUMN_NAME FROM information_schema.STATISTICS
                 WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'messages'
                   AND INDEX_NAME = ? AND SEQ_IN_INDEX = 1
                """, String.class, messages.key());
        assertThat(leadingColumns).as("所选索引必须以会话 ID 为首列").containsExactly("conversation_id");
        assertThat(messages.rowsExamined()).as("高选择性夹具不得退化为全索引扫描")
                .isBetween(1L, 400L);
        assertThat(messages.accessType()).isNotEqualTo("ALL");
        assertThat(reads.accessType()).isEqualTo("eq_ref");
        assertThat(reads.key()).isEqualTo("uk_message_reads");
        assertThat(ranked.accessType()).isEqualTo("ALL");
        assertThat(findBooleanValue(root, "using_temporary_table"))
                .as("必须显式读取 using_temporary_table")
                .isPresent();
        assertThat(findBooleanValue(root, "using_filesort"))
                .as("必须显式读取 using_filesort")
                .isPresent();
    }

    private static PlanNode onlyNode(List<PlanNode> nodes, String alias) {
        List<PlanNode> matches = nodes.stream().filter(node -> alias.equals(node.tableName())).toList();
        assertThat(matches).as("EXPLAIN 节点 %s 必须存在且唯一", alias).hasSize(1);
        return matches.get(0);
    }

    private static void collectPlanNodes(JsonNode node, List<PlanNode> out) {
        if (node == null) return;
        if (node.isObject()) {
            JsonNode table = node.get("table");
            if (table != null && table.isObject() && table.has("table_name")) {
                out.add(new PlanNode(
                        table.path("table_name").asText(),
                        table.path("access_type").asText(),
                        table.has("key") ? table.path("key").asText() : null,
                        table.path("rows_examined_per_scan").asLong(-1)));
            }
            node.fields().forEachRemaining(entry -> collectPlanNodes(entry.getValue(), out));
        } else if (node.isArray()) {
            node.forEach(child -> collectPlanNodes(child, out));
        }
    }

    private static Optional<Boolean> findBooleanValue(JsonNode node, String field) {
        if (node == null) return Optional.empty();
        if (node.isObject()) {
            if (node.has(field) && node.get(field).isBoolean()) {
                return Optional.of(node.get(field).asBoolean());
            }
            for (JsonNode child : node) {
                Optional<Boolean> found = findBooleanValue(child, field);
                if (found.isPresent()) return found;
            }
        } else if (node.isArray()) {
            for (JsonNode child : node) {
                Optional<Boolean> found = findBooleanValue(child, field);
                if (found.isPresent()) return found;
            }
        }
        return Optional.empty();
    }

    private record PlanNode(String tableName, String accessType, String key, long rowsExamined) {}
}