convex-realtime

Convex 实时功能

每个 useQuery 都是实时订阅。唯一的规则：让查询读取的文档集合保持精简并建立索引，因为读取集决定了查询何时重新运行。

适用场景

• 首次将 React 组件连接到 Convex
• 服务器上的数据发生变化，但页面没有更新
• 列表或仪表板因无关写入而反复重新渲染
• 加载更多列表在新数据到达时丢失或重复显示行
• 显示在线用户或正在输入的用户

订阅的工作方式

useQuery 会通过客户端的 websocket 建立订阅。服务器运行查询，记录其读取集（访问过的每个文档和索引范围），然后推送结果。当 mutation 提交的写入与该读取集重叠时，服务器会重新运行查询并推送新结果。没有重叠，就不会重新运行。结果按函数和参数缓存，因此两个组件调用 useQuery(api.tasks.list, { userId }) 时会共用一个订阅。所有活跃订阅都会在同一个数据库时间戳下更新，因此 UI 绝不会显示只应用了一部分的 mutation。要保证这一点，查询必须是确定性的，这也是为什么查询中不允许使用 Date.now()、Math.random() 和 fetch。

useQuery

undefined 表示正在加载。传入 "skip"，而不要用条件语句包裹钩子。

import { useQuery } from "convex/react";
import { api } from "../convex/_generated/api";
import { Id } from "../convex/_generated/dataModel";

function TaskList({ userId }: { userId: Id<"users"> | null }) {
  const tasks = useQuery(api.tasks.list, userId ? { userId } : "skip");

  if (userId === null) return <p>Select a user</p>;
  if (tasks === undefined) return <p>Loading</p>;

  return (
    <ul>
      {tasks.map((task) => (
        <li key={task._id}>{task.title}</li>
      ))}
    </ul>
  );
}

后端查询会返回经过验证的数据，并通过索引读取：

// convex/tasks.ts
import { query } from "./_generated/server";
import { v } from "convex/values";

export const list = query({
  args: { userId: v.id("users") },
  returns: v.array(
    v.object({
      _id: v.id("tasks"),
      _creationTime: v.number(),
      userId: v.id("users"),
      title: v.string(),
      completed: v.boolean(),
    }),
  ),
  handler: async (ctx, args) => {
    return await ctx.db
      .query("tasks")
      .withIndex("by_user", (q) => q.eq("userId", args.userId))
      .order("desc")
      .take(100);
  },
});

Mutations 和乐观更新

对于大多数 UI，直接调用 useMutation 就足够了。写入提交后，订阅通常会在一个往返时间内刷新。如果能明显感知到这个往返时间，就添加乐观更新：例如切换状态、重新排序、发送聊天消息。

import { useMutation } from "convex/react";
import { api } from "../convex/_generated/api";
import { Id } from "../convex/_generated/dataModel";

function useToggleTask(userId: Id<"users">) {
  return useMutation(api.tasks.toggle).withOptimisticUpdate(
    (localStore, args) => {
      const current = localStore.getQuery(api.tasks.list, { userId });
      if (current === undefined) return;
      localStore.setQuery(
        api.tasks.list,
        { userId },
        current.map((task) =>
          task._id === args.taskId
            ? { ...task, completed: !task.completed }
            : task,
        ),
      );
    },
  );
}

向列表插入内容也是同样的做法。创建一个临时文档，使用占位符 _id 和 _creationTime: Date.now()（这里允许这样做，因为代码在客户端运行），将其放到列表开头，再让服务器结果替换它。如果 mutation 抛出错误，Convex 会为你回滚本地存储。更新该 mutation 影响的每个查询，而不只是当前屏幕上的查询，否则其他查询在服务器响应前会显示过期数据。

保持实时的分页

// convex/messages.ts
import { query } from "./_generated/server";
import { v } from "convex/values";
import { paginationOptsValidator } from "convex/server";

export const listByChannel = query({
  args: { channelId: v.id("channels"), paginationOpts: paginationOptsValidator },
  // Return shape is fixed by .paginate(): { page, isDone, continueCursor }
  handler: async (ctx, args) => {
    return await ctx.db
      .query("messages")
      .withIndex("by_channel", (q) => q.eq("channelId", args.channelId))
      .order("desc")
      .paginate(args.paginationOpts);
  },
});

import { usePaginatedQuery } from "convex/react";
import { api } from "../convex/_generated/api";
import { Id } from "../convex/_generated/dataModel";

function MessageList({ channelId }: { channelId: Id<"channels"> }) {
  const { results, status, loadMore } = usePaginatedQuery(
    api.messages.listByChannel,
    { channelId },
    { initialNumItems: 30 },
  );

  return (
    <div>
      {results.map((m) => (
        <p key={m._id}>{m.body}</p>
      ))}
      {status === "CanLoadMore" && (
        <button onClick={() => loadMore(30)}>Load more</button>
      )}
      {status === "LoadingMore" && <p>Loading</p>}
    </div>
  );
}

