{"slug":"convex-realtime-patterns","title":"Convex Real-Time Patterns: Queries, Mutations, Actions, and Scheduling","tags":["convex","database","realtime","typescript","react","backend"],"agent_summary":"Convex real-time database patterns — reactive queries, transactional mutations, external-IO actions, React hooks, file storage, scheduled functions, and critical constraints (no v.map/v.set, no undefined, no unbounded collect).","trigger_phrases":["Convex","Convex query","Convex mutation","Convex action","Convex scheduling","useQuery Convex","useMutation Convex","Convex schema"],"runnable":false,"markdown":"\n## Overview\n\nConvex is a reactive database where queries are live subscriptions, mutations are atomic transactions, and actions handle external I/O. Every query auto-updates clients when its dependencies change.\n\n## Critical Constraints\n\nBefore writing Convex code, internalize these:\n\n- `v.map()` and `v.set()` — **NOT supported** as validator types\n- `undefined` — **invalid** in Convex (use `null` or optional fields)\n- `.collect()` without pagination — **loads entire table** into memory\n- `.filter()` without an index — **scans full table**\n- Mutations — **cannot call external APIs** (use actions instead)\n\n## Schema Definition\n\n```typescript\n// convex/schema.ts\nimport { defineSchema, defineTable } from \"convex/server\";\nimport { v } from \"convex/values\";\n\nexport default defineSchema({\n messages: defineTable({\n channelId: v.id(\"channels\"),\n text: v.string(),\n authorId: v.string(),\n deletedAt: v.optional(v.number()), // null not needed; optional handles absence\n }).index(\"by_channel\", [\"channelId\"]),\n\n channels: defineTable({\n name: v.string(),\n ownerId: v.string(),\n isPrivate: v.boolean(),\n }),\n});\n```\n\n## Queries (Reactive Subscriptions)\n\n```typescript\n// convex/messages.ts\nimport { query } from \"./_generated/server\";\nimport { v } from \"convex/values\";\n\nexport const listMessages = query({\n args: { channelId: v.id(\"channels\") },\n returns: v.array(v.object({\n _id: v.id(\"messages\"),\n text: v.string(),\n authorId: v.string(),\n })),\n handler: async (ctx, args) => {\n return await ctx.db\n .query(\"messages\")\n .withIndex(\"by_channel\", (q) => q.eq(\"channelId\", args.channelId))\n .filter((q) => q.eq(q.field(\"deletedAt\"), undefined)) // filter soft-deleted\n .order(\"desc\")\n .take(50); // ALWAYS use take() or paginate(), never collect()\n },\n});\n```\n\n## Mutations (Atomic Writes)\n\n```typescript\nexport const sendMessage = mutation({\n args: { channelId: v.id(\"channels\"), text: v.string() },\n returns: v.id(\"messages\"),\n handler: async (ctx, args) => {\n const identity = await ctx.auth.getUserIdentity();\n if (!identity) throw new Error(\"Not authenticated\");\n\n return await ctx.db.insert(\"messages\", {\n channelId: args.channelId,\n text: args.text,\n authorId: identity.subject,\n });\n },\n});\n\nexport const deleteMessage = mutation({\n args: { messageId: v.id(\"messages\") },\n returns: v.null(),\n handler: async (ctx, args) => {\n const identity = await ctx.auth.getUserIdentity();\n if (!identity) throw new Error(\"Not authenticated\");\n\n const message = await ctx.db.get(args.messageId);\n if (!message || message.authorId !== identity.subject) {\n throw new Error(\"Unauthorized\");\n }\n\n await ctx.db.patch(args.messageId, { deletedAt: Date.now() });\n return null;\n },\n});\n```\n\n## Actions (External I/O)\n\nActions can call external APIs, use fetch, and call AI models. They cannot write to the DB directly — they call mutations:\n\n```typescript\n// convex/ai.ts\nimport { action } from \"./_generated/server\";\nimport { api } from \"./_generated/api\";\nimport { v } from \"convex/values\";\n\nexport const generateReply = action({\n args: { messageId: v.id(\"messages\"), prompt: v.string() },\n returns: v.string(),\n handler: async (ctx, args) => {\n // External API call (allowed in actions)\n const response = await fetch(\"https://api.anthropic.com/v1/messages\", {\n method: \"POST\",\n headers: {\n \"x-api-key\": process.env.ANTHROPIC_API_KEY!,\n \"Content-Type\": \"application/json\",\n \"anthropic-version\": \"2023-06-01\",\n },\n body: JSON.stringify({\n model: \"claude-opus-4-6\",\n max_tokens: 1024,\n messages: [{ role: \"user\", content: args.prompt }],\n }),\n });\n\n const data = await response.json();\n const reply = data.content[0].text;\n\n // Write back via mutation (not direct DB access)\n await ctx.runMutation(api.messages.sendMessage, {\n channelId: args.messageId as any, // adjust for your schema\n text: reply,\n });\n\n return reply;\n },\n});\n```\n\n## Internal Functions\n\nUse `internal*` for sensitive operations that should not be callable from the client:\n\n```typescript\nimport { internalMutation, internalQuery } from \"./_generated/server\";\n\nexport const adminDeleteUser = internalMutation({\n args: { userId: v.string() },\n handler: async (ctx, args) => {\n // Only callable from other Convex functions, not from client\n await ctx.db.delete(args.userId as any);\n },\n});\n```\n\n## React Integration\n\n```typescript\n\"use client\";\nimport { useQuery, useMutation } from \"convex/react\";\nimport { api } from \"@/convex/_generated/api\";\n\nexport function MessageList({ channelId }: { channelId: Id<\"channels\"> }) {\n // Live subscription — auto-updates when data changes\n const messages = useQuery(api.messages.listMessages, { channelId });\n const send = useMutation(api.messages.sendMessage);\n\n if (messages === undefined) return
Loading...
;\n\n return (\n
\n {messages.map((msg) => (\n
{msg.text}
\n ))}\n \n
\n );\n}\n```\n\n## Scheduled Functions\n\n```typescript\n// convex/cron.ts\nimport { cronJobs } from \"convex/server\";\nimport { internal } from \"./_generated/api\";\n\nconst crons = cronJobs();\n\ncrons.daily(\n \"clean-deleted-messages\",\n { hourUTC: 2, minuteUTC: 0 },\n internal.messages.purgeDeleted,\n);\n\ncrons.interval(\n \"sync-analytics\",\n { minutes: 15 },\n internal.analytics.sync,\n);\n\nexport default crons;\n```\n\n## Pagination\n\n```typescript\nimport { paginationOptsValidator } from \"convex/server\";\n\nexport const paginatedMessages = query({\n args: { channelId: v.id(\"channels\"), paginationOpts: paginationOptsValidator },\n handler: async (ctx, args) => {\n return await ctx.db\n .query(\"messages\")\n .withIndex(\"by_channel\", (q) => q.eq(\"channelId\", args.channelId))\n .order(\"desc\")\n .paginate(args.paginationOpts);\n },\n});\n```\n\n```typescript\n// React: usePaginatedQuery\nconst { results, loadMore, status } = usePaginatedQuery(\n api.messages.paginatedMessages,\n { channelId },\n { initialNumItems: 20 },\n);\n```\n\n## File Storage\n\n```typescript\n// Upload from client\nconst sendImage = useMutation(api.messages.sendImage);\nconst generateUploadUrl = useMutation(api.storage.generateUploadUrl);\n\nasync function handleUpload(file: File) {\n const url = await generateUploadUrl();\n const result = await fetch(url, { method: \"POST\", body: file });\n const { storageId } = await result.json();\n await sendImage({ storageId, channelId });\n}\n```\n","html":"

