frontend/src/core/usecases/ProcessMessageUseCase.ts

/**
 * ProcessMessageUseCase - 处理消息用例
 * 处理消息的业务逻辑，协调 Repository 和 SSEClient
 * 纯业务逻辑，无UI依赖
 */

import { messageRepository } from '../../data/repositories/MessageRepository';
import { sseClient, SSEEventType } from '../../data/datasources/SSEClient';
import {
  Message,
  Conversation,
  GroupNotice,
  createMessage,
  isMessageFromCurrentUser,
  shouldIncrementUnread,
  buildGroupNoticeText,
} from '../entities/Message';
import { messageService } from '../../services/messageService';
import { api } from '../../services/api';

// 事件类型定义
export type MessageUseCaseEventType =
  | 'message_received'
  | 'message_recalled'
  | 'typing_status'
  | 'group_notice'
  | 'connection_changed'
  | 'error';

export interface MessageUseCaseEvent {
  type: MessageUseCaseEventType;
  payload: any;
  timestamp: number;
}

export type MessageUseCaseSubscriber = (event: MessageUseCaseEvent) => void;

// 已读状态保护记录
interface ReadStateRecord {
  timestamp: number;
  version: number;
  lastReadSeq: number;
  clearTimer?: NodeJS.Timeout;
}

class ProcessMessageUseCase {
  private subscribers: Set<MessageUseCaseSubscriber> = new Set();
  private unsubscribeFns: Array<() => void> = [];
  private processedMessageIds: Set<string> = new Set();
  private processedMessageIdsExpiry: Map<string, number> = new Map();
  private pendingReadMap: Map<string, ReadStateRecord> = new Map();
  private readStateVersion = 0;
  private pendingUserRequests: Map<string, Promise<any>> = new Map();
  private currentUserId: string | null = null;

  // 常量
  private readonly READ_STATE_PROTECTION_DELAY = 5000; // 5秒
  private readonly MESSAGE_ID_EXPIRY = 5 * 60 * 1000; // 5分钟

  /**
   * 初始化用例
   */
  initialize(currentUserId: string | null): void {
    this.currentUserId = currentUserId;
    this.setupSSEListeners();
  }

  /**
   * 销毁用例
   */
  destroy(): void {
    this.unsubscribeFns.forEach((fn) => fn());
    this.unsubscribeFns = [];
    this.subscribers.clear();
    this.processedMessageIds.clear();
    this.processedMessageIdsExpiry.clear();
    this.pendingReadMap.clear();
    this.pendingUserRequests.clear();
  }

  /**
   * 设置SSE监听器
   */
  private setupSSEListeners(): void {
    // 监听私聊消息
    const unsubChat = sseClient.on('chat', (message) => {
      this.handleNewMessage(message);
    });

    // 监听群聊消息
    const unsubGroupMessage = sseClient.on('group_message', (message) => {
      this.handleNewMessage(message);
    });

    // 监听私聊已读回执
    const unsubRead = sseClient.on('read', (message) => {
      this.handleReadReceipt(message);
    });

    // 监听群聊已读回执
    const unsubGroupRead = sseClient.on('group_read', (message) => {
      this.handleGroupReadReceipt(message);
    });

    // 监听私聊消息撤回
    const unsubRecall = sseClient.on('recall', (message) => {
      this.handleRecallMessage(message);
    });

    // 监听群聊消息撤回
    const unsubGroupRecall = sseClient.on('group_recall', (message) => {
      this.handleGroupRecallMessage(message);
    });

    // 监听群聊输入状态
    const unsubGroupTyping = sseClient.on('group_typing', (message) => {
      this.handleGroupTyping(message);
    });

    // 监听群通知
    const unsubGroupNotice = sseClient.on('group_notice', (message) => {
      this.handleGroupNotice(message);
    });

    // 监听连接状态
    const unsubConnected = sseClient.on('connected', () => {
      this.notifySubscribers({
        type: 'connection_changed',
        payload: { connected: true },
        timestamp: Date.now(),
      });
    });

    const unsubDisconnected = sseClient.on('disconnected', () => {
      this.notifySubscribers({
        type: 'connection_changed',
        payload: { connected: false },
        timestamp: Date.now(),
      });
    });

    this.unsubscribeFns = [
      unsubChat,
      unsubGroupMessage,
      unsubRead,
      unsubGroupRead,
      unsubRecall,
      unsubGroupRecall,
      unsubGroupTyping,
      unsubGroupNotice,
      unsubConnected,
      unsubDisconnected,
    ];
  }

