convex-performance-patterns

Convex 性能模式

概述

Convex 专为高性能而设计，但要实现最佳效果，需要遵循特定模式。本技能涵盖数据反规范化策略、索引设计、规避常见性能陷阱，以及使用 OCC（乐观并发控制）处理并发。

TypeScript：绝不使用 any 类型

**关键规则：**此代码库启用了 @typescript-eslint/no-explicit-any。使用 any 会导致构建失败。

何时使用此技能

在以下情况下使用此技能：

• 查询运行缓慢或导致过多重新渲染
• 设计索引以实现高效的数据访问
• 避免 N+1 查询模式
• 处理高争用写入（OCC 错误）
• 通过数据反规范化提升读取性能
• 优化响应式查询
• 使用计数器或聚合

核心性能原则

原则 1：查询复杂度应为 O(log n)，而不是 O(n)

Convex 查询应使用索引高效检索数据。如果你正在扫描整个表，说明做法不对。

原则 2：积极进行数据反规范化

Convex 不支持联接。应嵌入相关数据或维护查找表。

原则 3：尽量减少文档读取

查询中的每次文档读取都会产生一个依赖项。读取越少，重新渲染就越少。

原则 4：避免热点

频繁写入的单个文档会导致 OCC 冲突。

数据反规范化模式

模式 1：嵌入相关数据

❌ 错误：N+1 查询

export const getTeamWithMembers = query({
  args: { teamId: v.id("teams") },
  returns: v.null(),
  handler: async (ctx, args) => {
    const team = await ctx.db.get(args.teamId);
    if (!team) return null;

    // ❌ This triggers N additional reads, each causing re-renders
    const members = await Promise.all(
      team.memberIds.map((id) => ctx.db.get(id))
    );
    return { team, members };
  },
});

✅ 正确：将成员信息反规范化到团队中

// Schema: teams.members: v.array(v.object({ userId: v.id("users"), name: v.string(), avatar: v.string() }))
export const getTeamWithMembers = query({
  args: { teamId: v.id("teams") },
  returns: v.union(
    v.object({
      _id: v.id("teams"),
      _creationTime: v.number(),
      name: v.string(),
      members: v.array(
        v.object({
          userId: v.id("users"),
          name: v.string(),
          avatar: v.string(),
        })
      ),
    }),
    v.null()
  ),
  handler: async (ctx, args) => {
    return await ctx.db.get(args.teamId); // Single read, includes members
  },
});

模式 2：反规范化计数

切勿仅为计数而调用 .collect()。

❌ 错误：无界读取

const messages = await ctx.db
  .query("messages")
  .withIndex("by_channel", (q) => q.eq("channelId", channelId))
  .collect();
const count = messages.length;

✅ 正确：采用“99+”模式

const messages = await ctx.db
  .query("messages")
  .withIndex("by_channel", (q) => q.eq("channelId", channelId))
  .take(100);
const count = messages.length === 100 ? "99+" : String(messages.length);

✅ 最佳：反规范化计数器表

// Maintain a separate "channelStats" table with messageCount field
// Update it in the same mutation that inserts messages

export const getMessageCount = query({
  args: { channelId: v.id("channels") },
  returns: v.number(),
  handler: async (ctx, args) => {
    const stats = await ctx.db
      .query("channelStats")
      .withIndex("by_channel", (q) => q.eq("channelId", args.channelId))
      .unique();
    return stats?.messageCount ?? 0;
  },
});

export const addMessage = mutation({
  args: { channelId: v.id("channels"), content: v.string() },
  returns: v.id("messages"),
  handler: async (ctx, args) => {
    const messageId = await ctx.db.insert("messages", {
      channelId: args.channelId,
      content: args.content,
    });

    // Update denormalized count
    const stats = await ctx.db
      .query("channelStats")
      .withIndex("by_channel", (q) => q.eq("channelId", args.channelId))
      .unique();

    if (stats) {
      await ctx.db.patch(stats._id, { messageCount: stats.messageCount + 1 });
    } else {
      await ctx.db.insert("channelStats", {
        channelId: args.channelId,
        messageCount: 1,
      });
    }

    return messageId;
  },
});

