Firebase实时数据库未读消息计数更新与监听器异常问题排查

我在Next.js网站上实现1对1聊天功能时，遇到Firebase实时数据库未读消息计数更新失效、监听器无法正确工作的问题：编写了getUnreadMessages函数，通过onValue监听器监听各聊天的未读消息数并汇总总数，但调用setUnreadMessages更新特定聊天的计数后，监听器并未同步更新，返回的总数始终是过时数据。

Firebase版本为v10.8.1，以下是数据库结构、相关函数代码及react-query调用方式：

数据库结构 #

{
  "chats": {
    "-NsiGVfPascYlr1ukPVP": {
      "lastMessage": "sdf",
      "members": {
        "65dd85133a58aa82bc050081": true,
        "65e71757ff707f0da0b9e8d8": true
      },
      "timestamp": 1710242413808,
      "unreadMessages": {
        "65dd85133a58aa82bc050081": 0,
        "65e71757ff707f0da0b9e8d8": 5
      }
    },
    "one": {
      "lastMessage": "안녕",
      "members": {
        "65dd85133a58aa82bc050081": true,
        "65df0955c7a84d629f8a54a5": true
      },
      "timestamp": 1710220731677,
      "unreadMessages": {
        "65dd85133a58aa82bc050081": 0
      }
    }
  },
  "users": {
    "65dd85133a58aa82bc050081": {
      "chats": {
        "-NsiGVfPascYlr1ukPVP": true,
        "one": true
      }
    },
    "65df0955c7a84d629f8a54a5": {
      "chats": {
        "one": true
      }
    },
    "65e71757ff707f0da0b9e8d8": {
      "chats": {
        "-NsiGVfPascYlr1ukPVP": true
      }
    }
  }
}

getUnreadMessages函数 #

export async function getUnreadMessages({
  userId,
}: {
  userId: string;
}): Promise<number> {
  let totalUnreadMessages = 0;

  const userChatsRef = ref(realtimeDB, `users/${userId}/chats`);
  const snapshot = await get(userChatsRef);
  const chatIds = snapshot.val() || {};

  const promises: Promise<void>[] = [];

  for (const chatId in chatIds) {
    const userChatRefRef = ref(
      realtimeDB,
      `chats/${chatId}/unreadMessages/${userId}`
    );

    const promise = new Promise<void>((resolve, reject) => {
      onValue(
        userChatRefRef,
        (snapshot: DataSnapshot) => {
          const unreadMessagesCount = snapshot.exists() ? snapshot.val() : 0;
          totalUnreadMessages += unreadMessagesCount;
          resolve();
        },
        (error) => {
          console.error(`Error listening to chat ${chatId}:`, error);
          reject(error);
        }
      );
    });

    promises.push(promise);
  }

  await Promise.all(promises);
  return totalUnreadMessages;
}

setUnreadMessages函数 #

export async function setUnreadMessages({
  chatId,
  userId,
  isReset,
}: {
  chatId: string;
  userId: string;
  isReset?: boolean;
}): Promise<void> {
  if (!chatId || !userId) return;
  try {
    const unreadMessagesRef = ref(
      realtimeDB,
      `chats/${chatId}/unreadMessages/${userId}`
    );
    let unreadMessagesCount = 0;

    if (!isReset) {
      const unreadMessagesSnapshot = await get(unreadMessagesRef);

      if (unreadMessagesSnapshot.exists()) {
        unreadMessagesCount = unreadMessagesSnapshot.val() || 0;
      }
    }

    await set(
      unreadMessagesRef,
      isReset ? unreadMessagesCount : ++unreadMessagesCount
    );
    console.log(
      `Unread messages count for user ${userId} in chat ${chatId} reset.`
    );
  } catch (error) {
    console.error("Error resetting user unread messages count:", error);
    throw error;
  }
}

react-query调用方式 #

const { data: fetchedUnreadMessagesCount } = useQuery({
    queryKey: ["unreadMessagesCount"],
    queryFn: () => getUnreadMessages({ userId: session?.user.id }),
    enabled: !!session?.user.id,
  });

问题排查与解决方案 #

核心问题分析 #

1. 监听器逻辑错误：getUnreadMessages中用onValue创建监听器，但在第一个快照返回后就调用resolve()，导致监听器只执行一次初始数据获取，后续数据更新不会触发回调。而且totalUnreadMessages仅在初始时累加，后续更新无法同步到返回值。
2. React Query缓存未自动更新：当前getUnreadMessages是一次性返回值的异步函数，无法实时推送更新，React Query会缓存初始结果，不会自动刷新。