  /**
   * 处理新消息
   */
  private async handleNewMessage(message: any): Promise<void> {
    const { conversation_id, id, sender_id, seq, segments, created_at, _isAck } = message;

    // 消息去重检查
    if (this.isMessageProcessed(id)) {
      return;
    }
    this.markMessageAsProcessed(id);

    // 构造消息对象
    const newMessage = createMessage({
      id,
      conversationId: String(conversation_id),
      senderId: sender_id,
      seq,
      segments: segments || [],
      createdAt: created_at,
    });

    // 异步获取发送者信息（如果是群聊且不是当前用户发送的）
    if (message.type === 'group_message' && sender_id && sender_id !== this.currentUserId) {
      this.enrichMessageWithSender(newMessage);
    }

    // 异步保存到本地数据库
    const isRead = _isAck || sender_id === this.currentUserId;
    messageRepository.saveMessage(newMessage, isRead).catch((error) => {
      console.error('[ProcessMessageUseCase] 保存消息失败:', error);
    });

    // 通知订阅者
    this.notifySubscribers({
      type: 'message_received',
      payload: {
        message: newMessage,
        isCurrentUser: sender_id === this.currentUserId,
        isAck: _isAck,
      },
      timestamp: Date.now(),
    });
  }

  /**
   * 丰富消息的发送者信息
   */
  private async enrichMessageWithSender(message: Message): Promise<void> {
    try {
      const user = await this.getSenderInfo(message.senderId);
      if (user) {
        message.sender = user;
        // 重新通知，包含发送者信息
        this.notifySubscribers({
          type: 'message_received',
          payload: {
            message,
            isCurrentUser: false,
            isUpdated: true,
          },
          timestamp: Date.now(),
        });
      }
    } catch (error) {
      console.error('[ProcessMessageUseCase] 获取发送者信息失败:', error);
    }
  }

