convex-fundamentals

TypeScript Agent 的 Convex 基础知识

概述

Convex 是一项响应式后端即服务，提供实时数据同步、ACID 事务和 TypeScript 优先的开发体验。本技能涵盖核心心智模型、函数类型、分层架构模式和 HTTP action 实现。

TypeScript：绝不使用 any 类型

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

❌ 错误：

function handleData(data: any) { ... }
const items: any[] = [];

✅ 正确：

function handleData(data: { id: string; name: string }) { ... }
const items: Doc<"items">[] = [];

何时使用此技能

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

• 从头开始构建 Convex 后端应用
• 创建查询、mutation 或 action
• 理解不同函数类型之间的区别
• 实现分层架构模式
• 使用路由处理器设置 HTTP webhook
• 将其他后端迁移到 Convex
• 理解 Convex 的响应式数据模型

理念：为什么 Convex 感觉“别扭”（以及为什么这是正确的）

Convex 有意施加限制。当某种做法感觉像是反模式时，通常是 Convex 在提醒你重新考虑方案。

有意为之的摩擦

感觉不对的地方	Convex 想告诉你的
“我不能在查询上使用 .filter()”	定义索引。查询的复杂度应该是 O(log n)，而不是 O(n)。
“action 不能访问 ctx.db”	副作用和事务不能混在一起。通过 mutation 路由。
“我的 mutation 总是因 OCC 错误而失败”	你对同一文档的写入过于频繁。重新设计数据模型或使用 Workpool。
“我不能在 mutation 中调用 fetch()”	mutation 必须是确定性的。改为调度 action。
“每次文档变更时，查询都会重新运行”	你收集了过多数据。缩小索引范围或进行数据反规范化。
“我不能做联接”	进行数据反规范化。嵌入相关数据或使用查找表。

核心不变量

1. mutation 是 ACID 事务 — 所有写入要么一起成功，要么一起失败。
2. 查询是响应式的 — 任何读取的文档发生变化时，查询都会重新运行。
3. action 是非事务性的编排 — 不具备响应性，也没有 ctx.db。可以执行外部 I/O，并且必须通过 ctx.run* 调用查询和 mutation。
4. 调度与 mutation 是原子的 — 如果 mutation 失败，已调度的函数不会运行。

核心心智模型

Convex 中的一切都属于以下类别：

类型	可见性	用途
query	public	从 DB 读取，提供响应式订阅
mutation	public	写入 DB，执行 ACID 事务
action	public	外部 API、非确定性工作
httpAction	public（HTTP）	Webhook、第三方集成
internalQuery	private	供内部使用的读取原语
internalMutation	private	供内部使用的写入原语
internalAction	private	编排、外部调用

规则：仅导出精简的公共接口。所有实际逻辑都放在内部原语中。

internal* 变体与对应的公共函数具有相同的执行语义。唯一的区别是可见性——内部函数显示在 internal.* 下，而非 api.* 下，且不能从客户端调用。

分层架构

第 1 层：客户端接口（公共 API）

此处仅包含入口点。使用精简包装器进行校验、身份验证检查并委派处理。

// convex/jobs.ts — PUBLIC SURFACE
import { mutation, query } from "./_generated/server";
import { v } from "convex/values";
import { internal } from "./_generated/api";

// Public mutation: client entrypoint
export const startGeneration = mutation({
  args: { prompt: v.string() },
  returns: v.id("jobs"),
  handler: async (ctx, args) => {
    // 1. Auth check
    const identity = await ctx.auth.getUserIdentity();
    if (!identity) throw new Error("Unauthorized");

    // 2. Delegate to internal mutation
    const jobId = await ctx.runMutation(internal.jobs.createJob, {
      userId: identity.subject,
      prompt: args.prompt,
    });

    // 3. Schedule background work
    await ctx.scheduler.runAfter(0, internal.jobs.processJob, { jobId });

    return jobId;
  },
});

