From 554c6eaef4e4987d2a9d9a7672216ee348e9c427 Mon Sep 17 00:00:00 2001 From: Zhang Shun Date: Thu, 15 Oct 2026 14:35:25 +0800 Subject: [PATCH] =?UTF-8?q?fix(chat):=20=E4=BF=AE=E5=A4=8D=20MySQL=205.7?= =?UTF-8?q?=20=E4=BC=9A=E8=AF=9D=E6=9F=A5=E8=AF=A2=E5=B9=B6=E8=A1=A5?= =?UTF-8?q?=E9=BD=90=E9=AA=8C=E6=94=B6=E5=A5=91=E7=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/chat-realtime-api.md | 68 ++++++++++++++ docs/frontend-integration.md | 2 + docs/openapi-rc.json | 64 ++++++++++++++ .../messaging/MessageRepository.java | 28 +----- .../messaging/MessageUnreadQueries.java | 9 ++ .../messaging/MessageUnreadQueriesImpl.java | 88 +++++++++++++++++++ .../CompanyResumeDownloadIntegrationTest.java | 4 +- .../UnifiedRegistrationIntegrationTest.java | 2 +- ...ependentCrossDatabaseVerificationTest.java | 23 +++-- .../R01EMySqlQueryCountAndExplainTest.java | 82 +++++++++++++++-- .../MySql57MigrationConfigurationTest.java | 8 +- .../LxqModulesMySql57IntegrationTest.java | 64 +++++++++++++- .../LxqModulesMySqlIntegrationTest.java | 14 ++- 13 files changed, 406 insertions(+), 50 deletions(-) create mode 100644 docs/chat-realtime-api.md create mode 100644 mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageUnreadQueries.java create mode 100644 mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageUnreadQueriesImpl.java diff --git a/docs/chat-realtime-api.md b/docs/chat-realtime-api.md new file mode 100644 index 0000000..c2e349e --- /dev/null +++ b/docs/chat-realtime-api.md @@ -0,0 +1,68 @@ +# 轻松聘聊天实时接口 + +本文补充 `openapi-rc.json` 中的消息发送、事件补拉和消息字段。所有 HTTP 路径以 `/api/v1` 为前缀；下文省略该前缀。 + +## 连接和发送 + +通过 STOMP WebSocket 连接 `/api/v1/ws`，在 **STOMP CONNECT 帧**中设置 `Authorization: Bearer `。不得放在 URL 中，refresh token 不可代替 access token。连接成功后订阅唯一用户队列 `/user/queue/chat-events`，收到 `{"type":"CHAT_READY"}` 后开始补拉。业务消息继续通过 HTTP 接口发送，STOMP SEND 不开放。 + +`POST /conversations/messages` 请求： + +```json +{ + "conversationId": 1, + "content": "你好", + "clientMessageId": "example-message-001" +} +``` + +`clientMessageId` 为 1–64 位英文字母、数字、下划线或连字符；为兼容旧客户端可省略，但新客户端必须为一次逻辑发送生成固定标识，并在超时、断线或未收到响应时复用。同一会话、同一发送者下，相同标识和正文返回原消息，不重复生成事件；相同标识搭配不同正文返回 400。撤回后重试也不会恢复旧正文。 + +消息响应增加 `clientMessageId`（旧消息可为空）和 `version`（消息版本）。`MessageResponse` 在普通会话返回 `senderUserId`；人才邀请会话继续使用 `senderScope` 隐藏用户标识。撤回消息只返回“消息已撤回”占位。 + +## 事件补拉 + +`GET /conversations/{id}/events?after=0&size=50` 必须携带普通 HTTP Bearer access token，且当前账号必须属于该会话。`after` 是非负事件游标；`size` 默认 50，合法范围 1–100。 + +响应的 `data`： + +```json +{ + "items": [ + { + "eventId": 2, + "type": "MESSAGE_CREATED", + "conversationId": 1, + "occurredAt": "2026-10-15T10:00:00", + "actorScope": "OTHER_PARTICIPANT", + "payload": { + "id": 1, + "clientMessageId": "example-message-001", + "version": 0, + "conversationId": 1, + "content": "你好", + "status": "SENT", + "isRead": false + } + } + ], + "nextCursor": 2, + "hasMore": false +} +``` + +事件包括 `MESSAGE_CREATED`、`MESSAGE_READ`、`MESSAGE_RECALLED`。`actorScope` 为 `CURRENT_USER` 或 `OTHER_PARTICIPANT`；`payload.isRead` 表示当前接收用户的已读状态。payload 读取消息当前状态，而不是保存历史正文快照。 + +客户端按以下顺序恢复： + +1. 连接、鉴权、订阅并等待 `CHAT_READY`。 +2. 使用该账号、该会话已持久化的补拉游标调用事件接口；首次使用 0。 +3. 以 `eventId` 去重，以消息 `id` 和 `version` 合并，避免旧版本覆盖新状态。 +4. 本页成功合并后才持久化 `nextCursor`；`hasMore=true` 时继续请求，直到为 false。 +5. 补拉期间同时合并实时事件。实时事件 ID 可能乱序，不得直接用其最大值推进补拉游标。同版本的已读状态还需按 `actorScope` 和本地已读操作处理，不能仅按版本丢弃已读事件。 + +断线后重新获取有效 access token、重连、订阅，再重复上述流程。账号切换必须隔离游标、消息缓存和连接。 + +## 运行边界 + +这些接口沿用企业认证、会话成员、屏蔽和现有匹配准入规则；不会改变投递或开聊门槛。目前 WebSocket 使用单实例内存 broker，多实例部署需要共享消息分发能力。Nginx 示例文件须经部署配置和验证后才能使用。 diff --git a/docs/frontend-integration.md b/docs/frontend-integration.md index 469674d..f56566e 100644 --- a/docs/frontend-integration.md +++ b/docs/frontend-integration.md @@ -9,6 +9,8 @@ 所有接口统一前缀 `/api/v1`，OpenAPI 文档：`GET /api/v1/v3/api-docs`（Swagger UI：`/api/v1/swagger-ui/index.html`）。 +聊天实时连接、发送幂等、事件补拉和游标规则见 [聊天实时接口](chat-realtime-api.md)。 + ## 2. 后端启动（Windows） 先启动基础设施容器： diff --git a/docs/openapi-rc.json b/docs/openapi-rc.json index eb34d1d..2979388 100644 --- a/docs/openapi-rc.json +++ b/docs/openapi-rc.json @@ -2407,6 +2407,59 @@ } } }, + "/conversations/{id}/events": { + "get": { + "tags": [ + "消息" + ], + "summary": "补拉会话事件", + "description": "按持久化游标恢复新消息、已读、撤回；仅当前会话成员", + "operationId": "replayEvents", + "parameters": [ + { + "name": "id", + "in": "path", + "required": true, + "schema": { + "type": "integer", + "format": "int64" + } + }, + { + "name": "after", + "in": "query", + "required": false, + "schema": { + "type": "integer", + "format": "int64", + "default": 0 + } + }, + { + "name": "size", + "in": "query", + "required": false, + "schema": { + "type": "integer", + "format": "int32", + "default": 50 + } + } + ], + "responses": { + "200": { + "description": "OK", + "content": { + "*/*": { + "schema": { + "$ref": "#/components/schemas/ApiResponseMapStringObject" + } + } + } + } + } + } + }, "/conversations/{id}/messages": { "get": { "tags": [ @@ -15160,6 +15213,10 @@ "SendMessageRequest": { "type": "object", "properties": { + "clientMessageId": { + "type": "string", + "pattern": "[A-Za-z0-9_-]{1,64}" + }, "conversationId": { "type": "integer", "format": "int64" @@ -18003,6 +18060,13 @@ "type": "integer", "format": "int64" }, + "clientMessageId": { + "type": "string" + }, + "version": { + "type": "integer", + "format": "int64" + }, "conversationId": { "type": "integer", "format": "int64" diff --git a/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageRepository.java b/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageRepository.java index 9690a6d..3a04783 100644 --- a/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageRepository.java +++ b/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageRepository.java @@ -14,7 +14,7 @@ import java.time.LocalDateTime; import java.util.List; @Repository -public interface MessageRepository extends JpaRepository { +public interface MessageRepository extends JpaRepository, MessageUnreadQueries { @Lock(LockModeType.PESSIMISTIC_WRITE) java.util.Optional findByConversationIdAndSenderUserIdAndClientMessageId( @@ -29,32 +29,6 @@ public interface MessageRepository extends JpaRepository { Long getUnreadCount(); } - /** - * 批量统计各会话最近 500 条消息中的未读数，保持历史 countUnread 的窗口与撤回语义。 - */ - @Query(value = """ - 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 - ) AS row_no - FROM messages m - WHERE m.conversation_id IN (:conversationIds) - ) ranked - LEFT JOIN message_reads mr - ON mr.message_id = ranked.id - AND mr.reader_user_id = :readerUserId - WHERE ranked.row_no <= 500 - AND (ranked.status IS NULL OR ranked.status <> 'RECALLED') - AND mr.id IS NULL - GROUP BY ranked.conversation_id - """, nativeQuery = true) - List countUnreadByConversationIds( - @Param("conversationIds") List conversationIds, - @Param("readerUserId") Long readerUserId); - /** * 条件更新未命中后的当前读，用于区分并发请求已完成撤回与真实冲突。 * PESSIMISTIC_WRITE 强制读取提交后的最新行，避免 MySQL REPEATABLE READ 使用旧快照。 diff --git a/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageUnreadQueries.java b/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageUnreadQueries.java new file mode 100644 index 0000000..eb85b3c --- /dev/null +++ b/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageUnreadQueries.java @@ -0,0 +1,9 @@ +package com.mydreamjob.messaging; + +import java.util.List; + +/** 批量统计每个会话最近 500 条消息内的未读数，兼容 MySQL 5.7。 */ +public interface MessageUnreadQueries { + List countUnreadByConversationIds( + List conversationIds, Long readerUserId); +} diff --git a/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageUnreadQueriesImpl.java b/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageUnreadQueriesImpl.java new file mode 100644 index 0000000..23c184d --- /dev/null +++ b/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageUnreadQueriesImpl.java @@ -0,0 +1,88 @@ +package com.mydreamjob.messaging; + +import jakarta.persistence.EntityManager; +import jakarta.persistence.PersistenceContext; +import java.util.ArrayList; +import java.util.List; +import java.util.stream.IntStream; +import org.hibernate.Session; +import org.springframework.transaction.annotation.Transactional; + +/** Spring Data 仓库片段；只切换 SQL 能力，不改变会话权限和最近 500 条统计规则。 */ +public class MessageUnreadQueriesImpl implements MessageUnreadQueries { + @PersistenceContext private EntityManager entityManager; + private volatile Boolean legacyMysql; + + @Override + @Transactional(readOnly = true) + public List countUnreadByConversationIds( + List conversationIds, Long readerUserId) { + // 保持原 IN 查询的集合语义，避免重复会话 ID 使 UNION 分支重复计数。 + conversationIds = conversationIds.stream().distinct().toList(); + if (conversationIds.isEmpty()) return List.of(); + String sql = isLegacyMysql() ? legacySql(conversationIds.size()) : windowSql(conversationIds.size()); + var query = entityManager.createNativeQuery(sql); + for (int i = 0; i < conversationIds.size(); i++) query.setParameter(i + 1, conversationIds.get(i)); + query.setParameter(conversationIds.size() + 1, readerUserId); + List counts = new ArrayList<>(); + for (Object result : query.getResultList()) { + Object[] row = (Object[]) result; + counts.add(new UnreadCount(((Number) row[0]).longValue(), ((Number) row[1]).longValue())); + } + return counts; + } + + private boolean isLegacyMysql() { + if (legacyMysql == null) { + synchronized (this) { + if (legacyMysql == null) { + legacyMysql = entityManager.unwrap(Session.class).doReturningWork(connection -> { + var metadata = connection.getMetaData(); + return "MySQL".equals(metadata.getDatabaseProductName()) + && metadata.getDatabaseMajorVersion() < 8; + }); + } + } + } + return legacyMysql; + } + + private static String windowSql(int count) { + String placeholders = String.join(",", IntStream.rangeClosed(1, count).mapToObj(i -> "?" + i).toList()); + return """ + SELECT ranked.conversation_id, COUNT(*) + 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 = ?%d + WHERE ranked.row_no <= 500 + AND (ranked.status IS NULL OR ranked.status <> 'RECALLED') AND mr.id IS NULL + GROUP BY ranked.conversation_id + """.formatted(placeholders, count + 1); + } + + private static String legacySql(int count) { + // 每个分支在数据库内取最近 500 条；一次 SQL 避免逐会话往返和把全量历史加载进 Java。 + List branches = IntStream.rangeClosed(1, count).mapToObj(i -> """ + SELECT m.id, m.conversation_id, m.status FROM messages m + WHERE m.conversation_id = ?%d ORDER BY m.created_at DESC, m.id DESC LIMIT 500 + """.formatted(i).strip()).toList(); + String bounded = count == 1 ? branches.get(0) + : String.join(" UNION ALL ", branches.stream().map(branch -> "(" + branch + ")").toList()); + return """ + SELECT recent.conversation_id, COUNT(*) FROM (%s) recent + LEFT JOIN message_reads mr ON mr.message_id = recent.id AND mr.reader_user_id = ?%d + WHERE (recent.status IS NULL OR recent.status <> 'RECALLED') AND mr.id IS NULL + GROUP BY recent.conversation_id + """.formatted(bounded, count + 1); + } + + private record UnreadCount(Long conversationId, Long unreadCount) + implements MessageRepository.ConversationUnreadCount { + @Override public Long getConversationId() { return conversationId; } + @Override public Long getUnreadCount() { return unreadCount; } + } +} diff --git a/mydreamjob-server/src/test/java/com/mydreamjob/application/CompanyResumeDownloadIntegrationTest.java b/mydreamjob-server/src/test/java/com/mydreamjob/application/CompanyResumeDownloadIntegrationTest.java index 997927f..0694c8c 100644 --- a/mydreamjob-server/src/test/java/com/mydreamjob/application/CompanyResumeDownloadIntegrationTest.java +++ b/mydreamjob-server/src/test/java/com/mydreamjob/application/CompanyResumeDownloadIntegrationTest.java @@ -462,8 +462,8 @@ class CompanyResumeDownloadIntegrationTest { } assertThat(modulePaths).as("N03/N04/N05 新模块路径").isEqualTo(13); assertThat(moduleOperations).as("N03/N04/N05 新模块操作").isEqualTo(18); - assertThat(pathCount - modulePaths).as("185 paths 包含首次密码设置").isEqualTo(185); - assertThat(opCount - moduleOperations).as("223 operations 包含首次密码设置").isEqualTo(223); + assertThat(pathCount - modulePaths).as("186 paths 包含首次密码设置与聊天事件补拉").isEqualTo(186); + assertThat(opCount - moduleOperations).as("224 operations 包含首次密码设置与聊天事件补拉").isEqualTo(224); var download = paths.path("/company/candidates/{applicationId}/resume").path("get"); assertThat(download.isMissingNode()).as("必须存在企业简历下载 GET").isFalse(); diff --git a/mydreamjob-server/src/test/java/com/mydreamjob/auth/identity/UnifiedRegistrationIntegrationTest.java b/mydreamjob-server/src/test/java/com/mydreamjob/auth/identity/UnifiedRegistrationIntegrationTest.java index c87dc24..d12bbb1 100644 --- a/mydreamjob-server/src/test/java/com/mydreamjob/auth/identity/UnifiedRegistrationIntegrationTest.java +++ b/mydreamjob-server/src/test/java/com/mydreamjob/auth/identity/UnifiedRegistrationIntegrationTest.java @@ -210,7 +210,7 @@ class UnifiedRegistrationIntegrationTest { } } assertThat(moduleDenied).as("新模块全部拒绝未选择身份账号").isEqualTo(18); - assertThat(denied - moduleDenied).as("既有身份门禁范围不减少").isEqualTo(197); + assertThat(denied - moduleDenied).as("既有身份门禁及聊天事件补拉均受保护").isEqualTo(198); java.nio.file.Files.writeString(java.nio.file.Path.of("target/unified-identity-gate-result.json"), "{\"deniedOperations\":"+denied+",\"paths\":"+doc.path("paths").size()+"}"); } diff --git a/mydreamjob-server/src/test/java/com/mydreamjob/common/r01e/R01EIndependentCrossDatabaseVerificationTest.java b/mydreamjob-server/src/test/java/com/mydreamjob/common/r01e/R01EIndependentCrossDatabaseVerificationTest.java index 446649d..d6378f6 100644 --- a/mydreamjob-server/src/test/java/com/mydreamjob/common/r01e/R01EIndependentCrossDatabaseVerificationTest.java +++ b/mydreamjob-server/src/test/java/com/mydreamjob/common/r01e/R01EIndependentCrossDatabaseVerificationTest.java @@ -1202,11 +1202,11 @@ class R01EIndependentCrossDatabaseVerificationTest { // 4. MySQL EXPLAIN FORMAT=JSON // ══════════════════════════════════════════════════════════ - /** 与 MessageRepository.countUnreadByConversationIds 逐字一致的原生 SQL。 */ + /** 独立复刻 MySQL 8 未读统计分支，使用 JDBC 占位符。 */ static String unreadSql(int placeholders) { return "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) AS row_no " + + "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 (" + String.join(",", Collections.nCopies(placeholders, "?")) + ")) ranked " + "LEFT JOIN message_reads mr ON mr.message_id = ranked.id AND mr.reader_user_id = ? " @@ -1218,7 +1218,7 @@ class R01EIndependentCrossDatabaseVerificationTest { * EXPLAIN 得到的「表节点」信息。 * *

