Compare commits
2 Commits
439c25be9c
...
554c6eaef4
| Author | SHA1 | Date |
|---|---|---|
Zhang Shun
|
554c6eaef4 | 2 weeks ago |
|
Zhang Shun
|
6fff2a2217 | 2 weeks ago |
55 changed files with 1333 additions and 86 deletions
@ -0,0 +1,18 @@ |
|||||
|
# 实时聊天转发片段：合并到现有 HTTPS server 中，不要覆盖整个 Nginx 配置。 |
||||
|
# 部署前核对真实后端地址、端口以及 /api/v1 context-path。 |
||||
|
# 此示例仅用于单后端实例；多实例需要共享消息代理，不能仅添加负载均衡。 |
||||
|
# 前端使用 wss://你的业务域名/api/v1/ws；Token 放在 STOMP CONNECT 头，不放 URL。 |
||||
|
location = /api/v1/ws { |
||||
|
proxy_pass http://127.0.0.1:8080; |
||||
|
proxy_http_version 1.1; |
||||
|
proxy_set_header Upgrade $http_upgrade; |
||||
|
proxy_set_header Connection "upgrade"; |
||||
|
proxy_set_header Host $host; |
||||
|
proxy_set_header X-Forwarded-Proto $scheme; |
||||
|
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; |
||||
|
proxy_read_timeout 75s; |
||||
|
proxy_send_timeout 75s; |
||||
|
proxy_buffering off; |
||||
|
} |
||||
|
# 握手限流目前按直接连接来源计数；经过反向代理时来源可能都是网关。 |
||||
|
# 大规模部署需在网关增加按可信真实 IP 的限流，并核对后端限流配置。 |
||||
@ -0,0 +1,68 @@ |
|||||
|
# 轻松聘聊天实时接口 |
||||
|
|
||||
|
本文补充 `openapi-rc.json` 中的消息发送、事件补拉和消息字段。所有 HTTP 路径以 `/api/v1` 为前缀；下文省略该前缀。 |
||||
|
|
||||
|
## 连接和发送 |
||||
|
|
||||
|
通过 STOMP WebSocket 连接 `/api/v1/ws`，在 **STOMP CONNECT 帧**中设置 `Authorization: Bearer <accessToken>`。不得放在 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 示例文件须经部署配置和验证后才能使用。 |
||||
@ -0,0 +1,10 @@ |
|||||
|
package com.mydreamjob.common; |
||||
|
|
||||
|
/** 对外品牌统一入口；不改变包名、数据库、接口路径及部署标识。 */ |
||||
|
public final class ProductBrand { |
||||
|
public static final String DISPLAY_NAME = "轻松聘"; |
||||
|
public static final String API_TITLE = DISPLAY_NAME + " API"; |
||||
|
public static final String API_DESCRIPTION = DISPLAY_NAME + " - 大学生求职服务平台 API"; |
||||
|
|
||||
|
private ProductBrand() { } |
||||
|
} |
||||
@ -0,0 +1,9 @@ |
|||||
|
package com.mydreamjob.messaging; |
||||
|
|
||||
|
import java.util.List; |
||||
|
|
||||
|
/** 批量统计每个会话最近 500 条消息内的未读数，兼容 MySQL 5.7。 */ |
||||
|
public interface MessageUnreadQueries { |
||||
|
List<MessageRepository.ConversationUnreadCount> countUnreadByConversationIds( |
||||
|
List<Long> conversationIds, Long readerUserId); |
||||
|
} |
||||
@ -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<MessageRepository.ConversationUnreadCount> countUnreadByConversationIds( |
||||
|
List<Long> 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<MessageRepository.ConversationUnreadCount> 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<String> 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; } |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,19 @@ |
|||||
|
package com.mydreamjob.messaging.realtime; |
||||
|
|
||||
|
import jakarta.persistence.*; |
||||
|
import lombok.*; |
||||
|
import java.time.LocalDateTime; |
||||
|
|
||||
|
@Entity |
||||
|
@Table(name = "chat_events") |
||||
|
@Getter @Setter @Builder @NoArgsConstructor @AllArgsConstructor |
||||
|
public class ChatEvent { |
||||
|
@Id @GeneratedValue(strategy = GenerationType.IDENTITY) |
||||
|
private Long id; |
||||
|
@Column(name = "conversation_id", nullable = false) private Long conversationId; |
||||
|
@Column(name = "recipient_user_id", nullable = false) private Long recipientUserId; |
||||
|
@Column(name = "actor_user_id", nullable = false) private Long actorUserId; |
||||
|
@Column(name = "message_id", nullable = false) private Long messageId; |
||||
|
@Column(name = "event_type", nullable = false, length = 32) private String eventType; |
||||
|
@Column(name = "created_at", nullable = false) private LocalDateTime createdAt; |
||||
|
} |
||||
@ -0,0 +1,10 @@ |
|||||
|
package com.mydreamjob.messaging.realtime; |
||||
|
|
||||
|
import org.springframework.data.jpa.repository.JpaRepository; |
||||
|
import org.springframework.data.domain.Pageable; |
||||
|
import java.util.List; |
||||
|
|
||||
|
public interface ChatEventRepository extends JpaRepository<ChatEvent, Long> { |
||||
|
List<ChatEvent> findByConversationIdAndRecipientUserIdAndIdGreaterThanOrderByIdAsc( |
||||
|
Long conversationId, Long recipientUserId, Long cursor, Pageable pageable); |
||||
|
} |
||||
@ -0,0 +1,90 @@ |
|||||
|
package com.mydreamjob.messaging.realtime; |
||||
|
|
||||
|
import com.mydreamjob.messaging.*; |
||||
|
import lombok.RequiredArgsConstructor; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.context.ApplicationEventPublisher; |
||||
|
import org.springframework.messaging.simp.SimpMessagingTemplate; |
||||
|
import org.springframework.stereotype.Service; |
||||
|
import org.springframework.transaction.annotation.Propagation; |
||||
|
import org.springframework.transaction.annotation.Transactional; |
||||
|
import org.springframework.transaction.event.TransactionalEventListener; |
||||
|
import org.springframework.transaction.event.TransactionPhase; |
||||
|
import org.springframework.data.domain.PageRequest; |
||||
|
import java.time.Clock; |
||||
|
import java.time.LocalDateTime; |
||||
|
import java.util.*; |
||||
|
|
||||
|
/** 事件和业务写入在同一事务；只在提交后推送，发送失败由持久化游标恢复。 */ |
||||
|
@Service @RequiredArgsConstructor @Slf4j |
||||
|
public class ChatEventService { |
||||
|
private final ChatEventRepository events; |
||||
|
private final ConversationMemberRepository members; |
||||
|
private final ConversationRepository conversations; |
||||
|
private final MessageRepository messages; |
||||
|
private final MessageReadRepository reads; |
||||
|
private final MessageViewMapper mapper; |
||||
|
private final ApplicationEventPublisher publisher; |
||||
|
private final SimpMessagingTemplate socket; |
||||
|
private final Clock clock; |
||||
|
|
||||
|
public record CommittedEvent(Long eventId) {} |
||||
|
|
||||
|
@Transactional(propagation = Propagation.MANDATORY) |
||||
|
public void record(String type, Long conversationId, Long actorId, Long messageId) { |
||||
|
for (var member : members.findByConversationId(conversationId)) { |
||||
|
var event = events.save(ChatEvent.builder().conversationId(conversationId) |
||||
|
.recipientUserId(member.getUserId()).actorUserId(actorId).messageId(messageId) |
||||
|
.eventType(type).createdAt(LocalDateTime.now(clock)).build()); |
||||
|
publisher.publishEvent(new CommittedEvent(event.getId())); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) |
||||
|
@Transactional(propagation = Propagation.REQUIRES_NEW, readOnly = true) |
||||
|
public void push(CommittedEvent committed) { |
||||
|
try { |
||||
|
events.findById(committed.eventId()).ifPresent(event -> { |
||||
|
var view = view(event); |
||||
|
if (view != null) socket.convertAndSendToUser(event.getRecipientUserId().toString(), |
||||
|
"/queue/chat-events", view); |
||||
|
}); |
||||
|
} catch (Exception ignored) { |
||||
|
// 不把已提交的发送伪报为失败，不记录正文或身份凭据。
|
||||
|
log.warn("Chat event push failed; cursor replay is available"); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* 返回 items / nextCursor / hasMore；初始 after=0，size 最大值由入口限制为 100。 |
||||
|
* 客户端仅在补拉页成功合并之后持久化 nextCursor，并持续翻页直到 hasMore=false。 |
||||
|
* 实时帧可能晚到或乱序，不得用推送 eventId 直接推进补拉游标。 |
||||
|
* payload 是消息当前状态（并非历史正文快照），按 message.id + version 合并。 |
||||
|
* MESSAGE_READ 的 actorScope 表示谁执行已读；payload.isRead 仍表示当前用户已读。 |
||||
|
*/ |
||||
|
@Transactional(readOnly = true) |
||||
|
public Map<String, Object> replay(Long conversationId, Long userId, long after, int size) { |
||||
|
var rows = events.findByConversationIdAndRecipientUserIdAndIdGreaterThanOrderByIdAsc( |
||||
|
conversationId, userId, after, PageRequest.of(0, size + 1)); |
||||
|
boolean hasMore = rows.size() > size; |
||||
|
var page = rows.stream().limit(size).toList(); |
||||
|
var items = page.stream().map(this::view).filter(Objects::nonNull).toList(); |
||||
|
long next = page.isEmpty() ? after : page.get(page.size() - 1).getId(); |
||||
|
return Map.of("items", items, "nextCursor", next, "hasMore", hasMore); |
||||
|
} |
||||
|
|
||||
|
private Map<String, Object> view(ChatEvent event) { |
||||
|
// 防止成员移除之后继续读取旧事件。
|
||||
|
if (!members.existsByConversationIdAndUserId(event.getConversationId(), event.getRecipientUserId())) return null; |
||||
|
var message = messages.findById(event.getMessageId()).orElse(null); |
||||
|
var conversation = conversations.findById(event.getConversationId()).orElse(null); |
||||
|
if (message == null || conversation == null) return null; |
||||
|
var response = mapper.toResponse(message, event.getRecipientUserId(), |
||||
|
"TALENT_INVITATION".equals(conversation.getType())); |
||||
|
response.setIsRead(reads.existsByMessageIdAndReaderUserId(message.getId(), event.getRecipientUserId())); |
||||
|
return Map.of("eventId", event.getId(), "type", event.getEventType(), |
||||
|
"conversationId", event.getConversationId(), "occurredAt", event.getCreatedAt(), |
||||
|
"actorScope", Objects.equals(event.getActorUserId(), event.getRecipientUserId()) |
||||
|
? "CURRENT_USER" : "OTHER_PARTICIPANT", "payload", response); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,112 @@ |
|||||
|
package com.mydreamjob.messaging.realtime; |
||||
|
|
||||
|
import com.mydreamjob.auth.*; |
||||
|
import lombok.RequiredArgsConstructor; |
||||
|
import org.springframework.messaging.*; |
||||
|
import org.springframework.messaging.support.*; |
||||
|
import org.springframework.messaging.simp.stomp.*; |
||||
|
import org.springframework.security.authentication.UsernamePasswordAuthenticationToken; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
import java.time.Clock; |
||||
|
import java.util.*; |
||||
|
|
||||
|
/** 只允许 CONNECT 鉴权和自己的固定队列；不允许 STOMP SEND 绕过 HTTP 服务层。 */ |
||||
|
@Component @RequiredArgsConstructor @lombok.extern.slf4j.Slf4j |
||||
|
public class ChatSocketAuth implements ChannelInterceptor { |
||||
|
static final String CLAIMS = "chat.accessClaims"; |
||||
|
private final JwtTokenProvider tokens; |
||||
|
private final AuthSessionService sessions; |
||||
|
private final UserRepository users; |
||||
|
private final com.mydreamjob.company.CompanyRepository companies; |
||||
|
private final Clock clock; |
||||
|
private final java.util.concurrent.ConcurrentHashMap<String, AccessTokenClaims> connected = |
||||
|
new java.util.concurrent.ConcurrentHashMap<>(); |
||||
|
|
||||
|
@org.springframework.context.event.EventListener |
||||
|
public void disconnected(org.springframework.web.socket.messaging.SessionDisconnectEvent event) { |
||||
|
connected.remove(event.getSessionId()); |
||||
|
} |
||||
|
|
||||
|
boolean active(AccessTokenClaims claims) { |
||||
|
if (claims == null || !clock.instant().isBefore(claims.expiresAt())) return false; |
||||
|
if (!sessions.isSessionActive(claims.userId(), claims.sessionId(), clock.instant())) return false; |
||||
|
return users.findByIdWithRoles(claims.userId()).map(u -> { |
||||
|
if (u.getRole() == User.Role.UNASSIGNED) return false; |
||||
|
if (u.getRoles() != null && u.getRoles().contains(User.Role.COMPANY)) { |
||||
|
return companies.findByOwnerUserId(u.getId()) |
||||
|
.map(com.mydreamjob.company.VerifiedCompanyGuard::isVerified).orElse(false); |
||||
|
} |
||||
|
return true; |
||||
|
}).orElse(false); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public Message<?> preSend(Message<?> message, MessageChannel channel) { |
||||
|
var accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class); |
||||
|
if (accessor == null) throw denied(); |
||||
|
var attrs = accessor.getSessionAttributes(); |
||||
|
if (attrs == null) throw denied(); |
||||
|
if (accessor.getCommand() == StompCommand.DISCONNECT) return message; |
||||
|
try { |
||||
|
if (accessor.getCommand() == StompCommand.CONNECT) { |
||||
|
if (attrs.containsKey(CLAIMS)) throw denied(); |
||||
|
var auth = accessor.getNativeHeader("Authorization"); |
||||
|
if (auth == null || auth.size() != 1 || !auth.get(0).startsWith("Bearer ") |
||||
|
|| auth.get(0).length() > 8192) throw denied(); |
||||
|
var claims = tokens.parseAndValidateAccessToken(auth.get(0).substring(7), clock.instant()); |
||||
|
if (!active(claims)) throw denied(); |
||||
|
if (accessor.getSessionId() == null) throw denied(); |
||||
|
synchronized (connected) { |
||||
|
if (connected.values().stream().filter(c -> c.userId().equals(claims.userId())).count() >= 5) |
||||
|
throw denied(); |
||||
|
connected.put(accessor.getSessionId(), claims); |
||||
|
} |
||||
|
accessor.setUser(new UsernamePasswordAuthenticationToken(claims.userId(), null, List.of())); |
||||
|
attrs.put(CLAIMS, claims); |
||||
|
// 原始 Token 不进入会话属性或后续帧。
|
||||
|
accessor.removeNativeHeader("Authorization"); |
||||
|
} else { |
||||
|
if (!active((AccessTokenClaims) attrs.get(CLAIMS))) throw denied(); |
||||
|
if (accessor.getCommand() == StompCommand.SUBSCRIBE |
||||
|
&& !"/user/queue/chat-events".equals(accessor.getDestination())) throw denied(); |
||||
|
if (accessor.getCommand() != StompCommand.SUBSCRIBE |
||||
|
&& accessor.getCommand() != StompCommand.UNSUBSCRIBE |
||||
|
&& accessor.getCommand() != null) throw denied(); |
||||
|
} |
||||
|
synchronized (attrs) { |
||||
|
long now = clock.millis(); |
||||
|
long start = (long) attrs.getOrDefault("chat.rateStart", now); |
||||
|
int count = (int) attrs.getOrDefault("chat.rateCount", 0); |
||||
|
if (now - start >= 60000) { start = now; count = 0; } |
||||
|
if (++count > 120) throw denied(); |
||||
|
attrs.put("chat.rateStart", start); |
||||
|
attrs.put("chat.rateCount", count); |
||||
|
} |
||||
|
return message; |
||||
|
} catch (Exception ignored) { |
||||
|
throw denied(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private MessageDeliveryException denied() { |
||||
|
return new MessageDeliveryException("Chat connection rejected"); |
||||
|
} |
||||
|
|
||||
|
/** 推送时再次检查，防止过期、停用或登出后仍然接收消息。 */ |
||||
|
public ChannelInterceptor outbound() { |
||||
|
return new ChannelInterceptor() { |
||||
|
@Override public Message<?> preSend(Message<?> message, MessageChannel channel) { |
||||
|
var accessor = MessageHeaderAccessor.getAccessor(message, |
||||
|
org.springframework.messaging.simp.SimpMessageHeaderAccessor.class); |
||||
|
if (accessor == null || accessor.getMessageType() != org.springframework.messaging.simp.SimpMessageType.MESSAGE) |
||||
|
return message; |
||||
|
try { |
||||
|
if (!active(connected.get(accessor.getSessionId()))) { |
||||
|
log.warn("Chat outbound authorization rejected"); return null; |
||||
|
} |
||||
|
return message; |
||||
|
} catch (Exception ignored) { return null; } |
||||
|
} |
||||
|
}; |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,68 @@ |
|||||
|
package com.mydreamjob.messaging.realtime; |
||||
|
|
||||
|
import lombok.RequiredArgsConstructor; |
||||
|
import org.springframework.beans.factory.annotation.Value; |
||||
|
import org.springframework.context.annotation.*; |
||||
|
import org.springframework.messaging.simp.config.*; |
||||
|
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; |
||||
|
import org.springframework.web.socket.config.annotation.*; |
||||
|
|
||||
|
/** 当前使用单实例内置 broker；多实例部署必须改为共享 broker，不能仅添加负载均衡。 */ |
||||
|
@Configuration @EnableWebSocketMessageBroker @RequiredArgsConstructor |
||||
|
public class ChatSocketConfig implements WebSocketMessageBrokerConfigurer { |
||||
|
private final ChatSocketAuth auth; |
||||
|
private final ChatSocketConnections connections; |
||||
|
private final org.springframework.beans.factory.ObjectProvider<org.springframework.messaging.simp.SimpMessagingTemplate> socket; |
||||
|
@Value("${app.cors.allowed-origins:http://localhost:5173,http://localhost:3000}") |
||||
|
private String origins; |
||||
|
|
||||
|
@Bean public ThreadPoolTaskScheduler chatHeartbeatScheduler() { |
||||
|
var scheduler = new ThreadPoolTaskScheduler(); |
||||
|
scheduler.setPoolSize(2); |
||||
|
scheduler.setThreadNamePrefix("chat-heartbeat-"); |
||||
|
return scheduler; |
||||
|
} |
||||
|
@Override public void registerStompEndpoints(StompEndpointRegistry registry) { |
||||
|
registry.addEndpoint("/ws").addInterceptors(connections.handshakeGuard()) |
||||
|
.setAllowedOrigins(java.util.Arrays.stream(origins.split(",")) |
||||
|
.map(String::trim).filter(s -> !s.isEmpty()).toArray(String[]::new)); |
||||
|
} |
||||
|
@Override public void configureMessageBroker(MessageBrokerRegistry registry) { |
||||
|
registry.enableSimpleBroker("/queue").setHeartbeatValue(new long[]{10000, 10000}) |
||||
|
.setTaskScheduler(chatHeartbeatScheduler()); |
||||
|
registry.setUserDestinationPrefix("/user"); |
||||
|
registry.setApplicationDestinationPrefixes("/app"); |
||||
|
registry.setPreservePublishOrder(true); |
||||
|
configureBrokerChannel(registry.configureBrokerChannel()); |
||||
|
} |
||||
|
@Override public void configureClientInboundChannel(ChannelRegistration registration) { |
||||
|
registration.interceptors(auth); |
||||
|
} |
||||
|
@Override public void configureClientOutboundChannel(ChannelRegistration registration) { |
||||
|
registration.interceptors(auth.outbound()); |
||||
|
} |
||||
|
private void configureBrokerChannel(ChannelRegistration registration) { |
||||
|
registration.interceptors(new org.springframework.messaging.support.ChannelInterceptor() { |
||||
|
@Override public void afterSendCompletion(org.springframework.messaging.Message<?> message, |
||||
|
org.springframework.messaging.MessageChannel channel, boolean sent, Exception exception) { |
||||
|
var headers = org.springframework.messaging.simp.SimpMessageHeaderAccessor.wrap(message); |
||||
|
if (!sent || exception != null |
||||
|
|| headers.getMessageType() != org.springframework.messaging.simp.SimpMessageType.SUBSCRIBE |
||||
|
|| headers.getDestination() == null |
||||
|
|| !headers.getDestination().startsWith("/queue/chat-events-user") |
||||
|
|| headers.getUser() == null || headers.getSessionId() == null) return; |
||||
|
// 同步 broker channel 返回时订阅已经登记。客户端收到 READY 后再补拉，
|
||||
|
// 避免“补拉已结束但订阅尚未生效”的窗口漏掉消息。
|
||||
|
var ready = org.springframework.messaging.simp.SimpMessageHeaderAccessor.create( |
||||
|
org.springframework.messaging.simp.SimpMessageType.MESSAGE); |
||||
|
ready.setSessionId(headers.getSessionId()); ready.setLeaveMutable(true); |
||||
|
socket.getObject().convertAndSendToUser(headers.getUser().getName(), "/queue/chat-events", |
||||
|
java.util.Map.of("type", "CHAT_READY"), ready.getMessageHeaders()); |
||||
|
} |
||||
|
}); |
||||
|
} |
||||
|
@Override public void configureWebSocketTransport(WebSocketTransportRegistration registration) { |
||||
|
registration.setMessageSizeLimit(16384).setSendBufferSizeLimit(262144).setSendTimeLimit(10000) |
||||
|
.setTimeToFirstMessage(10000).addDecoratorFactory(connections::decorate); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,82 @@ |
|||||
|
package com.mydreamjob.messaging.realtime; |
||||
|
|
||||
|
import com.mydreamjob.auth.AccessTokenClaims; |
||||
|
import lombok.RequiredArgsConstructor; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
import org.springframework.scheduling.annotation.Scheduled; |
||||
|
import org.springframework.web.socket.*; |
||||
|
import org.springframework.web.socket.handler.WebSocketHandlerDecorator; |
||||
|
import java.time.Clock; |
||||
|
import java.util.concurrent.ConcurrentHashMap; |
||||
|
|
||||
|
/** 限制单实例连接数、每用户设备数；定期关闭未鉴权、过期或被撤销的连接。 */ |
||||
|
@Component @RequiredArgsConstructor |
||||
|
public class ChatSocketConnections { |
||||
|
private final ChatSocketAuth auth; |
||||
|
private final Clock clock; |
||||
|
private final ConcurrentHashMap<String, WebSocketSession> sessions = new ConcurrentHashMap<>(); |
||||
|
private final java.util.Map<String, long[]> attempts = new java.util.HashMap<>(); |
||||
|
|
||||
|
public org.springframework.web.socket.server.HandshakeInterceptor handshakeGuard() { |
||||
|
return new org.springframework.web.socket.server.HandshakeInterceptor() { |
||||
|
@Override public boolean beforeHandshake(org.springframework.http.server.ServerHttpRequest request, |
||||
|
org.springframework.http.server.ServerHttpResponse response, WebSocketHandler handler, |
||||
|
java.util.Map<String, Object> attributes) { |
||||
|
// 凭据不能进入 URL；只接受后续 CONNECT 的 Authorization 头。
|
||||
|
if (request.getURI().getRawQuery() != null) { |
||||
|
response.setStatusCode(org.springframework.http.HttpStatus.BAD_REQUEST); return false; |
||||
|
} |
||||
|
String source = request.getRemoteAddress() == null ? "unknown" |
||||
|
: request.getRemoteAddress().getAddress().getHostAddress(); |
||||
|
synchronized (attempts) { |
||||
|
long now = clock.millis(); |
||||
|
attempts.entrySet().removeIf(e -> now - e.getValue()[0] >= 60000); |
||||
|
if (!attempts.containsKey(source) && attempts.size() >= 1024) { |
||||
|
response.setStatusCode(org.springframework.http.HttpStatus.TOO_MANY_REQUESTS); return false; |
||||
|
} |
||||
|
var count = attempts.computeIfAbsent(source, key -> new long[]{now, 0}); |
||||
|
if (++count[1] > 40) { |
||||
|
response.setStatusCode(org.springframework.http.HttpStatus.TOO_MANY_REQUESTS); return false; |
||||
|
} |
||||
|
} |
||||
|
return true; |
||||
|
} |
||||
|
@Override public void afterHandshake(org.springframework.http.server.ServerHttpRequest request, |
||||
|
org.springframework.http.server.ServerHttpResponse response, WebSocketHandler handler, Exception exception) { } |
||||
|
}; |
||||
|
} |
||||
|
|
||||
|
public WebSocketHandler decorate(WebSocketHandler handler) { |
||||
|
return new WebSocketHandlerDecorator(handler) { |
||||
|
@Override public void afterConnectionEstablished(WebSocketSession session) throws Exception { |
||||
|
synchronized (sessions) { |
||||
|
if (sessions.size() >= 500) { session.close(CloseStatus.POLICY_VIOLATION); return; } |
||||
|
session.getAttributes().put("chat.openedAt", clock.millis()); |
||||
|
sessions.put(session.getId(), session); |
||||
|
} |
||||
|
try { super.afterConnectionEstablished(session); } |
||||
|
catch (Exception ex) { sessions.remove(session.getId()); throw ex; } |
||||
|
} |
||||
|
@Override public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception { |
||||
|
sessions.remove(session.getId()); |
||||
|
super.afterConnectionClosed(session, status); |
||||
|
} |
||||
|
}; |
||||
|
} |
||||
|
|
||||
|
@Scheduled(fixedDelay = 10000) |
||||
|
public void expire() { |
||||
|
var perUser = new java.util.HashMap<Long, Integer>(); |
||||
|
sessions.values().forEach(session -> { |
||||
|
try { |
||||
|
var claims = (AccessTokenClaims) session.getAttributes().get(ChatSocketAuth.CLAIMS); |
||||
|
boolean invalid = claims == null |
||||
|
? clock.millis() - (long) session.getAttributes().getOrDefault("chat.openedAt", 0L) > 10000 |
||||
|
: !auth.active(claims) || perUser.merge(claims.userId(), 1, Integer::sum) > 5; |
||||
|
if (invalid) session.close(CloseStatus.POLICY_VIOLATION); |
||||
|
} catch (Exception ignored) { |
||||
|
try { session.close(CloseStatus.SERVER_ERROR); } catch (Exception closeIgnored) { } |
||||
|
} |
||||
|
}); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,18 @@ |
|||||
|
ALTER TABLE messages ADD COLUMN client_message_id VARCHAR(64) NULL; |
||||
|
ALTER TABLE messages ADD COLUMN request_content_hash VARCHAR(64) NULL; |
||||
|
CREATE UNIQUE INDEX uk_messages_client_id ON messages(conversation_id, sender_user_id, client_message_id); |
||||
|
|
||||
|
-- 不复制聊天正文；补拉时读取消息当前状态，避免已撤回正文重新出现。 |
||||
|
CREATE TABLE chat_events ( |
||||
|
id BIGINT AUTO_INCREMENT PRIMARY KEY, |
||||
|
conversation_id BIGINT NOT NULL, |
||||
|
recipient_user_id BIGINT NOT NULL, |
||||
|
actor_user_id BIGINT NOT NULL, |
||||
|
message_id BIGINT NOT NULL, |
||||
|
event_type VARCHAR(32) NOT NULL, |
||||
|
created_at DATETIME(6) NOT NULL, |
||||
|
CONSTRAINT fk_chat_event_conversation FOREIGN KEY (conversation_id) REFERENCES conversations(id) ON DELETE CASCADE, |
||||
|
CONSTRAINT fk_chat_event_recipient FOREIGN KEY (recipient_user_id) REFERENCES users(id) ON DELETE CASCADE, |
||||
|
CONSTRAINT fk_chat_event_message FOREIGN KEY (message_id) REFERENCES messages(id) ON DELETE CASCADE, |
||||
|
INDEX idx_chat_events_cursor (conversation_id, recipient_user_id, id) |
||||
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; |
||||
@ -0,0 +1,23 @@ |
|||||
|
package com.mydreamjob.common; |
||||
|
|
||||
|
import com.mydreamjob.config.OpenApiSecurityConfig; |
||||
|
import com.mydreamjob.controller.ApiController; |
||||
|
import com.mydreamjob.common.logging.RequestIdAccessor; |
||||
|
import org.junit.jupiter.api.Test; |
||||
|
import org.springframework.mock.web.MockHttpServletRequest; |
||||
|
import static org.junit.jupiter.api.Assertions.*; |
||||
|
import static org.mockito.Mockito.mock; |
||||
|
|
||||
|
class ProductBrandTest { |
||||
|
@Test void apiInformationUsesNewBrandAndPreservesTechnicalName() { |
||||
|
var controller = new ApiController(null, mock(RequestIdAccessor.class)); |
||||
|
var info = controller.index(new MockHttpServletRequest()).getBody().getData(); |
||||
|
assertEquals("mydreamjob", info.get("name")); |
||||
|
assertEquals("轻松聘", info.get("displayName")); |
||||
|
assertEquals("轻松聘 - 大学生求职服务平台 API", info.get("description")); |
||||
|
assertFalse(info.toString().contains("职来")); |
||||
|
} |
||||
|
@Test void swaggerTitleUsesNewBrand() { |
||||
|
assertEquals("轻松聘 API", new OpenApiSecurityConfig().homepageOpenAPI().getInfo().getTitle()); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,202 @@ |
|||||
|
package com.mydreamjob.messaging.realtime; |
||||
|
|
||||
|
import com.fasterxml.jackson.databind.ObjectMapper; |
||||
|
import com.mydreamjob.auth.*; |
||||
|
import com.mydreamjob.auth.support.TestAuthTokens; |
||||
|
import com.mydreamjob.company.*; |
||||
|
import com.mydreamjob.messaging.*; |
||||
|
import com.mydreamjob.messaging.dto.SendMessageRequest; |
||||
|
import com.mydreamjob.common.exception.BusinessException; |
||||
|
import org.junit.jupiter.api.*; |
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
import org.springframework.boot.test.context.SpringBootTest; |
||||
|
import org.springframework.boot.test.web.server.LocalServerPort; |
||||
|
import org.springframework.boot.test.mock.mockito.MockBean; |
||||
|
import org.springframework.data.redis.core.RedisTemplate; |
||||
|
import org.springframework.test.context.ActiveProfiles; |
||||
|
import org.springframework.transaction.support.TransactionTemplate; |
||||
|
import org.springframework.transaction.PlatformTransactionManager; |
||||
|
import org.springframework.web.socket.client.standard.StandardWebSocketClient; |
||||
|
import org.springframework.web.socket.messaging.WebSocketStompClient; |
||||
|
import org.springframework.messaging.simp.stomp.*; |
||||
|
import org.springframework.messaging.converter.StringMessageConverter; |
||||
|
import java.lang.reflect.Type; |
||||
|
import java.util.*; |
||||
|
import java.util.concurrent.*; |
||||
|
import static org.junit.jupiter.api.Assertions.*; |
||||
|
|
||||
|
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, properties = { |
||||
|
"spring.datasource.url=jdbc:h2:mem:chatrealtime;MODE=MySQL;DB_CLOSE_DELAY=-1", |
||||
|
"spring.jpa.show-sql=false", "management.server.port=0"}) |
||||
|
@ActiveProfiles("integration-test") |
||||
|
class ChatRealtimeIntegrationTest { |
||||
|
@Autowired MessagingService service; |
||||
|
@Autowired UserRepository users; |
||||
|
@Autowired CompanyRepository companies; |
||||
|
@Autowired ConversationRepository conversations; |
||||
|
@Autowired ConversationMemberRepository members; |
||||
|
@Autowired MessageRepository messages; |
||||
|
@Autowired ChatEventRepository events; |
||||
|
@Autowired PlatformTransactionManager transactions; |
||||
|
@Autowired ObjectMapper mapper; |
||||
|
@Autowired org.springframework.messaging.simp.user.SimpUserRegistry connectedUsers; |
||||
|
@MockBean RedisTemplate<String, Object> redis; |
||||
|
@org.springframework.boot.test.mock.mockito.SpyBean ChatEventService eventService; |
||||
|
@LocalServerPort int port; |
||||
|
Long student, hr, conversation; |
||||
|
|
||||
|
@BeforeEach void setup() { |
||||
|
student = user(User.Role.STUDENT); hr = user(User.Role.COMPANY); |
||||
|
companies.save(Company.builder().name("实时聊天测试企业").ownerUserId(hr).verified(true).build()); |
||||
|
conversation = conversations.save(Conversation.builder().createdBy(hr).type("DIRECT").build()).getId(); |
||||
|
members.save(ConversationMember.builder().conversationId(conversation).userId(student).build()); |
||||
|
members.save(ConversationMember.builder().conversationId(conversation).userId(hr).build()); |
||||
|
} |
||||
|
private Long user(User.Role role) { |
||||
|
return users.save(User.builder().phone("rt-" + UUID.randomUUID().toString().substring(0, 12)) |
||||
|
.role(role).roles(Set.of(role)).build()).getId(); |
||||
|
} |
||||
|
private SendMessageRequest request(String key, String content) { |
||||
|
var req = new SendMessageRequest(); req.setConversationId(conversation); |
||||
|
req.setClientMessageId(key); req.setContent(content); return req; |
||||
|
} |
||||
|
@Test void idempotentRetryReturnsSameMessageAndDoesNotRepeatEvents() { |
||||
|
var first = service.sendMessage(student, request("retry-key", "hello")); |
||||
|
var second = service.sendMessage(student, request("retry-key", "hello")); |
||||
|
assertEquals(first.get("id"), second.get("id")); |
||||
|
var replay = service.replayEvents(conversation, hr, 0, 100); |
||||
|
assertEquals(1, ((List<?>) replay.get("items")).size()); |
||||
|
assertThrows(BusinessException.class, () -> service.sendMessage(student, request("retry-key", "changed"))); |
||||
|
} |
||||
|
@Test void concurrentRetryStoresOnlyOneMessage() throws Exception { |
||||
|
var pool = Executors.newFixedThreadPool(2); |
||||
|
try { |
||||
|
var a = pool.submit(() -> service.sendMessage(student, request("parallel-key", "same"))); |
||||
|
var b = pool.submit(() -> service.sendMessage(student, request("parallel-key", "same"))); |
||||
|
assertEquals(a.get(10, TimeUnit.SECONDS).get("id"), b.get(10, TimeUnit.SECONDS).get("id")); |
||||
|
} finally { pool.shutdownNow(); } |
||||
|
} |
||||
|
@Test void replayPaginatesWithoutLossAndIncludesReadAndRecall() { |
||||
|
var sent = service.sendMessage(student, request("event-key", "secret")); |
||||
|
service.markRead(conversation, hr); |
||||
|
service.recallMessage(student, conversation, (Long) sent.get("id"), null); |
||||
|
long cursor = 0; var found = new ArrayList<Map<?, ?>>(); |
||||
|
Map<String, Object> page; |
||||
|
do { |
||||
|
page = service.replayEvents(conversation, hr, cursor, 1); |
||||
|
for (var item : (List<?>) page.get("items")) found.add((Map<?, ?>) item); |
||||
|
cursor = ((Number) page.get("nextCursor")).longValue(); |
||||
|
} while ((Boolean) page.get("hasMore")); |
||||
|
assertEquals(List.of("MESSAGE_CREATED", "MESSAGE_READ", "MESSAGE_RECALLED"), |
||||
|
found.stream().map(e -> e.get("type")).toList()); |
||||
|
assertFalse(found.toString().contains("secret")); |
||||
|
} |
||||
|
@Test void unauthorizedAndInvalidCursorRejected() { |
||||
|
Long outsider = user(User.Role.STUDENT); |
||||
|
assertThrows(BusinessException.class, () -> service.replayEvents(conversation, outsider, 0, 50)); |
||||
|
assertThrows(BusinessException.class, () -> service.replayEvents(conversation, student, -1, 50)); |
||||
|
assertThrows(BusinessException.class, () -> service.replayEvents(conversation, student, 0, 101)); |
||||
|
} |
||||
|
@Test void rolledBackSendHasNoPersistedEvents() { |
||||
|
new TransactionTemplate(transactions).execute(status -> { |
||||
|
service.sendMessage(student, request("rollback-key", "discard")); status.setRollbackOnly(); return null; |
||||
|
}); |
||||
|
assertTrue(((List<?>) service.replayEvents(conversation, hr, 0, 50).get("items")).isEmpty()); |
||||
|
org.mockito.Mockito.verify(eventService, org.mockito.Mockito.never()).push(org.mockito.ArgumentMatchers.any()); |
||||
|
} |
||||
|
@Test void invitationReplayDoesNotExposeOtherUserId() { |
||||
|
var c = conversations.findById(conversation).orElseThrow(); c.setType("TALENT_INVITATION"); conversations.save(c); |
||||
|
service.sendMessage(student, request("private-key", "hello")); |
||||
|
var page = mapper.valueToTree(service.replayEvents(conversation, hr, 0, 50)); |
||||
|
assertFalse(page.path("items").get(0).path("payload").has("senderUserId")); |
||||
|
assertEquals("OTHER_PARTICIPANT", page.path("items").get(0).path("payload").path("senderScope").asText()); |
||||
|
} |
||||
|
@Test void realWebSocketReceivesBothDirectionsAfterCommit() throws Exception { |
||||
|
var client = new WebSocketStompClient(new StandardWebSocketClient()); |
||||
|
var scheduler = new org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler(); |
||||
|
scheduler.initialize(); client.setTaskScheduler(scheduler); |
||||
|
var converter = new org.springframework.messaging.converter.MappingJackson2MessageConverter(); |
||||
|
converter.setObjectMapper(mapper); client.setMessageConverter(converter); |
||||
|
var hrQueue = new LinkedBlockingQueue<String>(); var studentQueue = new LinkedBlockingQueue<String>(); |
||||
|
StompSession hrSocket = null, studentSocket = null; |
||||
|
try { |
||||
|
hrSocket = connect(client, hr, "COMPANY", hrQueue); |
||||
|
studentSocket = connect(client, student, "STUDENT", studentQueue); |
||||
|
service.sendMessage(student, request("socket-student", "hello HR")); |
||||
|
assertEquals(1, ((List<?>) service.replayEvents(conversation, hr, 0, 50).get("items")).size()); |
||||
|
org.mockito.Mockito.verify(eventService, org.mockito.Mockito.atLeastOnce()).push(org.mockito.ArgumentMatchers.any()); |
||||
|
assertEquals("hello HR", mapper.readTree(Objects.requireNonNull(hrQueue.poll(10, TimeUnit.SECONDS))) |
||||
|
.path("payload").path("content").asText()); |
||||
|
studentQueue.poll(10, TimeUnit.SECONDS); // 发送者自己的设备也接收事件。
|
||||
|
service.sendMessage(hr, request("socket-hr", "hello student")); |
||||
|
assertEquals("hello student", mapper.readTree(Objects.requireNonNull(studentQueue.poll(10, TimeUnit.SECONDS))) |
||||
|
.path("payload").path("content").asText()); |
||||
|
} finally { |
||||
|
if (hrSocket != null && hrSocket.isConnected()) hrSocket.disconnect(); |
||||
|
if (studentSocket != null && studentSocket.isConnected()) studentSocket.disconnect(); |
||||
|
client.stop(); |
||||
|
scheduler.shutdown(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
@Test void realReconnectReplaysMoreThanOnePageAndContinuesReceiving() throws Exception { |
||||
|
var client = new WebSocketStompClient(new StandardWebSocketClient()); |
||||
|
var scheduler = new org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler(); scheduler.initialize(); |
||||
|
client.setTaskScheduler(scheduler); |
||||
|
var converter = new org.springframework.messaging.converter.MappingJackson2MessageConverter(); |
||||
|
converter.setObjectMapper(mapper); client.setMessageConverter(converter); |
||||
|
var queue = new LinkedBlockingQueue<String>(); StompSession socket = null; |
||||
|
try { |
||||
|
socket = connect(client, hr, "COMPANY", queue); |
||||
|
service.sendMessage(student, request("online", "before disconnect")); |
||||
|
assertNotNull(queue.poll(10, TimeUnit.SECONDS)); |
||||
|
long cursor = ((Number) service.replayEvents(conversation, hr, 0, 50).get("nextCursor")).longValue(); |
||||
|
socket.disconnect(); |
||||
|
org.awaitility.Awaitility.await().atMost(10, TimeUnit.SECONDS) |
||||
|
.until(() -> connectedUsers.getUser(hr.toString()) == null); |
||||
|
for (int i = 0; i < 55; i++) service.sendMessage(student, request("offline-" + i, "offline " + i)); |
||||
|
socket = connect(client, hr, "COMPANY", queue); |
||||
|
var ids = new HashSet<Object>(); boolean hasMore; |
||||
|
do { |
||||
|
var page = service.replayEvents(conversation, hr, cursor, 10); |
||||
|
for (var value : (List<?>) page.get("items")) { |
||||
|
var event = (Map<?, ?>) value; |
||||
|
assertTrue(ids.add(event.get("eventId")), "不能重复补拉事件"); |
||||
|
} |
||||
|
cursor = ((Number) page.get("nextCursor")).longValue(); |
||||
|
hasMore = (Boolean) page.get("hasMore"); |
||||
|
} while (hasMore); |
||||
|
assertEquals(55, ids.size()); |
||||
|
queue.clear(); |
||||
|
service.sendMessage(student, request("after-reconnect", "live again")); |
||||
|
assertEquals("live again", mapper.readTree(Objects.requireNonNull(queue.poll(10, TimeUnit.SECONDS))) |
||||
|
.path("payload").path("content").asText()); |
||||
|
} finally { |
||||
|
if (socket != null && socket.isConnected()) socket.disconnect(); |
||||
|
client.stop(); scheduler.shutdown(); |
||||
|
} |
||||
|
} |
||||
|
private StompSession connect(WebSocketStompClient client, Long user, String role, |
||||
|
BlockingQueue<String> queue) throws Exception { |
||||
|
var headers = new StompHeaders(); headers.set("Authorization", "Bearer " + TestAuthTokens.accessToken(user, role)); |
||||
|
var session = client.connectAsync("ws://localhost:" + port + "/api/v1/ws", new org.springframework.web.socket.WebSocketHttpHeaders(), headers, |
||||
|
new StompSessionHandlerAdapter() {}).get(10, TimeUnit.SECONDS); |
||||
|
session.subscribe("/user/queue/chat-events", new StompFrameHandler() { |
||||
|
@Override public Type getPayloadType(StompHeaders h) { return Map.class; } |
||||
|
@Override public void handleFrame(StompHeaders h, Object payload) { |
||||
|
try { queue.add(mapper.writeValueAsString(payload)); } |
||||
|
catch (Exception ex) { throw new IllegalStateException("Invalid event payload", ex); } |
||||
|
} |
||||
|
}); |
||||
|
// 内置 broker 不支持 STOMP receipts；通过真实服务端订阅注册作为测试屏障。
|
||||
|
org.awaitility.Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> { |
||||
|
var connected = connectedUsers.getUser(user.toString()); |
||||
|
return connected != null && connected.getSessions().stream() |
||||
|
.anyMatch(s -> !s.getSubscriptions().isEmpty()); |
||||
|
}); |
||||
|
var ready = queue.poll(10, TimeUnit.SECONDS); |
||||
|
assertNotNull(ready, "必须收到订阅就绪帧再补拉历史"); |
||||
|
assertEquals("CHAT_READY", mapper.readTree(ready).path("type").asText()); |
||||
|
return session; |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,114 @@ |
|||||
|
package com.mydreamjob.messaging.realtime; |
||||
|
|
||||
|
import com.mydreamjob.auth.*; |
||||
|
import org.junit.jupiter.api.*; |
||||
|
import org.springframework.messaging.support.MessageBuilder; |
||||
|
import org.springframework.messaging.simp.stomp.*; |
||||
|
import java.time.*; |
||||
|
import java.util.*; |
||||
|
import static org.junit.jupiter.api.Assertions.*; |
||||
|
import static org.mockito.Mockito.*; |
||||
|
|
||||
|
class ChatSocketAuthTest { |
||||
|
private final Instant now = Instant.parse("2026-10-14T08:00:00Z"); |
||||
|
private JwtTokenProvider tokens; |
||||
|
private AuthSessionService sessions; |
||||
|
private UserRepository users; |
||||
|
private ChatSocketAuth guard; |
||||
|
private AccessTokenClaims claims; |
||||
|
|
||||
|
@BeforeEach void setup() { |
||||
|
tokens = mock(JwtTokenProvider.class); sessions = mock(AuthSessionService.class); |
||||
|
users = mock(UserRepository.class); |
||||
|
guard = new ChatSocketAuth(tokens, sessions, users, |
||||
|
mock(com.mydreamjob.company.CompanyRepository.class), Clock.fixed(now, ZoneOffset.UTC)); |
||||
|
claims = new AccessTokenClaims(1L, "session", "token", "STUDENT", now.minusSeconds(1), now.plusSeconds(60)); |
||||
|
when(tokens.parseAndValidateAccessToken("valid", now)).thenReturn(claims); |
||||
|
when(sessions.isSessionActive(1L, "session", now)).thenReturn(true); |
||||
|
when(users.findByIdWithRoles(1L)).thenReturn(Optional.of(User.builder().id(1L).build())); |
||||
|
} |
||||
|
private StompHeaderAccessor frame(StompCommand command, Map<String, Object> attrs) { |
||||
|
var frame = StompHeaderAccessor.create(command); |
||||
|
frame.setSessionId("ws-1"); frame.setSessionAttributes(attrs); frame.setLeaveMutable(true); |
||||
|
return frame; |
||||
|
} |
||||
|
private void send(StompHeaderAccessor frame) { |
||||
|
guard.preSend(MessageBuilder.createMessage(new byte[0], frame.getMessageHeaders()), null); |
||||
|
} |
||||
|
private Map<String, Object> connect() { |
||||
|
var attrs = new HashMap<String, Object>(); var frame = frame(StompCommand.CONNECT, attrs); |
||||
|
frame.setNativeHeader("Authorization", "Bearer valid"); send(frame); return attrs; |
||||
|
} |
||||
|
@Test void connectAuthenticatesAndRemovesRawToken() { |
||||
|
var frame = frame(StompCommand.CONNECT, new HashMap<>()); |
||||
|
frame.setNativeHeader("Authorization", "Bearer valid"); send(frame); |
||||
|
assertEquals("1", frame.getUser().getName()); |
||||
|
assertNull(frame.getFirstNativeHeader("Authorization")); |
||||
|
} |
||||
|
@Test void anonymousConnectRejected() { |
||||
|
assertThrows(Exception.class, () -> send(frame(StompCommand.CONNECT, new HashMap<>()))); |
||||
|
} |
||||
|
@Test void revokedSessionRejected() { |
||||
|
when(sessions.isSessionActive(1L, "session", now)).thenReturn(false); |
||||
|
assertThrows(Exception.class, this::connect); |
||||
|
} |
||||
|
@Test void expiredTokenRejected() { |
||||
|
when(tokens.parseAndValidateAccessToken("valid", now)).thenReturn( |
||||
|
new AccessTokenClaims(1L, "session", "token", "STUDENT", now.minusSeconds(60), now)); |
||||
|
assertThrows(Exception.class, this::connect); |
||||
|
} |
||||
|
@Test void unassignedIdentityRejected() { |
||||
|
when(users.findByIdWithRoles(1L)).thenReturn(Optional.of(User.builder().role(User.Role.UNASSIGNED).build())); |
||||
|
assertThrows(Exception.class, this::connect); |
||||
|
} |
||||
|
@Test void ownFixedQueueAllowed() { |
||||
|
var frame = frame(StompCommand.SUBSCRIBE, connect()); |
||||
|
frame.setDestination("/user/queue/chat-events"); assertDoesNotThrow(() -> send(frame)); |
||||
|
} |
||||
|
@Test void otherUserAndBrokerQueuesRejected() { |
||||
|
var attrs = connect(); |
||||
|
for (String path : List.of("/user/2/queue/chat-events", "/queue/chat-events", "/topic/all")) { |
||||
|
var frame = frame(StompCommand.SUBSCRIBE, attrs); frame.setDestination(path); |
||||
|
assertThrows(Exception.class, () -> send(frame)); |
||||
|
} |
||||
|
} |
||||
|
@Test void socketSendRejected() { |
||||
|
var frame = frame(StompCommand.SEND, connect()); frame.setDestination("/app/chat"); |
||||
|
assertThrows(Exception.class, () -> send(frame)); |
||||
|
} |
||||
|
@Test void databaseFailureFailsClosed() { |
||||
|
when(users.findByIdWithRoles(1L)).thenThrow(new IllegalStateException("private detail")); |
||||
|
var error = assertThrows(Exception.class, this::connect); |
||||
|
assertEquals("Chat connection rejected", error.getMessage()); |
||||
|
} |
||||
|
@Test void repeatedConnectRejected() { |
||||
|
var frame = frame(StompCommand.CONNECT, connect()); frame.setNativeHeader("Authorization", "Bearer valid"); |
||||
|
assertThrows(Exception.class, () -> send(frame)); |
||||
|
} |
||||
|
@Test void unverifiedCompanyRejected() { |
||||
|
when(users.findByIdWithRoles(1L)).thenReturn(Optional.of(User.builder().id(1L) |
||||
|
.role(User.Role.COMPANY).roles(Set.of(User.Role.COMPANY)).build())); |
||||
|
assertThrows(Exception.class, this::connect); |
||||
|
} |
||||
|
@Test void revokedSessionCannotSubscribeAgain() { |
||||
|
var attrs = connect(); when(sessions.isSessionActive(1L, "session", now)).thenReturn(false); |
||||
|
var frame = frame(StompCommand.SUBSCRIBE, attrs); frame.setDestination("/user/queue/chat-events"); |
||||
|
assertThrows(Exception.class, () -> send(frame)); |
||||
|
} |
||||
|
@Test void subscriptionFloodRejected() { |
||||
|
var attrs = connect(); var frame = frame(StompCommand.SUBSCRIBE, attrs); |
||||
|
frame.setDestination("/user/queue/chat-events"); |
||||
|
for (int i = 0; i < 119; i++) send(frame); |
||||
|
assertThrows(Exception.class, () -> send(frame)); |
||||
|
} |
||||
|
@Test void outboundChecksRevocationBeforeDelivery() { |
||||
|
connect(); |
||||
|
var headers = org.springframework.messaging.simp.SimpMessageHeaderAccessor.create( |
||||
|
org.springframework.messaging.simp.SimpMessageType.MESSAGE); |
||||
|
headers.setSessionId("ws-1"); |
||||
|
var message = MessageBuilder.createMessage(new byte[0], headers.getMessageHeaders()); |
||||
|
assertNotNull(guard.outbound().preSend(message, null)); |
||||
|
when(sessions.isSessionActive(1L, "session", now)).thenReturn(false); |
||||
|
assertNull(guard.outbound().preSend(message, null)); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,43 @@ |
|||||
|
package com.mydreamjob.messaging.realtime; |
||||
|
|
||||
|
import org.junit.jupiter.api.Test; |
||||
|
import org.springframework.mock.web.*; |
||||
|
import org.springframework.http.server.*; |
||||
|
import org.springframework.web.socket.*; |
||||
|
import java.time.*; |
||||
|
import java.util.*; |
||||
|
import static org.junit.jupiter.api.Assertions.*; |
||||
|
import static org.mockito.Mockito.*; |
||||
|
|
||||
|
class ChatSocketConnectionsTest { |
||||
|
private final Clock clock = Clock.fixed(Instant.parse("2026-10-14T08:00:00Z"), ZoneOffset.UTC); |
||||
|
@Test void urlCredentialsRejected() throws Exception { |
||||
|
var connections = new ChatSocketConnections(mock(ChatSocketAuth.class), clock); |
||||
|
var request = new MockHttpServletRequest("GET", "/ws"); request.setQueryString("token=secret"); |
||||
|
var response = new MockHttpServletResponse(); |
||||
|
assertFalse(connections.handshakeGuard().beforeHandshake(new ServletServerHttpRequest(request), |
||||
|
new ServletServerHttpResponse(response), mock(WebSocketHandler.class), new HashMap<>())); |
||||
|
assertEquals(400, response.getStatus()); |
||||
|
} |
||||
|
@Test void handshakeFloodRejected() throws Exception { |
||||
|
var connections = new ChatSocketConnections(mock(ChatSocketAuth.class), clock); |
||||
|
var guard = connections.handshakeGuard(); var request = new MockHttpServletRequest("GET", "/ws"); |
||||
|
var response = new MockHttpServletResponse(); |
||||
|
for (int i = 0; i < 40; i++) assertTrue(guard.beforeHandshake(new ServletServerHttpRequest(request), |
||||
|
new ServletServerHttpResponse(response), mock(WebSocketHandler.class), new HashMap<>())); |
||||
|
assertFalse(guard.beforeHandshake(new ServletServerHttpRequest(request), |
||||
|
new ServletServerHttpResponse(response), mock(WebSocketHandler.class), new HashMap<>())); |
||||
|
assertEquals(429, response.getStatus()); |
||||
|
} |
||||
|
@Test void unauthenticatedConnectionIsClosedAndReleased() throws Exception { |
||||
|
var connections = new ChatSocketConnections(mock(ChatSocketAuth.class), clock); |
||||
|
var session = mock(WebSocketSession.class); var attrs = new HashMap<String, Object>(); |
||||
|
when(session.getId()).thenReturn("ws-1"); when(session.getAttributes()).thenReturn(attrs); |
||||
|
var handler = connections.decorate(mock(WebSocketHandler.class)); |
||||
|
handler.afterConnectionEstablished(session); |
||||
|
attrs.put("chat.openedAt", clock.millis() - 20000); |
||||
|
connections.expire(); verify(session).close(CloseStatus.POLICY_VIOLATION); |
||||
|
handler.afterConnectionClosed(session, CloseStatus.POLICY_VIOLATION); |
||||
|
connections.expire(); verify(session, times(1)).close(CloseStatus.POLICY_VIOLATION); |
||||
|
} |
||||
|
} |
||||
Loading…
Reference in new issue