/** * 消息发送服务 * 处理消息发送相关逻辑 * 实现乐观更新:先显示消息,再发送到服务器 */ import type { MessageResponse, MessageSegment, ConversationResponse } from '../../../types/dto'; import { messageService } from '@/services/message'; import { messageRepository } from '@/database'; import type { IMessageSendService } from '../types'; import { useMessageStore } from '../store'; import { createErrorHandler } from '@/services/core'; const errorHandler = createErrorHandler('MessageSendService'); // 临时ID前缀,用于乐观更新 const TEMP_ID_PREFIX = 'temp_'; // 全局自增计数器:与 Date.now() + Math.random() 组合,彻底消除同一毫秒内 // 连续发送导致的临时 ID 碰撞风险(addOrReplaceMessage 按 id 幂等,碰撞会吞掉消息)。 let tempIdCounter = 0; export class MessageSendService implements IMessageSendService { private getCurrentUserId: () => string | null; constructor(getCurrentUserId: () => string | null) { this.getCurrentUserId = getCurrentUserId; } /** * 生成临时消息ID * 组合 timestamp + 自增计数器 + 随机串,保证全局唯一。 */ private generateTempId(): string { return `${TEMP_ID_PREFIX}${Date.now()}_${++tempIdCounter}_${Math.random().toString(36).slice(2, 11)}`; } /** * 发送消息(乐观更新模式) * 1. 立即创建临时消息并显示 * 2. 后台发送到服务器 * 3. 成功:更新为正式消息 * 4. 失败:标记为发送失败 * * @param detailType 会话类型,'private' 优先走 WebSocket(失败降级 HTTP), * 'group' 直接走 HTTP。默认 'private'。 */ async sendMessage( conversationId: string, segments: MessageSegment[], options?: { replyToId?: string; detailType?: 'private' | 'group' } ): Promise { const store = useMessageStore.getState(); const currentUserId = this.getCurrentUserId(); const detailType = options?.detailType ?? 'private'; // 1. 创建临时消息(乐观显示) const tempId = this.generateTempId(); const tempMessage: MessageResponse = { id: tempId, conversation_id: conversationId, sender_id: currentUserId || '', seq: 0, // 临时消息seq为0 segments, reply_to_id: options?.replyToId, created_at: new Date().toISOString(), status: 'pending', // 标记为发送中 }; // 2. 立即添加到store,用户可以马上看到 store.addOrReplaceMessage(conversationId, tempMessage); // Outbox:临时消息落库(status='pending'),保证 App 重启后能恢复未完成的发送 this.persistMessage(tempMessage); // 3. 更新会话最后消息(乐观更新) this.updateConversationWithNewMessage(conversationId, tempMessage); try { // 4. 调用API发送 const response = await messageService.sendMessage(conversationId, { segments, reply_to_id: options?.replyToId, }, detailType); if (response) { // 5. 发送成功:用服务器返回的数据更新临时消息 const successMessage: MessageResponse = { id: response.id, conversation_id: conversationId, sender_id: currentUserId || '', seq: response.seq, segments, reply_to_id: options?.replyToId, created_at: new Date().toISOString(), status: 'normal', }; // 删除临时消息,添加正式消息 store.removeMessage(conversationId, tempId); store.addOrReplaceMessage(conversationId, successMessage); // 更新会话最后消息 this.updateConversationWithNewMessage(conversationId, successMessage); // Outbox:成功后持久化正式消息(INSERT OR REPLACE 按 id 覆盖), // 并删除 tempId 的 DB 记录,避免重启后重复恢复 this.persistMessage(successMessage); messageRepository.delete(tempId).catch((e: unknown) => errorHandler(e)); return successMessage; } // 如果服务器没有返回有效响应,标记为失败 const failedMessage: MessageResponse = { ...tempMessage, status: 'failed', }; // 注意:addOrReplaceMessage 按 id 幂等(已存在则不更新),改用 patchMessages 更新 status const patches = new Map>(); patches.set(String(failedMessage.id), { status: 'failed' }); store.patchMessages(conversationId, patches); // Outbox:失败消息落库 this.persistMessage(failedMessage); return null; } catch (error) { errorHandler(error); // 6. 发送失败:更新消息状态为失败 const failedMessage: MessageResponse = { ...tempMessage, status: 'failed', }; // addOrReplaceMessage 按 id 幂等不更新,改用 patchMessages 更新 status const patches = new Map>(); patches.set(String(failedMessage.id), { status: 'failed' }); store.patchMessages(conversationId, patches); // Outbox:失败消息落库 this.persistMessage(failedMessage); return null; } } /** * 重试发送失败的消息 * * 实现要点: * 1. 先移除旧的失败消息(临时 ID),避免与即将插入的新 pending 消息并存出现两条。 * 2. 复用原 segments 与 replyToId 调用 sendMessage,由它重新走完整乐观更新流程 * (插入新 pending → 发送 → 成功替换为正式消息 / 失败再次标记 failed)。 * 3. detailType 无法从失败消息反推,默认 private; * 群聊失败消息重试走 HTTP(sendMessage 的 group 分支)也能成功。 */ async retrySendMessage( conversationId: string, failedMessage: MessageResponse ): Promise { if (failedMessage.status !== 'failed') { console.warn('[MessageSendService] 只能重试发送失败的消息'); return null; } const store = useMessageStore.getState(); // 移除旧的失败消息,避免 sendMessage 新插入的 pending 消息与之并存 store.removeMessage(conversationId, String(failedMessage.id)); // Outbox:同步删除 DB 中旧的 failed 记录,避免重启后 getPendingOrFailed 重复恢复 // (sendMessage 会用新 tempId 重新落库 pending,重试成功后正常落库 normal 并删新 tempId) messageRepository.delete(String(failedMessage.id)) .catch((e: unknown) => errorHandler(e)); // 重新发送:sendMessage 会重新插入 pending 临时消息并完成后续流程 return this.sendMessage(conversationId, failedMessage.segments, { replyToId: failedMessage.reply_to_id, }); } /** * 持久化消息到本地数据库(Outbox 机制)。 * 把 MessageResponse(DTO)映射为 CachedMessage(DB 行)并落库(INSERT OR REPLACE 按 id 幂等)。 * 用于 pending/failed/normal 三态消息的持久化,保证 App 重启后发送状态可恢复。 * 落库失败不阻断 UI(fire-and-forget + 日志),最坏情况是重启后丢失该条未完成发送。 */ private persistMessage(message: MessageResponse): void { const segments = message.segments || []; const textContent = segments .filter((s: any) => s.type === 'text') .map((s: any) => s.data?.text || '') .join('') || ''; messageRepository.saveMessage({ id: String(message.id), conversationId: message.conversation_id, senderId: message.sender_id, content: textContent, type: segments.find((s: any) => s.type === 'image') ? 'image' : 'text', isRead: true, createdAt: message.created_at, seq: message.seq, status: message.status, segments, }).catch((e: unknown) => errorHandler(e)); } /** * 更新会话的最后消息信息 */ private updateConversationWithNewMessage( conversationId: string, message: MessageResponse ): void { const store = useMessageStore.getState(); const conversation = store.getConversation(conversationId); const createdAt = message.created_at || new Date().toISOString(); if (conversation) { // 更新现有会话 const updatedConv: ConversationResponse = { ...conversation, last_seq: message.seq > 0 ? message.seq : conversation.last_seq, last_message: message, last_message_at: createdAt, updated_at: createdAt, }; store.updateConversation(updatedConv); } } }