每个已加载的页面都是独立的订阅，因此列表中任意位置的插入和删除都会即时显示，无需重新获取数据。status 的值为 LoadingFirstPage、CanLoadMore、LoadingMore 或 Exhausted。要实现无限滚动，可在哨兵元素上使用 IntersectionObserver，并在 status === "CanLoadMore" 时调用 loadMore。游标只能来自 .paginate()。不要通过切分 .collect() 的结果来实现偏移分页。

使用 useQueries 并行加载

同一个组件中的多个 useQuery 调用本来就会并行加载，并保持彼此一致。当查询集合是动态的（例如列表中的每个 id 对应一个查询）时，再使用 useQueries。

import { useQueries } from "convex/react";
import { api } from "../convex/_generated/api";
import { Id } from "../convex/_generated/dataModel";

function Avatars({ userIds }: { userIds: Array<Id<"users">> }) {
  const users = useQueries(
    Object.fromEntries(
      userIds.map((userId) => [userId, { query: api.users.get, args: { userId } }]),
    ),
  );

  return (
    <div>
      {userIds.map((id) => {
        const user = users[id];
        if (user === undefined) return <span key={id}>...</span>;
        if (user instanceof Error) return <span key={id}>!</span>;
        return <img key={id} src={user.avatarUrl} alt={user.name} />;
      })}
    </div>
  );
}

加载时，每个值都是 undefined；失败时是一个 Error；否则就是查询结果。

避免频繁重新运行

如果组件不关心某些写入，查询却因这些写入而重新运行，这就是频繁重新运行。先从服务器端着手解决。

• 缩小读取集。使用带有相等条件或范围条件的 withIndex 只读取一个切片。.filter() 或直接使用 .query("table") 会读取整张表，因此对该表的任何写入都会使查询重新运行。
• 只返回组件会渲染的内容。联接五张表的查询会在这五张表中的任意一张发生变化时重新运行。
• 查询中绝不要使用 Date.now()。将时间作为参数传入并进行取整，以保持参数稳定：useQuery(api.tasks.overdue, { now: Math.floor(Date.now() / 60000) * 60000 }) 每分钟重新运行一次，而不是每次渲染都重新运行。
• 使用 .take(n) 或分页限制结果数量。在不断增长的表上使用 .collect()，读取集也会不断增长。
• 将频繁变化的字段拆分到单独的表中。用户文档上的 lastSeen 时间戳会导致读取用户的所有查询都重新运行。将其放入有独立查询的 presence 表中。
• 对快速输入（打字、拖动）触发的 mutation 进行防抖，避免订阅被中间状态淹没。

在线状态

在线状态是每个客户端发送的一次心跳变更，以及对近期有心跳的行进行的查询。它保存在独立的表中，因此不会影响其他内容。请使用 @convex-dev/presence 组件，而不是自行构建。它负责处理心跳、断开连接后的清理，以及 React hook。

// convex/convex.config.ts
import { defineApp } from "convex/server";
import presence from "@convex-dev/presence/convex.config";

const app = defineApp();
app.use(presence);
export default app;

使用 npm install @convex-dev/presence 安装，然后按照软件包 README 配置服务端封装和 usePresence hook。如果必须手动实现，请将心跳保存在独立的表中，在服务端进行去重：如果上次心跳距今时间很短，就提前返回；读取时使用 roomId 索引。

常见错误

错误	为什么会出问题	应当这样做
if (userId) useQuery(...)	不同渲染之间 hook 调用顺序会改变	useQuery(fn, userId ? args : "skip")
将 undefined 视为空结果	加载状态会被渲染为“无结果”	先检查 === undefined
在查询中使用 Date.now()	结果不确定，会破坏缓存	将 now 作为参数传入，并进行取整
对大型表使用 .filter()	整个表都会进入读取集	添加索引并使用 withIndex
为信息流使用 .collect()	读取集会随表增长	使用 .take(n) 或 .paginate()
只对一个查询进行乐观更新	服务端响应前，其他视图会显示陈旧数据	在回调中更新所有受影响的查询
通过切片 .collect() 的结果进行偏移分页	插入行后，分页内容会漂移	使用 .paginate() 和 usePaginatedQuery
在用户文档中保存 lastSeen	每次心跳都会导致所有用户查询重新运行	使用独立的在线状态表或组件

检查清单

• 每个 useQuery 在读取结果前都会处理 undefined
• 条件查询使用 "skip"，而不是有条件地调用 hook
• 每个后端查询都有 args 和 returns 验证器
• 每个查询都通过 withIndex 读取大型表，没有使用 .filter()
• 所有查询中都没有 Date.now() 或 Math.random()
• 可能增长的列表使用 .paginate() 和 usePaginatedQuery
• 乐观更新会触及变更影响的所有查询
• 心跳等高频更新字段保存在独立的表中
• 客户端会对快速连续的输入变更进行防抖

文档

• https://docs.convex.dev/llms.txt
• https://docs.convex.dev/client/react
• https://docs.convex.dev/client/react/optimistic-updates
• https://docs.convex.dev/database/pagination
• https://www.convex.dev/components/presence