注意：MySQL 的 {@code EXPLAIN FORMAT=JSON} 里 {@code table_name} 是**别名**。 - * 被测语句逐字取自 {@code MessageRepository.countUnreadByConversationIds}， + * 被测语句独立复刻 {@code MessageUnreadQueriesImpl} 的 MySQL 8 分支， * 其别名为 {@code messages m}、{@code message_reads mr}、派生表 {@code ranked}， * 故此处表节点分别为 {@code m} / {@code mr} / {@code ranked}。

*/ @@ -1341,9 +1341,20 @@ class R01EIndependentCrossDatabaseVerificationTest { assertThat(m.accessType()) .as("%s：基础表 messages 不得全表扫描（access_type=%s, key=%s, possible_keys=%s）", label, m.accessType(), m.key(), m.possibleKeys()) - .isNotEqualTo("ALL"); - assertThat(m.key()).as("%s：messages 必须命中既有索引", label) - .isEqualTo("idx_messages_conversation_created"); + .isIn("ref", "range"); + assertThat(m.key()).as("%s：messages 必须命中会话过滤索引", label).isNotNull(); + var firstColumns = MySql.jdbc.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, m.key()); + assertThat(firstColumns).as("%s：所选索引首列必须是 conversation_id", label) + .containsExactly("conversation_id"); + long targetRows = Long.parseLong(label.substring(label.indexOf("targetRows=") + "targetRows=".length())); + assertThat(m.rowsExamined()).as("%s：必须给出扫描行数估计", label).isInstanceOf(Number.class); + assertThat(((Number) m.rowsExamined()).longValue()) + .as("%s：高选择性场景不得扫描大量无关消息", label) + .isBetween(1L, Math.max(10L, targetRows * 2)); } for (TableNode r : reads) { assertThat(r.accessType()) diff --git a/mydreamjob-server/src/test/java/com/mydreamjob/common/r01e/R01EMySqlQueryCountAndExplainTest.java b/mydreamjob-server/src/test/java/com/mydreamjob/common/r01e/R01EMySqlQueryCountAndExplainTest.java index a2f78e4..7194d6c 100644 --- a/mydreamjob-server/src/test/java/com/mydreamjob/common/r01e/R01EMySqlQueryCountAndExplainTest.java +++ b/mydreamjob-server/src/test/java/com/mydreamjob/common/r01e/R01EMySqlQueryCountAndExplainTest.java @@ -40,8 +40,8 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; *
  • {@code EXPLAIN FORMAT=JSON} 的索引使用、access type、temporary / filesort 情况。
    • * * - *

    EXPLAIN 使用与 {@code MessageRepository.countUnreadByConversationIds} **逐字相同**的 SQL - * 与同一批绑定参数，避免解释一条并不存在的语句。

    + *

    EXPLAIN 独立复刻 {@code MessageUnreadQueriesImpl} 的 MySQL 8 窗口分支； + * 筛选、排序和窗口规则一致，使用 JDBC 占位符与同一批绑定参数。

    */ @Testcontainers(disabledWithoutDocker = true) @SpringBootTest @@ -63,14 +63,14 @@ class R01EMySqlQueryCountAndExplainTest { registry.add("spring.datasource.password", MYSQL::getPassword); } - /** 与 MessageRepository 中原生查询**逐字一致**（仅换行为便于阅读）。 */ + /** 独立复刻 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 + ORDER BY m.created_at DESC, m.id DESC ) AS row_no FROM messages m WHERE m.conversation_id IN (%s) @@ -248,6 +248,54 @@ class R01EMySqlQueryCountAndExplainTest { 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 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 nodes = new ArrayList<>(); + collectPlanNodes(root, nodes); + return onlyNode(nodes, "m"); } @Test @@ -256,14 +304,21 @@ class R01EMySqlQueryCountAndExplainTest { 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"}}, + {"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); @@ -277,7 +332,7 @@ class R01EMySqlQueryCountAndExplainTest { .isInstanceOf(AssertionError.class); } - private static void assertExplainPlan(JsonNode root) { + private void assertExplainPlan(JsonNode root) { List nodes = new ArrayList<>(); collectPlanNodes(root, nodes); PlanNode messages = onlyNode(nodes, "m"); @@ -285,7 +340,15 @@ class R01EMySqlQueryCountAndExplainTest { PlanNode ranked = onlyNode(nodes, "ranked"); assertThat(messages.accessType()).isIn("ref", "range"); - assertThat(messages.key()).isEqualTo("idx_messages_conversation_created"); + 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"); @@ -312,7 +375,8 @@ class R01EMySqlQueryCountAndExplainTest { out.add(new PlanNode( table.path("table_name").asText(), table.path("access_type").asText(), - table.has("key") ? table.path("key").asText() : null)); + 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()) { @@ -339,5 +403,5 @@ class R01EMySqlQueryCountAndExplainTest { return Optional.empty(); } - private record PlanNode(String tableName, String accessType, String key) {} + private record PlanNode(String tableName, String accessType, String key, long rowsExamined) {} } diff --git a/mydreamjob-server/src/test/java/com/mydreamjob/config/MySql57MigrationConfigurationTest.java b/mydreamjob-server/src/test/java/com/mydreamjob/config/MySql57MigrationConfigurationTest.java index 414b184..ca7cd11 100644 --- a/mydreamjob-server/src/test/java/com/mydreamjob/config/MySql57MigrationConfigurationTest.java +++ b/mydreamjob-server/src/test/java/com/mydreamjob/config/MySql57MigrationConfigurationTest.java @@ -27,7 +27,13 @@ class MySql57MigrationConfigurationTest { var provider = config.getResourceProvider(); assertThat(provider).isNotNull(); var resources = provider.getResources("V", new String[]{".sql"}); - assertThat(resources).hasSize(54); + var originals = new org.springframework.core.io.support.PathMatchingResourcePatternResolver() + .getResources("classpath*:db/migration/V*.sql"); + var filenames = java.util.Arrays.stream(originals).map(org.springframework.core.io.Resource::getFilename) + .collect(java.util.stream.Collectors.toSet()); + assertThat(filenames).contains("V49__skill_cloud_boolean_type.sql", "V55__chat_realtime_events.sql"); + assertThat(resources).extracting(org.flywaydb.core.api.resource.LoadableResource::getFilename) + .containsExactlyInAnyOrderElementsOf(filenames); for (var resource : resources) { String path = "V49__skill_cloud_boolean_type.sql".equals(resource.getFilename()) ? "db/mysql57/" : "db/migration/"; diff --git a/mydreamjob-server/src/test/java/com/mydreamjob/integration/LxqModulesMySql57IntegrationTest.java b/mydreamjob-server/src/test/java/com/mydreamjob/integration/LxqModulesMySql57IntegrationTest.java index a42634c..0adb306 100644 --- a/mydreamjob-server/src/test/java/com/mydreamjob/integration/LxqModulesMySql57IntegrationTest.java +++ b/mydreamjob-server/src/test/java/com/mydreamjob/integration/LxqModulesMySql57IntegrationTest.java @@ -43,6 +43,56 @@ class LxqModulesMySql57IntegrationTest { @Autowired JdbcTemplate jdbc; @Autowired MockMvc mvc; @Autowired ObjectMapper mapper; + @Autowired com.mydreamjob.messaging.MessageRepository messages; + @Autowired jakarta.persistence.EntityManagerFactory entityManagerFactory; + + @Test void recentWindowUnreadUsesOneBoundedQueryOnMysql57() { + String phone = "q57-" + java.util.UUID.randomUUID().toString().substring(0, 8); + jdbc.update("INSERT INTO users (phone,role,is_active,is_verified) VALUES (?,'STUDENT',1,1)", phone); + Long user = jdbc.queryForObject("SELECT id FROM users WHERE phone=?", Long.class, phone); + java.util.List conversations = new java.util.ArrayList<>(); + try { + for (int c = 0; c < 2; c++) { + String title = phone + "-" + c; + jdbc.update("INSERT INTO conversations (title,type,created_by,created_at,updated_at) " + + "VALUES (?,'DIRECT',?,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)", title, user); + Long id = jdbc.queryForObject("SELECT id FROM conversations WHERE title=?", Long.class, title); + conversations.add(id); + java.util.List batch = new java.util.ArrayList<>(); + for (int i = 0; i < 600; i++) batch.add(new Object[]{id,user,"legacy-" + i, + i % 20 == 0 ? "RECALLED" : "SENT"}); + jdbc.batchUpdate("INSERT INTO messages (conversation_id,sender_user_id,content,status,created_at) " + + "VALUES (?,?,?,?,'2026-10-15 08:00:00')", batch); + jdbc.update("INSERT INTO message_reads (message_id,reader_user_id,read_at) " + + "SELECT id,?,CURRENT_TIMESTAMP FROM messages WHERE conversation_id=? " + + "AND content IN ('legacy-50','legacy-150','legacy-250','legacy-350','legacy-450','legacy-550')", user,id); + } + var stats = entityManagerFactory.unwrap(org.hibernate.SessionFactory.class).getStatistics(); + boolean wasEnabled = stats.isStatisticsEnabled(); + stats.setStatisticsEnabled(true); + try { + long before = stats.getPrepareStatementCount(); + var counts = messages.countUnreadByConversationIds(conversations, user); + assertThat(stats.getPrepareStatementCount() - before).as("两会话合计只发一条 SQL").isEqualTo(1); + assertThat(counts).hasSize(2); + assertThat(counts).extracting(com.mydreamjob.messaging.MessageRepository.ConversationUnreadCount::getConversationId) + .containsExactlyInAnyOrderElementsOf(conversations); + assertThat(counts).extracting(com.mydreamjob.messaging.MessageRepository.ConversationUnreadCount::getUnreadCount) + .containsOnly(470L); + assertThat(messages.countUnreadByConversationIds(java.util.List.of(), user)).isEmpty(); + assertThat(messages.countUnreadByConversationIds(java.util.List.of(conversations.get(0)), user)) + .extracting(com.mydreamjob.messaging.MessageRepository.ConversationUnreadCount::getUnreadCount) + .containsExactly(470L); + assertThat(messages.countUnreadByConversationIds( + java.util.List.of(conversations.get(0), conversations.get(0)), user)) + .extracting(com.mydreamjob.messaging.MessageRepository.ConversationUnreadCount::getUnreadCount) + .containsExactly(470L); + } finally { stats.setStatisticsEnabled(wasEnabled); } + } finally { + for (Long id : conversations) jdbc.update("DELETE FROM conversations WHERE id=?", id); + jdbc.update("DELETE FROM users WHERE id=?", user); + } + } @Test void fullMigrationAndSeedsValidateOnActualServerVersion() { assertThat(jdbc.queryForObject("select version()", String.class)).startsWith("5.7.36"); @@ -58,8 +108,18 @@ class LxqModulesMySql57IntegrationTest { JsonNode runtime = mapper.readTree(mvc.perform(get("/v3/api-docs")) .andExpect(status().isOk()).andReturn().getResponse().getContentAsByteArray()); JsonNode contract = mapper.readTree(Files.readString(Path.of("../docs/openapi-rc.json"))); - assertThat(operations(runtime)).hasSize(241).isEqualTo(operations(contract)); - assertThat(runtime.path("paths").size()).isEqualTo(198); + assertThat(operations(runtime)).hasSize(242).isEqualTo(operations(contract)); + assertThat(runtime.path("paths").size()).isEqualTo(199); + assertThat(runtime.path("paths").path("/conversations/{id}/events")) + .isEqualTo(contract.path("paths").path("/conversations/{id}/events")); + for (String field : Set.of("clientMessageId", "version")) { + String pointer = "/components/schemas/MessageResponse/properties/" + field; + assertThat(runtime.at(pointer).isMissingNode()).as(pointer).isFalse(); + assertThat(contract.at(pointer)).as(pointer).isEqualTo(runtime.at(pointer)); + } + String requestField = "/components/schemas/SendMessageRequest/properties/clientMessageId"; + assertThat(runtime.at(requestField).isMissingNode()).isFalse(); + assertThat(contract.at(requestField)).isEqualTo(runtime.at(requestField)); } private Set operations(JsonNode document) { diff --git a/mydreamjob-server/src/test/java/com/mydreamjob/integration/LxqModulesMySqlIntegrationTest.java b/mydreamjob-server/src/test/java/com/mydreamjob/integration/LxqModulesMySqlIntegrationTest.java index 2cfd33b..2cf9ebb 100644 --- a/mydreamjob-server/src/test/java/com/mydreamjob/integration/LxqModulesMySqlIntegrationTest.java +++ b/mydreamjob-server/src/test/java/com/mydreamjob/integration/LxqModulesMySqlIntegrationTest.java @@ -59,8 +59,18 @@ class LxqModulesMySqlIntegrationTest { JsonNode runtime = mapper.readTree(mvc.perform(get("/v3/api-docs")) .andExpect(status().isOk()).andReturn().getResponse().getContentAsByteArray()); JsonNode contract = mapper.readTree(Files.readString(Path.of("../docs/openapi-rc.json"))); - assertThat(operations(runtime)).hasSize(241).isEqualTo(operations(contract)); - assertThat(runtime.path("paths").size()).isEqualTo(198); + assertThat(operations(runtime)).hasSize(242).isEqualTo(operations(contract)); + assertThat(runtime.path("paths").size()).isEqualTo(199); + assertThat(runtime.path("paths").path("/conversations/{id}/events")) + .isEqualTo(contract.path("paths").path("/conversations/{id}/events")); + for (String field : Set.of("clientMessageId", "version")) { + String pointer = "/components/schemas/MessageResponse/properties/" + field; + assertThat(runtime.at(pointer).isMissingNode()).as(pointer).isFalse(); + assertThat(contract.at(pointer)).as(pointer).isEqualTo(runtime.at(pointer)); + } + String requestField = "/components/schemas/SendMessageRequest/properties/clientMessageId"; + assertThat(runtime.at(requestField).isMissingNode()).isFalse(); + assertThat(contract.at(requestField)).isEqualTo(runtime.at(requestField)); } private Set operations(JsonNode document) {