修复方案 #

1. 重构实时监听函数

改为返回可订阅的数据流，确保数据更新时能同步通知：

export function subscribeToUnreadMessages(
  userId: string,
  onUpdate: (total: number) => void
): () => void {
  let unsubscribeHandlers: (() => void)[] = [];

  // 监听用户聊天列表变化，同步更新监听器
  const unsubscribeChats = onValue(ref(realtimeDB, `users/${userId}/chats`), async (snapshot) => {
    // 移除旧监听器
    unsubscribeHandlers.forEach(unsub => unsub());
    unsubscribeHandlers = [];

    const chatIds = snapshot.val() || {};
    const chatKeys = Object.keys(chatIds);

    // 初始化总数
    const initialCounts = await Promise.all(chatKeys.map(async (cid) => {
      const snap = await get(ref(realtimeDB, `chats/${cid}/unreadMessages/${userId}`));
      return snap.exists() ? snap.val() : 0;
    }));
    onUpdate(initialCounts.reduce((sum, count) => sum + count, 0));

    // 为每个聊天添加未读数监听器
    chatKeys.forEach(chatId => {
      const unreadRef = ref(realtimeDB, `chats/${chatId}/unreadMessages/${userId}`);
      const unsubscribe = onValue(unreadRef, async () => {
        // 每次更新时重新计算总数
        const counts = await Promise.all(chatKeys.map(async (cid) => {
          const snap = await get(ref(realtimeDB, `chats/${cid}/unreadMessages/${userId}`));
          return snap.exists() ? snap.val() : 0;
        }));
        onUpdate(counts.reduce((sum, count) => sum + count, 0));
      });
      unsubscribeHandlers.push(unsubscribe);
    });
  });

  return () => {
    unsubscribeChats();
    unsubscribeHandlers.forEach(unsub => unsub());
  };
}

2. 组件中结合React Query实现实时更新

import { useEffect } from 'react';
import { useQueryClient, useQuery } from '@tanstack/react-query';

function ChatComponent() {
  const queryClient = useQueryClient();
  const session = ...; // 获取用户会话信息

  useEffect(() => {
    if (!session?.user.id) return;

    // 订阅未读数更新，同步到React Query缓存
    const unsubscribe = subscribeToUnreadMessages(
      session.user.id,
      (total) => queryClient.setQueryData(["unreadMessagesCount"], total)
    );

    return unsubscribe;
  }, [session?.user.id, queryClient]);

  const { data: fetchedUnreadMessagesCount } = useQuery({
    queryKey: ["unreadMessagesCount"],
    queryFn: async () => {
      // 初始加载计算总数
      const userChatsRef = ref(realtimeDB, `users/${session.user.id}/chats`);
      const snapshot = await get(userChatsRef);
      const chatIds = snapshot.val() || {};
      const counts = await Promise.all(Object.keys(chatIds).map(async (cid) => {
        const snap = await get(ref(realtimeDB, `chats/${cid}/unreadMessages/${session.user.id}`));
        return snap.exists() ? snap.val() : 0;
      }));
      return counts.reduce((sum, count) => sum + count, 0);
    },
    enabled: !!session?.user.id,
  });

  return <div>未读消息：{fetchedUnreadMessagesCount}</div>;
}

3. 优化未读数更新的原子性

当前get+set存在并发冲突风险，改用Firebase原子操作：

export async function setUnreadMessages({
  chatId,
  userId,
  isReset,
}: {
  chatId: string;
  userId: string;
  isReset?: boolean;
}): Promise<void> {
  if (!chatId || !userId) return;
  try {
    const unreadMessagesRef = ref(
      realtimeDB,
      `chats/${chatId}/unreadMessages/${userId}`
    );

    if (isReset) {
      await set(unreadMessagesRef, 0);
    } else {
      // 原子递增，避免并发覆盖
      await increment(unreadMessagesRef, 1);
    }
    console.log(
      `Unread messages count for user ${userId} in chat ${chatId} ${isReset ? 'reset' : 'incremented'}.`
    );
  } catch (error) {
    console.error("Error updating unread messages count:", error);
    throw error;
  }
}

内容的提问来源于stack exchange，提问作者Mijeong Won