admin/danyang-match
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(chat): Phase 09 复验 —— 修复 WebSocket 下行死链路 / Unmatch 后可发消息 / Block 未落库 / 聊天 IDOR

Browse Source 
对 Phase 09（Text Chat）执行 KEEP-VERIFY/EXTEND 复验：不采信历史 ✅，逐条核对
REST 发送 + WebSocket 接收、幂等、重连、未读计数、非 active Match 禁聊、Unmatch/
Block 即时禁聊、活动双确认会话复用。共发现并修复 4 处真实缺陷 + 1 处规范不符。

D1（Critical）WebSocket 下行整条链路是死代码
  - 无 HandshakeInterceptor，无任一环节把 userId 写入 session attributes，
    extractUserId 恒返回 null；sendToUser/sendToConversation 全仓库零调用点。
  - 新增 ChatHandshakeInterceptor：从 ?token= 或 Authorization: Bearer 解析 JWT，
    校验通过后写 session attr userId；失败 fail-closed 拒绝握手；仅接受 type=access。
  - 重写 ChatWebSocketHandler：按规范 {event,data} 信封推送；未认证连接直接
    close(NOT_ACCEPTABLE)；同账号重连由新连接接管、旧连接 close(NORMAL)；
    上行仅支持 ping->pong，禁止客户端自订阅。
  - ChatService.sendText 落库+递增未读后推送 message.new；RiskNotice 同推。
  - 移除 WebSocketConfig 的 .withSockJS()（SockJS 改写端点致原生 WS 无法直连）。

D2（Critical）Unmatch 后仍可发消息
  - unmatch 只改 Match，不联动会话；sendText 仅校验 conv.status==active。
  - 新增 ChatService.assertMatchAlive(conv)：按 matchId 查 Match，状态非
    {active,getting_to_know} 抛 "匹配已解除，无法发送消息"。sendText 在 conv.status
    校验之后调用。
  - MatchController.unmatch 落库后调 chatService.syncFromMatch 推 conversation.closed。
  - syncFromMatch 重写：alive 恢复 active；非 alive 只推 conversation.closed，
    不改 status（会话仍以灰显态留在列表，属未授权产品决策，登记待决）。

D3（Critical）RelationService.block 调静态 MatchService.unmatch 只改内存态不落库
  - 结果：Block 后 matches.status 仍为 active。
  - 补 matchMapper.updateById(m) 持久化 + chatService.closeConversationForMatch(
    CONVERSATION_STATUS_BLOCKED) 通知对端。RelationshipService.block 同步补会话关闭通知。

D4（Critical 安全）聊天 4 个接口无成员校验 -> IDOR 越权读私聊
  - MessageController.load/markRead、ConversationController.detail/unread 补
    chatService.isMember(conversationId, userId) 校验；非成员返回
    ApiResponse.notFound()（body 级 success=false code=NOT_FOUND，HTTP 200，
    与 MatchController.authorizedMatch 同约定），不区分"不存在"与"无权"以防探测。

D5（Medium）上行订阅/信封不符规范
  - REST API 接口详细设计规范.md 第124-126章要求禁止客户端自订阅，事件名
    message.new/message.read/conversation.closed，信封 {event,data}。已对齐。

测试保真度 + 测试补齐
  - schema-h2.sql 新增 uk_message_client_id(sender_id, client_message_id)，
    对齐生产 V004:71；H2 不支持过滤索引故省略 WHERE（NULL 语义等价）。
    否则 service 的 SELECT-then-INSERT 竞态在测试中无 DB 兜底，形成测试盲区。
  - 新增 ChatPhase09IntegrationTest（13 用例）：幂等重试不重复投递、推送信封字段、
    未读增减清零、Match unmatched 拒发、Unmatch/Block 端点即时禁聊+收 closed 事件、
    活动双确认复用同一消息体系、未形成 Match 无法建会话、非成员读消息/详情/未读
    被拒且不泄露正文、DB 唯一索引兜底重复 clientMessageId、无幂等键系统消息可共存。
  - 新增 ChatWebSocketHandlerTest（7 用例）：userId 路由、信封格式、未认证关连、
    断开不推、重连替换旧连接、上行仅 ping。

验证证据
  - 全量回归（离线容器 danyang-match-backend-test，mvn -o test）：
    328 tests / 0 failures / 0 errors / 3 skipped / BUILD SUCCESS（基线 308，+20）。
  - 负向验证：临时注释 assertMatchAlive + 两处 isMember，恰好 3 个用例变红，
    证明守卫有效；随后还原并 grep 确认无残留。

文档
  - 新增 Phase-09-V11-Recovery-Completion-Report-2026-09-16.md（完整报告）。
  - Phase-09-V11-Completion-Report.md 顶部加状态注记，推翻其 4 条 WebSocket 结论。
  - 计划/STATUS/JIRA_BACKLOG 同步 Phase 09 完成态与测试总数（328）。