// Public query: reactive subscription
export const getJob = query({
  args: { jobId: v.id("jobs") },
  returns: v.union(
    v.object({
      _id: v.id("jobs"),
      _creationTime: v.number(),
      status: v.string(),
      prompt: v.string(),
      result: v.optional(v.string()),
    }),
    v.null()
  ),
  handler: async (ctx, args) => {
    return await ctx.db.get(args.jobId);
  },
});

第 2 层：领域逻辑（内部原语）

真正的逻辑所在之处。这些是后端的私有 API。

// convex/jobs.ts — INTERNAL PRIMITIVES (same file, different exports)
import { internalMutation, internalQuery } from "./_generated/server";
import { v } from "convex/values";

export const createJob = internalMutation({
  args: {
    userId: v.string(),
    prompt: v.string(),
  },
  returns: v.id("jobs"),
  handler: async (ctx, args) => {
    return await ctx.db.insert("jobs", {
      userId: args.userId,
      prompt: args.prompt,
      status: "pending",
    });
  },
});

export const markComplete = internalMutation({
  args: {
    jobId: v.id("jobs"),
    result: v.string(),
  },
  returns: v.null(),
  handler: async (ctx, args) => {
    await ctx.db.patch(args.jobId, {
      status: "completed",
      result: args.result,
    });
    return null;
  },
});

export const markFailed = internalMutation({
  args: {
    jobId: v.id("jobs"),
    error: v.string(),
  },
  returns: v.null(),
  handler: async (ctx, args) => {
    await ctx.db.patch(args.jobId, {
      status: "failed",
      error: args.error,
    });
    return null;
  },
});

export const getById = internalQuery({
  args: { jobId: v.id("jobs") },
  returns: v.union(
    v.object({
      _id: v.id("jobs"),
      _creationTime: v.number(),
      userId: v.string(),
      prompt: v.string(),
      status: v.string(),
      result: v.optional(v.string()),
      error: v.optional(v.string()),
    }),
    v.null()
  ),
  handler: async (ctx, args) => {
    return await ctx.db.get(args.jobId);
  },
});

第 3 层：编排（操作 + 调度器）

用于多步骤、外部或长时间运行的工作。

// convex/jobs.ts — ORCHESTRATION
import { internalAction } from "./_generated/server";
import { v } from "convex/values";
import { internal } from "./_generated/api";

export const processJob = internalAction({
  args: { jobId: v.id("jobs") },
  returns: v.null(),
  handler: async (ctx, args) => {
    // 1. Read current state
    const job = await ctx.runQuery(internal.jobs.getById, {
      jobId: args.jobId,
    });
    if (!job || job.status !== "pending") return null;

    try {
      // 2. External API call (non-deterministic)
      const result = await fetch("https://api.example.com/generate", {
        method: "POST",
        body: JSON.stringify({ prompt: job.prompt }),
      });
      const data = await result.json();

      // 3. Write result via mutation
      await ctx.runMutation(internal.jobs.markComplete, {
        jobId: args.jobId,
        result: data.output,
      });
    } catch (e) {
      await ctx.runMutation(internal.jobs.markFailed, {
        jobId: args.jobId,
        error: String(e),
      });
    }

    return null;
  },
});

HTTP 操作（Webhooks）

HTTP 操作必须放在 convex/http.ts 中，并使用 httpRouter：

// convex/http.ts
import { httpRouter } from "convex/server";
import { httpAction } from "./_generated/server";
import { internal } from "./_generated/api";

const http = httpRouter();

http.route({
  path: "/webhooks/stripe",
  method: "POST",
  handler: httpAction(async (ctx, req) => {
    // 1. Validate signature
    const signature = req.headers.get("stripe-signature");
    if (!signature) {
      return new Response("Missing signature", { status: 401 });
    }

    // 2. Parse body
    const body = await req.text();

    // 3. Delegate to internal mutation
    await ctx.runMutation(internal.billing.handleStripeWebhook, {
      signature,
      body,
    });

    return new Response(null, { status: 200 });
  }),
});

