frontend/src/stores/message/services/MessageSyncService.ts

/**
 * 消息同步服务
 * 处理消息和会话的服务器同步逻辑
 */

import type { ConversationResponse, MessageResponse } from '../../../types/dto';
import { messageService } from '@/services/message';
import { messageRepository, conversationRepository } from '@/database';
import { MessageMapper } from '@/data/mappers';
import {
  type IConversationListPagedSource,
  SqliteConversationListPagedSource,
  createRemoteConversationListSource,
  CONVERSATION_LIST_PAGE_SIZE,
} from '../sources';
import type { IMessageSyncService, MessageManagerConversationListDeps, IUserCacheService } from '../types';
import { useMessageStore, normalizeConversationId, mergeMessagesById } from '../store';
import { ReadReceiptManager } from './ReadReceiptManager';

export class MessageSyncService implements IMessageSyncService {
  private getCurrentUserId: () => string | null;
  private readReceiptManager: ReadReceiptManager;
  private userCacheService: IUserCacheService;

  /** 远端会话列表（游标或页码） */
  private readonly remoteConversationListSource: IConversationListPagedSource;
  /** 本地会话列表缓存（SQLite） */
  private readonly localConversationListSource: IConversationListPagedSource;

  /** 正在加载会话列表下一页 */
  private loadingMoreConversations = false;
    /** fetchUnreadCount 去重：复用 in-flight promise */
  private fetchUnreadCountPromise: Promise<void> | null = null;

  constructor(
    getCurrentUserId: () => string | null,
    readReceiptManager: ReadReceiptManager,
    userCacheService: IUserCacheService,
    deps?: MessageManagerConversationListDeps
  ) {
    this.getCurrentUserId = getCurrentUserId;
    this.readReceiptManager = readReceiptManager;
    this.userCacheService = userCacheService;

    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> {
    const store = useMessageStore.getState();
    
    if (store.isLoading() && !forceRefresh) {
      return;
    }

    if (this.shouldDeferConversationRefresh(source, forceRefresh)) {
      if (__DEV__) {
        console.log('[MessageSyncService] defer fetchConversations', {
          source,
          activeConversationId: store.getActiveConversation(),
        });
      }
      return;
    }

    if (__DEV__) {
      console.log('[MessageSyncService] fetchConversations', {
        source,
        forceRefresh,
        activeConversationId: store.getActiveConversation(),
      });
    }

    // 非强制刷新且内存为空：先用本地源暖机
    const currentState = store.getState();
    if (!forceRefresh && currentState.conversations.size === 0) {
      const warmed = await this.hydrateConversationsFromLocalSource();
      if (warmed) {
        this.persistConversationListCache();
      }
    }

    store.setLoading(true);

    try {
      this.remoteConversationListSource.restart();
      const page = await this.remoteConversationListSource.loadNext();

      // 使用 zustand set 函数正确更新状态
      const newConversations = new Map<string, ConversationResponse>();
      page.items.forEach(conv => {
        const normalizedConv = this.normalizeConversationFromFetch(conv);
        newConversations.set(normalizedConv.id, normalizedConv);
      });
      
      // 合并更新：conversations + totalUnread 在同一次 set 中完成，消除中间状态窗口
      const totalUnread = Array.from(newConversations.values()).reduce(
        (sum, conv) => sum + (conv.unread_count || 0),
        0,
      );
      store.setConversationsWithUnread(newConversations, totalUnread, store.getUnreadCount().system);
      
      this.persistConversationListCache();
    } catch (error) {
      console.error('[MessageSyncService] 获取会话列表失败:', error);
      if (store.getConversations().length === 0) {
        const recovered = await this.hydrateConversationsFromLocalSource();
        if (recovered) {
          this.persistConversationListCache();
        }
      }
    } finally {
      store.setLoading(false);
    }
  }

  /**
   * 会话列表加载下一页
   */
  async loadMoreConversations(): Promise<void> {
    const store = useMessageStore.getState();
    
    if (!this.remoteConversationListSource.hasMore) {
      return;
    }
    if (this.loadingMoreConversations || store.isLoading()) {
      return;
    }

    this.loadingMoreConversations = true;
    try {
      const page = await this.remoteConversationListSource.loadNext();

      page.items.forEach(conv => {
        const normalizedConv = this.normalizeConversationFromFetch(conv);
        store.updateConversation(normalizedConv);
      });

      this.recomputeConversationTotalUnread();
      this.persistConversationListCache();
    } catch (error) {
      console.error('[MessageSyncService] 加载更多会话失败:', error);
    } finally {
      this.loadingMoreConversations = false;
    }
  }

  /**
   * 获取单个会话详情
   */
  async fetchConversationDetail(conversationId: string): Promise<ConversationResponse | null> {
    const store = useMessageStore.getState();
    const normalizedId = normalizeConversationId(conversationId);
    
    try {
      const detail = await messageService.getConversationById(normalizedId);
      if (detail) {
        const normalizedDetail = this.normalizeConversationFromFetch(detail as ConversationResponse);
        store.updateConversation(normalizedDetail);
        return normalizedDetail;
      }
      return null;
    } catch (error) {
      console.error('[MessageSyncService] 获取会话详情失败:', error);
      return null;
    }
  }

  /**
   * 获取会话消息（增量同步）
   */
  async fetchMessages(conversationId: string, afterSeq?: number): Promise<void> {
    const store = useMessageStore.getState();
    
    // 防止重复加载
    if (store.isLoadingMessages(conversationId)) {
      return;
    }

    store.setLoadingMessages(conversationId, true);

    try {
      const existingMessagesAtStart = store.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 messageRepository.getByConversation(conversationId, 20);

            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,
              }));

