diff --git a/backend/src/main/java/com/danyang/match/platform/chat/config/WebSocketConfig.java b/backend/src/main/java/com/danyang/match/platform/chat/config/WebSocketConfig.java new file mode 100644 index 0000000..d4407c6 --- /dev/null +++ b/backend/src/main/java/com/danyang/match/platform/chat/config/WebSocketConfig.java @@ -0,0 +1,36 @@ +package com.danyang.match.platform.chat.config; + +import com.danyang.match.platform.chat.websocket.ChatWebSocketHandler; +import lombok.RequiredArgsConstructor; +import org.springframework.context.annotation.Configuration; +import org.springframework.web.socket.config.annotation.EnableWebSocket; +import org.springframework.web.socket.config.annotation.WebSocketConfigurer; +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 解除 + * + * 鉴权: 握手时 URL 带 JWT token（与 REST 同密钥） + */ +@Configuration +@EnableWebSocket +@RequiredArgsConstructor +public class WebSocketConfig implements WebSocketConfigurer { + + private final ChatWebSocketHandler chatHandler; + + @Override + public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { + registry.addHandler(chatHandler, "/ws/chat") + .setAllowedOriginPatterns("*") + .withSockJS(); // SockJS fallback for miniapp 兼容 + } +} diff --git a/backend/src/main/java/com/danyang/match/platform/chat/controller/ConversationController.java b/backend/src/main/java/com/danyang/match/platform/chat/controller/ConversationController.java new file mode 100644 index 0000000..abd1054 --- /dev/null +++ b/backend/src/main/java/com/danyang/match/platform/chat/controller/ConversationController.java @@ -0,0 +1,77 @@ +package com.danyang.match.platform.chat.controller; + +import com.danyang.match.platform.chat.entity.Conversation; +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.shared.dto.ApiResponse; +import com.danyang.match.platform.shared.security.principal.UserPrincipal; +import lombok.Data; +import lombok.RequiredArgsConstructor; +import org.springframework.security.core.annotation.AuthenticationPrincipal; +import org.springframework.web.bind.annotation.*; + +import java.util.List; +import java.util.stream.Collectors; + +/** + * Conversation 端点 + * + * POST /api/v1/conversations/from-match/{matchId} — 从 Match 创建或获取 Conversation + * GET /api/v1/conversations/{id} — 详情 + * GET /api/v1/conversations — 我的活跃 Conversation 列表 + */ +@RestController +@RequestMapping("/conversations") +@RequiredArgsConstructor +public class ConversationController { + + private final ChatService chatService; + private final MatchMapper matchMapper; + + /** + * 从 Match 创建/获取 Conversation + * 前端拿到 mutual Match 后调这个 + * + * P09-V11-T01: Activity Match / user_like Match / matchmaker Match + * 三种 source 的 Match 都能建 Conversation。 + */ + @PostMapping("/from-match/{matchId}") + public ApiResponse fromMatch(@PathVariable String matchId, + @AuthenticationPrincipal UserPrincipal principal) { + if (principal == null) return ApiResponse.unauthorized(); + Match match = matchMapper.findById(matchId).orElse(null); + if (match == null) return ApiResponse.notFound(); + // 权限：必须是 Match 参与者 + if (!match.getUserAId().equals(principal.id()) && !match.getUserBId().equals(principal.id())) { + return ApiResponse.forbidden(); + } + Conversation conv = chatService.getOrCreateConversation(matchId, match.getUserAId(), match.getUserBId()); + return ApiResponse.ok(ConversationResponse.from(conv, match)); + } + + @GetMapping + public ApiResponse> list(@AuthenticationPrincipal UserPrincipal principal) { + if (principal == null) return ApiResponse.unauthorized(); + List convIds = List.of(); // TODO: 通过 memberMapper 拿 + return ApiResponse.ok(List.of()); + } + + @Data + public static class ConversationResponse { + private String id; + private String matchId; + private String status; + private String matchSource; + private String peerUserId; + + public static ConversationResponse from(Conversation conv, Match match) { + ConversationResponse r = new ConversationResponse(); + r.id = conv.getId(); + r.matchId = conv.getMatchId(); + r.status = conv.getStatus(); + r.matchSource = match.getSource(); + return r; + } + } +} diff --git a/backend/src/main/java/com/danyang/match/platform/chat/controller/MessageController.java b/backend/src/main/java/com/danyang/match/platform/chat/controller/MessageController.java new file mode 100644 index 0000000..18ec569 --- /dev/null +++ b/backend/src/main/java/com/danyang/match/platform/chat/controller/MessageController.java @@ -0,0 +1,94 @@ +package com.danyang.match.platform.chat.controller; + +import com.danyang.match.platform.chat.entity.Message; +import com.danyang.match.platform.chat.service.ChatService; +import com.danyang.match.platform.shared.dto.ApiResponse; +import com.danyang.match.platform.shared.security.principal.UserPrincipal; +import lombok.Data; +import lombok.RequiredArgsConstructor; +import org.springframework.security.core.annotation.AuthenticationPrincipal; +import org.springframework.web.bind.annotation.*; + +import java.util.List; +import java.util.Map; + +/** + * Message 端点（HTTP REST + 幂等） + * + * POST /api/v1/conversations/{id}/messages — 发消息 + * GET /api/v1/conversations/{id}/messages — 拉历史 + * PUT /api/v1/conversations/{id}/messages/read — 标记已读 + * + * 实时推送走 WebSocket（Phase 09 同时提供，Phase 10 Chat Safety 加风控） + */ +@RestController +@RequestMapping("/conversations") +@RequiredArgsConstructor +public class MessageController { + + private final ChatService chatService; + + @PostMapping("/{id}/messages") + public ApiResponse send(@PathVariable String id, + @RequestBody SendRequest req, + @AuthenticationPrincipal UserPrincipal principal) { + if (principal == null) return ApiResponse.unauthorized(); + Message msg = chatService.sendText(id, principal.id(), req.content, req.clientMessageId); + return ApiResponse.ok(MessageResponse.from(msg)); + } + + @GetMapping("/{id}/messages") + public ApiResponse> load(@PathVariable String id, + @RequestParam(defaultValue = "30") int limit, + @RequestParam(defaultValue = "0") int offset, + @AuthenticationPrincipal UserPrincipal principal) { + if (principal == null) return ApiResponse.unauthorized(); + List messages = chatService.loadMessages(id, Math.min(limit, 100), offset); + return ApiResponse.ok(messages.stream().map(MessageResponse::from).toList()); + } + + @PutMapping("/{id}/messages/read") + public ApiResponse markRead(@PathVariable String id, + @RequestBody Map body, + @AuthenticationPrincipal UserPrincipal principal) { + if (principal == null) return ApiResponse.unauthorized(); + String lastReadId = body.get("lastReadMessageId"); + if (lastReadId != null) { + chatService.markAsRead(id, principal.id(), lastReadId); + } + return ApiResponse.ok(); + } + + // ==================== DTO ==================== + + @Data + public static class SendRequest { + private String content; + private String clientMessageId; // 幂等键 + } + + @Data + public static class MessageResponse { + private String id; + private String conversationId; + private String senderId; + private String clientMessageId; + private String messageType; + private String content; + private String status; + private java.time.Instant createdAt; + + public static MessageResponse from(Message m) { + MessageResponse r = new MessageResponse(); + r.id = m.getId(); + r.conversationId = m.getConversationId(); + r.senderId = m.getSenderId(); + r.clientMessageId = m.getClientMessageId(); + r.messageType = m.getMessageType(); + r.content = m.getContent(); + r.status = m.getStatus(); + r.createdAt = m.getCreatedAt(); + return r; + } + } +} diff --git a/backend/src/main/java/com/danyang/match/platform/chat/entity/Conversation.java b/backend/src/main/java/com/danyang/match/platform/chat/entity/Conversation.java new file mode 100644 index 0000000..965a7f8 --- /dev/null +++ b/backend/src/main/java/com/danyang/match/platform/chat/entity/Conversation.java @@ -0,0 +1,26 @@ +package com.danyang.match.platform.chat.entity; + +import com.baomidou.mybatisplus.annotation.TableName; +import com.danyang.match.platform.shared.base.BaseEntity; +import lombok.Data; +import lombok.EqualsAndHashCode; + +/** + * 会话（一对一聊天） + * 对应 DB: conversations + * + * 一个 Conversation 绑定一个 Match（match_id UNIQUE） + * - Match active / getting_to_know → conversation active + * - Match unmatched → conversation unmatched + * - 任一方 Block → conversation blocked + */ +@Data +@EqualsAndHashCode(callSuper = true) +@TableName("conversations") +public class Conversation extends BaseEntity { + + private String matchId; + private String status; // active / unmatched / blocked + private String lastMessageId; + private java.time.Instant lastMessageAt; +} diff --git a/backend/src/main/java/com/danyang/match/platform/chat/entity/ConversationMember.java b/backend/src/main/java/com/danyang/match/platform/chat/entity/ConversationMember.java new file mode 100644 index 0000000..371e688 --- /dev/null +++ b/backend/src/main/java/com/danyang/match/platform/chat/entity/ConversationMember.java @@ -0,0 +1,23 @@ +package com.danyang.match.platform.chat.entity; + +import com.baomidou.mybatisplus.annotation.TableName; +import lombok.Data; + +import java.io.Serializable; +import java.time.Instant; + +/** + * 会话成员（复合主键 conversation_id + user_id） + * 对应 DB: conversation_members + */ +@Data +@TableName("conversation_members") +public class ConversationMember implements Serializable { + + private String conversationId; + private String userId; + private String lastReadMessageId; + private Integer unreadCount; + private Boolean canSend; + private Instant joinedAt; +} diff --git a/backend/src/main/java/com/danyang/match/platform/chat/entity/Message.java b/backend/src/main/java/com/danyang/match/platform/chat/entity/Message.java new file mode 100644 index 0000000..4d69ac9 --- /dev/null +++ b/backend/src/main/java/com/danyang/match/platform/chat/entity/Message.java @@ -0,0 +1,28 @@ +package com.danyang.match.platform.chat.entity; + +import com.baomidou.mybatisplus.annotation.TableName; +import com.danyang.match.platform.shared.base.BaseEntity; +import lombok.Data; +import lombok.EqualsAndHashCode; + +/** + * 消息 + * 对应 DB: messages + * + * message_type: text / image / system / risk_notice + * status: sent / delivered / read / failed + */ +@Data +@EqualsAndHashCode(callSuper = true) +@TableName("messages") +public class Message extends BaseEntity { + + private String conversationId; + private String senderId; + private String clientMessageId; // 客户端生成的幂等 ID + private String messageType; + private String content; + private String fileId; + private String status; + private String riskLevel; // 内容风控标记，Phase 10 用 +} diff --git a/backend/src/main/java/com/danyang/match/platform/chat/mapper/ConversationMapper.java b/backend/src/main/java/com/danyang/match/platform/chat/mapper/ConversationMapper.java new file mode 100644 index 0000000..e2c6110 --- /dev/null +++ b/backend/src/main/java/com/danyang/match/platform/chat/mapper/ConversationMapper.java @@ -0,0 +1,16 @@ +package com.danyang.match.platform.chat.mapper; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import com.danyang.match.platform.chat.entity.Conversation; +import org.apache.ibatis.annotations.Mapper; +import org.apache.ibatis.annotations.Param; +import org.apache.ibatis.annotations.Select; + +import java.util.Optional; + +@Mapper +public interface ConversationMapper extends BaseMapper { + + @Select("SELECT * FROM conversations WHERE match_id = #{matchId} LIMIT 1") + Optional findByMatchId(@Param("matchId") String matchId); +} diff --git a/backend/src/main/java/com/danyang/match/platform/chat/mapper/ConversationMemberMapper.java b/backend/src/main/java/com/danyang/match/platform/chat/mapper/ConversationMemberMapper.java new file mode 100644 index 0000000..d7c2640 --- /dev/null +++ b/backend/src/main/java/com/danyang/match/platform/chat/mapper/ConversationMemberMapper.java @@ -0,0 +1,46 @@ +package com.danyang.match.platform.chat.mapper; + +import com.danyang.match.platform.chat.entity.ConversationMember; +import org.apache.ibatis.annotations.Insert; +import org.apache.ibatis.annotations.Mapper; +import org.apache.ibatis.annotations.Param; +import org.apache.ibatis.annotations.Select; +import org.apache.ibatis.annotations.Update; + +import java.time.Instant; +import java.util.List; +import java.util.Optional; + +@Mapper +public interface ConversationMemberMapper { + + @Select("SELECT * FROM conversation_members WHERE conversation_id = #{convId} AND user_id = #{userId}") + Optional find(@Param("convId") String convId, @Param("userId") String userId); + + @Select("SELECT * FROM conversation_members WHERE conversation_id = #{convId}") + List findByConversation(@Param("convId") String convId); + + @Select("SELECT conversation_id FROM conversation_members WHERE user_id = #{userId} AND conversation_id IN (" + + "SELECT id FROM conversations WHERE status = 'active' ORDER BY last_message_at DESC" + + ") LIMIT 100") + List findActiveConversationIdsByUser(@Param("userId") String userId); + + @Insert("INSERT INTO conversation_members (conversation_id, user_id, last_read_message_id, unread_count, can_send, joined_at) " + + "VALUES (#{m.conversationId}, #{m.userId}, #{m.lastReadMessageId}, #{m.unreadCount}, #{m.canSend}, #{m.joinedAt}) " + + "ON CONFLICT (conversation_id, user_id) DO NOTHING") + void insertMemberIfAbsent(@Param("m") ConversationMember member); + + @Update("UPDATE conversation_members SET last_read_message_id = #{lastReadId}, unread_count = 0 " + + "WHERE conversation_id = #{convId} AND user_id = #{userId}") + void updateLastRead(@Param("convId") String convId, + @Param("userId") String userId, + @Param("lastReadId") String lastReadId); + + @Update("UPDATE conversation_members SET unread_count = 0 WHERE conversation_id = #{convId} AND user_id = #{userId}") + void resetUnread(@Param("convId") String convId, @Param("userId") String userId); + + @Update("UPDATE conversation_members SET unread_count = #{newCount} WHERE conversation_id = #{convId} AND user_id = #{userId}") + void incrementUnread(@Param("convId") String convId, + @Param("userId") String userId, + @Param("newCount") int newCount); +} diff --git a/backend/src/main/java/com/danyang/match/platform/chat/mapper/MessageMapper.java b/backend/src/main/java/com/danyang/match/platform/chat/mapper/MessageMapper.java new file mode 100644 index 0000000..8d3c594 --- /dev/null +++ b/backend/src/main/java/com/danyang/match/platform/chat/mapper/MessageMapper.java @@ -0,0 +1,34 @@ +package com.danyang.match.platform.chat.mapper; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import com.danyang.match.platform.chat.entity.Message; +import org.apache.ibatis.annotations.Mapper; +import org.apache.ibatis.annotations.Param; +import org.apache.ibatis.annotations.Select; + +import java.time.Instant; +import java.util.List; + +@Mapper +public interface MessageMapper extends BaseMapper { + + @Select("SELECT * FROM messages WHERE conversation_id = #{convId} ORDER BY created_at DESC LIMIT #{limit} OFFSET #{offset}") + List findRecentByConversation(@Param("convId") String convId, + @Param("limit") int limit, + @Param("offset") int offset); + + @Select("SELECT * FROM messages WHERE conversation_id = #{convId} AND id > #{afterId} ORDER BY created_at ASC LIMIT #{limit}") + List findAfter(@Param("convId") String convId, + @Param("afterId") String afterId, + @Param("limit") int limit); + + @Select("SELECT COUNT(*) FROM messages WHERE conversation_id = #{convId} AND sender_id <> #{userId} AND id > #{lastReadId}") + int countUnread(@Param("convId") String convId, + @Param("userId") String userId, + @Param("lastReadId") String lastReadId); + + @Select("SELECT * FROM messages WHERE conversation_id = #{convId} AND sender_id = #{senderId} AND client_message_id = #{clientMsgId} LIMIT 1") + Optional findByClientMessageId(@Param("convId") String convId, + @Param("senderId") String senderId, + @Param("clientMsgId") String clientMsgId); +} diff --git a/backend/src/main/java/com/danyang/match/platform/chat/service/ChatService.java b/backend/src/main/java/com/danyang/match/platform/chat/service/ChatService.java new file mode 100644 index 0000000..4a26597 --- /dev/null +++ b/backend/src/main/java/com/danyang/match/platform/chat/service/ChatService.java @@ -0,0 +1,189 @@ +package com.danyang.match.platform.chat.service; + +import com.danyang.match.platform.chat.entity.Conversation; +import com.danyang.match.platform.chat.entity.ConversationMember; +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.match.entity.Match; +import com.danyang.match.platform.match.mapper.MatchMapper; +import com.danyang.match.platform.match.mapper.UserRelationMapper; +import com.danyang.match.platform.shared.constant.Constants; +import lombok.RequiredArgsConstructor; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +import java.time.Instant; +import java.util.List; +import java.util.Optional; +import java.util.UUID; + +/** + * Chat 服务 + * + * 核心不变量： + * - INV-005: 未 Mutual 不可聊天 — 只有 Match active/getting_to_know 才能发消息 + * - INV-004: Block 优先级最高 — 双方任一 block → conversation blocked，禁发 + * - P09-V11-T01: Activity Match source 兼容 — Conversation 只绑 match_id，不关心 source + */ +@Service +@RequiredArgsConstructor +public class ChatService { + + private static final Logger log = LoggerFactory.getLogger(ChatService.class); + + private final ConversationMapper conversationMapper; + private final ConversationMemberMapper memberMapper; + private final MessageMapper messageMapper; + private final MatchMapper matchMapper; + private final UserRelationMapper relationMapper; + + // ==================== Conversation 创建 ==================== + + /** + * 为 Match 创建或复用 Conversation + * + * P09-V11-T01: 不管 Match source 是 user_like / activity_mutual / matchmaker， + * 只要 Match active 就能建 Conversation。 + * + * 幂等：Conversation 已存在 → 直接返回 + */ + @Transactional + public Conversation getOrCreateConversation(String matchId, String userA, String userB) { + Optional existing = conversationMapper.findByMatchId(matchId); + if (existing.isPresent()) return existing.get(); + + // 检查 Match 状态 + Match match = matchMapper.findById(matchId) + .orElseThrow(() -> new IllegalArgumentException("Match 不存在: " + matchId)); + if (!Constants.MATCH_STATUS_ACTIVE.equals(match.getStatus()) + && !Constants.MATCH_STATUS_GETTING_TO_KNOW.equals(match.getStatus())) { + throw new IllegalStateException("Match 已失效，无法创建会话"); + } + + Conversation conv = new Conversation(); + conv.setId(UUID.randomUUID().toString()); + conv.setMatchId(matchId); + conv.setStatus(Constants.CONVERSATION_STATUS_ACTIVE); + conversationMapper.insert(conv); + + // 两个成员 + for (String uid : List.of(userA, userB)) { + ConversationMember m = new ConversationMember(); + m.setConversationId(conv.getId()); + m.setUserId(uid); + m.setUnreadCount(0); + m.setCanSend(true); + m.setJoinedAt(Instant.now()); + // conversation_members 无 BaseMapper，用原生 insert + memberMapper.insertMemberIfAbsent(m); + } + + log.info("Conversation created: {} for match {}, users {}+{}", conv.getId(), matchId, userA, userB); + return conv; + } + + // ==================== 发消息 ==================== + + /** + * 发送文本消息 + * + * 校验： + * 1. 用户必须是 Conversation 成员 + * 2. Conversation 必须 active（未 unmatched / blocked） + * 3. 不能发空消息 + */ + @Transactional + public Message sendText(String conversationId, String senderId, String content, String clientMessageId) { + if (content == null || content.isBlank()) { + throw new IllegalArgumentException("消息内容不能为空"); + } + + Conversation conv = conversationMapper.findById(conversationId) + .orElseThrow(() -> new IllegalArgumentException("会话不存在: " + conversationId)); + + // 成员校验 + Optional senderMember = memberMapper.find(conversationId, senderId); + if (senderMember.isEmpty()) { + throw new IllegalStateException("您不是此会话成员"); + } + if (!Boolean.TRUE.equals(senderMember.getCanSend())) { + throw new IllegalStateException("当前被禁言"); + } + + // 会话状态校验 + if (!Constants.CONVERSATION_STATUS_ACTIVE.equals(conv.getStatus())) { + throw new IllegalStateException("会话已关闭"); + } + + // 幂等：client_message_id 已存在 → 返回已有 + if (clientMessageId != null) { + Optional dup = messageMapper.findByClientMessageId(conversationId, senderId, clientMessageId); + if (dup.isPresent()) return dup.get(); + } + + Message msg = new Message(); + msg.setId(UUID.randomUUID().toString()); + msg.setConversationId(conversationId); + msg.setSenderId(senderId); + msg.setClientMessageId(clientMessageId); + msg.setMessageType(Constants.MESSAGE_TYPE_TEXT); + msg.setContent(content); + msg.setStatus(Constants.MESSAGE_STATUS_SENT); + messageMapper.insert(msg); + + // 更新 Conversation last_message + conv.setLastMessageId(msg.getId()); + conv.setLastMessageAt(Instant.now()); + conversationMapper.updateById(conv); + + // 对方 unread +1 + List members = memberMapper.findByConversation(conversationId); + for (ConversationMember m : members) { + if (!m.getUserId().equals(senderId)) { + int unread = m.getUnreadCount() == null ? 0 : m.getUnreadCount(); + memberMapper.incrementUnread(conversationId, m.getUserId(), unread + 1); + } + } + + log.info("Message sent: conv={}, sender={}, id={}", conversationId, senderId, msg.getId()); + return msg; + } + + // ==================== 拉消息 ==================== + + public List loadMessages(String conversationId, int limit, int offset) { + return messageMapper.findRecentByConversation(conversationId, limit, offset); + } + + /** 拉取某个游标之后的新消息（增量同步） */ + public List loadAfter(String conversationId, String afterId, int limit) { + return messageMapper.findAfter(conversationId, afterId, limit); + } + + // ==================== 标记已读 ==================== + + @Transactional + public void markAsRead(String conversationId, String userId, String lastReadMessageId) { + memberMapper.updateLastRead(conversationId, userId, lastReadMessageId); + memberMapper.resetUnread(conversationId, userId); + } + + // ==================== 状态联动 ==================== + + /** Match 状态变更时同步更新 Conversation 状态 */ + @Transactional + public void syncFromMatch(String matchId, String matchStatus) { + 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 -> {} + } + conversationMapper.updateById(conv); + }); + } +} diff --git a/backend/src/main/java/com/danyang/match/platform/chat/websocket/ChatWebSocketHandler.java b/backend/src/main/java/com/danyang/match/platform/chat/websocket/ChatWebSocketHandler.java new file mode 100644 index 0000000..4abb37b --- /dev/null +++ b/backend/src/main/java/com/danyang/match/platform/chat/websocket/ChatWebSocketHandler.java @@ -0,0 +1,89 @@ +package com.danyang.match.platform.chat.websocket; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Component; +import org.springframework.web.socket.CloseStatus; +import org.springframework.web.socket.TextMessage; +import org.springframework.web.socket.WebSocketSession; +import org.springframework.web.socket.handler.TextWebSocketHandler; + +import java.io.IOException; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +/** + * WebSocket Handler — 实时消息推送 + * + * 简易实现：内存 Map 维护 userId → Session 映射 + * 多实例部署时需要 Redis Pub/Sub 扩展，Phase 22+ 再做 + */ +@Component +public class ChatWebSocketHandler extends TextWebSocketHandler { + + private static final Logger log = LoggerFactory.getLogger(ChatWebSocketHandler.class); + + /** userId -> WebSocketSession */ + private final Map sessions = new ConcurrentHashMap<>(); + + /** userId -> Set 当前订阅的会话 */ + private final Map> subscriptions = new ConcurrentHashMap<>(); + + @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()); + } + } + + @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()); + } + + @Override + public void afterConnectionClosed(WebSocketSession session, CloseStatus status) { + 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) { + 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()); + } + } + } + + /** 推送消息给双方会话成员 */ + public void sendToConversation(String convId, String userA, String userB, String json) { + sendToUser(userA, json); + sendToUser(userB, json); + } + + /** 广播到所有连接（慎用） */ + public void broadcast(String json) { + sessions.values().forEach(s -> { + if (s.isOpen()) { + try { s.sendMessage(new TextMessage(json)); } catch (IOException ignored) {} + } + }); + } + + private String extractUserId(WebSocketSession session) { + Object attr = session.getAttributes().get("userId"); + return attr != null ? attr.toString() : null; + } +}