export default http;

模型层模式

将业务逻辑提取到可复用的模型函数中。函数保持精简；由模型负责实际工作。

为什么需要模型层？

1. 复用 — 在查询、变更和操作中使用相同的逻辑
2. 可测试性 — 纯函数，易于进行单元测试
3. 职责分离 — 在一个地方进行身份验证检查，在另一个地方处理业务逻辑
4. 类型安全 — 使用 QueryCtx 和 MutationCtx 类型表示上下文

模式：模型函数

// convex/model/users.ts — MODEL LAYER
import { QueryCtx, MutationCtx } from "../_generated/server";
import { Id } from "../_generated/dataModel";

// Read helper: reusable across queries
export async function getCurrentUser(ctx: QueryCtx) {
  const identity = await ctx.auth.getUserIdentity();
  if (!identity) throw new Error("Unauthorized");

  const user = await ctx.db
    .query("users")
    .withIndex("by_tokenIdentifier", (q) =>
      q.eq("tokenIdentifier", identity.tokenIdentifier)
    )
    .unique();

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

// Access control helper
export async function ensureHasAccess(
  ctx: QueryCtx,
  { conversationId }: { conversationId: Id<"conversations"> }
) {
  const user = await getCurrentUser(ctx);
  const conversation = await ctx.db.get(conversationId);
  if (!conversation || !conversation.members.includes(user._id)) {
    throw new Error("Unauthorized");
  }
  return { user, conversation };
}

// Write helper: encapsulates business logic
export async function addMessage(
  ctx: MutationCtx,
  {
    conversationId,
    content,
  }: { conversationId: Id<"conversations">; content: string }
) {
  const { user } = await ensureHasAccess(ctx, { conversationId });

  const messageId = await ctx.db.insert("messages", {
    conversationId,
    authorId: user._id,
    content,
    createdAt: Date.now(),
  });

  // Update denormalized lastMessageAt
  await ctx.db.patch(conversationId, { lastMessageAt: Date.now() });

  return messageId;
}

模式：精简的函数包装器

// convex/conversations.ts — THIN WRAPPERS
import { mutation, query } from "./_generated/server";
import { v } from "convex/values";
import * as Conversations from "./model/conversations";

// Query: delegates to model
export const listMessages = query({
  args: { conversationId: v.id("conversations") },
  returns: v.array(
    v.object({
      _id: v.id("messages"),
      content: v.string(),
      authorId: v.id("users"),
      createdAt: v.number(),
    })
  ),
  handler: async (ctx, args) => {
    return Conversations.listMessages(ctx, args);
  },
});

// Mutation: delegates to model
export const sendMessage = mutation({
  args: { conversationId: v.id("conversations"), content: v.string() },
  returns: v.id("messages"),
  handler: async (ctx, args) => {
    return Conversations.addMessage(ctx, args);
  },
});

文件结构

优先采用扁平结构。Convex 使用基于文件的路由：

convex/
  schema.ts              # Database schema
  http.ts                # HTTP actions (webhooks)
  auth.ts                # Auth helpers
  crons.ts               # Cron job definitions

  # Domain files (public + internal in same file)
  users.ts               # api.users.* + internal.users.*
  jobs.ts
  billing.ts
  audio.ts

  # Model layer (business logic)
  model/
    users.ts             # User business logic helpers
    conversations.ts     # Conversation logic
    billing.ts           # Billing logic

  # Shared utilities (NOT Convex functions)
  _lib/
    validators.ts        # Shared validators
    constants.ts         # Constants

路由：

• convex/users.ts → api.users.functionName / internal.users.functionName
• convex/audio/stems.ts → api.audio.stems.functionName
• convex/model/*.ts → 不参与路由（仅用于导入辅助函数）

函数类型快速参考

导入模式

// Public (client-callable)
import { query, mutation, action } from "./_generated/server";

// Private (internal only)
import {
  internalQuery,
  internalMutation,
  internalAction,
} from "./_generated/server";

// HTTP
import { httpAction } from "./_generated/server";
import { httpRouter } from "convex/server";

函数引用

import { api, internal } from "./_generated/api";

// Public: api.fileName.functionName
await ctx.runQuery(api.users.getUser, { userId });

// Internal: internal.fileName.functionName
await ctx.runMutation(internal.jobs.createJob, { data });

数据库操作

// Read
const doc = await ctx.db.get(id);
const docs = await ctx.db
  .query("table")
  .withIndex("by_x", (q) => q.eq("x", value))
  .collect();
const single = await ctx.db
  .query("table")
  .withIndex("by_x", (q) => q.eq("x", value))
  .unique();

// Write
const id = await ctx.db.insert("table", { field: value });
await ctx.db.patch(id, { field: newValue });
await ctx.db.replace(id, { ...fullDocument });
await ctx.db.delete(id);

常见问题

问题 1：在变更中调用 fetch()

❌ 错误：

export const createOrder = mutation({
  args: { productId: v.string() },
  returns: v.null(),
  handler: async (ctx, args) => {
    // ❌ Mutations must be deterministic - no external calls!
    const price = await fetch(
      "https://api.stripe.com/prices/" + args.productId
    );
    await ctx.db.insert("orders", { productId: args.productId, price });
    return null;
  },
});

✅ 正确：

// Mutation creates order, schedules action for external call
export const createOrder = mutation({
  args: { productId: v.string() },
  returns: v.id("orders"),
  handler: async (ctx, args) => {
    const orderId = await ctx.db.insert("orders", {
      productId: args.productId,
      status: "pending",
    });
    await ctx.scheduler.runAfter(0, internal.orders.fetchPrice, { orderId });
    return orderId;
  },
});

// Action handles external API call
export const fetchPrice = internalAction({
  args: { orderId: v.id("orders") },
  returns: v.null(),
  handler: async (ctx, args) => {
    const order = await ctx.runQuery(internal.orders.getById, {
      orderId: args.orderId,
    });
    const price = await fetch(
      "https://api.stripe.com/prices/" + order.productId
    );
    await ctx.runMutation(internal.orders.updatePrice, {
      orderId: args.orderId,
      price: await price.json(),
    });
    return null;
  },
});

问题 2：在操作中访问 ctx.db

❌ 错误：

export const processData = action({
  args: { id: v.string() },
  returns: v.null(),
  handler: async (ctx, args) => {
    // ❌ Actions don't have ctx.db!
    const data = await ctx.db.get(args.id);
    return null;
  },
});

✅ 正确：

export const processData = action({
  args: { id: v.id("data") },
  returns: v.null(),
  handler: async (ctx, args) => {
    // ✅ Use ctx.runQuery to read data
    const data = await ctx.runQuery(internal.data.getById, { id: args.id });

    // Do external work...
    const result = await fetch("...");

    // ✅ Use ctx.runMutation to write data
    await ctx.runMutation(internal.data.update, {
      id: args.id,
      result: await result.json(),
    });

    return null;
  },
});

问题 3：缺少返回值验证器

❌ 错误：

export const foo = mutation({
  args: {},
  handler: async (ctx) => {
    // implicitly returns undefined
  },
});

✅ 正确：

export const foo = mutation({
  args: {},
  returns: v.null(),
  handler: async (ctx) => {
    return null;
  },
});

决策指南

1. 此函数是否从数据库读取数据？ → 使用 query 或 internalQuery
2. 此函数是否向数据库写入数据？ → 使用 mutation 或 internalMutation
3. 此函数是否调用外部 API？ → 使用 action 或 internalAction
4. 此函数是否由客户端调用？ → 使用公共变体（query、mutation、action）
5. 此函数是否仅供内部使用？ → 使用 internal* 变体
6. 这是 Webhook 端点吗？ → 在 convex/http.ts 中使用 httpAction