Phase 10 未启动，按计划每阶段出报告后停止等待确认。
This commit is contained in:
WorkBuddy
2026-09-16 16:42:48 +08:00
parent 4e478ce6ba
commit 03a03c7c11
17 changed files with 1295 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/main/java/com/danyang/match/platform/chat/config/WebSocketConfig.java
+27 -10
View File
@@ -1,5 +1,6 @@
package com.danyang.match.platform.chat.config;
import com.danyang.match.platform.chat.websocket.ChatHandshakeInterceptor;
import com.danyang.match.platform.chat.websocket.ChatWebSocketHandler;
import lombok.RequiredArgsConstructor;
import org.springframework.context.annotation.Configuration;
@@ -10,15 +11,28 @@ import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry
/**
* WebSocket 配置 — 实时消息推送
*
* 端点: /ws/chat?token={jwt}
* 协议: JSON (type + payload)
* - type="message" → 新消息推送
* - type="typing" → 对方正在输入
* - type="read" → 已读回执
* - type="match" → 新 Match 通知
* - type="unmatch" → Match 解除
* <p>端点: {@code /ws/chat?token={jwt}}（原生 WebSocket，非 SockJS）
*
* 鉴权: 握手时 URL 带 JWT token（与 REST 同密钥）
* <p>协议：事件信封对齐《REST API 接口详细设计规范》第 125 章，形如
* <pre>{ "event": "message.new", "data": { ... } }</pre>
* 事件名沿用规范第 124 章清单：
* <ul>
* <li>{@code message.new} — 新消息</li>
* <li>{@code message.read} — 已读回执</li>
* <li>{@code conversation.closed} — 会话关闭（Unmatch / Block）</li>
* <li>{@code match.created} — 新 Match</li>
* <li>{@code notification.new} — 通知</li>
* <li>{@code account.status_changed} — 账号状态变更</li>
* </ul>
*
* <p>鉴权：握手阶段由 {@link ChatHandshakeInterceptor} 校验 {@code token} 并把 userId
* 绑定到连接（规范第 126 章）。
*
* <p><b>关于 SockJS</b>：此前本配置调用了 {@code .withSockJS()}。SockJS 会把端点改写为
* {@code /ws/chat/**} 的 SockJS 协议族，原生 WebSocket 客户端（小程序
* {@code wx.connectSocket}、规范第 124 章给的 {@code /wss/chat}）<b>无法直接连接</b>，
* 与类注释自述的 {@code /ws/chat?token=} 原生契约自相矛盾。
* 本项目两端都是原生 WebSocket 客户端，故移除 SockJS。
*/
@Configuration
@EnableWebSocket
@@ -26,11 +40,14 @@ import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry
public class WebSocketConfig implements WebSocketConfigurer {
private final ChatWebSocketHandler chatHandler;
private final ChatHandshakeInterceptor chatHandshakeInterceptor;
@Override
public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
registry.addHandler(chatHandler, "/ws/chat")
.setAllowedOriginPatterns("*")
.withSockJS(); // SockJS fallback for miniapp 兼容
.addInterceptors(chatHandshakeInterceptor)
// 小程序 / 原生客户端握手不带 Origin，需允许任意来源；
// 真正的访问控制由握手 token 鉴权承担，不以 Origin 作为安全边界。
.setAllowedOriginPatterns("*");
}
}
backend/src/main/java/com/danyang/match/platform/chat/controller/ConversationController.java
+4
View File
@@ -92,6 +92,8 @@ public class ConversationController {
public ApiResponse<ConversationDTO> detail(@PathVariable String id,
@AuthenticationPrincipal UserPrincipal principal) {
if (principal == null) return ApiResponse.unauthorized();
// 对象级鉴权：非会话成员一律 404 —— 否则可凭 conversationId 读到对方昵称、最后一条消息
if (!chatService.isMember(id, principal.id())) return ApiResponse.notFound();
ConversationDTO dto = chatService.getConversation(principal.id(), id);
if (dto == null) return ApiResponse.notFound();
return ApiResponse.ok(dto);
@@ -105,6 +107,8 @@ public class ConversationController {
public ApiResponse<Map<String, Object>> unread(@PathVariable String id,
@AuthenticationPrincipal UserPrincipal principal) {
if (principal == null) return ApiResponse.unauthorized();
// 对象级鉴权：非成员不得读取他人会话未读
if (!chatService.isMember(id, principal.id())) return ApiResponse.notFound();
return ApiResponse.ok(Map.of("unreadCount", chatService.getUnreadCount(principal.id(), id)));
}
backend/src/main/java/com/danyang/match/platform/chat/controller/MessageController.java
+4
View File
@@ -59,6 +59,8 @@ public class MessageController {
@RequestParam(defaultValue = "0") int offset,
@AuthenticationPrincipal UserPrincipal principal) {
if (principal == null) return ApiResponse.unauthorized();
// 对象级鉴权：非会话成员一律 404（不区分"不存在"与"无权"，避免探测他人会话）
if (!chatService.isMember(id, principal.id())) return ApiResponse.notFound();
int size = Math.min(Math.max(pageSize, 1), MAX_PAGE_SIZE);
List<Message> messages;
@@ -82,6 +84,8 @@ public class MessageController {
@RequestBody Map<String, String> body,
@AuthenticationPrincipal UserPrincipal principal) {
if (principal == null) return ApiResponse.unauthorized();
// 对象级鉴权：非成员不得改写他人会话的已读状态
if (!chatService.isMember(id, principal.id())) return ApiResponse.notFound();
String lastReadId = body.get("lastReadMessageId");
if (lastReadId != null) {
chatService.markAsRead(id, principal.id(), lastReadId);
backend/src/main/java/com/danyang/match/platform/chat/service/ChatService.java
+154 -5
View File
@@ -7,6 +7,7 @@ import com.danyang.match.platform.chat.entity.Message;
import com.danyang.match.platform.chat.mapper.ConversationMapper;
import com.danyang.match.platform.chat.mapper.ConversationMemberMapper;
import com.danyang.match.platform.chat.mapper.MessageMapper;
import com.danyang.match.platform.chat.websocket.ChatWebSocketHandler;
import com.danyang.match.platform.match.entity.Match;
import com.danyang.match.platform.match.mapper.MatchMapper;
import com.danyang.match.platform.match.mapper.UserRelationMapper;
@@ -19,7 +20,9 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.time.Instant;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.UUID;
@@ -31,6 +34,7 @@ import java.util.UUID;
* - INV-004: Block 优先级最高 — 双方任一 block → conversation blocked，禁发
* - P09-V11-T01: Activity Match source 兼容 — Conversation 只绑 match_id，不关心 source
* - Phase 10 Chat Safety: 频控 + 内容扫描 + RiskNotice
* - Phase 09: REST send + WebSocket receive —— 落库成功后按规范第 125 章事件信封推送
*/
@Service
@RequiredArgsConstructor
@@ -45,6 +49,7 @@ public class ChatService {
private final UserRelationMapper relationMapper;
private final UserProfileMapper userProfileMapper;
private final ChatSafetyService safetyService;
private final ChatWebSocketHandler webSocketHandler;
// ==================== Conversation 创建 ====================
@@ -110,6 +115,11 @@ public class ChatService {
throw new IllegalStateException("会话已关闭");
}
// INV-005 服务端权威校验：未 Mutual / 已 Unmatch / 已成功牵手退出 → 不可发消息。
// 不能只信 conv.status —— 历史缺陷里 Unmatch 只改 Match 表、从不关闭会话，
// 于是会话仍是 active，仅凭 conversation.status 会放行已解除匹配的双方继续聊天。
assertMatchAlive(conv);
// 幂等重放优先：同一 clientMessageId 已成功落库则直接返回原消息，不重复走安全校验
if (clientMessageId != null) {
Optional<Message> dup = messageMapper.findByClientMessageId(conversationId, senderId, clientMessageId);
@@ -171,6 +181,14 @@ public class ChatService {
}
}
// Phase 09: REST send + WebSocket receive —— 落库成功后推送 message.new 给对方。
// 幂等重放已在上面 return，走不到这里，所以正常会话不会重复投递；
// 对方离线（无活跃连接）时静默跳过，不抛异常、不影响已提交的事务。
if (peerId != null) {
webSocketHandler.sendEvent(peerId, ChatWebSocketHandler.EVENT_MESSAGE_NEW,
buildMessageNewData(conversationId, msg));
}
// 3. warning 级内容 → 给对方推 RiskNotice 系统消息（P10-V11-T01/T02）
if (scan.riskLevel.equals(Constants.RISK_LEVEL_WARNING) && peerId != null) {
sendSystemRiskNotice(conversationId, senderId, peerId, scan.reason);
@@ -196,11 +214,28 @@ public class ChatService {
notice.setContent(ChatSafetyService.buildRiskNoticeBody(reason != null ? reason : "检测到可疑内容"));
notice.setStatus(Constants.MESSAGE_STATUS_SENT);
messageMapper.insert(notice);
// 系统提示同样需要实时到达对方，否则用户要重进会话才看到反诈提示
if (peerId != null) {
webSocketHandler.sendEvent(peerId, ChatWebSocketHandler.EVENT_MESSAGE_NEW,
buildMessageNewData(conversationId, notice));
}
log.info("RiskNotice sent to {} in conv {}", peerId, conversationId);
}
// ==================== 拉消息 ====================
/**
* 会话成员判定 —— 供控制器做对象级鉴权（AGENTS 第 7 条：工作人员及无关用户不得查看私人聊天）。
*
* <p>历史缺陷：{@code MessageController.load/markRead}、{@code ConversationController.detail/unread}
* 此前<b>完全没有成员校验</b>，任何已登录用户只要拿到 conversationId 就能读取他人完整聊天记录
* （典型 IDOR）。调用方一律以 404 响应非成员，避免被用来探测会话是否存在。
*/
public boolean isMember(String conversationId, String userId) {
if (conversationId == null || userId == null) return false;
return memberMapper.find(conversationId, userId).isPresent();
}
public List<Message> loadMessages(String conversationId, int limit, int offset) {
return messageMapper.findRecentByConversation(conversationId, limit, offset);
}
@@ -219,18 +254,132 @@ public class ChatService {
// ==================== 状态联动 ====================
/**
* 会话状态跟随 Match 状态联动，并在关闭时实时通知双方。
*
* <p>历史缺陷（本方法修复）：本方法此前<b>无任何调用点</b> ——
* {@code MatchController.unmatch} 只把 Match 改成 unmatched 就结束，
* {@code RelationService.block} 连 Match 状态都没落库，
* 于是会话永远停留在 {@code active}，Unmatch/Block 后仍可继续发消息。
*
* @param matchId 会话绑定的 Match
* @param matchStatus Match 的新状态
*/
@Transactional
public void syncFromMatch(String matchId, String matchStatus) {
if (matchId == null) return;
conversationMapper.findByMatchId(matchId).ifPresent(conv -> {
switch (matchStatus) {
case Constants.MATCH_STATUS_UNMATCHED -> conv.setStatus(Constants.CONVERSATION_STATUS_UNMATCHED);
case Constants.MATCH_STATUS_ACTIVE, Constants.MATCH_STATUS_GETTING_TO_KNOW -> conv.setStatus(Constants.CONVERSATION_STATUS_ACTIVE);
default -> {}
boolean alive = Constants.MATCH_STATUS_ACTIVE.equals(matchStatus)
|| Constants.MATCH_STATUS_GETTING_TO_KNOW.equals(matchStatus);
if (alive) {
// 恢复可聊（如 prior Unmatch 后活动互选复用同一 Match）
if (!Constants.CONVERSATION_STATUS_ACTIVE.equals(conv.getStatus())) {
conv.setStatus(Constants.CONVERSATION_STATUS_ACTIVE);
conversationMapper.updateById(conv);
log.info("Conversation reopened: conv={}, match={}", conv.getId(), matchId);
}
return;
}
conversationMapper.updateById(conv);
// 不再可聊：只推事件，不改 conversation.status（原因见 closeConversationForMatch）
notifyConversationClosed(conv.getId(), matchId, Constants.CONVERSATION_STATUS_UNMATCHED);
});
}
/**
* Match 失效（Unmatch / Block）时通知双方会话已关闭。
*
* <p><b>为什么只推事件、不改 conversation.status</b>：
* {@code ConversationMapper.findActiveByUser} 过滤 {@code status = 'active'}，
* 一旦把 status 改成 {@code unmatched}/{@code blocked}，会话就会从"我的消息"列表里消失，
* 这等于静默改变了列表可见性 —— 而 {@code conversation-item} 组件恰恰要在列表里
* 渲染"会话已停用 / 会话已停用（已拉黑）"的灰显态，AGENTS 也要求 Block 保留历史可查。
* 因此关闭语义由"事件 + 发送时校验"承载，而非由 status 承载：
* <ul>
* <li>发送权限：{@link #assertMatchAlive} + 拉黑关系双向校验，与 status 无关，无法绕过；</li>
* <li>实时表现：本方法推送 {@code conversation.closed}，客户端据此立即灰显；</li>
* <li>读接口：{@code toConversationDTO} 已经用 {@code matchAlive && !blocked}
* 算出 {@code canSendMessage=false} 与 {@code relationStatus}。</li>
* </ul>
* 列表可见性是否应在关闭后收回，属产品决策，未在本阶段授权范围内，故保持现状并登记为待决项。
*
* @param matchId Match id
* @param reasonStatus 通知里携带的状态：{@code unmatched} 或 {@code blocked}
*/
@Transactional
public void closeConversationForMatch(String matchId, String reasonStatus) {
if (matchId == null) return;
conversationMapper.findByMatchId(matchId)
.ifPresent(conv -> notifyConversationClosed(conv.getId(), matchId, reasonStatus));
}
/** 会话关闭后通知全部成员（规范第 124 章 conversation.closed） */
private void notifyConversationClosed(String conversationId, String matchId, String status) {
Map<String, Object> data = new LinkedHashMap<>();
data.put("conversationId", conversationId);
data.put("matchId", matchId);
data.put("status", status);
memberMapper.findByConversation(conversationId)
.forEach(m -> webSocketHandler.sendEvent(
m.getUserId(), ChatWebSocketHandler.EVENT_CONVERSATION_CLOSED, data));
log.info("Conversation closed: conv={}, match={}, status={}", conversationId, matchId, status);
}
/**
* INV-005 权威校验：会话绑定的 Match 必须仍处于可聊天状态。
*
* <p>仅当 Match 已不存在、或状态不在
* {active, getting_to_know} 之内时拒绝。没有绑定 Match 的会话不受此限
* （预留给将来非 Match 来源的会话）。
*/
private void assertMatchAlive(Conversation conv) {
String matchId = conv.getMatchId();
if (matchId == null) return;
Match match = matchMapper.findById(matchId).orElse(null);
boolean alive = match != null
&& (Constants.MATCH_STATUS_ACTIVE.equals(match.getStatus())
|| Constants.MATCH_STATUS_GETTING_TO_KNOW.equals(match.getStatus()));
if (!alive) {
log.warn("Message rejected by MATCH_NOT_ALIVE: conv={}, match={}, status={}",
conv.getId(), matchId, match == null ? "missing" : match.getStatus());
throw new IllegalStateException("匹配已解除，无法发送消息");
}
}
/**
* 构造规范第 125 章 message.new 的 data 负载。
*
* <p>字段同时满足两份契约：
* <ul>
* <li>规范示例用 {@code id / type / createdAt}；</li>
* <li>小程序 {@code types/index.d.ts} 的 {@code Message} 用
* {@code messageId / messageType / sentAt(epoch 毫秒)}，
* 且 {@code stores/messageStore.js#onIncomingMessage} 按这些名字取值并做去重。</li>
* </ul>
* 故输出为两者并集，避免任一消费方断链。
*/
private Map<String, Object> buildMessageNewData(String conversationId, Message msg) {
Map<String, Object> message = new LinkedHashMap<>();
message.put("messageId", msg.getId());
message.put("id", msg.getId());
message.put("senderId", msg.getSenderId());
message.put("messageType", msg.getMessageType());
message.put("type", msg.getMessageType());
message.put("content", msg.getContent());
message.put("fileId", msg.getFileId());
message.put("clientMessageId", msg.getClientMessageId());
message.put("status", msg.getStatus());
message.put("riskLevel", msg.getRiskLevel());
message.put("createdAt", msg.getCreatedAt());
message.put("sentAt", msg.getCreatedAt() == null ? null : msg.getCreatedAt().toEpochMilli());
Map<String, Object> data = new LinkedHashMap<>();
data.put("conversationId", conversationId);
data.put("message", message);
return data;
}
// ==================== 会话查询（原 messaging 模块能力合并） ====================
/**
backend/src/main/java/com/danyang/match/platform/chat/websocket/ChatHandshakeInterceptor.java
+130
View File
@@ -0,0 +1,130 @@
package com.danyang.match.platform.chat.websocket;
import com.danyang.match.platform.auth.token.JwtTokenService;
import lombok.RequiredArgsConstructor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.server.ServerHttpRequest;
import org.springframework.http.server.ServerHttpResponse;
import org.springframework.http.server.ServletServerHttpRequest;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.server.HandshakeInterceptor;
import org.springframework.web.util.UriComponentsBuilder;
import jakarta.servlet.http.HttpServletRequest;
import java.util.Map;
/**
* WebSocket 握手鉴权（规范第 126 章 WebSocket 安全原则）。
*
* <p>规范要求：<b>Token 认证 / 连接绑定 userId</b>。本拦截器在握手阶段完成这两件事，
* 只有认证通过才放行，并把 userId 写入 session attributes 供
* {@link ChatWebSocketHandler} 识别连接归属。
*
* <p>历史缺陷（本类修复）：此前<b>全局没有任何 HandshakeInterceptor</b>，也没有任何代码
* 往 session attributes 里写 userId，导致 {@code ChatWebSocketHandler.extractUserId()}
* 恒返回 null、{@code sessions} 映射永远为空 —— <b>整条 WebSocket 下行推送静默失效</b>，
* 且未认证连接可以随意建立（违反规范第 126 章）。
*
* <p>鉴权方式（按优先级）：
* <ol>
* <li>握手 URL 查询参数 {@code ?token=xxx}（小程序 {@code wx.connectSocket} 最易实现）</li>
* <li>握手请求头 {@code Authorization: Bearer xxx}</li>
* </ol>
*
* <p>仅接受<b>用户</b> access token（{@code type=access}）。运营端 token
* （{@code type=operator_access}）一律拒绝 —— AGENTS 第 7 条：普通工作人员不能查看私人聊天。
*/
@Component
@RequiredArgsConstructor
public class ChatHandshakeInterceptor implements HandshakeInterceptor {
private static final Logger log = LoggerFactory.getLogger(ChatHandshakeInterceptor.class);
/** session attributes 中承载用户身份的键，与 ChatWebSocketHandler 约定一致 */
public static final String ATTR_USER_ID = "userId";
private static final String QUERY_PARAM_TOKEN = "token";
private static final String BEARER_PREFIX = "Bearer ";
private static final String TOKEN_TYPE_USER_ACCESS = "access";
private final JwtTokenService jwtTokenService;
@Override
public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response,
WebSocketHandler wsHandler, Map<String, Object> attributes) {
String token = resolveToken(request);
if (token == null || token.isBlank()) {
log.warn("WS handshake rejected: 缺少 token");
return false;
}
// fail-closed：token 校验内部会查 Redis 黑名单，任何异常（如 Redis 不可用）
// 都必须拒绝握手，否则会在握手阶段抛异常而不是给出可判定的拒绝。
try {
if (!jwtTokenService.validateToken(token)) {
log.warn("WS handshake rejected: token 无效或已进黑名单");
return false;
}
String type = jwtTokenService.parseToken(token).get("type", String.class);
if (!TOKEN_TYPE_USER_ACCESS.equals(type)) {
log.warn("WS handshake rejected: 非用户 token, type={}", type);
return false;
}
String userId = jwtTokenService.getUserIdFromToken(token);
if (userId == null || userId.isBlank()) {
log.warn("WS handshake rejected: token 无 subject");
return false;
}
attributes.put(ATTR_USER_ID, userId);
log.info("WS handshake accepted: userId={}", userId);
return true;
} catch (Exception e) {
log.warn("WS handshake rejected: token 校验异常 {}", e.getMessage());
return false;
}
}
@Override
public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response,
WebSocketHandler wsHandler, Exception exception) {
if (exception != null) {
log.warn("WS handshake error: {}", exception.getMessage());
}
}
/**
* 从查询参数或 Authorization 头解析 token。
*
* <p>查询参数优先：小程序端 WebSocket 请求头支持有限，规范给的也是
* {@code /wss/chat} 直连形式。
*/
private String resolveToken(ServerHttpRequest request) {
String fromQuery = UriComponentsBuilder.fromUri(request.getURI())
.build()
.getQueryParams()
.getFirst(QUERY_PARAM_TOKEN);
if (fromQuery != null && !fromQuery.isBlank()) {
return fromQuery;
}
String header = request.getHeaders().getFirst("Authorization");
if (header != null && header.startsWith(BEARER_PREFIX)) {
return header.substring(BEARER_PREFIX.length());
}
// 兼容 Servlet 容器下未透传到 header 映射的场景
if (request instanceof ServletServerHttpRequest servletRequest) {
HttpServletRequest raw = servletRequest.getServletRequest();
String rawHeader = raw.getHeader("Authorization");
if (rawHeader != null && rawHeader.startsWith(BEARER_PREFIX)) {
return rawHeader.substring(BEARER_PREFIX.length());
}
}
return null;
}
}
backend/src/main/java/com/danyang/match/platform/chat/websocket/ChatWebSocketHandler.java
+97 -34
View File
@@ -1,5 +1,8 @@
package com.danyang.match.platform.chat.websocket;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
@@ -9,81 +12,141 @@ import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import java.io.IOException;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
/**
* WebSocket Handler — 实时消息推送
* WebSocket Handler — 实时消息推送（下行）
*
* 简易实现：内存 Map 维护 userId → Session 映射
* 多实例部署时需要 Redis Pub/Sub 扩展，Phase 22+ 再做
* <p>简易实现：内存 Map 维护 userId → Session 映射。
* 多实例部署时需要 Redis Pub/Sub 扩展（当前为单实例）。
*
* <p><b>协议</b>：事件信封按《REST API 接口详细设计规范》第 125 章：
* <pre>{ "event": "message.new", "data": { ... } }</pre>
*
* <p><b>鉴权与订阅边界</b>（规范第 126 章）：
* <ul>
* <li>Token 认证 + 连接绑定 userId —— 由 {@link ChatHandshakeInterceptor} 在握手阶段完成；
* 本类只信任 session attributes 里的 userId，绝不接受客户端自报身份。</li>
* <li>服务端决定订阅权限 —— 上行帧<b>不做订阅处理</b>。规范明确禁止
* "客户端自行订阅任意 conversationId"，故上行只支持 {@code ping} 保活，
* 其余一律忽略。消息落库与推送范围完全由服务端按会话成员计算。</li>
* </ul>
*/
@Component
public class ChatWebSocketHandler extends TextWebSocketHandler {
private static final Logger log = LoggerFactory.getLogger(ChatWebSocketHandler.class);
/** userId -> WebSocketSession */
/** 事件名（规范第 124 章清单） */
public static final String EVENT_MESSAGE_NEW = "message.new";
public static final String EVENT_MESSAGE_READ = "message.read";
public static final String EVENT_CONVERSATION_CLOSED = "conversation.closed";
private static final String UPLINK_PING = "ping";
private static final String EVENT_PONG = "pong";
/** userId -> WebSocketSession（同一用户重连时覆盖旧连接） */
private final Map<String, WebSocketSession> sessions = new ConcurrentHashMap<>();
/** userId -> Set<conversationId> 当前订阅的会话 */
private final Map<String, Set<String>> subscriptions = new ConcurrentHashMap<>();
/** Instant 等 JSR310 类型需要模块支持，否则推送序列化抛 InvalidDefinitionException */
private final ObjectMapper objectMapper = new ObjectMapper().registerModule(new JavaTimeModule());
@Override
public void afterConnectionEstablished(WebSocketSession session) {
String userId = extractUserId(session);
if (userId != null) {
sessions.put(userId, session);
log.info("WS connected: userId={}, session={}", userId, session.getId());
if (userId == null) {
// 理论上不可达：握手拦截器已拒绝无 userId 的连接。兜底关闭，避免留下匿名会话。
log.warn("WS connected without userId, closing: session={}", session.getId());
try {
session.close(CloseStatus.NOT_ACCEPTABLE);
} catch (IOException ignored) {
// 连接已断开
}
return;
}
WebSocketSession previous = sessions.put(userId, session);
if (previous != null && previous.isOpen() && !previous.getId().equals(session.getId())) {
// 同账号多端：以最新连接为推送目标，旧连接主动关闭（避免重复投递）
try {
previous.close(CloseStatus.NORMAL.withReason("replaced by new session"));
} catch (IOException ignored) {
// 旧连接已断开
}
}
log.info("WS connected: userId={}, session={}", userId, session.getId());
}
@Override
protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
// 客户端上行消息目前只处理 ping/pong 和 typing/read 类事件
// 发消息走 HTTP REST 端点（有完整校验 + 风控 + 幂等）
// WS 只用于下行推送
log.debug("WS received from {}: {}", session.getId(), message.getPayload());
// 发消息走 HTTP REST（有完整校验 + 风控 + 幂等）；WS 只用于下行推送。
// 上行仅支持 ping 保活；不处理订阅类指令（规范第 126 章：不允许客户端自行订阅任意 conversationId）。
String payload = message.getPayload();
if (payload == null) return;
String trimmed = payload.trim();
if (UPLINK_PING.equalsIgnoreCase(trimmed) || trimmed.contains("\"ping\"")) {
session.sendMessage(new TextMessage(buildEnvelope(EVENT_PONG, Map.of("ts", System.currentTimeMillis()))));
return;
}
log.debug("WS 上行帧已忽略（非 ping）: session={}, payload={}", session.getId(), trimmed);
}
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
// 仅当该 session 仍是当前登记连接时才移除，避免重连竞态把新连接误删
sessions.entrySet().removeIf(e -> e.getValue().getId().equals(session.getId()));
log.info("WS disconnected: session={}, status={}", session.getId(), status);
}
// ==================== 下行推送 API ====================
/** 推送消息给指定用户 */
public void sendToUser(String userId, String json) {
/**
* 按规范信封推送事件给指定用户。
*
* @param userId 目标用户；未连接时静默跳过（不抛异常，不影响主业务事务）
* @param event 事件名，见规范第 124 章
* @param data 事件负载
*/
public void sendEvent(String userId, String event, Object data) {
if (userId == null || event == null) return;
WebSocketSession session = sessions.get(userId);
if (session != null && session.isOpen()) {
try {
session.sendMessage(new TextMessage(json));
} catch (IOException e) {
log.warn("WS send failed to {}: {}", userId, e.getMessage());
}
if (session == null || !session.isOpen()) return;
try {
session.sendMessage(new TextMessage(buildEnvelope(event, data)));
} catch (IOException e) {
log.warn("WS send failed to {}: {}", userId, e.getMessage());
}
}
/** 推送消息给双方会话成员 */
public void sendToConversation(String convId, String userA, String userB, String json) {
sendToUser(userA, json);
sendToUser(userB, json);
/** 批量推送同一事件给多个用户（如会话双方） */
public void sendEventToUsers(Collection<String> userIds, String event, Object data) {
if (userIds == null) return;
for (String uid : userIds) {
sendEvent(uid, event, data);
}
}
/** 广播到所有连接（慎用） */
public void broadcast(String json) {
sessions.values().forEach(s -> {
if (s.isOpen()) {
try { s.sendMessage(new TextMessage(json)); } catch (IOException ignored) {}
}
});
/** 当前在线连接数（健康检查 / 排障用） */
public int onlineCount() {
return sessions.size();
}
private String buildEnvelope(String event, Object data) {
Map<String, Object> envelope = new LinkedHashMap<>();
envelope.put("event", event);
envelope.put("data", data);
try {
return objectMapper.writeValueAsString(envelope);
} catch (JsonProcessingException e) {
log.warn("WS envelope serialize failed: event={}, err={}", event, e.getMessage());
return "{\"event\":\"" + event + "\",\"data\":null}";
}
}
private String extractUserId(WebSocketSession session) {
Object attr = session.getAttributes().get("userId");
Object attr = session.getAttributes().get(ChatHandshakeInterceptor.ATTR_USER_ID);
return attr != null ? attr.toString() : null;
}
}
backend/src/main/java/com/danyang/match/platform/match/controller/MatchController.java
+6
View File
@@ -1,11 +1,13 @@
package com.danyang.match.platform.match.controller;
import com.danyang.match.platform.chat.service.ChatService;
import com.danyang.match.platform.match.entity.Match;
import com.danyang.match.platform.match.mapper.MatchMapper;
import com.danyang.match.platform.match.mapper.UserRelationMapper;
import com.danyang.match.platform.match.service.MatchService;
import com.danyang.match.platform.match.service.RelationService;
import com.danyang.match.platform.relationship.service.RelationshipStatusService;
import com.danyang.match.platform.shared.constant.Constants;
import com.danyang.match.platform.shared.dto.ApiResponse;
import com.danyang.match.platform.shared.dto.CursorPage;
import com.danyang.match.platform.shared.security.principal.UserPrincipal;
@@ -42,6 +44,7 @@ public class MatchController {
private final UserRelationMapper relationMapper;
private final RelationService relationService;
private final RelationshipStatusService relationshipStatusService;
private final ChatService chatService;
@GetMapping
public ApiResponse<CursorPage<MatchResponse>> list(@AuthenticationPrincipal UserPrincipal principal) {
@@ -78,6 +81,9 @@ public class MatchController {
}
MatchService.unmatch(m, principal.id());
matchMapper.updateById(m);
// Phase 09: Unmatch 必须即时禁聊 —— 同步关闭会话并向双方推送 conversation.closed。
// 历史缺陷：此处只改 Match 状态、不联动会话，导致会话仍为 active、对方还能继续发消息。
chatService.syncFromMatch(m.getId(), Constants.MATCH_STATUS_UNMATCHED);
return ApiResponse.ok();
}
backend/src/main/java/com/danyang/match/platform/match/service/RelationService.java
+12 -3
View File
@@ -1,6 +1,7 @@
package com.danyang.match.platform.match.service;
import com.danyang.match.platform.capability.service.CapabilityService;
import com.danyang.match.platform.chat.service.ChatService;
import com.danyang.match.platform.match.entity.Match;
import com.danyang.match.platform.match.entity.UserRelation;
import com.danyang.match.platform.match.mapper.MatchMapper;
@@ -35,6 +36,7 @@ public class RelationService {
private final MatchMapper matchMapper;
private final UserMapper userMapper;
private final CapabilityService capabilityService;
private final ChatService chatService;
// ==================== 公共查询 ====================
@@ -156,9 +158,16 @@ public class RelationService {
relationMapper.insert(block);
// 自动 Unmatch active Match
matchMapper.findActiveBetween(userId, targetUserId).ifPresent(m ->
MatchService.unmatch(m, userId)
);
matchMapper.findActiveBetween(userId, targetUserId).ifPresent(m -> {
MatchService.unmatch(m, userId);
// 历史缺陷 1：MatchService.unmatch 只是静态工具方法，仅改内存态、不落库，
// 此处先前没有 updateById，"自动 Unmatch" 实际是空操作 ——
// Block 之后 Match 仍留在 active，双方还显示为已匹配。
matchMapper.updateById(m);
// 历史缺陷 2：Block 后会话从未被通知，对方界面不会立即停用。
// Block 优先级最高，故通知状态用 blocked 而非 unmatched。
chatService.closeConversationForMatch(m.getId(), Constants.CONVERSATION_STATUS_BLOCKED);
});
log.info("User {} blocked {}", userId, targetUserId);
}
backend/src/main/java/com/danyang/match/platform/relationship/service/RelationshipService.java
+2
View File
@@ -168,6 +168,8 @@ public class RelationshipService {
match.setStatus(Constants.MATCH_STATUS_UNMATCHED);
match.setUnmatchedAt(Instant.now());
matchMapper.updateById(match);
// Phase 09: Block 优先级最高 —— 通知双方会话已停用（此前无任何联动）
chatService.closeConversationForMatch(match.getId(), Constants.CONVERSATION_STATUS_BLOCKED);
});
log.info("拉黑用户: userId={}, targetUserId={}", userId, targetUserId);