refactor(message): modularize MessageManager architecture
All checks were successful
Frontend CI / ota-android (push) Successful in 11m20s
Frontend CI / build-and-push-web (push) Successful in 34m19s
Frontend CI / build-android-apk (push) Successful in 57m45s

重构消息管理模块,将单体 MessageManager 拆分为职责单一的子模块:

- MessageStateManager: 纯状态管理
- MessageDeduplication: 消息去重服务
- UserCacheService: 用户缓存服务
- ReadReceiptManager: 已读回执管理
- ConversationOperations: 会话操作
- MessageSendService: 消息发送
- MessageSyncService: 消息同步
- WSMessageHandler: WebSocket 消息处理

新增类型定义和常量文件,保持原有 API 向后兼容。修复 CallScreen 状态值错误,补充 PostDetailScreen 缺失样式,优化 GroupInfoPanel 导入。
This commit is contained in:
lafay
2026-03-31 16:13:24 +08:00
parent 2ef267a897
commit 1bee7ea551
16 changed files with 3128 additions and 3258 deletions

View File

@@ -32,7 +32,7 @@ const CallScreen: React.FC = () => {
return '通话功能暂不支持网页端';
case 'connected':
return '通话功能暂不支持网页端';
case 'ending':
case 'ended':
return '通话已结束';
default:
return '';

View File

@@ -1902,6 +1902,10 @@ function createPostDetailStyles(colors: AppColors) {
alignItems: 'center',
padding: spacing.xs,
},
editButtonText: {
marginLeft: 2,
fontSize: fontSizes.sm,
},
reportButtonInline: {
flexDirection: 'row',
alignItems: 'center',

View File

@@ -17,7 +17,7 @@ import {
} from 'react-native';
import { MaterialCommunityIcons } from '@expo/vector-icons';
import { Avatar, Text } from '../../../../components/common';
import { useAppColors, spacing, fontSizes, shadows, type AppColors } from '../../../../theme';
import { useAppColors, spacing, fontSizes, shadows, borderRadius, type AppColors } from '../../../../theme';
import { GroupMemberResponse, GroupResponse, GroupAnnouncementResponse } from '../../../../types/dto';
import { useBreakpointGTE } from '../../../../hooks/useResponsive';
import { groupManager } from '../../../../stores/groupManager';

View File

@@ -0,0 +1,259 @@
/**
* 会话操作服务
* 处理会话的 CRUD 操作
*/
import type { ConversationResponse } from '../../types/dto';
import { messageService } from '../../services/messageService';
import { deleteConversation } from '../../services/database';
import type { IConversationOperations, IMessageStateManager } from './types';
export class ConversationOperations implements IConversationOperations {
private stateManager: IMessageStateManager;
constructor(stateManager: IMessageStateManager) {
this.stateManager = stateManager;
}
/**
* 创建私聊会话
*/
async createConversation(userId: string): Promise<ConversationResponse | null> {
try {
const conversation = await messageService.createConversation(userId);
// 添加到会话列表
this.stateManager.updateConversation(conversation);
// 通知更新
this.stateManager.notifySubscribers({
type: 'conversations_updated',
payload: { conversations: this.stateManager.getConversations() },
timestamp: Date.now(),
});
return conversation;
} catch (error) {
console.error('[ConversationOperations] 创建会话失败:', error);
this.stateManager.notifySubscribers({
type: 'error',
payload: { error, context: 'createConversation' },
timestamp: Date.now(),
});
return null;
}
}
/**
* 本地更新会话信息
*/
updateConversation(conversationId: string, updates: Partial<ConversationResponse>): void {
const conversation = this.stateManager.getConversation(conversationId);
if (!conversation) {
console.warn('[ConversationOperations] 更新会话失败,会话不存在:', conversationId);
return;
}
const updatedConv: ConversationResponse = {
...conversation,
...updates,
};
this.stateManager.updateConversation(updatedConv);
// 通知更新
this.stateManager.notifySubscribers({
type: 'conversations_updated',
payload: { conversations: this.stateManager.getConversations() },
timestamp: Date.now(),
});
}
/**
* 仅自己删除会话后的本地状态回收
*/
removeConversation(conversationId: string): void {
this.stateManager.removeConversation(conversationId);
// 通知更新
this.stateManager.notifySubscribers({
type: 'conversations_updated',
payload: { conversations: this.stateManager.getConversations() },
timestamp: Date.now(),
});
const unreadCount = this.stateManager.getUnreadCount();
this.stateManager.notifySubscribers({
type: 'unread_count_updated',
payload: {
totalUnreadCount: unreadCount.total,
systemUnreadCount: unreadCount.system,
},
timestamp: Date.now(),
});
// 删除本地数据库中的会话
deleteConversation(conversationId).catch(error => {
console.error('[ConversationOperations] 删除本地会话失败:', error);
});
}
/**
* 清空会话列表
*/
clearConversations(): void {
this.stateManager.reset();
this.stateManager.notifySubscribers({
type: 'conversations_updated',
payload: { conversations: [] },
timestamp: Date.now(),
});
this.stateManager.notifySubscribers({
type: 'unread_count_updated',
payload: {
totalUnreadCount: 0,
systemUnreadCount: 0,
},
timestamp: Date.now(),
});
}
/**
* 更新单个会话未读数
*/
updateUnreadCount(conversationId: string, count: number): void {
const conversation = this.stateManager.getConversation(conversationId);
if (!conversation) {
console.warn('[ConversationOperations] 更新未读数失败,会话不存在:', conversationId);
return;
}
const prevUnreadCount = conversation.unread_count || 0;
const diff = count - prevUnreadCount;
// 更新会话未读数
const updatedConv: ConversationResponse = {
...conversation,
unread_count: count,
};
this.stateManager.updateConversation(updatedConv);
// 更新总未读数
const currentUnread = this.stateManager.getUnreadCount();
this.stateManager.setUnreadCount(
Math.max(0, currentUnread.total + diff),
currentUnread.system
);
// 通知更新
this.stateManager.notifySubscribers({
type: 'unread_count_updated',
payload: {
totalUnreadCount: this.stateManager.getUnreadCount().total,
systemUnreadCount: this.stateManager.getUnreadCount().system,
},
timestamp: Date.now(),
});
this.stateManager.notifySubscribers({
type: 'conversations_updated',
payload: { conversations: this.stateManager.getConversations() },
timestamp: Date.now(),
});
}
/**
* 增加会话未读数
*/
incrementUnreadCount(conversationId: string): void {
const conversation = this.stateManager.getConversation(conversationId);
if (!conversation) return;
const prevUnreadCount = conversation.unread_count || 0;
const newUnreadCount = prevUnreadCount + 1;
// 创建新对象而不是直接修改,确保状态一致性
const updatedConv: ConversationResponse = {
...conversation,
unread_count: newUnreadCount,
};
this.stateManager.updateConversation(updatedConv);
const currentUnread = this.stateManager.getUnreadCount();
this.stateManager.setUnreadCount(
currentUnread.total + 1,
currentUnread.system
);
// 通知更新
this.stateManager.notifySubscribers({
type: 'unread_count_updated',
payload: {
totalUnreadCount: this.stateManager.getUnreadCount().total,
systemUnreadCount: this.stateManager.getUnreadCount().system,
},
timestamp: Date.now(),
});
this.stateManager.notifySubscribers({
type: 'conversations_updated',
payload: { conversations: this.stateManager.getConversations() },
timestamp: Date.now(),
});
}
/**
* 设置系统消息未读数
*/
setSystemUnreadCount(count: number): void {
const currentUnread = this.stateManager.getUnreadCount();
this.stateManager.setUnreadCount(currentUnread.total, count);
this.stateManager.notifySubscribers({
type: 'unread_count_updated',
payload: {
totalUnreadCount: currentUnread.total,
systemUnreadCount: count,
},
timestamp: Date.now(),
});
}
/**
* 增加系统消息未读数
*/
incrementSystemUnreadCount(): void {
const currentUnread = this.stateManager.getUnreadCount();
this.stateManager.setUnreadCount(currentUnread.total, currentUnread.system + 1);
this.stateManager.notifySubscribers({
type: 'unread_count_updated',
payload: {
totalUnreadCount: currentUnread.total,
systemUnreadCount: currentUnread.system + 1,
},
timestamp: Date.now(),
});
}
/**
* 减少系统消息未读数
*/
decrementSystemUnreadCount(count = 1): void {
const currentUnread = this.stateManager.getUnreadCount();
const newSystemCount = Math.max(0, currentUnread.system - count);
this.stateManager.setUnreadCount(currentUnread.total, newSystemCount);
this.stateManager.notifySubscribers({
type: 'unread_count_updated',
payload: {
totalUnreadCount: currentUnread.total,
systemUnreadCount: newSystemCount,
},
timestamp: Date.now(),
});
}
}

View File

@@ -0,0 +1,61 @@
/**
* 消息去重服务
* 处理消息去重逻辑,防止重复处理相同的消息
*/
import type { IMessageDeduplication } from './types';
import {
PROCESSED_MESSAGE_ID_EXPIRY,
PROCESSED_MESSAGE_ID_MAX_SIZE,
} from './constants';
export class MessageDeduplication implements IMessageDeduplication {
// 已处理消息ID集合用于去重防止ACK消息和正常消息重复处理
private processedMessageIds: Set<string> = new Set();
private processedMessageIdsExpiry: Map<string, number> = new Map();
/**
* 检查消息是否已处理
*/
isMessageProcessed(messageId: string): boolean {
return this.processedMessageIds.has(messageId);
}
/**
* 标记消息已处理
*/
markMessageAsProcessed(messageId: string): void {
this.processedMessageIds.add(messageId);
this.processedMessageIdsExpiry.set(messageId, Date.now());
// 定期清理
if (this.processedMessageIds.size > PROCESSED_MESSAGE_ID_MAX_SIZE) {
this.cleanupProcessedMessageIds();
}
}
/**
* 清理过期的消息ID防止内存泄漏
*/
cleanupProcessedMessageIds(): void {
const now = Date.now();
for (const [id, timestamp] of this.processedMessageIdsExpiry.entries()) {
if (now - timestamp > PROCESSED_MESSAGE_ID_EXPIRY) {
this.processedMessageIds.delete(id);
this.processedMessageIdsExpiry.delete(id);
}
}
}
/**
* 重置所有已处理的消息ID
* 用于测试或需要清除缓存的场景
*/
reset(): void {
this.processedMessageIds.clear();
this.processedMessageIdsExpiry.clear();
}
}
// 单例导出
export const messageDeduplication = new MessageDeduplication();

View File

@@ -1,221 +1,326 @@
/**
* MessageManager - 重构版
* MessageManager - 消息管理核心模块(重构版
*
* 作为协调者,整合各个专注的服务模块
* 保持与原始版本完全兼容的 API
*/
import { messageStateManager, MessageStateManager } from './MessageStateManager';
import { wsMessageHandler, WSMessageHandler } from './WSMessageHandler';
import { messageSyncService, MessageSyncService } from './MessageSyncService';
import { readReceiptManager, ReadReceiptManager } from './ReadReceiptManager';
import { messageRepository } from '../../data/repositories/MessageRepository';
import type { ConversationResponse, MessageResponse, MessageSegment, UserDTO } from '../../types/dto';
import { messageService } from '../../services/messageService';
import {
wsService,
WSChatMessage,
WSGroupChatMessage,
WSReadMessage,
WSGroupReadMessage,
WSRecallMessage,
WSGroupRecallMessage,
WSGroupTypingMessage,
WSGroupNoticeMessage,
GroupNoticeType,
} from '../../services/wsService';
import {
saveMessage,
saveMessagesBatch,
getMessagesByConversation,
getMaxSeq,
getMinSeq,
getMessagesBeforeSeq,
markConversationAsRead,
updateConversationCacheUnreadCount,
CachedMessage,
getUserCache,
saveUserCache,
updateMessageStatus,
deleteConversation as deleteConversationFromDb,
saveConversationsWithRelatedCache,
} from '../../services/database';
import { api } from '../../services/api';
import { vibrateOnMessage } from '../../services/messageVibrationService';
import { useAuthStore } from '../authStore';
import type { Message, Conversation } from '../../core/entities/Message';
import type { ConversationResponse, MessageResponse } from '../../types/dto';
import {
type IConversationListPagedSource,
SqliteConversationListPagedSource,
createRemoteConversationListSource,
CONVERSATION_LIST_PAGE_SIZE,
} from '../conversationListSources';
// 导入新模块
import { MessageStateManager } from './MessageStateManager';
import { MessageDeduplication } from './MessageDeduplication';
import { UserCacheService } from './UserCacheService';
import { ReadReceiptManager } from './ReadReceiptManager';
import { ConversationOperations } from './ConversationOperations';
import { MessageSendService } from './MessageSendService';
import { MessageSyncService } from './MessageSyncService';
import { WSMessageHandler } from './WSMessageHandler';
// 重新导出类型,保持兼容性
export type {
MessageEventType,
MessageEvent,
MessageSubscriber,
} from './MessageStateManager';
MessageManagerState,
ReadStateRecord,
MessageManagerConversationListDeps,
} from './types';
export class MessageManager {
// ==================== MessageManager 类 ====================
class MessageManager {
// 子模块
private stateManager: MessageStateManager;
private wsHandler: WSMessageHandler;
private deduplication: MessageDeduplication;
private userCacheService: UserCacheService;
private readReceiptManager: ReadReceiptManager;
private conversationOps: ConversationOperations;
private sendService: MessageSendService;
private syncService: MessageSyncService;
private readManager: ReadReceiptManager;
private wsHandler: WSMessageHandler;
private initialized: boolean = false;
private authUnsubscribe: (() => void) | null = null;
private wsUnsubscribe: (() => void) | null = null;
// 本地状态
private currentUserId: string | null = null;
private authUnsubscribe: (() => void) | null = null;
private initializePromise: Promise<void> | null = null;
private activatingConversationTasks: Map<string, Promise<void>> = new Map();
constructor() {
this.stateManager = messageStateManager;
this.wsHandler = wsMessageHandler;
this.syncService = messageSyncService;
this.readManager = readReceiptManager;
constructor(deps?: import('./types').MessageManagerConversationListDeps) {
// 初始化子模块
this.stateManager = new MessageStateManager();
this.deduplication = new MessageDeduplication();
this.userCacheService = new UserCacheService();
this.readReceiptManager = new ReadReceiptManager(this.stateManager);
this.conversationOps = new ConversationOperations(this.stateManager);
this.sendService = new MessageSendService(this.stateManager, () => this.getCurrentUserId());
this.syncService = new MessageSyncService(
this.stateManager,
() => this.getCurrentUserId(),
deps
);
this.wsHandler = new WSMessageHandler(
this.stateManager,
this.deduplication,
this.userCacheService,
() => this.getCurrentUserId(),
{
markAsRead: (id, seq) => this.markAsRead(id, seq),
fetchConversations: (force, source) => this.fetchConversations(force, source),
fetchUnreadCount: () => this.fetchUnreadCount(),
fetchMessages: (id) => this.fetchMessages(id),
}
);
this.readManager.setStateManager(this.stateManager);
this.setupWSHandlers();
// 监听认证状态变化
this.initAuthListener();
}
private initAuthListener(): void {
// ==================== 私有工具方法 ====================
private initAuthListener() {
const authState = useAuthStore.getState();
this.currentUserId = authState.currentUser?.id || null;
if (this.currentUserId && !this.initialized) {
if (this.currentUserId && !this.stateManager.getState().isInitialized) {
this.initialize();
}
this.authUnsubscribe = useAuthStore.subscribe((state) => {
const nextUserId = state.currentUser?.id || null;
this.currentUserId = nextUserId;
if (!this.authUnsubscribe) {
this.authUnsubscribe = useAuthStore.subscribe((state) => {
const nextUserId = state.currentUser?.id || null;
const prevUserId = this.currentUserId;
this.currentUserId = nextUserId;
if (nextUserId && !this.initialized) {
this.initialize();
}
});
if (nextUserId && !this.stateManager.getState().isInitialized) {
this.initialize();
}
if (nextUserId && this.stateManager.getActiveConversation()) {
this.activateConversation(this.stateManager.getActiveConversation()!, { forceSync: true }).catch(error => {
console.error('[MessageManager] 登录态就绪后重激活会话失败:', error);
});
}
if (!nextUserId && prevUserId && this.stateManager.getState().isInitialized) {
this.destroy();
}
});
}
}
private setupWSHandlers(): void {
this.wsUnsubscribe = this.wsHandler.subscribe((event) => {
switch (event.type) {
case 'chat_message':
case 'group_message':
this.handleNewMessage(event.payload);
break;
case 'read_receipt':
case 'group_read_receipt':
this.handleReadReceipt(event.payload);
break;
case 'message_recalled':
case 'group_message_recalled':
this.handleMessageRecall(event.payload);
break;
case 'typing':
this.handleTypingStatus(event.payload);
break;
case 'group_notice':
this.handleGroupNotice(event.payload);
break;
}
});
private getCurrentUserId(): string | null {
if (!this.currentUserId) {
this.currentUserId = useAuthStore.getState().currentUser?.id || null;
}
return this.currentUserId;
}
private normalizeConversationId(conversationId: string | number | null | undefined): string {
return conversationId == null ? '' : String(conversationId);
}
// ==================== 初始化与销毁 ====================
async initialize(): Promise<void> {
if (this.initialized) return;
if (this.stateManager.getState().isInitialized) {
return;
}
if (this.initializePromise) {
await this.initializePromise;
return;
}
this.initializePromise = (async () => {
try {
this.currentUserId = this.getCurrentUserId();
if (!this.currentUserId) {
console.warn('[MessageManager] 用户未登录,延迟初始化');
return;
}
this.wsHandler.setBootstrapping(true);
// 初始化SSE监听
this.wsHandler.connect();
// 加载会话列表
await this.fetchConversations(false, 'initialize');
// 加载未读数
await this.fetchUnreadCount();
this.stateManager.setInitialized(true);
// 处理缓冲的 SSE 事件
await this.wsHandler.flushBufferedSSEEvents();
this.wsHandler.setBootstrapping(false);
this.stateManager.notifySubscribers({
type: 'conversations_updated',
payload: { conversations: this.stateManager.getConversations() },
timestamp: Date.now(),
});
if (this.stateManager.getActiveConversation()) {
await this.fetchMessages(this.stateManager.getActiveConversation()!);
}
} catch (error) {
console.error('[MessageManager] 初始化失败:', error);
this.wsHandler.setBootstrapping(false);
this.stateManager.notifySubscribers({
type: 'error',
payload: { error, context: 'initialize' },
timestamp: Date.now(),
});
}
})();
try {
console.log('[MessageManager] Initializing...');
this.wsHandler.connect();
const conversations = await this.syncService.syncConversations();
this.stateManager.setConversations(conversations);
this.initialized = true;
console.log('[MessageManager] Initialized successfully');
} catch (error) {
console.error('[MessageManager] Initialization failed:', error);
throw error;
await this.initializePromise;
} finally {
this.initializePromise = null;
}
}
async activateConversation(conversationId: string): Promise<void> {
this.stateManager.setCurrentConversation(conversationId);
const existing = this.stateManager.getMessages(conversationId);
if (existing.length === 0) {
await this.loadMessages(conversationId);
}
const conversation = this.stateManager.getConversation(conversationId);
if (conversation && conversation.unreadCount > 0) {
await this.readManager.markAsRead(conversationId, conversation.lastSeq);
}
destroy(): void {
this.wsHandler.disconnect();
this.stateManager.reset();
this.syncService.restartSources();
this.readReceiptManager.reset();
this.deduplication.reset();
this.userCacheService.reset();
this.activatingConversationTasks.clear();
this.initializePromise = null;
}
deactivateConversation(): void {
this.stateManager.setCurrentConversation(null);
// ==================== 数据获取方法(代理到 syncService====================
async fetchConversations(forceRefresh = false, source: string = 'unknown'): Promise<void> {
return this.syncService.fetchConversations(forceRefresh, source);
}
async loadMessages(conversationId: string): Promise<Message[]> {
const localMessages = await this.syncService.loadLocalMessages(conversationId);
if (localMessages.length > 0) {
this.stateManager.setMessages(conversationId, localMessages);
}
const syncResult = await this.syncService.syncMessages(conversationId);
if (syncResult.messages.length > 0) {
this.stateManager.appendMessages(conversationId, syncResult.messages);
return [...localMessages, ...syncResult.messages];
}
return localMessages;
async loadMoreConversations(): Promise<void> {
return this.syncService.loadMoreConversations();
}
async loadHistory(conversationId: string): Promise<Message[]> {
const messages = this.stateManager.getMessages(conversationId);
const oldestSeq = messages.length > 0 ? Math.min(...messages.map(m => m.seq)) : 0;
const history = await this.syncService.loadHistoryMessages(
conversationId,
oldestSeq,
20
);
if (history.length > 0) {
this.stateManager.prependMessages(conversationId, history);
}
return history;
async fetchConversationDetail(conversationId: string): Promise<ConversationResponse | null> {
return this.syncService.fetchConversationDetail(conversationId);
}
async fetchMessages(conversationId: string, afterSeq?: number): Promise<void> {
return this.syncService.fetchMessages(conversationId, afterSeq);
}
async loadMoreMessages(conversationId: string, beforeSeq: number, limit = 20): Promise<MessageResponse[]> {
return this.syncService.loadMoreMessages(conversationId, beforeSeq, limit);
}
async fetchUnreadCount(): Promise<void> {
return this.syncService.fetchUnreadCount();
}
// ==================== 数据操作方法(代理到各服务)====================
async sendMessage(
conversationId: string,
content: string,
type: 'private' | 'group' = 'private'
): Promise<Message> {
throw new Error('sendMessage not implemented - use existing implementation');
segments: MessageSegment[],
options?: { replyToId?: string }
): Promise<MessageResponse | null> {
return this.sendService.sendMessage(conversationId, segments, options);
}
private async handleNewMessage(message: Message): Promise<void> {
await messageRepository.saveMessage(message, true);
this.stateManager.addMessage(message.conversationId, message);
const currentConvId = this.stateManager.getCurrentConversationId();
if (currentConvId !== message.conversationId) {
const conv = this.stateManager.getConversation(message.conversationId);
if (conv) {
this.stateManager.updateConversation(message.conversationId, {
unreadCount: (conv.unreadCount || 0) + 1,
lastSeq: message.seq,
lastMessageAt: message.createdAt,
});
}
}
async markAsRead(conversationId: string, seq: number): Promise<void> {
return this.readReceiptManager.markAsRead(conversationId, seq);
}
private handleReadReceipt(payload: any): void {
const { conversationId, lastReadSeq } = payload;
this.stateManager.updateConversation(conversationId, {
unreadCount: 0,
});
async markAllAsRead(): Promise<void> {
return this.readReceiptManager.markAllAsRead();
}
private async handleMessageRecall(payload: any): Promise<void> {
const { messageId, conversationId } = payload;
await messageRepository.updateMessageStatus(messageId, 'recalled', true);
this.stateManager.updateMessage(conversationId, messageId, {
status: 'recalled',
});
async createConversation(userId: string): Promise<ConversationResponse | null> {
return this.conversationOps.createConversation(userId);
}
private handleTypingStatus(payload: any): void {
const { groupId, userIds } = payload;
this.stateManager.setTypingUsers(groupId, userIds);
updateConversation(conversationId: string, updates: Partial<ConversationResponse>): void {
return this.conversationOps.updateConversation(conversationId, updates);
}
private handleGroupNotice(notice: any): void {
console.log('[MessageManager] Group notice:', notice);
removeConversation(conversationId: string): void {
return this.conversationOps.removeConversation(conversationId);
}
subscribe(callback: any): () => void {
return this.stateManager.subscribe(callback);
clearConversations(): void {
return this.conversationOps.clearConversations();
}
getConversations(): Conversation[] {
updateUnreadCount(conversationId: string, count: number): void {
return this.conversationOps.updateUnreadCount(conversationId, count);
}
setSystemUnreadCount(count: number): void {
return this.conversationOps.setSystemUnreadCount(count);
}
incrementSystemUnreadCount(): void {
return this.conversationOps.incrementSystemUnreadCount();
}
decrementSystemUnreadCount(count?: number): void {
return this.conversationOps.decrementSystemUnreadCount(count);
}
// ==================== 状态查询方法(代理到 stateManager====================
getConversations(): ConversationResponse[] {
return this.stateManager.getConversations();
}
getMessages(conversationId: string): Message[] {
getConversation(conversationId: string): ConversationResponse | null {
return this.stateManager.getConversation(conversationId);
}
getMessages(conversationId: string): MessageResponse[] {
return this.stateManager.getMessages(conversationId);
}
@@ -224,28 +329,172 @@ export class MessageManager {
}
isConnected(): boolean {
return this.wsHandler.isConnected();
return this.stateManager.isConnected();
}
reset(): void {
this.initialized = false;
this.stateManager.reset();
this.readManager.reset();
this.wsHandler.disconnect();
isLoading(): boolean {
return this.stateManager.isLoading();
}
destroy(): void {
if (this.authUnsubscribe) {
this.authUnsubscribe();
this.authUnsubscribe = null;
isLoadingMessages(conversationId: string): boolean {
return this.stateManager.isLoadingMessages(conversationId);
}
getTypingUsers(groupId: string): string[] {
return this.stateManager.getTypingUsers(groupId);
}
isMuted(groupId: string): boolean {
return this.stateManager.isMuted(groupId);
}
setMutedStatus(groupId: string, isMuted: boolean): void {
return this.stateManager.setMutedStatus(groupId, isMuted);
}
// ==================== 活动会话管理 ====================
setActiveConversation(conversationId: string | null): void {
const normalizedId = conversationId ? this.normalizeConversationId(conversationId) : null;
this.stateManager.setCurrentConversation(normalizedId);
}
getActiveConversation(): string | null {
return this.stateManager.getActiveConversation();
}
async activateConversation(conversationId: string, options?: { forceSync?: boolean }): Promise<void> {
const normalizedId = this.normalizeConversationId(conversationId);
if (!normalizedId) return;
await this.initialize();
this.setActiveConversation(normalizedId);
// 先下发当前内存快照
const snapshot = this.getMessages(normalizedId);
this.stateManager.notifySubscribers({
type: 'messages_updated',
payload: {
conversationId: normalizedId,
messages: [...snapshot],
source: 'activation_snapshot',
},
timestamp: Date.now(),
});
const existingTask = this.activatingConversationTasks.get(normalizedId);
if (existingTask && !options?.forceSync) {
await existingTask;
return;
}
if (this.wsUnsubscribe) {
this.wsUnsubscribe();
this.wsUnsubscribe = null;
const task = (async () => {
await this.fetchMessages(normalizedId);
})();
this.activatingConversationTasks.set(normalizedId, task);
try {
await task;
} finally {
if (this.activatingConversationTasks.get(normalizedId) === task) {
this.activatingConversationTasks.delete(normalizedId);
}
}
this.reset();
}
// ==================== 订阅机制 ====================
subscribe(subscriber: import('./types').MessageSubscriber): () => void {
const unsubscribe = this.stateManager.subscribe(subscriber);
// 立即发送当前状态给新订阅者(与原始版本保持一致)
subscriber({
type: 'conversations_updated',
payload: { conversations: this.stateManager.getConversations() },
timestamp: Date.now(),
});
const unreadCount = this.stateManager.getUnreadCount();
subscriber({
type: 'unread_count_updated',
payload: {
totalUnreadCount: unreadCount.total,
systemUnreadCount: unreadCount.system,
},
timestamp: Date.now(),
});
subscriber({
type: 'connection_changed',
payload: { connected: this.stateManager.isConnected() },
timestamp: Date.now(),
});
// 如果当前有活动会话,立即发送该会话的消息给新订阅者
const activeConversationId = this.stateManager.getActiveConversation();
if (activeConversationId) {
const currentMessages = this.stateManager.getMessages(activeConversationId);
if (currentMessages.length > 0) {
subscriber({
type: 'messages_updated',
payload: {
conversationId: activeConversationId,
messages: [...currentMessages],
source: 'initial_sync',
},
timestamp: Date.now(),
});
}
}
return unsubscribe;
}
subscribeToMessages(conversationId: string, callback: (messages: MessageResponse[]) => void): () => void {
return this.stateManager.subscribe((event) => {
if (event.type === 'messages_updated' && event.payload.conversationId === conversationId) {
callback(event.payload.messages);
}
});
}
// ==================== 其他方法 ====================
canLoadMoreConversations(): boolean {
return this.syncService.canLoadMoreConversations();
}
invalidateCache(_type?: 'all' | 'list' | 'unread'): void {
// 兼容性方法,新架构不需要外部缓存
}
async requestConversationListRefresh(
source: string,
options?: { force?: boolean; allowDefer?: boolean }
): Promise<void> {
const forceRefresh = options?.force ?? true;
const allowDefer = options?.allowDefer ?? true;
if (allowDefer && this.shouldDeferConversationRefresh(source, forceRefresh)) {
// 延迟刷新逻辑在 syncService 中处理
return;
}
await this.fetchConversations(forceRefresh, source);
}
private shouldDeferConversationRefresh(source: string, forceRefresh: boolean): boolean {
if (!this.stateManager.getActiveConversation()) return false;
if (!forceRefresh) return false;
return source === 'sse-reconnect' || source === 'prefetch' || source === 'hooks-initial-refresh';
}
}
// ==================== 单例导出 ====================
export const messageManager = new MessageManager();
export default messageManager;
// 导出类以支持类型检查和测试
export { MessageManager };
export default messageManager;

View File

@@ -0,0 +1,144 @@
/**
* 消息发送服务
* 处理消息发送相关逻辑
*/
import type { MessageResponse, MessageSegment, ConversationResponse } from '../../types/dto';
import { messageService } from '../../services/messageService';
import { saveMessage } from '../../services/database';
import type { IMessageSendService, IMessageStateManager } from './types';
export class MessageSendService implements IMessageSendService {
private stateManager: IMessageStateManager;
private getCurrentUserId: () => string | null;
constructor(stateManager: IMessageStateManager, getCurrentUserId: () => string | null) {
this.stateManager = stateManager;
this.getCurrentUserId = getCurrentUserId;
}
/**
* 发送消息
*/
async sendMessage(
conversationId: string,
segments: MessageSegment[],
options?: { replyToId?: string }
): Promise<MessageResponse | null> {
try {
// 调用API发送
const response = await messageService.sendMessage(conversationId, {
segments,
reply_to_id: options?.replyToId,
});
if (response) {
const currentUserId = this.getCurrentUserId();
// 添加到本地消息列表
const existingMessages = this.stateManager.getMessages(conversationId);
const newMessage: MessageResponse = {
id: response.id,
conversation_id: conversationId,
sender_id: currentUserId || '',
seq: response.seq,
segments,
created_at: new Date().toISOString(),
status: 'normal',
};
const updatedMessages = this.mergeMessagesById(existingMessages, [newMessage]);
this.stateManager.setMessages(conversationId, updatedMessages);
// 关键:立即广播消息列表更新
this.stateManager.notifySubscribers({
type: 'messages_updated',
payload: {
conversationId,
messages: updatedMessages,
newMessage,
source: 'send',
},
timestamp: Date.now(),
});
// 更新会话最后消息
this.updateConversationWithNewMessage(conversationId, newMessage);
// 通知消息已发送
this.stateManager.notifySubscribers({
type: 'message_sent',
payload: { conversationId, message: newMessage },
timestamp: Date.now(),
});
// 保存到本地数据库
const textContent = segments
?.filter((s: any) => s.type === 'text')
.map((s: any) => s.data?.text || '')
.join('') || '';
saveMessage({
id: response.id,
conversationId,
senderId: currentUserId || '',
content: textContent,
type: segments?.find((s: any) => s.type === 'image') ? 'image' : 'text',
isRead: true,
createdAt: newMessage.created_at,
seq: response.seq,
status: 'normal',
segments,
}).catch(console.error);
return newMessage;
}
return null;
} catch (error) {
console.error('[MessageSendService] 发送消息失败:', error);
this.stateManager.notifySubscribers({
type: 'error',
payload: { error, context: 'sendMessage' },
timestamp: Date.now(),
});
return null;
}
}
/**
* 更新会话的最后消息信息
*/
private updateConversationWithNewMessage(
conversationId: string,
message: MessageResponse
): void {
const conversation = this.stateManager.getConversation(conversationId);
const createdAt = message.created_at || new Date().toISOString();
if (conversation) {
// 更新现有会话
const updatedConv: ConversationResponse = {
...conversation,
last_seq: message.seq,
last_message: message,
last_message_at: createdAt,
updated_at: createdAt,
};
this.stateManager.updateConversation(updatedConv);
}
}
/**
* 按 message_id 合并消息并按 seq 排序
*/
private mergeMessagesById(base: MessageResponse[], incoming: MessageResponse[]): MessageResponse[] {
if (incoming.length === 0) return base;
const merged = new Map<string, MessageResponse>();
[...base, ...incoming].forEach(msg => {
merged.set(String(msg.id), msg);
});
return Array.from(merged.values()).sort((a, b) => a.seq - b.seq);
}
}

View File

@@ -1,301 +1,260 @@
/**
* 消息状态管理器
* 只负责管理状态,不包含业务逻辑
* 纯状态管理类,不包含业务逻辑
* 负责管理所有消息相关的内存状态
*/
import type { Message, Conversation, UnreadCount } from '../../core/entities/Message';
import type {
ConversationResponse,
MessageResponse,
} from '../../types/dto';
import type {
MessageManagerState,
MessageSubscriber,
MessageEvent,
IMessageStateManager,
} from './types';
export type MessageEventType =
| 'conversations_updated'
| 'messages_updated'
| 'unread_count_updated'
| 'connection_changed'
| 'message_sent'
| 'message_read'
| 'message_received'
| 'message_recalled'
| 'typing_status'
| 'group_notice'
| 'error';
export interface MessageEvent {
type: MessageEventType;
payload: any;
timestamp: number;
}
export type MessageSubscriber = (event: MessageEvent) => void;
export interface MessageState {
conversations: Map<string, Conversation>;
conversationList: Conversation[];
messagesMap: Map<string, Message[]>;
unreadCount: UnreadCount;
isSSEConnected: boolean;
currentConversationId: string | null;
isLoading: boolean;
typingUsersMap: Map<string, string[]>;
mutedStatusMap: Map<string, boolean>;
}
export class MessageStateManager {
private state: MessageState;
private subscribers: Set<MessageSubscriber>;
private readStateVersion: number = 0;
private pendingReadMap: Map<string, { timestamp: number; version: number }> = new Map();
export class MessageStateManager implements IMessageStateManager {
private state: MessageManagerState;
constructor() {
this.state = {
conversations: new Map(),
conversationList: [],
messagesMap: new Map(),
unreadCount: { total: 0, system: 0 },
isSSEConnected: false,
totalUnreadCount: 0,
systemUnreadCount: 0,
isWSConnected: false,
currentConversationId: null,
isLoading: false,
isLoadingConversations: false,
loadingMessagesSet: new Set(),
isInitialized: false,
subscribers: new Set(),
typingUsersMap: new Map(),
mutedStatusMap: new Map(),
};
this.subscribers = new Set();
}
// ==================== 状态获取 ====================
// ==================== 获取状态 ====================
getState(): MessageState {
getState(): MessageManagerState {
return this.state;
}
getConversations(): Conversation[] {
getConversations(): ConversationResponse[] {
return this.state.conversationList;
}
getConversation(id: string): Conversation | undefined {
return this.state.conversations.get(id);
getConversation(conversationId: string): ConversationResponse | null {
const normalizedId = this.normalizeConversationId(conversationId);
return this.state.conversations.get(normalizedId) || null;
}
getMessages(conversationId: string): Message[] {
return this.state.messagesMap.get(conversationId) || [];
getMessages(conversationId: string): MessageResponse[] {
const normalizedId = this.normalizeConversationId(conversationId);
return this.state.messagesMap.get(normalizedId) || [];
}
getUnreadCount(): UnreadCount {
return this.state.unreadCount;
}
getCurrentConversationId(): string | null {
return this.state.currentConversationId;
getUnreadCount(): { total: number; system: number } {
return {
total: this.state.totalUnreadCount,
system: this.state.systemUnreadCount,
};
}
isConnected(): boolean {
return this.state.isSSEConnected;
return this.state.isWSConnected;
}
isLoading(): boolean {
return this.state.isLoadingConversations;
}
isLoadingMessages(conversationId: string): boolean {
const normalizedId = this.normalizeConversationId(conversationId);
return this.state.loadingMessagesSet.has(normalizedId);
}
getTypingUsers(groupId: string): string[] {
return this.state.typingUsersMap.get(groupId) || [];
}
getMutedStatus(groupId: string): boolean {
isMuted(groupId: string): boolean {
return this.state.mutedStatusMap.get(groupId) || false;
}
// ==================== 状态更新 ====================
getActiveConversation(): string | null {
return this.state.currentConversationId;
}
setConversations(conversations: Conversation[]): void {
this.state.conversations.clear();
this.state.conversationList = conversations;
// ==================== 设置状态 ====================
setConversations(conversations: Map<string, ConversationResponse>): void {
this.state.conversations = conversations;
this.updateConversationList();
}
updateConversation(conversation: ConversationResponse): void {
const id = this.normalizeConversationId(conversation.id);
this.state.conversations.set(id, { ...conversation, id });
this.updateConversationList();
}
removeConversation(conversationId: string): void {
const normalizedId = this.normalizeConversationId(conversationId);
const target = this.state.conversations.get(normalizedId);
const removedUnread = target?.unread_count || 0;
this.state.conversations.delete(normalizedId);
this.state.messagesMap.delete(normalizedId);
this.state.totalUnreadCount = Math.max(0, this.state.totalUnreadCount - removedUnread);
for (const conv of conversations) {
this.state.conversations.set(conv.id, conv);
if (this.state.currentConversationId === normalizedId) {
this.state.currentConversationId = null;
}
this.updateUnreadCount();
this.notify({ type: 'conversations_updated', payload: conversations });
this.updateConversationList();
}
addConversation(conversation: Conversation): void {
this.state.conversations.set(conversation.id, conversation);
this.state.conversationList = Array.from(this.state.conversations.values());
this.updateUnreadCount();
this.notify({ type: 'conversations_updated', payload: this.state.conversationList });
setMessages(conversationId: string, messages: MessageResponse[]): void {
const normalizedId = this.normalizeConversationId(conversationId);
this.state.messagesMap.set(normalizedId, messages);
}
updateConversation(id: string, updates: Partial<Conversation>): void {
const existing = this.state.conversations.get(id);
if (existing) {
const updated = { ...existing, ...updates };
this.state.conversations.set(id, updated);
this.state.conversationList = Array.from(this.state.conversations.values());
this.updateUnreadCount();
this.notify({ type: 'conversations_updated', payload: this.state.conversationList });
addMessage(conversationId: string, message: MessageResponse): void {
const normalizedId = this.normalizeConversationId(conversationId);
const existing = this.state.messagesMap.get(normalizedId) || [];
const exists = existing.some(m => String(m.id) === String(message.id));
if (!exists) {
const updated = this.mergeMessagesById(existing, [message]);
this.state.messagesMap.set(normalizedId, updated);
}
}
removeConversation(id: string): void {
this.state.conversations.delete(id);
this.state.messagesMap.delete(id);
this.state.conversationList = Array.from(this.state.conversations.values());
this.updateUnreadCount();
this.notify({ type: 'conversations_updated', payload: this.state.conversationList });
}
setMessages(conversationId: string, messages: Message[]): void {
this.state.messagesMap.set(conversationId, messages);
this.notify({ type: 'messages_updated', payload: { conversationId, messages } });
}
addMessage(conversationId: string, message: Message): void {
const messages = this.state.messagesMap.get(conversationId) || [];
const existingIndex = messages.findIndex(m => m.id === message.id);
if (existingIndex >= 0) {
messages[existingIndex] = message;
} else {
messages.push(message);
}
this.state.messagesMap.set(conversationId, messages);
this.notify({ type: 'messages_updated', payload: { conversationId, messages } });
}
prependMessages(conversationId: string, newMessages: Message[]): void {
const existing = this.state.messagesMap.get(conversationId) || [];
const merged = [...newMessages, ...existing];
const unique = this.deduplicateMessages(merged);
this.state.messagesMap.set(conversationId, unique);
this.notify({ type: 'messages_updated', payload: { conversationId, messages: unique } });
}
appendMessages(conversationId: string, newMessages: Message[]): void {
const existing = this.state.messagesMap.get(conversationId) || [];
const merged = [...existing, ...newMessages];
const unique = this.deduplicateMessages(merged);
this.state.messagesMap.set(conversationId, unique);
this.notify({ type: 'messages_updated', payload: { conversationId, messages: unique } });
}
updateMessage(conversationId: string, messageId: string, updates: Partial<Message>): void {
const messages = this.state.messagesMap.get(conversationId);
updateMessage(conversationId: string, messageId: string, updates: Partial<MessageResponse>): void {
const normalizedId = this.normalizeConversationId(conversationId);
const messages = this.state.messagesMap.get(normalizedId);
if (!messages) return;
const index = messages.findIndex(m => m.id === messageId);
if (index >= 0) {
messages[index] = { ...messages[index], ...updates };
this.notify({ type: 'messages_updated', payload: { conversationId, messages } });
}
const updated = messages.map(m =>
String(m.id) === String(messageId) ? { ...m, ...updates } : m
);
this.state.messagesMap.set(normalizedId, updated);
}
setUnreadCount(count: UnreadCount): void {
this.state.unreadCount = count;
this.notify({ type: 'unread_count_updated', payload: count });
setUnreadCount(total: number, system: number): void {
this.state.totalUnreadCount = total;
this.state.systemUnreadCount = system;
}
setSSEConnected(connected: boolean): void {
this.state.isSSEConnected = connected;
this.notify({ type: 'connection_changed', payload: connected });
this.state.isWSConnected = connected;
}
setCurrentConversation(id: string | null): void {
this.state.currentConversationId = id;
setCurrentConversation(conversationId: string | null): void {
this.state.currentConversationId = conversationId;
}
setLoading(loading: boolean): void {
this.state.isLoading = loading;
this.state.isLoadingConversations = loading;
}
setTypingUsers(groupId: string, userIds: string[]): void {
this.state.typingUsersMap.set(groupId, userIds);
this.notify({ type: 'typing_status', payload: { groupId, userIds } });
setLoadingMessages(conversationId: string, loading: boolean): void {
const normalizedId = this.normalizeConversationId(conversationId);
if (loading) {
this.state.loadingMessagesSet.add(normalizedId);
} else {
this.state.loadingMessagesSet.delete(normalizedId);
}
}
setMutedStatus(groupId: string, muted: boolean): void {
this.state.mutedStatusMap.set(groupId, muted);
setTypingUsers(groupId: string, users: string[]): void {
this.state.typingUsersMap.set(groupId, users);
}
// ==================== 已读状态保护 ====================
beginReadOperation(conversationId: string): number {
this.readStateVersion++;
this.pendingReadMap.set(conversationId, {
timestamp: Date.now(),
version: this.readStateVersion,
});
return this.readStateVersion;
setMutedStatus(groupId: string, isMuted: boolean): void {
this.state.mutedStatusMap.set(groupId, isMuted);
}
endReadOperation(conversationId: string): void {
setTimeout(() => {
this.pendingReadMap.delete(conversationId);
}, 5000);
}
isReadOperationPending(conversationId: string): boolean {
const record = this.pendingReadMap.get(conversationId);
if (!record) return false;
return Date.now() - record.timestamp < 5000;
}
getReadStateVersion(): number {
return this.readStateVersion;
setInitialized(initialized: boolean): void {
this.state.isInitialized = initialized;
}
// ==================== 订阅机制 ====================
subscribe(callback: MessageSubscriber): () => void {
this.subscribers.add(callback);
return () => this.subscribers.delete(callback);
subscribe(subscriber: MessageSubscriber): () => void {
this.state.subscribers.add(subscriber);
return () => {
this.state.subscribers.delete(subscriber);
};
}
private notify(event: Omit<MessageEvent, 'timestamp'>): void {
const fullEvent: MessageEvent = {
...event,
timestamp: Date.now(),
};
this.subscribers.forEach(callback => {
notifySubscribers(event: MessageEvent): void {
this.state.subscribers.forEach(subscriber => {
try {
callback(fullEvent);
subscriber(event);
} catch (error) {
console.error('[MessageStateManager] Subscriber error:', error);
console.error('[MessageStateManager] 订阅者执行失败:', error);
}
});
}
// ==================== 工具方法 ====================
private updateUnreadCount(): void {
let total = 0;
let system = 0;
for (const conv of this.state.conversations.values()) {
total += conv.unreadCount || 0;
}
this.state.unreadCount = { total, system };
this.notify({ type: 'unread_count_updated', payload: this.state.unreadCount });
private normalizeConversationId(conversationId: string | number | null | undefined): string {
return conversationId == null ? '' : String(conversationId);
}
private deduplicateMessages(messages: Message[]): Message[] {
const seen = new Set<string>();
return messages.filter(msg => {
if (seen.has(msg.id)) return false;
seen.add(msg.id);
return true;
/**
* 更新会话列表排序
* 排序规则:置顶优先,再按最后消息时间排序
*/
private updateConversationList(): void {
const list = Array.from(this.state.conversations.values()).sort((a, b) => {
const aPinned = a.is_pinned ? 1 : 0;
const bPinned = b.is_pinned ? 1 : 0;
if (aPinned !== bPinned) {
return bPinned - aPinned;
}
const aTime = new Date(a.last_message_at || a.updated_at || 0).getTime();
const bTime = new Date(b.last_message_at || b.updated_at || 0).getTime();
return bTime - aTime;
});
this.state.conversationList = list;
}
/**
* 按 message_id 合并消息并按 seq 排序
* 后者覆盖前者,可用来吸收更完整的同 ID 消息体
*/
private mergeMessagesById(base: MessageResponse[], incoming: MessageResponse[]): MessageResponse[] {
if (incoming.length === 0) return base;
const merged = new Map<string, MessageResponse>();
[...base, ...incoming].forEach(msg => {
merged.set(String(msg.id), msg);
});
return Array.from(merged.values()).sort((a, b) => a.seq - b.seq);
}
// ==================== 重置 ====================
reset(): void {
this.state = {
conversations: new Map(),
conversationList: [],
messagesMap: new Map(),
unreadCount: { total: 0, system: 0 },
isSSEConnected: false,
currentConversationId: null,
isLoading: false,
typingUsersMap: new Map(),
mutedStatusMap: new Map(),
};
this.pendingReadMap.clear();
this.readStateVersion = 0;
this.state.conversations.clear();
this.state.conversationList = [];
this.state.messagesMap.clear();
this.state.totalUnreadCount = 0;
this.state.systemUnreadCount = 0;
this.state.isWSConnected = false;
this.state.currentConversationId = null;
this.state.isLoadingConversations = false;
this.state.loadingMessagesSet.clear();
this.state.isInitialized = false;
this.state.typingUsersMap.clear();
this.state.mutedStatusMap.clear();
}
}
export const messageStateManager = new MessageStateManager();
// 单例导出
export const messageStateManager = new MessageStateManager();

View File

@@ -1,180 +1,643 @@
/**
* 消息同步服务
* 负责从服务器同步消息和会话
* 处理消息和会话的服务器同步逻辑
*/
import type { ConversationResponse, MessageResponse } from '../../types/dto';
import { messageService } from '../../services/messageService';
import { messageRepository } from '../../data/repositories/MessageRepository';
import type { Message, Conversation } from '../../core/entities/Message';
import type { ConversationListResponse, MessageListResponse } from '../../types/dto';
import {
getMessagesByConversation,
getMaxSeq,
saveMessagesBatch,
saveConversationsWithRelatedCache,
getMessagesBeforeSeq,
} from '../../services/database';
import {
type IConversationListPagedSource,
SqliteConversationListPagedSource,
createRemoteConversationListSource,
CONVERSATION_LIST_PAGE_SIZE,
} from '../conversationListSources';
import type { IMessageSyncService, IMessageStateManager, MessageManagerConversationListDeps } from './types';
export interface SyncOptions {
force?: boolean;
lastSeq?: number;
}
export class MessageSyncService implements IMessageSyncService {
private stateManager: IMessageStateManager;
private getCurrentUserId: () => string | null;
export interface SyncResult {
conversations: Conversation[];
messages: Message[];
hasMore: boolean;
}
/** 远端会话列表(游标或页码) */
private readonly remoteConversationListSource: IConversationListPagedSource;
/** 本地会话列表缓存SQLite */
private readonly localConversationListSource: IConversationListPagedSource;
export class MessageSyncService {
private syncingConversations: boolean = false;
private syncingMessages: Map<string, boolean> = new Map();
/** 正在加载会话列表下一页 */
private loadingMoreConversations = false;
/** 聊天页活跃期间延迟的会话列表刷新 */
private deferredConversationRefresh = false;
async syncConversations(): Promise<Conversation[]> {
if (this.syncingConversations) {
console.log('[MessageSyncService] Already syncing conversations');
constructor(
stateManager: IMessageStateManager,
getCurrentUserId: () => string | null,
deps?: MessageManagerConversationListDeps
) {
this.stateManager = stateManager;
this.getCurrentUserId = getCurrentUserId;
const pageSize = deps?.remoteListPageSize ?? CONVERSATION_LIST_PAGE_SIZE;
this.remoteConversationListSource =
deps?.remoteConversationListSource ??
createRemoteConversationListSource(
deps?.remoteListKind === 'offset' ? 'offset' : 'cursor',
pageSize
);
this.localConversationListSource =
deps?.localConversationListSource ?? new SqliteConversationListPagedSource();
}
/**
* 获取会话列表
*/
async fetchConversations(forceRefresh = false, source: string = 'unknown'): Promise<void> {
if (this.stateManager.isLoading() && !forceRefresh) {
return;
}
if (this.shouldDeferConversationRefresh(source, forceRefresh)) {
this.deferredConversationRefresh = true;
if (__DEV__) {
console.log('[MessageSyncService] defer fetchConversations', {
source,
activeConversationId: this.stateManager.getActiveConversation(),
});
}
return;
}
if (__DEV__) {
console.log('[MessageSyncService] fetchConversations', {
source,
forceRefresh,
activeConversationId: this.stateManager.getActiveConversation(),
});
}
// 非强制刷新且内存为空:先用本地源暖机
const currentState = this.stateManager.getState();
if (!forceRefresh && currentState.conversations.size === 0) {
const warmed = await this.hydrateConversationsFromLocalSource();
if (warmed) {
this.emitConversationListAndUnreadUpdates();
}
}
this.stateManager.setLoading(true);
const emitLoadingToUi = this.stateManager.getConversations().length === 0;
if (emitLoadingToUi) {
this.stateManager.notifySubscribers({
type: 'conversations_loading',
payload: { loading: true },
timestamp: Date.now(),
});
}
try {
this.remoteConversationListSource.restart();
const page = await this.remoteConversationListSource.loadNext();
currentState.conversations.clear();
page.items.forEach(conv => {
const normalizedConv = this.normalizeConversationFromFetch(conv);
currentState.conversations.set(normalizedConv.id, normalizedConv);
});
this.updateConversationList();
this.recomputeConversationTotalUnread();
this.emitConversationListAndUnreadUpdates();
} catch (error) {
console.error('[MessageSyncService] 获取会话列表失败:', error);
if (this.stateManager.getConversations().length === 0) {
const recovered = await this.hydrateConversationsFromLocalSource();
if (recovered) {
this.emitConversationListAndUnreadUpdates();
}
}
this.stateManager.notifySubscribers({
type: 'error',
payload: { error, context: 'fetchConversations' },
timestamp: Date.now(),
});
} finally {
this.stateManager.setLoading(false);
if (emitLoadingToUi) {
this.stateManager.notifySubscribers({
type: 'conversations_loading',
payload: { loading: false },
timestamp: Date.now(),
});
}
}
}
/**
* 会话列表加载下一页
*/
async loadMoreConversations(): Promise<void> {
if (!this.remoteConversationListSource.hasMore) {
return;
}
if (this.loadingMoreConversations || this.stateManager.isLoading()) {
return;
}
this.loadingMoreConversations = true;
try {
const page = await this.remoteConversationListSource.loadNext();
page.items.forEach(conv => {
const normalizedConv = this.normalizeConversationFromFetch(conv);
this.stateManager.updateConversation(normalizedConv);
});
this.recomputeConversationTotalUnread();
this.emitConversationListAndUnreadUpdates();
} catch (error) {
console.error('[MessageSyncService] 加载更多会话失败:', error);
this.stateManager.notifySubscribers({
type: 'error',
payload: { error, context: 'loadMoreConversations' },
timestamp: Date.now(),
});
} finally {
this.loadingMoreConversations = false;
}
}
/**
* 获取单个会话详情
*/
async fetchConversationDetail(conversationId: string): Promise<ConversationResponse | null> {
const normalizedId = this.normalizeConversationId(conversationId);
try {
const detail = await messageService.getConversationById(normalizedId);
if (detail) {
const normalizedDetail = this.normalizeConversationFromFetch(detail as ConversationResponse);
this.stateManager.updateConversation(normalizedDetail);
return normalizedDetail;
}
return null;
} catch (error) {
console.error('[MessageSyncService] 获取会话详情失败:', error);
return null;
}
}
/**
* 获取会话消息(增量同步)
*/
async fetchMessages(conversationId: string, afterSeq?: number): Promise<void> {
// 防止重复加载
if (this.stateManager.isLoadingMessages(conversationId)) {
return;
}
this.stateManager.setLoadingMessages(conversationId, true);
try {
const existingMessagesAtStart = this.stateManager.getMessages(conversationId);
const hasInMemoryMessages = existingMessagesAtStart.length > 0;
let baselineMaxSeq = hasInMemoryMessages
? existingMessagesAtStart.reduce((max, m) => Math.max(max, m.seq || 0), 0)
: 0;
if (!afterSeq) {
// 先从本地数据库加载
if (!hasInMemoryMessages) {
try {
const localMessages = await getMessagesByConversation(conversationId, 20);
const localMaxSeq = await getMaxSeq(conversationId);
baselineMaxSeq = localMaxSeq;
if (localMessages.length > 0) {
const formattedMessages: MessageResponse[] = localMessages.map(m => ({
id: m.id,
conversation_id: m.conversationId,
sender_id: m.senderId,
seq: m.seq,
segments: m.segments || [],
status: m.status as any,
created_at: m.createdAt,
}));
this.stateManager.setMessages(conversationId, formattedMessages);
this.stateManager.notifySubscribers({
type: 'messages_updated',
payload: {
conversationId,
messages: formattedMessages,
source: 'local',
},
timestamp: Date.now(),
});
} else {
// 冷启动兜底:先下发空列表事件
this.stateManager.setMessages(conversationId, []);
this.stateManager.notifySubscribers({
type: 'messages_updated',
payload: {
conversationId,
messages: [],
source: 'local_empty',
},
timestamp: Date.now(),
});
}
} catch (error) {
console.warn('[MessageSyncService] 读取本地消息失败:', error);
}
}
// 服务端快照 + 增量同步
try {
const snapshotResp = await messageService.getMessages(conversationId, undefined, undefined, 50);
const snapshotMessages = snapshotResp?.messages || [];
if (snapshotMessages.length > 0) {
const existingMessages = this.stateManager.getMessages(conversationId);
const mergedSnapshot = this.mergeMessagesById(existingMessages, snapshotMessages);
this.stateManager.setMessages(conversationId, mergedSnapshot);
this.stateManager.notifySubscribers({
type: 'messages_updated',
payload: {
conversationId,
messages: mergedSnapshot,
newMessages: snapshotMessages,
source: 'server_snapshot',
},
timestamp: Date.now(),
});
// 持久化到本地
saveMessagesBatch(snapshotMessages.map((m: any) => ({
id: m.id,
conversationId: m.conversation_id || conversationId,
senderId: m.sender_id,
content: m.content,
type: m.type || 'text',
isRead: m.is_read || false,
createdAt: m.created_at,
seq: m.seq,
status: m.status || 'normal',
segments: m.segments,
}))).catch(error => {
console.error('[MessageSyncService] 保存快照消息到本地失败:', error);
});
}
// 增量补齐
const snapshotMaxSeq = snapshotMessages.reduce((max, m) => Math.max(max, m.seq || 0), 0);
if (snapshotMaxSeq > baselineMaxSeq) {
const incrementalResp = await messageService.getMessages(conversationId, baselineMaxSeq);
const newMessages = incrementalResp?.messages || [];
if (newMessages.length > 0) {
const existingMessages = this.stateManager.getMessages(conversationId);
const mergedMessages = this.mergeMessagesById(existingMessages, newMessages);
this.stateManager.setMessages(conversationId, mergedMessages);
this.stateManager.notifySubscribers({
type: 'messages_updated',
payload: {
conversationId,
messages: mergedMessages,
newMessages,
source: 'server_incremental',
},
timestamp: Date.now(),
});
saveMessagesBatch(newMessages.map((m: any) => ({
id: m.id,
conversationId: m.conversation_id || conversationId,
senderId: m.sender_id,
content: m.content,
type: m.type || 'text',
isRead: m.is_read || false,
createdAt: m.created_at,
seq: m.seq,
status: m.status || 'normal',
segments: m.segments,
}))).catch(error => {
console.error('[MessageSyncService] 保存增量消息到本地失败:', error);
});
}
}
} catch (error) {
console.error('[MessageSyncService] 快照/增量同步失败:', error);
}
} else {
// 指定了 afterSeq
const response = await messageService.getMessages(conversationId, afterSeq);
if (response?.messages && response.messages.length > 0) {
const newMessages = response.messages;
const existingMessages = this.stateManager.getMessages(conversationId);
const mergedMessages = this.mergeMessagesById(existingMessages, newMessages);
this.stateManager.setMessages(conversationId, mergedMessages);
this.stateManager.notifySubscribers({
type: 'messages_updated',
payload: {
conversationId,
messages: mergedMessages,
newMessages,
source: 'server',
},
timestamp: Date.now(),
});
saveMessagesBatch(newMessages.map((m: any) => ({
id: m.id,
conversationId: m.conversation_id || conversationId,
senderId: m.sender_id,
content: m.content,
type: m.type || 'text',
isRead: m.is_read || false,
createdAt: m.created_at,
seq: m.seq,
status: m.status || 'normal',
segments: m.segments,
}))).catch(error => {
console.error('[MessageSyncService] 保存消息到本地失败:', error);
});
}
}
} catch (error) {
console.error('[MessageSyncService] 获取消息失败:', error);
this.stateManager.notifySubscribers({
type: 'error',
payload: { error, context: 'fetchMessages', conversationId },
timestamp: Date.now(),
});
} finally {
this.stateManager.setLoadingMessages(conversationId, false);
}
}
/**
* 加载更多历史消息
*/
async loadMoreMessages(conversationId: string, beforeSeq: number, limit = 20): Promise<MessageResponse[]> {
try {
// 先从本地获取
const localMessages = await getMessagesBeforeSeq(conversationId, beforeSeq, limit);
if (localMessages.length >= limit) {
const formattedMessages: MessageResponse[] = [...localMessages].reverse().map(m => ({
id: m.id,
conversation_id: m.conversationId,
sender_id: m.senderId,
seq: m.seq,
segments: m.segments || [],
status: m.status as any,
created_at: m.createdAt,
}));
const existingMessages = this.stateManager.getMessages(conversationId);
const mergedMessages = this.mergeOlderMessages(existingMessages, formattedMessages);
this.stateManager.setMessages(conversationId, mergedMessages);
this.stateManager.notifySubscribers({
type: 'messages_updated',
payload: { conversationId, messages: mergedMessages, source: 'local_history' },
timestamp: Date.now(),
});
return formattedMessages;
}
// 本地数据不足,从服务端获取
const response = await messageService.getMessages(conversationId, undefined, beforeSeq, limit);
if (response?.messages && response.messages.length > 0) {
const serverMessages = response.messages;
await saveMessagesBatch(serverMessages.map((m: any) => ({
id: m.id,
conversationId: m.conversation_id || conversationId,
senderId: m.sender_id,
content: m.content,
type: m.type || 'text',
isRead: m.is_read || false,
createdAt: m.created_at,
seq: m.seq,
status: m.status || 'normal',
segments: m.segments,
})));
const existingMessages = this.stateManager.getMessages(conversationId);
const mergedMessages = this.mergeOlderMessages(existingMessages, serverMessages);
this.stateManager.setMessages(conversationId, mergedMessages);
this.stateManager.notifySubscribers({
type: 'messages_updated',
payload: { conversationId, messages: mergedMessages, source: 'server_history' },
timestamp: Date.now(),
});
return serverMessages;
}
return [];
} catch (error) {
console.error('[MessageSyncService] 加载更多消息失败:', error);
return [];
}
}
this.syncingConversations = true;
/**
* 获取未读数
*/
async fetchUnreadCount(): Promise<void> {
try {
const response: ConversationListResponse = await messageService.getConversations();
const conversations = this.mapConversations(response.list || []);
return conversations;
const [unreadData, systemUnreadData] = await Promise.all([
messageService.getUnreadCount(),
messageService.getSystemUnreadCount(),
]);
const totalUnread = unreadData?.total_unread_count ?? 0;
const systemUnread = systemUnreadData?.unread_count ?? 0;
this.stateManager.setUnreadCount(totalUnread, systemUnread);
// 服务端汇总未读为 0 时,清掉内存中残留的红点
if (totalUnread === 0) {
const currentState = this.stateManager.getState();
let anyCleared = false;
for (const [cid, conv] of currentState.conversations) {
if ((conv.unread_count || 0) > 0) {
currentState.conversations.set(cid, { ...conv, unread_count: 0 });
anyCleared = true;
}
}
if (anyCleared) {
this.updateConversationList();
this.persistConversationListCache();
this.stateManager.notifySubscribers({
type: 'conversations_updated',
payload: { conversations: this.stateManager.getConversations() },
timestamp: Date.now(),
});
}
}
this.stateManager.notifySubscribers({
type: 'unread_count_updated',
payload: {
totalUnreadCount: totalUnread,
systemUnreadCount: systemUnread,
},
timestamp: Date.now(),
});
} catch (error) {
console.error('[MessageSyncService] Failed to sync conversations:', error);
throw error;
} finally {
this.syncingConversations = false;
console.error('[MessageSyncService] 获取未读数失败:', error);
}
}
async syncMessages(
conversationId: string,
options: SyncOptions = {}
): Promise<SyncResult> {
const key = conversationId;
/**
* 检查是否可加载更多会话
*/
canLoadMoreConversations(): boolean {
return this.remoteConversationListSource.hasMore;
}
/**
* 重启数据源
*/
restartSources(): void {
this.remoteConversationListSource.restart();
this.localConversationListSource.restart();
this.loadingMoreConversations = false;
}
// ==================== 私有工具方法 ====================
private shouldDeferConversationRefresh(source: string, forceRefresh: boolean): boolean {
if (!this.stateManager.getActiveConversation()) return false;
if (!forceRefresh) return false;
return source === 'sse-reconnect' || source === 'prefetch' || source === 'hooks-initial-refresh';
}
private normalizeConversationId(conversationId: string | number | null | undefined): string {
return conversationId == null ? '' : String(conversationId);
}
private normalizeConversationFromFetch(conv: ConversationResponse): ConversationResponse {
const id = this.normalizeConversationId(conv.id);
return {
...conv,
id,
};
}
private updateConversationList(): void {
const currentState = this.stateManager.getState();
const list = Array.from(currentState.conversations.values()).sort((a, b) => {
const aPinned = a.is_pinned ? 1 : 0;
const bPinned = b.is_pinned ? 1 : 0;
if (aPinned !== bPinned) {
return bPinned - aPinned;
}
const aTime = new Date(a.last_message_at || a.updated_at || 0).getTime();
const bTime = new Date(b.last_message_at || b.updated_at || 0).getTime();
return bTime - aTime;
});
currentState.conversationList = list;
}
private recomputeConversationTotalUnread(): void {
const currentState = this.stateManager.getState();
currentState.totalUnreadCount = Array.from(currentState.conversations.values()).reduce(
(sum, conv) => sum + (conv.unread_count || 0),
0
);
}
private emitConversationListAndUnreadUpdates(): void {
this.persistConversationListCache();
this.stateManager.notifySubscribers({
type: 'conversations_updated',
payload: { conversations: this.stateManager.getConversations() },
timestamp: Date.now(),
});
if (this.syncingMessages.get(key)) {
console.log('[MessageSyncService] Already syncing messages for:', conversationId);
return { conversations: [], messages: [], hasMore: false };
const unreadCount = this.stateManager.getUnreadCount();
this.stateManager.notifySubscribers({
type: 'unread_count_updated',
payload: {
totalUnreadCount: unreadCount.total,
systemUnreadCount: unreadCount.system,
},
timestamp: Date.now(),
});
}
private persistConversationListCache(): void {
const currentState = this.stateManager.getState();
const list = Array.from(currentState.conversations.values());
const users = list.flatMap(conv => [
...(conv.participants || []),
...(conv.last_message?.sender ? [conv.last_message.sender] : []),
]);
const groups = list
.map(conv => conv.group)
.filter((group): group is NonNullable<ConversationResponse['group']> => Boolean(group));
saveConversationsWithRelatedCache(list, users, groups).catch(error => {
console.error('[MessageSyncService] 持久化会话列表失败:', error);
});
}
private async hydrateConversationsFromLocalSource(): Promise<boolean> {
this.localConversationListSource.restart();
const page = await this.localConversationListSource.loadNext();
if (!page.items.length) {
return false;
}
this.syncingMessages.set(key, true);
const currentState = this.stateManager.getState();
currentState.conversations.clear();
page.items.forEach(conv => {
const normalizedConv = this.normalizeConversationFromFetch(conv);
currentState.conversations.set(normalizedConv.id, normalizedConv);
});
this.updateConversationList();
this.recomputeConversationTotalUnread();
return true;
}
try {
const lastSeq = options.lastSeq ?? await messageRepository.getMaxSeq(conversationId);
const response: MessageListResponse = await messageService.getMessages(
conversationId,
lastSeq,
undefined,
50
);
private mergeMessagesById(base: MessageResponse[], incoming: MessageResponse[]): MessageResponse[] {
if (incoming.length === 0) return base;
const merged = new Map<string, MessageResponse>();
[...base, ...incoming].forEach(msg => {
merged.set(String(msg.id), msg);
});
return Array.from(merged.values()).sort((a, b) => a.seq - b.seq);
}
const messages = this.mapMessages(response.messages || []);
if (messages.length > 0) {
await messageRepository.saveMessages(messages, true);
}
private mergeOlderMessages(existing: MessageResponse[], incoming: MessageResponse[]): MessageResponse[] {
if (incoming.length === 0) return existing;
return {
conversations: [],
messages,
hasMore: messages.length >= 50,
};
} catch (error) {
console.error('[MessageSyncService] Failed to sync messages:', error);
throw error;
} finally {
this.syncingMessages.delete(key);
const incomingAsc = [...incoming].sort((a, b) => a.seq - b.seq);
if (existing.length === 0) return incomingAsc;
const existingMinSeq = existing[0]?.seq ?? Number.MAX_SAFE_INTEGER;
const incomingMaxSeq = incomingAsc[incomingAsc.length - 1]?.seq ?? Number.MIN_SAFE_INTEGER;
if (incomingMaxSeq < existingMinSeq) {
return [...incomingAsc, ...existing];
}
}
async loadHistoryMessages(
conversationId: string,
beforeSeq: number,
limit: number = 20
): Promise<Message[]> {
try {
const response: MessageListResponse = await messageService.getMessages(
conversationId,
undefined,
beforeSeq,
limit
);
const messages = this.mapMessages(response.messages || []);
if (messages.length > 0) {
await messageRepository.saveMessages(messages, true);
}
return messages;
} catch (error) {
console.error('[MessageSyncService] Failed to load history:', error);
throw error;
}
}
async loadLocalMessages(
conversationId: string,
limit: number = 50
): Promise<Message[]> {
return messageRepository.getMessagesByConversation(conversationId, limit);
}
async loadHistoryFromLocal(
conversationId: string,
beforeSeq: number,
limit: number = 20
): Promise<Message[]> {
return messageRepository.getMessagesBeforeSeq(conversationId, beforeSeq, limit);
}
private mapConversations(list: ConversationListResponse['list']): Conversation[] {
return list.map(conv => ({
id: conv.id,
type: conv.type || 'private',
isPinned: conv.is_pinned || false,
lastSeq: conv.last_seq || 0,
lastMessageAt: conv.last_message_at || new Date().toISOString(),
unreadCount: conv.unread_count || 0,
participants: (conv.participants || []).map(p => ({
id: p.id,
username: p.username,
avatar: p.avatar || undefined,
nickname: p.nickname,
})),
group: conv.group ? {
id: String(conv.group.id),
name: conv.group.name,
avatar: conv.group.avatar,
} : undefined,
createdAt: conv.created_at || new Date().toISOString(),
updatedAt: conv.updated_at || new Date().toISOString(),
}));
}
private mapMessages(messages: MessageListResponse['messages']): Message[] {
return messages.map(msg => ({
id: msg.id,
conversationId: msg.conversation_id,
senderId: msg.sender_id,
seq: msg.seq,
segments: msg.segments || [],
createdAt: msg.created_at,
status: msg.status || 'normal',
sender: msg.sender ? {
id: msg.sender.id,
username: msg.sender.username,
avatar: msg.sender.avatar || undefined,
nickname: msg.sender.nickname,
} : undefined,
}));
}
isSyncingConversations(): boolean {
return this.syncingConversations;
}
isSyncingMessages(conversationId: string): boolean {
return this.syncingMessages.get(conversationId) || false;
return this.mergeMessagesById(existing, incomingAsc);
}
}
export const messageSyncService = new MessageSyncService();

View File

@@ -1,92 +1,313 @@
/**
* 已读回执管理器
* 负责处理消息已读状态的同步
* 理消息已读状态的同步和保护
*/
import type { ConversationResponse } from '../../types/dto';
import { messageService } from '../../services/messageService';
import { messageRepository } from '../../data/repositories/MessageRepository';
import type { MessageStateManager } from './MessageStateManager';
import { markConversationAsRead, updateConversationCacheUnreadCount } from '../../services/database';
import type { ReadStateRecord, IReadReceiptManager, IMessageStateManager } from './types';
import { READ_STATE_PROTECTION_DELAY } from './constants';
export interface ReadReceiptResult {
conversationId: string;
lastReadSeq: number;
success: boolean;
}
export class ReadReceiptManager implements IReadReceiptManager {
private stateManager: IMessageStateManager;
/**
* 正在进行中的已读 API 请求集合
* 用于防止 fetchConversations 的服务器数据覆盖乐观已读状态
*/
private pendingReadMap: Map<string, ReadStateRecord> = new Map();
/**
* 全局已读状态版本号,每次标记已读时递增
*/
private readStateVersion: number = 0;
export class ReadReceiptManager {
private pendingOperations: Map<string, Promise<void>> = new Map();
private stateManager: MessageStateManager | null = null;
setStateManager(manager: MessageStateManager): void {
this.stateManager = manager;
constructor(stateManager: IMessageStateManager) {
this.stateManager = stateManager;
}
async markAsRead(
conversationId: string,
lastSeq: number
): Promise<ReadReceiptResult> {
const existing = this.pendingOperations.get(conversationId);
if (existing) {
await existing;
/**
* 标记会话已读
* 关键方法:确保已读状态在所有组件间同步
*/
async markAsRead(conversationId: string, seq: number): Promise<void> {
const normalizedId = this.normalizeConversationId(conversationId);
const conversation = this.stateManager.getConversation(normalizedId);
if (!conversation) {
console.warn('[ReadReceiptManager] 会话不存在:', normalizedId);
return;
}
if (!this.stateManager) {
throw new Error('StateManager not set');
const prevUnreadCount = conversation.unread_count || 0;
const existingRecord = this.pendingReadMap.get(normalizedId);
// 使用 seq 去重:同一会话若已上报到更大/相同 seq则跳过重复上报
if (existingRecord && seq <= existingRecord.lastReadSeq) {
return;
}
const version = this.stateManager.beginReadOperation(conversationId);
// 清除可能存在的旧定时器
if (existingRecord?.clearTimer) {
clearTimeout(existingRecord.clearTimer);
}
const promise = this.executeMarkAsRead(conversationId, lastSeq, version);
this.pendingOperations.set(conversationId, promise);
// 递增全局版本号
this.readStateVersion++;
const currentVersion = this.readStateVersion;
// 1. 标记此会话有进行中的已读请求
this.pendingReadMap.set(normalizedId, {
timestamp: Date.now(),
version: currentVersion,
lastReadSeq: seq,
});
// 2. 乐观更新本地状态
const updatedConv: ConversationResponse = {
...conversation,
unread_count: 0,
my_last_read_seq: Math.max(Number(conversation.my_last_read_seq || 0), seq),
};
this.stateManager.updateConversation(updatedConv);
const currentUnread = this.stateManager.getUnreadCount();
const newTotalUnread = Math.max(0, currentUnread.total - prevUnreadCount);
this.stateManager.setUnreadCount(newTotalUnread, currentUnread.system);
// 3. 更新本地数据库
markConversationAsRead(normalizedId).catch(console.error);
// 4. 立即通知所有订阅者
this.stateManager.notifySubscribers({
type: 'message_read',
payload: {
conversationId: normalizedId,
unreadCount: 0,
totalUnreadCount: newTotalUnread,
},
timestamp: Date.now(),
});
this.stateManager.notifySubscribers({
type: 'unread_count_updated',
payload: {
totalUnreadCount: newTotalUnread,
systemUnreadCount: currentUnread.system,
},
timestamp: Date.now(),
});
// 5. 调用 API完成后设置延迟清除保护
try {
await messageService.markAsRead(normalizedId, seq);
} catch (error) {
console.error('[ReadReceiptManager] 标记已读API失败:', error);
// 失败时回滚状态
this.stateManager.updateConversation(conversation);
this.stateManager.setUnreadCount(currentUnread.total, currentUnread.system);
this.stateManager.notifySubscribers({
type: 'message_read',
payload: {
conversationId: normalizedId,
unreadCount: prevUnreadCount,
totalUnreadCount: currentUnread.total,
},
timestamp: Date.now(),
});
// API 失败时立即清除保护
this.pendingReadMap.delete(normalizedId);
return;
}
// API 成功后,设置延迟清除保护
const clearTimer = setTimeout(() => {
const record = this.pendingReadMap.get(normalizedId);
if (record && record.version === currentVersion) {
this.pendingReadMap.delete(normalizedId);
}
}, READ_STATE_PROTECTION_DELAY);
// 更新记录保存定时器ID
this.pendingReadMap.set(normalizedId, {
timestamp: Date.now(),
version: currentVersion,
lastReadSeq: seq,
clearTimer,
});
}
/**
* 标记所有消息已读
*/
async markAllAsRead(): Promise<void> {
const conversations = this.stateManager.getState().conversations;
const prevTotalUnread = this.stateManager.getUnreadCount().total;
const currentSystem = this.stateManager.getUnreadCount().system;
// 乐观更新
conversations.forEach(conv => {
conv.unread_count = 0;
});
this.stateManager.setUnreadCount(0, currentSystem);
this.stateManager.notifySubscribers({
type: 'unread_count_updated',
payload: {
totalUnreadCount: 0,
systemUnreadCount: currentSystem,
},
timestamp: Date.now(),
});
try {
await promise;
return { conversationId, lastReadSeq: lastSeq, success: true };
// 标记系统消息已读
await messageService.markAllSystemMessagesRead();
// 标记所有会话已读
const promises: Promise<any>[] = [];
conversations.forEach(conv => {
if ((conv.unread_count || 0) > 0) {
promises.push(messageService.markAsRead(conv.id, conv.last_seq));
}
});
await Promise.all(promises);
} catch (error) {
console.error('[ReadReceiptManager] Failed to mark as read:', error);
return { conversationId, lastReadSeq: lastSeq, success: false };
} finally {
this.pendingOperations.delete(conversationId);
this.stateManager.endReadOperation(conversationId);
}
}
console.error('[ReadReceiptManager] 标记所有已读失败:', error);
// 回滚
this.stateManager.setUnreadCount(prevTotalUnread, currentSystem);
private async executeMarkAsRead(
conversationId: string,
lastSeq: number,
version: number
): Promise<void> {
await messageService.markAsRead(conversationId, lastSeq);
await messageRepository.markConversationAsRead(conversationId);
if (this.stateManager) {
this.stateManager.updateConversation(conversationId, {
unreadCount: 0,
this.stateManager.notifySubscribers({
type: 'unread_count_updated',
payload: {
totalUnreadCount: prevTotalUnread,
systemUnreadCount: currentSystem,
},
timestamp: Date.now(),
});
}
}
async markAllAsRead(): Promise<void> {
if (!this.stateManager) return;
const conversations = this.stateManager.getConversations();
const unreadConversations = conversations.filter(c => c.unreadCount > 0);
await Promise.all(
unreadConversations.map(conv =>
this.markAsRead(conv.id, conv.lastSeq)
)
);
/**
* 开始已读操作,返回版本号
*/
beginReadOperation(conversationId: string, seq: number): number {
this.readStateVersion++;
const currentVersion = this.readStateVersion;
this.pendingReadMap.set(conversationId, {
timestamp: Date.now(),
version: currentVersion,
lastReadSeq: seq,
});
return currentVersion;
}
isPending(conversationId: string): boolean {
return this.pendingOperations.has(conversationId);
/**
* 结束已读操作
*/
endReadOperation(conversationId: string, version: number, success: boolean): void {
if (!success) {
this.pendingReadMap.delete(conversationId);
return;
}
// API 成功后,设置延迟清除保护
const clearTimer = setTimeout(() => {
const record = this.pendingReadMap.get(conversationId);
if (record && record.version === version) {
this.pendingReadMap.delete(conversationId);
}
}, READ_STATE_PROTECTION_DELAY);
const existingRecord = this.pendingReadMap.get(conversationId);
if (existingRecord) {
this.pendingReadMap.set(conversationId, {
...existingRecord,
clearTimer,
});
}
}
/**
* 检查是否有进行中的已读操作
*/
isReadOperationPending(conversationId: string): boolean {
return this.pendingReadMap.has(conversationId);
}
/**
* 获取当前已读状态版本号
*/
getReadStateVersion(): number {
return this.readStateVersion;
}
/**
* 获取待处理的已读映射
*/
getPendingReadMap(): Map<string, ReadStateRecord> {
return this.pendingReadMap;
}
/**
* 规范化会话ID
*/
private normalizeConversationId(conversationId: string | number | null | undefined): string {
return conversationId == null ? '' : String(conversationId);
}
/**
* 根据已读保护状态智能合并会话
*/
applyConversationFromFetch(conv: ConversationResponse): ConversationResponse {
const id = this.normalizeConversationId(conv.id);
const readRecord = this.pendingReadMap.get(id);
let next: ConversationResponse;
if (readRecord) {
const shouldPreserveLocalRead = (conv.unread_count || 0) > 0;
next = shouldPreserveLocalRead
? { ...conv, id, unread_count: 0 }
: { ...conv, id };
} else {
next = { ...conv, id };
}
return this.normalizeConversationUnreadFromReadCursor(next);
}
/**
* 当已读游标已追上 last_seq 时,强制未读为 0
*/
normalizeConversationUnreadFromReadCursor(conv: ConversationResponse): ConversationResponse {
const lastSeq = Number(conv.last_seq || 0);
const myRead = Number(conv.my_last_read_seq ?? 0);
if (lastSeq > 0 && myRead >= lastSeq && (conv.unread_count || 0) > 0) {
return { ...conv, unread_count: 0 };
}
return conv;
}
/**
* 重置所有状态
*/
reset(): void {
this.pendingOperations.clear();
// 清除所有定时器
for (const record of this.pendingReadMap.values()) {
if (record.clearTimer) {
clearTimeout(record.clearTimer);
}
}
this.pendingReadMap.clear();
this.readStateVersion = 0;
}
}
export const readReceiptManager = new ReadReceiptManager();

View File

@@ -0,0 +1,119 @@
/**
* 用户信息缓存服务
* 管理用户信息的缓存和获取
*/
import type { MessageResponse, UserDTO } from '../../types/dto';
import { api } from '../../services/api';
import { getUserCache, saveUserCache } from '../../services/database';
import type { IUserCacheService } from './types';
export class UserCacheService implements IUserCacheService {
// 正在获取用户信息的请求映射(用于去重)
private pendingUserRequests: Map<string, Promise<UserDTO | null>> = new Map();
/**
* 获取用户信息(带缓存和去重)
*/
async getSenderInfo(userId: string): Promise<UserDTO | null> {
// 1. 先检查本地缓存
const cachedUser = await getUserCache(userId);
if (cachedUser) {
return cachedUser;
}
// 2. 检查是否已有正在进行的请求
const pendingRequest = this.pendingUserRequests.get(userId);
if (pendingRequest) {
return pendingRequest;
}
// 3. 发起新请求
const request = this.fetchUserInfo(userId);
this.pendingUserRequests.set(userId, request);
try {
const user = await request;
return user;
} finally {
// 请求完成后清理
this.pendingUserRequests.delete(userId);
}
}
/**
* 从服务器获取用户信息
*/
private async fetchUserInfo(userId: string): Promise<UserDTO | null> {
try {
const response = await api.get<UserDTO>(`/users/${userId}`);
if (response.code === 0 && response.data) {
// 缓存到本地数据库
await saveUserCache(response.data);
return response.data;
}
return null;
} catch (error) {
console.error(`[UserCacheService] 获取用户信息失败: ${userId}`, error);
return null;
}
}
/**
* 批量异步填充消息的 sender 信息
* 填充完成后调用 notifyUpdate 更新内存并通知订阅者
*/
enrichMessagesWithSenderInfo(
conversationId: string,
messages: MessageResponse[],
notifyUpdate: (conversationId: string, messages: MessageResponse[]) => void
): void {
// 收集所有需要查询的唯一 sender_id排除系统用户
const senderIds = [...new Set(
messages
.filter(m => m.sender_id && m.sender_id !== '10000' && !m.sender)
.map(m => m.sender_id)
)];
if (senderIds.length === 0) return;
// 并发获取所有 sender 信息
Promise.all(senderIds.map(id => this.getSenderInfo(id).then(user => ({ id, user }))))
.then(results => {
const senderMap = new Map<string, UserDTO>();
results.forEach(({ id, user }) => {
if (user) senderMap.set(id, user);
});
if (senderMap.size === 0) return;
let changed = false;
const updated = messages.map(m => {
if (!m.sender && m.sender_id && senderMap.has(m.sender_id)) {
changed = true;
return { ...m, sender: senderMap.get(m.sender_id) };
}
return m;
});
if (!changed) return;
// 调用回调通知更新
notifyUpdate(conversationId, updated);
})
.catch(error => {
console.error('[UserCacheService] 批量获取 sender 信息失败:', error);
});
}
/**
* 清除所有待处理的请求
* 用于测试或需要重置的场景
*/
reset(): void {
this.pendingUserRequests.clear();
}
}
// 单例导出
export const userCacheService = new UserCacheService();

View File

@@ -1,148 +1,669 @@
/**
* WS 消息处理器
* 只负责处理 WS 消息,不管理状态
* WebSocket 消息处理器
* 处理所有来自 WebSocket 的消息事件
*/
import { wsService, WSChatMessage, WSGroupChatMessage, WSReadMessage, WSGroupReadMessage, WSRecallMessage, WSGroupRecallMessage, WSGroupTypingMessage, WSGroupNoticeMessage, GroupNoticeType, WSMessage } from '../../services/wsService';
import type { Message, GroupNotice } from '../../core/entities/Message';
import type {
WSChatMessage,
WSGroupChatMessage,
WSReadMessage,
WSGroupReadMessage,
WSRecallMessage,
WSGroupRecallMessage,
WSGroupTypingMessage,
WSGroupNoticeMessage,
} from '../../services/wsService';
import { wsService, GroupNoticeType } from '../../services/wsService';
import { vibrateOnMessage } from '../../services/messageVibrationService';
import { saveMessage, updateMessageStatus } from '../../services/database';
import type { MessageResponse, ConversationResponse } from '../../types/dto';
import type {
IWSMessageHandler,
IMessageStateManager,
IMessageDeduplication,
IUserCacheService,
HandleNewMessageOptions,
} from './types';
import { MAX_FLUSH_ITERATIONS } from './constants';
export type WSEventType =
| 'chat_message'
| 'group_message'
| 'read_receipt'
| 'group_read_receipt'
| 'message_recalled'
| 'group_message_recalled'
| 'typing'
| 'group_notice';
export class WSMessageHandler implements IWSMessageHandler {
private stateManager: IMessageStateManager;
private deduplication: IMessageDeduplication;
private userCacheService: IUserCacheService;
private getCurrentUserId: () => string | null;
private markAsReadCallback: (conversationId: string, seq: number) => Promise<void>;
private fetchConversationsCallback: (forceRefresh: boolean, source: string) => Promise<void>;
private fetchUnreadCountCallback: () => Promise<void>;
private fetchMessagesCallback: (conversationId: string) => Promise<void>;
export interface WSEvent {
type: WSEventType;
payload: any;
raw: any;
}
private sseUnsubscribe: (() => void) | null = null;
private isBootstrapping: boolean = false;
private lastReconnectSyncAt: number = 0;
export type WSEventHandler = (event: WSEvent) => void;
// 初始化阶段缓冲来自 SSE 的消息事件
private bufferedChatMessages: Array<(WSChatMessage | WSGroupChatMessage) & { _isAck?: boolean }> = [];
private bufferedReadReceipts: Array<WSReadMessage | WSGroupReadMessage> = [];
export class WSMessageHandler {
private unsubscribeFns: Array<() => void> = [];
private handlers: Set<WSEventHandler> = new Set();
constructor(
stateManager: IMessageStateManager,
deduplication: IMessageDeduplication,
userCacheService: IUserCacheService,
getCurrentUserId: () => string | null,
callbacks: {
markAsRead: (conversationId: string, seq: number) => Promise<void>;
fetchConversations: (forceRefresh: boolean, source: string) => Promise<void>;
fetchUnreadCount: () => Promise<void>;
fetchMessages: (conversationId: string) => Promise<void>;
}
) {
this.stateManager = stateManager;
this.deduplication = deduplication;
this.userCacheService = userCacheService;
this.getCurrentUserId = getCurrentUserId;
this.markAsReadCallback = callbacks.markAsRead;
this.fetchConversationsCallback = callbacks.fetchConversations;
this.fetchUnreadCountCallback = callbacks.fetchUnreadCount;
this.fetchMessagesCallback = callbacks.fetchMessages;
}
/**
* 初始化 SSE 监听器
*/
connect(): void {
if (this.unsubscribeFns.length > 0) return;
if (this.sseUnsubscribe) {
this.sseUnsubscribe();
}
// 监听私聊消息
const unsubChat = wsService.on('chat', (message) => {
this.emit('chat_message', this.parseChatMessage(message), message);
wsService.on('chat', (message: WSChatMessage) => {
if (!this.stateManager.getState().isInitialized || this.isBootstrapping) {
this.bufferedChatMessages.push(message);
return;
}
this.handleNewMessage(message);
});
// 监听群聊消息
const unsubGroupMessage = wsService.on('group_message', (message) => {
this.emit('group_message', this.parseGroupMessage(message), message);
wsService.on('group_message', (message: WSGroupChatMessage) => {
if (!this.stateManager.getState().isInitialized || this.isBootstrapping) {
this.bufferedChatMessages.push(message);
return;
}
this.handleNewMessage(message);
});
// 监听私聊已读回执
const unsubRead = wsService.on('read', (message) => {
this.emit('read_receipt', message, message);
wsService.on('read', (message: WSReadMessage) => {
if (!this.stateManager.getState().isInitialized || this.isBootstrapping) {
this.bufferedReadReceipts.push(message);
return;
}
this.handleReadReceipt(message);
});
// 监听群聊已读回执
const unsubGroupRead = wsService.on('group_read', (message) => {
this.emit('group_read_receipt', message, message);
wsService.on('group_read', (message: WSGroupReadMessage) => {
if (!this.stateManager.getState().isInitialized || this.isBootstrapping) {
this.bufferedReadReceipts.push(message);
return;
}
this.handleGroupReadReceipt(message);
});
// 监听私聊消息撤回
const unsubRecall = wsService.on('recall', (message) => {
this.emit('message_recalled', message, message);
wsService.on('recall', (message: WSRecallMessage) => {
this.handleRecallMessage(message);
});
// 监听群聊消息撤回
const unsubGroupRecall = wsService.on('group_recall', (message) => {
this.emit('group_message_recalled', message, message);
wsService.on('group_recall', (message: WSGroupRecallMessage) => {
this.handleGroupRecallMessage(message);
});
// 监听群聊输入状态
const unsubGroupTyping = wsService.on('group_typing', (message) => {
this.emit('typing', message, message);
wsService.on('group_typing', (message: WSGroupTypingMessage) => {
this.handleGroupTyping(message);
});
// 监听群通知
const unsubGroupNotice = wsService.on('group_notice', (message) => {
this.emit('group_notice', this.parseGroupNotice(message), message);
wsService.on('group_notice', (message: WSGroupNoticeMessage) => {
this.handleGroupNotice(message);
});
this.unsubscribeFns = [
unsubChat,
unsubGroupMessage,
unsubRead,
unsubGroupRead,
unsubRecall,
unsubGroupRecall,
unsubGroupTyping,
unsubGroupNotice,
];
}
// 监听连接状态
wsService.onConnect(() => {
this.stateManager.setSSEConnected(true);
this.stateManager.notifySubscribers({
type: 'connection_changed',
payload: { connected: true },
timestamp: Date.now(),
});
disconnect(): void {
this.unsubscribeFns.forEach((fn) => fn());
this.unsubscribeFns = [];
this.handlers.clear();
}
subscribe(handler: WSEventHandler): () => void {
this.handlers.add(handler);
return () => this.handlers.delete(handler);
}
private emit(type: WSEventType, payload: any, raw: any): void {
const event: WSEvent = { type, payload, raw };
this.handlers.forEach(handler => {
try {
handler(event);
} catch (error) {
console.error('[WSMessageHandler] Handler error:', error);
// 冷启动/重连兜底
const now = Date.now();
if (now - this.lastReconnectSyncAt > 1500) {
this.lastReconnectSyncAt = now;
const activeConversation = this.stateManager.getActiveConversation();
if (!activeConversation) {
this.fetchConversationsCallback(true, 'sse-reconnect').catch(error => {
console.error('[WSMessageHandler] 连接后同步会话失败:', error);
});
} else {
this.fetchUnreadCountCallback().catch(error => {
console.error('[WSMessageHandler] 连接后同步未读数失败:', error);
});
}
if (activeConversation) {
this.fetchMessagesCallback(activeConversation).catch(error => {
console.error('[WSMessageHandler] 连接后同步当前会话消息失败:', error);
});
}
}
});
wsService.onDisconnect(() => {
this.stateManager.setSSEConnected(false);
this.stateManager.notifySubscribers({
type: 'connection_changed',
payload: { connected: false },
timestamp: Date.now(),
});
});
}
private parseChatMessage(data: WSChatMessage): Message {
return {
id: data.id || '',
conversationId: data.conversation_id || '',
senderId: data.sender_id || '',
seq: data.seq || 0,
segments: data.segments || [],
createdAt: data.created_at || new Date().toISOString(),
status: 'normal',
};
}
private parseGroupMessage(data: WSGroupChatMessage): Message {
return {
id: data.id || '',
conversationId: data.conversation_id || '',
senderId: data.sender_id || '',
seq: data.seq || 0,
segments: data.segments || [],
createdAt: data.created_at || new Date().toISOString(),
status: 'normal',
};
}
private parseGroupNotice(data: WSGroupNoticeMessage): GroupNotice {
return {
type: data.notice_type || 'member_join',
groupId: String(data.group_id || ''),
data: data.data || {},
timestamp: data.timestamp || Date.now(),
messageId: data.message_id,
seq: data.seq,
};
/**
* 断开 SSE 连接
*/
disconnect(): void {
if (this.sseUnsubscribe) {
this.sseUnsubscribe();
this.sseUnsubscribe = null;
}
}
/**
* 检查是否已连接
*/
isConnected(): boolean {
return wsService.isConnected();
return this.stateManager.isConnected();
}
/**
* 设置启动状态
*/
setBootstrapping(value: boolean): void {
this.isBootstrapping = value;
}
/**
* 初始化完成后,处理缓冲的 SSE 事件
*/
async flushBufferedSSEEvents(): Promise<void> {
for (let i = 0; i < MAX_FLUSH_ITERATIONS; i++) {
const hasBuffered = this.bufferedChatMessages.length > 0 || this.bufferedReadReceipts.length > 0;
if (!hasBuffered) return;
const chatEvents = this.bufferedChatMessages;
const readEvents = this.bufferedReadReceipts;
this.bufferedChatMessages = [];
this.bufferedReadReceipts = [];
// 处理消息
for (const m of chatEvents) {
await this.handleNewMessage(m, {
suppressUnreadIncrement: true,
suppressVibration: true,
suppressConversationUpdates: true,
suppressAutoMarkAsRead: true,
});
}
// 处理已读回执
for (const r of readEvents) {
if (r.type === 'read') {
this.handleReadReceipt(r);
} else {
this.handleGroupReadReceipt(r as WSGroupReadMessage);
}
}
// 使用服务器权威刷新
await this.fetchConversationsCallback(true, 'sse-buffer-flush');
await this.fetchUnreadCountCallback();
}
}
/**
* 处理新消息SSE推送
*/
async handleNewMessage(
message: (WSChatMessage | WSGroupChatMessage) & { _isAck?: boolean },
options?: HandleNewMessageOptions
): Promise<void> {
const { conversation_id, id, sender_id, seq, segments, created_at, _isAck } = message;
const normalizedConversationId = this.normalizeConversationId(conversation_id);
const currentUserId = this.getCurrentUserId();
const suppressUnreadIncrement = options?.suppressUnreadIncrement ?? false;
const suppressVibration = options?.suppressVibration ?? false;
const suppressConversationUpdates = options?.suppressConversationUpdates ?? false;
const suppressAutoMarkAsRead = options?.suppressAutoMarkAsRead ?? false;
// 消息去重检查
if (this.deduplication.isMessageProcessed(id)) {
return;
}
this.deduplication.markMessageAsProcessed(id);
// 对于群聊消息,获取发送者信息
if (message.type === 'group_message' && sender_id && sender_id !== currentUserId && sender_id !== '10000') {
this.userCacheService.getSenderInfo(sender_id).then(user => {
if (user) {
const currentMessages = this.stateManager.getMessages(normalizedConversationId);
if (currentMessages) {
const updatedMessages = currentMessages.map(m =>
m.id === id ? { ...m, sender: user } : m
);
this.stateManager.setMessages(normalizedConversationId, updatedMessages);
this.stateManager.notifySubscribers({
type: 'messages_updated',
payload: {
conversationId: normalizedConversationId,
messages: [...updatedMessages],
},
timestamp: Date.now(),
});
}
}
}).catch(error => {
console.error('[WSMessageHandler] 获取发送者信息失败:', { userId: sender_id, error });
});
}
// 构造消息对象
const newMessage: MessageResponse = {
id,
conversation_id: normalizedConversationId,
sender_id,
seq,
segments: segments || [],
created_at,
status: 'normal',
};
// 立即更新内存中的消息列表
const existingMessages = this.stateManager.getMessages(normalizedConversationId);
const messageExists = existingMessages.some(m => String(m.id) === String(id));
if (!messageExists) {
const updatedMessages = this.mergeMessagesById(existingMessages, [newMessage]);
this.stateManager.setMessages(normalizedConversationId, updatedMessages);
this.stateManager.notifySubscribers({
type: 'messages_updated',
payload: {
conversationId: normalizedConversationId,
messages: updatedMessages,
newMessage,
},
timestamp: Date.now(),
});
}
// 更新会话信息
if (!suppressConversationUpdates) {
this.updateConversationWithNewMessage(normalizedConversationId, newMessage, created_at);
}
// 更新未读数
const isCurrentUserMessage = sender_id === currentUserId;
const isActiveConversation = normalizedConversationId === this.stateManager.getActiveConversation();
const shouldIncrementUnread =
!suppressUnreadIncrement &&
!isCurrentUserMessage &&
!isActiveConversation &&
!!currentUserId &&
!_isAck;
if (shouldIncrementUnread) {
if (!suppressVibration) {
const vibrationType = message.type === 'group_message' ? 'group_message' : 'chat';
vibrateOnMessage(vibrationType).catch(() => {});
}
this.incrementUnreadCount(normalizedConversationId);
}
// 如果是当前活动会话,自动标记已读
if (isActiveConversation && !isCurrentUserMessage && !suppressAutoMarkAsRead) {
this.markAsReadCallback(normalizedConversationId, seq).catch(() => {});
}
// 异步保存到本地数据库
const textContent = segments?.filter((s: any) => s.type === 'text').map((s: any) => s.data?.text || '').join('') || '';
saveMessage({
id,
conversationId: normalizedConversationId,
senderId: sender_id || '',
content: textContent,
type: segments?.find((s: any) => s.type === 'image') ? 'image' : 'text',
isRead: isActiveConversation,
createdAt: new Date(created_at).toISOString(),
seq,
status: 'normal',
segments,
}).catch(error => {
console.error('[WSMessageHandler] 保存消息到本地失败:', error);
});
// 通知收到新消息
this.stateManager.notifySubscribers({
type: 'message_received',
payload: {
conversationId: conversation_id,
message: newMessage,
isCurrentUser: isCurrentUserMessage,
},
timestamp: Date.now(),
});
}
/**
* 处理私聊已读回执
*/
handleReadReceipt(message: WSReadMessage): void {
// 可以在这里处理对方已读的状态更新
}
/**
* 处理群聊已读回执
*/
handleGroupReadReceipt(message: WSGroupReadMessage): void {
// 群聊已读回执处理
}
/**
* 处理私聊消息撤回
*/
handleRecallMessage(message: WSRecallMessage): void {
const { conversation_id, message_id } = message;
const normalizedConversationId = this.normalizeConversationId(conversation_id);
this.markMessageAsRecalled(normalizedConversationId, message_id);
this.syncConversationLastMessageOnRecall(normalizedConversationId, message_id);
this.stateManager.notifySubscribers({
type: 'message_recalled',
payload: { conversationId: normalizedConversationId, messageId: message_id },
timestamp: Date.now(),
});
updateMessageStatus(message_id, 'recalled', true).catch(error => {
console.error('[WSMessageHandler] 更新本地消息撤回状态失败:', error);
});
}
/**
* 处理群聊消息撤回
*/
handleGroupRecallMessage(message: WSGroupRecallMessage): void {
const { conversation_id, message_id } = message;
const normalizedConversationId = this.normalizeConversationId(conversation_id);
this.markMessageAsRecalled(normalizedConversationId, message_id);
this.syncConversationLastMessageOnRecall(normalizedConversationId, message_id);
this.stateManager.notifySubscribers({
type: 'message_recalled',
payload: { conversationId: normalizedConversationId, messageId: message_id, isGroup: true },
timestamp: Date.now(),
});
updateMessageStatus(message_id, 'recalled', true).catch(error => {
console.error('[WSMessageHandler] 更新本地消息撤回状态失败:', error);
});
}
/**
* 处理群聊输入状态
*/
handleGroupTyping(message: WSGroupTypingMessage): void {
const { group_id, user_id, is_typing } = message;
const groupIdStr = String(group_id);
const currentTypingUsers = this.stateManager.getTypingUsers(groupIdStr);
let updatedTypingUsers: string[];
if (is_typing) {
if (!currentTypingUsers.includes(user_id)) {
updatedTypingUsers = [...currentTypingUsers, user_id];
} else {
updatedTypingUsers = currentTypingUsers;
}
} else {
updatedTypingUsers = currentTypingUsers.filter(id => id !== user_id);
}
if (updatedTypingUsers.length !== currentTypingUsers.length) {
this.stateManager.setTypingUsers(groupIdStr, updatedTypingUsers);
this.stateManager.notifySubscribers({
type: 'typing_status',
payload: { groupId: groupIdStr, typingUsers: updatedTypingUsers },
timestamp: Date.now(),
});
}
}
/**
* 处理群通知
*/
handleGroupNotice(message: WSGroupNoticeMessage): void {
const { notice_type, group_id, data, timestamp, message_id, seq } = message;
const groupIdStr = String(group_id);
// 处理禁言/解除禁言通知
if (notice_type === 'muted' || notice_type === 'unmuted') {
const mutedUserId = data?.user_id;
const currentUserId = this.getCurrentUserId();
if (mutedUserId && mutedUserId === currentUserId) {
const isMuted = notice_type === 'muted';
this.stateManager.setMutedStatus(groupIdStr, isMuted);
}
}
// 补一条系统消息到当前会话消息流
if (message_id && typeof seq === 'number' && seq > 0) {
const conversationId = this.findConversationIdByGroupId(groupIdStr);
if (conversationId) {
const existingMessages = this.stateManager.getMessages(conversationId);
const exists = existingMessages.some(m => String(m.id) === String(message_id));
if (!exists) {
const noticeText = this.buildGroupNoticeText(notice_type, data);
const systemNoticeMessage: MessageResponse = {
id: String(message_id),
conversation_id: conversationId,
sender_id: '10000',
seq,
segments: noticeText
? [{ type: 'text', data: { text: noticeText } as any }]
: [],
status: 'normal',
category: 'notification',
created_at: new Date(timestamp || Date.now()).toISOString(),
};
const updatedMessages = [...existingMessages, systemNoticeMessage].sort((a, b) => a.seq - b.seq);
this.stateManager.setMessages(conversationId, updatedMessages);
this.stateManager.notifySubscribers({
type: 'messages_updated',
payload: {
conversationId,
messages: updatedMessages,
newMessage: systemNoticeMessage,
source: 'group_notice',
},
timestamp: Date.now(),
});
}
}
}
// 通知订阅者群通知
this.stateManager.notifySubscribers({
type: 'group_notice',
payload: {
groupId: groupIdStr,
noticeType: notice_type,
data,
timestamp,
messageId: message_id,
seq,
},
timestamp: Date.now(),
});
}
// ==================== 私有工具方法 ====================
private normalizeConversationId(conversationId: string | number | null | undefined): string {
return conversationId == null ? '' : String(conversationId);
}
private mergeMessagesById(base: MessageResponse[], incoming: MessageResponse[]): MessageResponse[] {
if (incoming.length === 0) return base;
const merged = new Map<string, MessageResponse>();
[...base, ...incoming].forEach(msg => {
merged.set(String(msg.id), msg);
});
return Array.from(merged.values()).sort((a, b) => a.seq - b.seq);
}
private updateConversationWithNewMessage(
conversationId: string,
message: MessageResponse,
createdAt: string
): void {
const conversation = this.stateManager.getConversation(conversationId);
if (conversation) {
const updatedConv: ConversationResponse = {
...conversation,
last_seq: message.seq,
last_message: message,
last_message_at: createdAt,
updated_at: createdAt,
};
this.stateManager.updateConversation(updatedConv);
}
this.stateManager.notifySubscribers({
type: 'conversations_updated',
payload: { conversations: this.stateManager.getConversations() },
timestamp: Date.now(),
});
}
private incrementUnreadCount(conversationId: string): void {
const conversation = this.stateManager.getConversation(conversationId);
if (conversation) {
const prevUnreadCount = conversation.unread_count || 0;
const updatedConv: ConversationResponse = {
...conversation,
unread_count: prevUnreadCount + 1,
};
this.stateManager.updateConversation(updatedConv);
const currentUnread = this.stateManager.getUnreadCount();
this.stateManager.setUnreadCount(currentUnread.total + 1, currentUnread.system);
this.stateManager.notifySubscribers({
type: 'unread_count_updated',
payload: {
totalUnreadCount: this.stateManager.getUnreadCount().total,
systemUnreadCount: this.stateManager.getUnreadCount().system,
},
timestamp: Date.now(),
});
this.stateManager.notifySubscribers({
type: 'conversations_updated',
payload: { conversations: this.stateManager.getConversations() },
timestamp: Date.now(),
});
}
}
private markMessageAsRecalled(conversationId: string, messageId: string): void {
const messages = this.stateManager.getMessages(conversationId);
if (!messages) return;
let changed = false;
const updatedMessages: MessageResponse[] = messages.map((m): MessageResponse => {
if (String(m.id) !== String(messageId) || m.status === 'recalled') {
return m;
}
changed = true;
return {
...m,
status: 'recalled' as MessageResponse['status'],
segments: [],
};
});
if (!changed) return;
this.stateManager.setMessages(conversationId, updatedMessages);
this.stateManager.notifySubscribers({
type: 'messages_updated',
payload: { conversationId, messages: updatedMessages },
timestamp: Date.now(),
});
}
private syncConversationLastMessageOnRecall(conversationId: string, messageId: string): void {
const conversation = this.stateManager.getConversation(conversationId);
if (!conversation?.last_message) return;
if (String(conversation.last_message.id) !== String(messageId)) return;
if (conversation.last_message.status === 'recalled') return;
const updatedConversation: ConversationResponse = {
...conversation,
last_message: {
...conversation.last_message,
status: 'recalled',
},
};
this.stateManager.updateConversation(updatedConversation);
}
private findConversationIdByGroupId(groupId: string): string | null {
const conversations = this.stateManager.getConversations();
for (const conv of conversations) {
if (String(conv.group?.id || '') === groupId) {
return conv.id;
}
}
return null;
}
private buildGroupNoticeText(noticeType: GroupNoticeType, data: any): string {
const username = data?.username || '用户';
switch (noticeType) {
case 'member_join':
return `"${username}" 加入了群聊`;
case 'member_leave':
return `"${username}" 退出了群聊`;
case 'member_removed':
return `"${username}" 被移出群聊`;
case 'muted':
return `"${username}" 已被管理员禁言`;
case 'unmuted':
return `"${username}" 已被管理员解除禁言`;
default:
return '';
}
}
}
export const wsMessageHandler = new WSMessageHandler();

View File

@@ -0,0 +1,40 @@
/**
* 消息管理常量定义模块
* 包含所有与消息管理相关的常量
*/
/**
* 已读状态保护延迟时间(毫秒)
* API 完成后,继续保护一段时间,防止 fetchConversations 的旧数据覆盖
*/
export const READ_STATE_PROTECTION_DELAY = 5000; // 5秒
/**
* 已处理消息ID过期时间毫秒
* 用于消息去重机制,防止内存泄漏
*/
export const PROCESSED_MESSAGE_ID_EXPIRY = 5 * 60 * 1000; // 5分钟
/**
* 已处理消息ID集合最大大小
* 超过此大小时触发清理
*/
export const PROCESSED_MESSAGE_ID_MAX_SIZE = 1000;
/**
* 缓冲SSE事件刷新最大迭代次数
* 防止在线流持续不断导致长时间阻塞初始化
*/
export const MAX_FLUSH_ITERATIONS = 3;
/**
* 会话列表刷新延迟检查间隔(毫秒)
* 用于防抖处理
*/
export const CONVERSATION_REFRESH_THROTTLE = 1000;
/**
* 重连同步最小间隔(毫秒)
* 防止频繁重连导致的重复同步
*/
export const MIN_RECONNECT_SYNC_INTERVAL = 1500;

View File

@@ -1,19 +1,66 @@
/**
* 消息模块统一导出
*
* 提供向后兼容的 API 导出
*/
// 导出主要的管理器
// ==================== 主管理器 ====================
export { messageManager, MessageManager } from './MessageManager';
// 导出状态管理器
export { messageStateManager, MessageStateManager } from './MessageStateManager';
export type { MessageState, MessageEvent, MessageSubscriber, MessageEventType } from './MessageStateManager';
// ==================== 类型导出 ====================
// 从 types.ts 导出所有类型
export type {
MessageEventType,
MessageEvent,
MessageSubscriber,
MessageManagerState,
ReadStateRecord,
MessageManagerConversationListDeps,
HandleNewMessageOptions,
IMessageStateManager,
IMessageDeduplication,
IUserCacheService,
IReadReceiptManager,
IConversationOperations,
IMessageSendService,
IMessageSyncService,
IWSMessageHandler,
} from './types';
// 导出同步服务
export { messageSyncService, MessageSyncService } from './MessageSyncService';
// ==================== 常量导出 ====================
export {
READ_STATE_PROTECTION_DELAY,
PROCESSED_MESSAGE_ID_EXPIRY,
PROCESSED_MESSAGE_ID_MAX_SIZE,
MAX_FLUSH_ITERATIONS,
CONVERSATION_REFRESH_THROTTLE,
MIN_RECONNECT_SYNC_INTERVAL,
} from './constants';
// 导出WS处理器
export { wsMessageHandler, WSMessageHandler } from './WSMessageHandler';
// ==================== 子模块导出(可选使用)====================
// 状态管理器
export { MessageStateManager, messageStateManager } from './MessageStateManager';
// 导出已读管理器
export { readReceiptManager, ReadReceiptManager } from './ReadReceiptManager';
// 消息去重服务
export { MessageDeduplication, messageDeduplication } from './MessageDeduplication';
// 用户缓存服务
export { UserCacheService, userCacheService } from './UserCacheService';
// 已读回执管理器
export { ReadReceiptManager } from './ReadReceiptManager';
// 会话操作服务
export { ConversationOperations } from './ConversationOperations';
// 消息发送服务
export { MessageSendService } from './MessageSendService';
// 消息同步服务
export { MessageSyncService } from './MessageSyncService';
// WebSocket 消息处理器
export { WSMessageHandler } from './WSMessageHandler';
// ==================== 默认导出 ====================
export { default } from './MessageManager';

259
src/stores/message/types.ts Normal file
View File

@@ -0,0 +1,259 @@
/**
* 消息管理类型定义模块
* 包含所有与消息管理相关的类型、接口和类型别名
*/
import type { ConversationResponse, MessageResponse, UserDTO } from '../../types/dto';
import type { IConversationListPagedSource } from '../conversationListSources';
// ==================== 事件类型 ====================
/**
* 消息事件类型枚举
* 定义了所有可能的消息事件类型
*/
export type MessageEventType =
| 'conversations_updated'
| 'conversations_loading'
| 'messages_updated'
| 'unread_count_updated'
| 'connection_changed'
| 'message_sent'
| 'message_read'
| 'message_received'
| 'message_recalled'
| 'typing_status'
| 'group_notice'
| 'error';
/**
* 消息事件接口
* 所有订阅者接收的事件对象结构
*/
export interface MessageEvent {
type: MessageEventType;
payload: any;
timestamp: number;
}
/**
* 消息订阅者类型
*/
export type MessageSubscriber = (event: MessageEvent) => void;
// ==================== 状态接口 ====================
/**
* 消息管理器内部状态接口
* 定义了所有需要管理的状态字段
*/
export interface MessageManagerState {
// 会话相关
conversations: Map<string, ConversationResponse>;
conversationList: ConversationResponse[];
// 消息相关 - 按会话ID存储
messagesMap: Map<string, MessageResponse[]>;
// 未读数
totalUnreadCount: number;
systemUnreadCount: number;
// 连接状态
isWSConnected: boolean;
// 当前活动会话ID用户正在查看的会话
currentConversationId: string | null;
// 加载状态
isLoadingConversations: boolean;
loadingMessagesSet: Set<string>; // 正在加载消息的会话ID集合
// 初始化状态
isInitialized: boolean;
// 订阅者
subscribers: Set<MessageSubscriber>;
// 输入状态 - 按群组ID存储正在输入的用户ID列表
typingUsersMap: Map<string, string[]>;
// 当前用户的禁言状态 - 按群组ID存储
mutedStatusMap: Map<string, boolean>;
}
// ==================== 已读状态相关 ====================
/**
* 已读状态记录接口
* 用于跟踪已读操作的状态和保护机制
*/
export interface ReadStateRecord {
/** 标记已读的时间戳 */
timestamp: number;
/** 状态版本号,用于防止旧数据覆盖新数据 */
version: number;
/** 已上报的已读序号(用于去重) */
lastReadSeq: number;
/** 清除保护的定时器ID */
clearTimer?: NodeJS.Timeout;
}
// ==================== 依赖注入接口 ====================
/**
* 可注入会话列表源,便于测试或替换实现
*/
export interface MessageManagerConversationListDeps {
remoteConversationListSource?: IConversationListPagedSource;
localConversationListSource?: IConversationListPagedSource;
/**
* 未传 remoteConversationListSource 时:默认 `cursor`/conversations/cursor
* `offset` 为常规页码(/conversations?page=&page_size=)。
*/
remoteListKind?: 'cursor' | 'offset';
/** 与 remoteListKind 搭配,默认与 CONVERSATION_LIST_PAGE_SIZE 一致 */
remoteListPageSize?: number;
}
// ==================== 消息处理选项 ====================
/**
* 处理新消息的选项
*/
export interface HandleNewMessageOptions {
/** 是否抑制未读数增加 */
suppressUnreadIncrement?: boolean;
/** 是否抑制震动 */
suppressVibration?: boolean;
/** 是否抑制会话更新 */
suppressConversationUpdates?: boolean;
/** 是否抑制自动标记已读 */
suppressAutoMarkAsRead?: boolean;
}
// ==================== 服务接口 ====================
/**
* 状态管理器接口
* 定义状态管理器需要实现的方法
*/
export interface IMessageStateManager {
// 获取状态
getState(): MessageManagerState;
getConversations(): ConversationResponse[];
getConversation(conversationId: string): ConversationResponse | null;
getMessages(conversationId: string): MessageResponse[];
getUnreadCount(): { total: number; system: number };
isConnected(): boolean;
isLoading(): boolean;
isLoadingMessages(conversationId: string): boolean;
getTypingUsers(groupId: string): string[];
isMuted(groupId: string): boolean;
getActiveConversation(): string | null;
// 设置状态
setConversations(conversations: Map<string, ConversationResponse>): void;
updateConversation(conversation: ConversationResponse): void;
removeConversation(conversationId: string): void;
setMessages(conversationId: string, messages: MessageResponse[]): void;
addMessage(conversationId: string, message: MessageResponse): void;
updateMessage(conversationId: string, messageId: string, updates: Partial<MessageResponse>): void;
setUnreadCount(total: number, system: number): void;
setSSEConnected(connected: boolean): void;
setCurrentConversation(conversationId: string | null): void;
setLoading(loading: boolean): void;
setLoadingMessages(conversationId: string, loading: boolean): void;
setTypingUsers(groupId: string, users: string[]): void;
setMutedStatus(groupId: string, isMuted: boolean): void;
setInitialized(initialized: boolean): void;
// 订阅机制
subscribe(subscriber: MessageSubscriber): () => void;
notifySubscribers(event: MessageEvent): void;
// 重置
reset(): void;
}
/**
* 消息去重服务接口
*/
export interface IMessageDeduplication {
isMessageProcessed(messageId: string): boolean;
markMessageAsProcessed(messageId: string): void;
cleanupProcessedMessageIds(): void;
}
/**
* 用户缓存服务接口
*/
export interface IUserCacheService {
getSenderInfo(userId: string): Promise<UserDTO | null>;
enrichMessagesWithSenderInfo(
conversationId: string,
messages: MessageResponse[],
notifyUpdate: (conversationId: string, messages: MessageResponse[]) => void
): void;
}
/**
* 已读回执管理器接口
*/
export interface IReadReceiptManager {
markAsRead(conversationId: string, seq: number): Promise<void>;
markAllAsRead(): Promise<void>;
beginReadOperation(conversationId: string, seq: number): number;
endReadOperation(conversationId: string, version: number, success: boolean): void;
isReadOperationPending(conversationId: string): boolean;
getReadStateVersion(): number;
getPendingReadMap(): Map<string, ReadStateRecord>;
}
/**
* 会话操作接口
*/
export interface IConversationOperations {
createConversation(userId: string): Promise<ConversationResponse | null>;
updateConversation(conversationId: string, updates: Partial<ConversationResponse>): void;
removeConversation(conversationId: string): void;
clearConversations(): void;
updateUnreadCount(conversationId: string, count: number): void;
incrementUnreadCount(conversationId: string): void;
setSystemUnreadCount(count: number): void;
incrementSystemUnreadCount(): void;
decrementSystemUnreadCount(count?: number): void;
}
/**
* 消息发送服务接口
*/
export interface IMessageSendService {
sendMessage(
conversationId: string,
segments: any[],
options?: { replyToId?: string }
): Promise<MessageResponse | null>;
}
/**
* 消息同步服务接口
*/
export interface IMessageSyncService {
fetchConversations(forceRefresh?: boolean, source?: string): Promise<void>;
loadMoreConversations(): Promise<void>;
fetchConversationDetail(conversationId: string): Promise<ConversationResponse | null>;
fetchMessages(conversationId: string, afterSeq?: number): Promise<void>;
loadMoreMessages(conversationId: string, beforeSeq: number, limit?: number): Promise<MessageResponse[]>;
fetchUnreadCount(): Promise<void>;
canLoadMoreConversations(): boolean;
}
/**
* WebSocket 消息处理器接口
*/
export interface IWSMessageHandler {
connect(): void;
disconnect(): void;
isConnected(): boolean;
}

File diff suppressed because it is too large Load Diff