模式 3：反规范化布尔字段

需要根据计算条件进行筛选时，应对结果进行反规范化：

// Schema
export default defineSchema({
  posts: defineTable({
    body: v.string(),
    tags: v.array(v.string()),
    // Denormalized: computed on write
    isImportant: v.boolean(),
  }).index("by_important", ["isImportant"]),
});

// Mutation: compute on write
export const createPost = mutation({
  args: { body: v.string(), tags: v.array(v.string()) },
  returns: v.id("posts"),
  handler: async (ctx, args) => {
    return await ctx.db.insert("posts", {
      body: args.body,
      tags: args.tags,
      isImportant: args.tags.includes("important"), // Denormalize!
    });
  },
});

// Query: O(log n) lookup
export const getImportantPosts = query({
  args: {},
  returns: v.array(
    v.object({
      _id: v.id("posts"),
      _creationTime: v.number(),
      body: v.string(),
      isImportant: v.boolean(),
    })
  ),
  handler: async (ctx) => {
    return await ctx.db
      .query("posts")
      .withIndex("by_important", (q) => q.eq("isImportant", true))
      .collect();
  },
});

索引设计

复合索引策略

索引支持前缀搜索。设计复合索引，以满足多个查询的需求。

// Schema
export default defineSchema({
  messages: defineTable({
    channelId: v.id("channels"),
    authorId: v.id("users"),
    content: v.string(),
    isDeleted: v.boolean(),
  })
    // ✅ This single index serves THREE query patterns:
    // 1. All messages in channel: .eq("channelId", id)
    // 2. Messages by author in channel: .eq("channelId", id).eq("authorId", id)
    // 3. Non-deleted messages by author: .eq("channelId", id).eq("authorId", id).eq("isDeleted", false)
    .index("by_channel_author_deleted", ["channelId", "authorId", "isDeleted"]),
});

// ❌ REDUNDANT: Don't create by_channel if you have by_channel_author_deleted
// The compound index can serve channel-only queries by partial prefix match

索引命名约定

包含所有字段：by_field1_and_field2_and_field3

.index("by_channel", ["channelId"])
.index("by_channel_and_author", ["channelId", "authorId"])
.index("by_user_and_status_and_createdAt", ["userId", "status", "createdAt"])

避免使用 Filter

绝不使用 .filter()。使用索引或在 TypeScript 中筛选。

❌ 错误：filter() 会扫描整个表

const activeUsers = await ctx.db
  .query("users")
  .filter((q) => q.eq(q.field("status"), "active"))
  .collect();

✅ 正确：基于索引

const activeUsers = await ctx.db
  .query("users")
  .withIndex("by_status", (q) => q.eq("status", "active"))
  .collect();

✅ 可以接受：数据集较小、筛选条件复杂

// Only if the dataset is bounded!
const allUsers = await ctx.db.query("users").take(1000);
const filtered = allUsers.filter(
  (u) => u.status === "active" && u.role !== "bot"
);

并发与 OCC（乐观并发控制）

Convex 使用 OCC 处理事务。当两个 mutation 同时读取并写入同一文档时，其中一个会自动重试。

问题：热点

❌ 错误：总是发生冲突的计数器

export const incrementCounter = mutation({
  args: {},
  returns: v.null(),
  handler: async (ctx) => {
    const counter = await ctx.db.query("counters").unique();
    await ctx.db.patch(counter!._id, { count: counter!.count + 1 });
    return null;
  },
});

// If 100 users click at once, 99 will retry → cascading OCC errors

解决方案 1：分片

将热点数据拆分到多个文档中：

// Schema: counterShards table
export default defineSchema({
  counterShards: defineTable({
    shardId: v.number(),
    delta: v.number(),
  }).index("by_shard", ["shardId"]),
});

// On write: pick random shard
export const incrementCounter = mutation({
  args: {},
  returns: v.null(),
  handler: async (ctx) => {
    const shardId = Math.floor(Math.random() * 10);
    await ctx.db.insert("counterShards", { shardId, delta: 1 });
    return null;
  },
});