              store.setMessages(conversationId, formattedMessages);
            } else {
              store.setMessages(conversationId, []);
            }
          } catch (error) {
            console.warn('[MessageSyncService] 读取本地消息失败:', error);
          }
        }

        // 更新 baseline：合并本地DB后重新计算
        const currentMessages = store.getMessages(conversationId);
        baselineMaxSeq = currentMessages.reduce((max, m) => Math.max(max, m.seq || 0), 0);

        // 策略：有本地数据时仅增量同步，无数据时拉快照
        if (baselineMaxSeq > 0) {
          // 增量同步：只拉 baselineMaxSeq 之后的消息
          try {
            const incrementalResp = await messageService.getMessages(conversationId, baselineMaxSeq);
            const newMessages = incrementalResp?.messages || [];

            if (newMessages.length > 0) {
              const existingMsgs = store.getMessages(conversationId);
              const mergedMessages = mergeMessagesById(existingMsgs, newMessages);
              store.setMessages(conversationId, mergedMessages);

              messageRepository.saveMessagesBatch(
                MessageMapper.toCachedMessages(newMessages, conversationId)
              ).catch(error => {
                console.error('[MessageSyncService] 保存增量消息到本地失败:', error);
              });
            }
          } catch (error) {
            console.error('[MessageSyncService] 增量同步失败:', error);
          }
        } else {
          // 冷启动：本地无数据，拉快照
          try {
            const snapshotResp = await messageService.getMessages(conversationId, undefined, undefined, 50);
            const snapshotMessages = snapshotResp?.messages || [];

            if (snapshotMessages.length > 0) {
              store.setMessages(conversationId, snapshotMessages);

              messageRepository.saveMessagesBatch(
                MessageMapper.toCachedMessages(snapshotMessages, conversationId)
              ).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 = store.getMessages(conversationId);
          const mergedMessages = mergeMessagesById(existingMessages, newMessages);

          store.setMessages(conversationId, mergedMessages);

            messageRepository.saveMessagesBatch(
              MessageMapper.toCachedMessages(newMessages, conversationId)
            ).catch(error => {
              console.error('[MessageSyncService] 保存消息到本地失败:', error);
            });
        }
      }
    } catch (error) {
      console.error('[MessageSyncService] 获取消息失败:', error);
    } finally {
      // 异步填充用户信息（不阻塞消息显示）
      const currentMessages = store.getMessages(conversationId);
      if (currentMessages.length > 0) {
        this.userCacheService.enrichMessagesWithSenderInfo(
          conversationId,
          currentMessages,
          (convId, enrichedMessages) => {
            store.setMessages(convId, enrichedMessages);
          }
        );
      }
      store.setLoadingMessages(conversationId, false);
    }
  }

  /**
   * 加载更多历史消息
   */
  async loadMoreMessages(conversationId: string, beforeSeq: number, limit = 20): Promise<MessageResponse[]> {
    const store = useMessageStore.getState();
    
    try {
      // 先从本地获取
      const localMessages = await messageRepository.getBeforeSeq(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 = store.getMessages(conversationId);
        const mergedMessages = this.mergeOlderMessages(existingMessages, formattedMessages);
        store.setMessages(conversationId, mergedMessages);

        return formattedMessages;
      }

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

      if (response?.messages && response.messages.length > 0) {
        const serverMessages = response.messages;

        await messageRepository.saveMessagesBatch(
          MessageMapper.toCachedMessages(serverMessages, conversationId)
        );

        const existingMessages = store.getMessages(conversationId);
        const mergedMessages = this.mergeOlderMessages(existingMessages, serverMessages);
        store.setMessages(conversationId, mergedMessages);

        return serverMessages;
      }

      return [];
    } catch (error) {
      console.error('[MessageSyncService] 加载更多消息失败:', error);
      return [];
    }
  }

  /**
   * 获取未读数（去重：复用 in-flight promise）
   */
  async fetchUnreadCount(): Promise<void> {
    if (this.fetchUnreadCountPromise) {
      return this.fetchUnreadCountPromise;
    }
    this.fetchUnreadCountPromise = this.doFetchUnreadCount();
    try {
      await this.fetchUnreadCountPromise;
    } finally {
      this.fetchUnreadCountPromise = null;
    }
  }

  private async doFetchUnreadCount(): Promise<void> {
    const store = useMessageStore.getState();

    try {
      const [unreadData, systemUnreadData] = await Promise.all([
        messageService.getUnreadCount(),
        messageService.getSystemUnreadCount(),
      ]);

      const totalUnread = unreadData?.total_unread_count ?? 0;
      const systemUnread = systemUnreadData?.unread_count ?? 0;

      store.setUnreadCount(totalUnread, systemUnread);

      // 获取最后一条系统通知的时间
      if (systemUnread > 0) {
        try {
          const latestSystemMsg = await messageService.getSystemMessagesCursor({ cursor: '', page_size: 1 });
          if (latestSystemMsg?.list?.[0]?.created_at) {
            store.setLastSystemMessageAt(latestSystemMsg.list[0].created_at);
          }
        } catch {
          // 获取最后系统通知时间失败，不影响主流程
        }
      }

      // 服务端汇总未读为 0：不主动清零本地未读，等 fetchConversations 提供权威数据
      // 避免后端缓存竞态导致的未读数抖动
    } catch (error) {
      console.error('[MessageSyncService] 获取未读数失败:', error);
    }
  }

  /**
   * 基于 seq 的轻量增量同步（用于重连场景，替代全量刷新）
   * 返回 true 表示同步成功，false 表示需要退化为全量刷新
   */
  async syncBySeq(): Promise<boolean> {
    try {
      const serverItems = await messageService.getSyncData();
      if (!serverItems || serverItems.length === 0) return false;

      const store = useMessageStore.getState();
      const localConversations = store.conversations;

      // 没有本地数据，无法增量同步
      if (localConversations.size === 0) return false;

      // 构建 serverMap
      const serverMap = new Map<string, { max_seq: number; last_message_at: string }>();
      for (const item of serverItems) {
        serverMap.set(normalizeConversationId(item.id), item);
      }

      // 对比差异
      const staleIds: string[] = [];
      for (const [id, conv] of localConversations) {
        const serverItem = serverMap.get(id);
        if (!serverItem) continue;
        const localMaxSeq = Number(conv.last_seq || 0);
        const serverMaxSeq = Number(serverItem.max_seq || 0);
        if (serverMaxSeq > localMaxSeq) {
          staleIds.push(id);
        }
      }

      // 差异过大（>50%），退化为全量刷新
      if (staleIds.length > localConversations.size * 0.5) {
        return false;
      }

      // 对有差异的会话更新数据
      if (staleIds.length > 0) {
        const detailPromises = staleIds.map(id =>
          this.fetchConversationDetail(id).catch(() => {})
        );
        await Promise.all(detailPromises);

        // 重新读取 store，避免 stale 引用
        const activeId = useMessageStore.getState().getActiveConversation();
        if (activeId && staleIds.includes(activeId)) {
          const localMaxSeq = useMessageStore.getState().getMessages(activeId).reduce(
            (max, m) => Math.max(max, m.seq || 0), 0
          );
          if (localMaxSeq > 0) {
            await this.fetchMessages(activeId, localMaxSeq);
          }
        }
      }

      return true;
    } catch (error) {
      console.error('[MessageSyncService] syncBySeq 失败:', error);
      return false;
    }
  }

  /**
   * 检查是否可加载更多会话
   */
  canLoadMoreConversations(): boolean {
    return this.remoteConversationListSource.hasMore;
  }

  /**
   * 重启数据源
   */
  restartSources(): void {
    this.remoteConversationListSource.restart();
    this.localConversationListSource.restart();
    this.loadingMoreConversations = false;
  }

  // ==================== 私有工具方法 ====================

  private shouldDeferConversationRefresh(source: string, forceRefresh: boolean): boolean {
    const store = useMessageStore.getState();
    if (!store.getActiveConversation()) return false;
    if (!forceRefresh) return false;
    // hooks-initial-refresh 是冷启动兜底刷新，不应该被延迟
    return source === 'sse-reconnect' || source === 'prefetch';
  }

  private normalizeConversationFromFetch(conv: ConversationResponse): ConversationResponse {
    const id = normalizeConversationId(conv.id);
    // 应用已读保护逻辑
    return this.readReceiptManager.applyConversationFromFetch({
      ...conv,
      id,
    });
  }

  private recomputeConversationTotalUnread(): void {
    const store = useMessageStore.getState();
    const totalUnread = Array.from(store.getState().conversations.values()).reduce(
      (sum, conv) => sum + (conv.unread_count || 0),
      0
    );
    // 使用 zustand set 函数正确更新状态
    store.setUnreadCount(totalUnread, store.getUnreadCount().system);
  }

  private persistConversationListCache(): void {
    const store = useMessageStore.getState();
    const currentState = store.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));

    conversationRepository.saveWithRelated(list, users, groups).catch(error => {
      console.error('[MessageSyncService] 持久化会话列表失败:', error);
    });
  }

  private async hydrateConversationsFromLocalSource(): Promise<boolean> {
    const store = useMessageStore.getState();
    this.localConversationListSource.restart();
    const page = await this.localConversationListSource.loadNext();
    if (!page.items.length) {
      return false;
    }

    // 使用 zustand set 函数正确更新状态
    const newConversations = new Map<string, ConversationResponse>();
    page.items.forEach(conv => {
      const normalizedConv = this.normalizeConversationFromFetch(conv);
      newConversations.set(normalizedConv.id, normalizedConv);
    });
    
    // 合并更新：conversations + totalUnread 在同一次 set 中完成
    const totalUnread = Array.from(newConversations.values()).reduce(
      (sum, conv) => sum + (conv.unread_count || 0),
      0,
    );
    store.setConversationsWithUnread(newConversations, totalUnread, store.getUnreadCount().system);
    
    return true;
  }

  private mergeOlderMessages(existing: MessageResponse[], incoming: MessageResponse[]): MessageResponse[] {
    if (incoming.length === 0) return existing;

    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];
    }

    return mergeMessagesById(existing, incomingAsc);
  }
}