Overview

\n

Convex is a reactive database where queries are live subscriptions, mutations are atomic transactions, and actions handle external I/O. Every query auto-updates clients when its dependencies change.

\n

Critical Constraints

\n

Before writing Convex code, internalize these:

\n\n

Schema Definition

\n
// convex/schema.ts\nimport { defineSchema, defineTable } from \"convex/server\";\nimport { v } from \"convex/values\";\n\nexport default defineSchema({\n  messages: defineTable({\n    channelId: v.id(\"channels\"),\n    text: v.string(),\n    authorId: v.string(),\n    deletedAt: v.optional(v.number()),  // null not needed; optional handles absence\n  }).index(\"by_channel\", [\"channelId\"]),\n\n  channels: defineTable({\n    name: v.string(),\n    ownerId: v.string(),\n    isPrivate: v.boolean(),\n  }),\n});\n
\n

Queries (Reactive Subscriptions)

\n
// convex/messages.ts\nimport { query } from \"./_generated/server\";\nimport { v } from \"convex/values\";\n\nexport const listMessages = query({\n  args: { channelId: v.id(\"channels\") },\n  returns: v.array(v.object({\n    _id: v.id(\"messages\"),\n    text: v.string(),\n    authorId: v.string(),\n  })),\n  handler: async (ctx, args) => {\n    return await ctx.db\n      .query(\"messages\")\n      .withIndex(\"by_channel\", (q) => q.eq(\"channelId\", args.channelId))\n      .filter((q) => q.eq(q.field(\"deletedAt\"), undefined))  // filter soft-deleted\n      .order(\"desc\")\n      .take(50);  // ALWAYS use take() or paginate(), never collect()\n  },\n});\n
\n

Mutations (Atomic Writes)