// On read: sum all shards
export const getCount = query({
  args: {},
  returns: v.number(),
  handler: async (ctx) => {
    const shards = await ctx.db.query("counterShards").collect();
    return shards.reduce((sum, s) => sum + s.delta, 0);
  },
});

解决方案 2：Workpool（convex-helpers）

通过串行化写入来避免冲突：

import { Workpool } from "@convex-dev/workpool";
import { components } from "./_generated/api";

const counterPool = new Workpool(components.counterWorkpool, {
  maxParallelism: 1, // Serialize all counter updates
});

export const incrementCounter = mutation({
  args: {},
  returns: v.null(),
  handler: async (ctx) => {
    await counterPool.enqueueMutation(ctx, internal.counters.doIncrement, {});
    return null;
  },
});

解决方案 3：Aggregate 组件

对于计数/求和，使用 Convex Aggregate 组件：

import { Aggregate } from "@convex-dev/aggregate";

// Atomic increments without OCC conflicts
await aggregate.insert(ctx, "pageViews", 1);
const total = await aggregate.sum(ctx);

何时使用 Workpool，何时使用 Scheduler

• 对于无需协调的一次性后台任务，使用 ctx.scheduler。
• 需要控制并发、扇出并行处理，或通过串行化来避免 OCC 冲突时，使用 Workpool。

事务边界

合并读取

在 action 中多次调用 ctx.runQuery 不具备事务性：

❌ 错误：查询之间存在竞态条件

export const processTeam = action({
  args: { teamId: v.id("teams") },
  returns: v.null(),
  handler: async (ctx, args) => {
    const team = await ctx.runQuery(internal.teams.getTeam, {
      teamId: args.teamId,
    });
    const owner = await ctx.runQuery(internal.users.getUser, {
      userId: team.ownerId,
    });
    // Owner might have changed between the two queries!
    return null;
  },
});

✅ 正确：使用单个事务性查询

export const processTeam = action({
  args: { teamId: v.id("teams") },
  returns: v.null(),
  handler: async (ctx, args) => {
    const teamWithOwner = await ctx.runQuery(internal.teams.getTeamWithOwner, {
      teamId: args.teamId,
    });
    // Team and owner fetched atomically
    return null;
  },
});

批量写入

在 action 中执行多次 mutation 并非原子操作：

❌ 错误：可能发生部分失败

export const createUsers = action({
  args: { users: v.array(v.object({ name: v.string() })) },
  returns: v.null(),
  handler: async (ctx, args) => {
    for (const user of args.users) {
      await ctx.runMutation(internal.users.insert, { user });
    }
    // If third insert fails, first two still exist!
    return null;
  },
});

✅ 正确：使用单个事务

export const createUsers = mutation({
  args: { users: v.array(v.object({ name: v.string() })) },
  returns: v.array(v.id("users")),
  handler: async (ctx, args) => {
    const ids: Id<"users">[] = [];
    for (const user of args.users) {
      ids.push(
        await ctx.db.insert("users", { name: user.name, createdAt: Date.now() })
      );
    }
    return ids; // All succeed or all fail together
  },
});

查询优化

使用带合理限制的 take()

// ❌ BAD: Unbounded collect
const allMessages = await ctx.db
  .query("messages")
  .withIndex("by_channel", (q) => q.eq("channelId", channelId))
  .collect();

// ✅ GOOD: Bounded with take()
const recentMessages = await ctx.db
  .query("messages")
  .withIndex("by_channel", (q) => q.eq("channelId", channelId))
  .order("desc")
  .take(50);

并行获取数据

export const getDashboard = query({
  args: { userId: v.id("users") },
  returns: v.object({
    user: v.object({ _id: v.id("users"), name: v.string() }),
    stats: v.object({ messageCount: v.number(), channelCount: v.number() }),
  }),
  handler: async (ctx, args) => {
    // Fetch in parallel - both queries run simultaneously
    const [user, stats] = await Promise.all([
      ctx.db.get(args.userId),
      ctx.db
        .query("userStats")
        .withIndex("by_user", (q) => q.eq("userId", args.userId))
        .unique(),
    ]);

    if (!user) throw new Error("User not found");

    return {
      user: { _id: user._id, name: user.name },
      stats: stats ?? { messageCount: 0, channelCount: 0 },
    };
  },
});