  /**
   * 获取发送者信息（带缓存和去重）
   */
  private async getSenderInfo(userId: string): Promise<any | null> {
    // 先检查本地缓存
    const cachedUser = await messageRepository.getUserCache(userId);
    if (cachedUser) {
      return cachedUser;
    }

    // 检查是否已有正在进行的请求
    const pendingRequest = this.pendingUserRequests.get(userId);
    if (pendingRequest) {
      return pendingRequest;
    }

    // 发起新请求
    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<any | null> {
    try {
      const response = await api.get(`/users/${userId}`);
      if (response.code === 0 && response.data) {
        await messageRepository.saveUserCache(response.data);
        return response.data;
      }
      return null;
    } catch (error) {
      console.error(`[ProcessMessageUseCase] 获取用户信息失败: ${userId}`, error);
      return null;
    }
  }

  /**
   * 检查消息是否已处理
   */
  private isMessageProcessed(messageId: string): boolean {
    return this.processedMessageIds.has(messageId);
  }

  /**
   * 标记消息已处理
   */
  private markMessageAsProcessed(messageId: string): void {
    this.processedMessageIds.add(messageId);
    this.processedMessageIdsExpiry.set(messageId, Date.now());

    // 定期清理
    if (this.processedMessageIds.size > 1000) {
      this.cleanupProcessedMessageIds();
    }
  }

  /**
   * 清理过期的消息ID
   */
  private cleanupProcessedMessageIds(): void {
    const now = Date.now();
    for (const [id, timestamp] of this.processedMessageIdsExpiry.entries()) {
      if (now - timestamp > this.MESSAGE_ID_EXPIRY) {
        this.processedMessageIds.delete(id);
        this.processedMessageIdsExpiry.delete(id);
      }
    }
  }

  /**
   * 处理私聊已读回执
   */
  private handleReadReceipt(message: any): void {
    // 可以在这里处理对方已读的状态更新
  }

  /**
   * 处理群聊已读回执
   */
  private handleGroupReadReceipt(message: any): void {
    // 可以在这里处理群聊已读状态更新
  }

  /**
   * 处理私聊消息撤回
   */
  private handleRecallMessage(message: any): Promise<void> {
    const { conversation_id, message_id } = message;

    // 更新本地数据库状态
    messageRepository
      .updateMessageStatus(message_id, 'recalled', true)
      .catch((error) => {
        console.error('[ProcessMessageUseCase] 更新撤回状态失败:', error);
      });

    // 通知订阅者
    this.notifySubscribers({
      type: 'message_recalled',
      payload: {
        conversationId: String(conversation_id),
        messageId: message_id,
        isGroup: false,
      },
      timestamp: Date.now(),
    });

    return Promise.resolve();
  }

  /**
   * 处理群聊消息撤回
   */
  private handleGroupRecallMessage(message: any): Promise<void> {
    const { conversation_id, message_id } = message;

    // 更新本地数据库状态
    messageRepository
      .updateMessageStatus(message_id, 'recalled', true)
      .catch((error) => {
        console.error('[ProcessMessageUseCase] 更新撤回状态失败:', error);
      });

    // 通知订阅者
    this.notifySubscribers({
      type: 'message_recalled',
      payload: {
        conversationId: String(conversation_id),
        messageId: message_id,
        isGroup: true,
      },
      timestamp: Date.now(),
    });

    return Promise.resolve();
  }

  /**
   * 处理群聊输入状态
   */
  private handleGroupTyping(message: any): void {
    const { group_id, user_id, is_typing } = message;

    this.notifySubscribers({
      type: 'typing_status',
      payload: {
        groupId: String(group_id),
        userId: user_id,
        isTyping: is_typing,
      },
      timestamp: Date.now(),
    });
  }

  /**
   * 处理群通知
   */
  private handleGroupNotice(message: any): void {
    const { notice_type, group_id, data, timestamp, message_id, seq } = message;
    const groupIdStr = String(group_id);

    const notice: GroupNotice = {
      type: notice_type,
      groupId: groupIdStr,
      data: data || {},
      timestamp: timestamp || Date.now(),
      messageId: message_id,
      seq,
    };

    // 通知订阅者
    this.notifySubscribers({
      type: 'group_notice',
      payload: {
        notice,
        text: buildGroupNoticeText(notice),
      },
      timestamp: Date.now(),
    });
  }

  /**
   * 标记会话已读
   */
  async markAsRead(conversationId: string, seq: number): Promise<void> {
    const existingRecord = this.pendingReadMap.get(conversationId);

    // 使用seq去重
    if (existingRecord && seq <= existingRecord.lastReadSeq) {
      return;
    }

    // 清除可能存在的旧定时器
    if (existingRecord?.clearTimer) {
      clearTimeout(existingRecord.clearTimer);
    }

    // 递增全局版本号
    this.readStateVersion++;
    const currentVersion = this.readStateVersion;

    // 记录已读状态
    this.pendingReadMap.set(conversationId, {
      timestamp: Date.now(),
      version: currentVersion,
      lastReadSeq: seq,
    });

    // 更新本地数据库
    await messageRepository.markConversationAsRead(conversationId);

    // 调用API
    try {
      await messageService.markAsRead(conversationId, seq);
    } catch (error) {
      console.error('[ProcessMessageUseCase] 标记已读API失败:', error);
      this.pendingReadMap.delete(conversationId);
      throw error;
    }

    // 设置延迟清除保护
    const clearTimer = setTimeout(() => {
      const record = this.pendingReadMap.get(conversationId);
      if (record && record.version === currentVersion) {
        this.pendingReadMap.delete(conversationId);
      }
    }, this.READ_STATE_PROTECTION_DELAY);

    this.pendingReadMap.set(conversationId, {
      timestamp: Date.now(),
      version: currentVersion,
      lastReadSeq: seq,
      clearTimer,
    });
  }

  /**
   * 检查会话是否有进行中的已读请求
   */
  hasPendingRead(conversationId: string): boolean {
    return this.pendingReadMap.has(conversationId);
  }

  /**
   * 获取进行中的已读记录
   */
  getPendingReadRecord(conversationId: string): ReadStateRecord | undefined {
    return this.pendingReadMap.get(conversationId);
  }

  /**
   * 发送消息
   */
  async sendMessage(
    conversationId: string,
    segments: any[],
    options?: { replyToId?: string }
  ): Promise<Message | null> {
    try {
      const response = await messageService.sendMessage(conversationId, {
        segments,
        reply_to_id: options?.replyToId,
      });

      if (response) {
        const message = createMessage({
          id: response.id,
          conversationId,
          senderId: this.currentUserId || '',
          seq: response.seq,
          segments,
          createdAt: new Date().toISOString(),
        });

        // 保存到本地数据库
        await messageRepository.saveMessage(message, true);

        return message;
      }

      return null;
    } catch (error) {
      console.error('[ProcessMessageUseCase] 发送消息失败:', error);
      throw error;
    }
  }

  /**
   * 获取本地消息
   */
  async getLocalMessages(conversationId: string, limit: number = 20): Promise<Message[]> {
    return messageRepository.getMessagesByConversation(conversationId, limit);
  }

  /**
   * 获取历史消息
   */
  async getHistoryMessages(
    conversationId: string,
    beforeSeq: number,
    limit: number = 20
  ): Promise<Message[]> {
    // 先从本地获取
    const localMessages = await messageRepository.getMessagesBeforeSeq(
      conversationId,
      beforeSeq,
      limit
    );

    if (localMessages.length >= limit) {
      return localMessages;
    }

    // 本地数据不足，从服务端获取
    try {
      const response = await messageService.getMessages(
        conversationId,
        undefined,
        beforeSeq,
        limit
      );

      if (response?.messages && response.messages.length > 0) {
        const serverMessages = response.messages.map((m: any) =>
          createMessage({
            id: m.id,
            conversationId: m.conversation_id || conversationId,
            senderId: m.sender_id,
            seq: m.seq,
            segments: m.segments || [],
            createdAt: m.created_at,
            status: m.status || 'normal',
          })
        );

        // 保存到本地
        await messageRepository.saveMessages(serverMessages, false);

        return serverMessages;
      }
    } catch (error) {
      console.error('[ProcessMessageUseCase] 获取历史消息失败:', error);
    }

    return localMessages;
  }

  /**
   * 从服务端同步消息
   */
  async syncMessagesFromServer(
    conversationId: string,
    afterSeq?: number
  ): Promise<Message[]> {
    try {
      const response = await messageService.getMessages(conversationId, afterSeq);

      if (response?.messages && response.messages.length > 0) {
        const messages = response.messages.map((m: any) =>
          createMessage({
            id: m.id,
            conversationId: m.conversation_id || conversationId,
            senderId: m.sender_id,
            seq: m.seq,
            segments: m.segments || [],
            createdAt: m.created_at,
            status: m.status || 'normal',
          })
        );

        // 保存到本地
        await messageRepository.saveMessages(messages, false);

        return messages;
      }
    } catch (error) {
      console.error('[ProcessMessageUseCase] 同步消息失败:', error);
    }

    return [];
  }

  /**
   * 订阅事件
   */
  subscribe(subscriber: MessageUseCaseSubscriber): () => void {
    this.subscribers.add(subscriber);

    return () => {
      this.subscribers.delete(subscriber);
    };
  }

  /**
   * 通知订阅者
   */
  private notifySubscribers(event: MessageUseCaseEvent): void {
    this.subscribers.forEach((subscriber) => {
      try {
        subscriber(event);
      } catch (error) {
        console.error('[ProcessMessageUseCase] 订阅者执行失败:', error);
      }
    });
  }

  /**
   * 获取当前用户ID
   */
  getCurrentUserId(): string | null {
    return this.currentUserId;
  }

  /**
   * 设置当前用户ID
   */
  setCurrentUserId(userId: string | null): void {
    this.currentUserId = userId;
  }
}

export const processMessageUseCase = new ProcessMessageUseCase();
export default processMessageUseCase;