\n
export const sendMessage = mutation({\n  args: { channelId: v.id(\"channels\"), text: v.string() },\n  returns: v.id(\"messages\"),\n  handler: async (ctx, args) => {\n    const identity = await ctx.auth.getUserIdentity();\n    if (!identity) throw new Error(\"Not authenticated\");\n\n    return await ctx.db.insert(\"messages\", {\n      channelId: args.channelId,\n      text: args.text,\n      authorId: identity.subject,\n    });\n  },\n});\n\nexport const deleteMessage = mutation({\n  args: { messageId: v.id(\"messages\") },\n  returns: v.null(),\n  handler: async (ctx, args) => {\n    const identity = await ctx.auth.getUserIdentity();\n    if (!identity) throw new Error(\"Not authenticated\");\n\n    const message = await ctx.db.get(args.messageId);\n    if (!message || message.authorId !== identity.subject) {\n      throw new Error(\"Unauthorized\");\n    }\n\n    await ctx.db.patch(args.messageId, { deletedAt: Date.now() });\n    return null;\n  },\n});\n
\n

Actions (External I/O)

\n

Actions can call external APIs, use fetch, and call AI models. They cannot write to the DB directly — they call mutations:

\n
// convex/ai.ts\nimport { action } from \"./_generated/server\";\nimport { api } from \"./_generated/api\";\nimport { v } from \"convex/values\";\n\nexport const generateReply = action({\n  args: { messageId: v.id(\"messages\"), prompt: v.string() },\n  returns: v.string(),\n  handler: async (ctx, args) => {\n    // External API call (allowed in actions)\n    const response = await fetch(\"https://api.anthropic.com/v1/messages\", {\n      method: \"POST\",\n      headers: {\n        \"x-api-key\": process.env.ANTHROPIC_API_KEY!,\n        \"Content-Type\": \"application/json\",\n        \"anthropic-version\": \"2023-06-01\",\n      },\n      body: JSON.stringify({\n        model: \"claude-opus-4-6\",\n        max_tokens: 1024,\n        messages: [{ role: \"user\", content: args.prompt }],\n      }),\n    });\n\n    const data = await response.json();\n    const reply = data.content[0].text;\n\n    // Write back via mutation (not direct DB access)\n    await ctx.runMutation(api.messages.sendMessage, {\n      channelId: args.messageId as any,  // adjust for your schema\n      text: reply,\n    });\n\n    return reply;\n  },\n});\n
\n

Internal Functions

\n

Use internal* for sensitive operations that should not be callable from the client:

\n
import { internalMutation, internalQuery } from \"./_generated/server\";\n\nexport const adminDeleteUser = internalMutation({\n  args: { userId: v.string() },\n  handler: async (ctx, args) => {\n    // Only callable from other Convex functions, not from client\n    await ctx.db.delete(args.userId as any);\n  },\n});\n
\n

React Integration

\n
\"use client\";\nimport { useQuery, useMutation } from \"convex/react\";\nimport { api } from \"@/convex/_generated/api\";\n\nexport function MessageList({ channelId }: { channelId: Id<\"channels\"> }) {\n  // Live subscription — auto-updates when data changes\n  const messages = useQuery(api.messages.listMessages, { channelId });\n  const send = useMutation(api.messages.sendMessage);\n\n  if (messages === undefined) return <div>Loading...</div>;\n\n  return (\n    <div>\n      {messages.map((msg) => (\n        <div key={msg._id}>{msg.text}</div>\n      ))}\n      <button onClick={() => send({ channelId, text: \"Hello!\" })}>\n        Send\n      </button>\n    </div>\n  );\n}\n
\n

Scheduled Functions

\n
// convex/cron.ts\nimport { cronJobs } from \"convex/server\";\nimport { internal } from \"./_generated/api\";\n\nconst crons = cronJobs();\n\ncrons.daily(\n  \"clean-deleted-messages\",\n  { hourUTC: 2, minuteUTC: 0 },\n  internal.messages.purgeDeleted,\n);\n\ncrons.interval(\n  \"sync-analytics\",\n  { minutes: 15 },\n  internal.analytics.sync,\n);\n\nexport default crons;\n
\n

Pagination

\n
import { paginationOptsValidator } from \"convex/server\";\n\nexport const paginatedMessages = query({\n  args: { channelId: v.id(\"channels\"), paginationOpts: paginationOptsValidator },\n  handler: async (ctx, args) => {\n    return await ctx.db\n      .query(\"messages\")\n      .withIndex(\"by_channel\", (q) => q.eq(\"channelId\", args.channelId))\n      .order(\"desc\")\n      .paginate(args.paginationOpts);\n  },\n});\n
\n
// React: usePaginatedQuery\nconst { results, loadMore, status } = usePaginatedQuery(\n  api.messages.paginatedMessages,\n  { channelId },\n  { initialNumItems: 20 },\n);\n
\n

File Storage

\n
// Upload from client\nconst sendImage = useMutation(api.messages.sendImage);\nconst generateUploadUrl = useMutation(api.storage.generateUploadUrl);\n\nasync function handleUpload(file: File) {\n  const url = await generateUploadUrl();\n  const result = await fetch(url, { method: \"POST\", body: file });\n  const { storageId } = await result.json();\n  await sendImage({ storageId, channelId });\n}\n
\n"}