只需一条记录时避免收集全部数据

// ❌ BAD: Collecting then taking first
const users = await ctx.db
  .query("users")
  .withIndex("by_email", (q) => q.eq("email", email))
  .collect();
const user = users[0];

// ✅ GOOD: Use .first() or .unique()
const user = await ctx.db
  .query("users")
  .withIndex("by_email", (q) => q.eq("email", email))
  .first();

// For exactly-one semantics (throws if multiple)
const user = await ctx.db
  .query("users")
  .withIndex("by_email", (q) => q.eq("email", email))
  .unique();

常见陷阱

陷阱 1：N+1 查询模式

❌ 错误：

const posts = await ctx.db.query("posts").take(10);
const postsWithAuthors = await Promise.all(
  posts.map(async (post) => ({
    ...post,
    author: await ctx.db.get(post.authorId), // N additional queries!
  }))
);

✅ 正确：进行数据反规范化或批量处理

// Option 1: Denormalize author info into posts
// Schema: posts.author: v.object({ id: v.id("users"), name: v.string() })

// Option 2: Batch fetch with getAll (from convex-helpers)
import { getAll } from "convex-helpers/server/relationships";

const posts = await ctx.db.query("posts").take(10);
const authorIds = [...new Set(posts.map((p) => p.authorId))];
const authors = await getAll(ctx.db, authorIds);
const authorMap = new Map(authors.map((a) => [a._id, a]));

const postsWithAuthors = posts.map((post) => ({
  ...post,
  author: authorMap.get(post.authorId),
}));

陷阱 2：不使用索引的无界查询

❌ 错误：

// Full table scan!
const allItems = await ctx.db.query("items").collect();

✅ 正确：

// With pagination or limits
const items = await ctx.db.query("items").take(100);

// Or with index if filtering
const items = await ctx.db
  .query("items")
  .withIndex("by_status", (q) => q.eq("status", "active"))
  .take(100);

陷阱 3：单文档热点

❌ 错误：

// Global counter - constant OCC conflicts under load
const global = await ctx.db.query("globals").unique();
await ctx.db.patch(global!._id, { viewCount: global!.viewCount + 1 });

✅ 正确：使用分片或聚合

// Sharded counter
const shardId = Math.floor(Math.random() * 10);
await ctx.db.insert("viewShards", { shardId, delta: 1, timestamp: Date.now() });

// Periodic aggregation job consolidates shards

性能检查清单

部署前，请确认：

• 所有查询都使用索引（数据库上不使用 .filter()）
• 不存在未使用 take(n) 的无界 .collect() 调用
• 相关数据已反规范化，以避免 N+1 模式
• 高频写入的文档使用分片或 Workpool
• 复合索引可服务于多种查询模式
• 没有冗余索引（复合索引已覆盖前缀）
• 计数使用反规范化计数器，而不是 .collect().length
• 变更操作在单个事务中批量处理相关写入

快速参考

查询模式

模式	方法	使用场景
按 ID 获取	ctx.db.get(id)	查找单个文档
获取多个	ctx.db.query().collect()	获取多个文档（使用 take(n)）
获取第一个	ctx.db.query().first()	获取第一个匹配的文档
获取唯一项	ctx.db.query().unique()	获取唯一文档（有多个时会抛出异常）
索引查询	.withIndex("name", q => ...)	高效的筛选查询

索引使用

// Equality on all fields
.withIndex("by_a_b_c", (q) => q.eq("a", 1).eq("b", 2).eq("c", 3))

// Prefix match (uses first N fields)
.withIndex("by_a_b_c", (q) => q.eq("a", 1).eq("b", 2))

// Range on last field
.withIndex("by_a_b_c", (q) => q.eq("a", 1).eq("b", 2).gt("c", 0))

// Cannot skip fields in the middle!
// ❌ .withIndex("by_a_b_c", (q) => q.eq("a", 1).eq("c", 3))