Convex Real-Time Patterns: Queries, Mutations, Actions, and Scheduling

Overview

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.

Critical Constraints

Before writing Convex code, internalize these:

• v.map() and v.set() — NOT supported as validator types
• undefined — invalid in Convex (use null or optional fields)
• .collect() without pagination — loads entire table into memory
• .filter() without an index — scans full table
• Mutations — cannot call external APIs (use actions instead)

Schema Definition

// convex/schema.ts
import { defineSchema, defineTable } from "convex/server";
import { v } from "convex/values";

export default defineSchema({
  messages: defineTable({
    channelId: v.id("channels"),
    text: v.string(),
    authorId: v.string(),
    deletedAt: v.optional(v.number()),  // null not needed; optional handles absence
  }).index("by_channel", ["channelId"]),

  channels: defineTable({
    name: v.string(),
    ownerId: v.string(),
    isPrivate: v.boolean(),
  }),
});

Queries (Reactive Subscriptions)

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

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

Mutations (Atomic Writes)

export const sendMessage = mutation({
  args: { channelId: v.id("channels"), text: v.string() },
  returns: v.id("messages"),
  handler: async (ctx, args) => {
    const identity = await ctx.auth.getUserIdentity();
    if (!identity) throw new Error("Not authenticated");

    return await ctx.db.insert("messages", {
      channelId: args.channelId,
      text: args.text,
      authorId: identity.subject,
    });
  },
});

export const deleteMessage = mutation({
  args: { messageId: v.id("messages") },
  returns: v.null(),
  handler: async (ctx, args) => {
    const identity = await ctx.auth.getUserIdentity();
    if (!identity) throw new Error("Not authenticated");

    const message = await ctx.db.get(args.messageId);
    if (!message || message.authorId !== identity.subject) {
      throw new Error("Unauthorized");
    }

    await ctx.db.patch(args.messageId, { deletedAt: Date.now() });
    return null;
  },
});

Actions (External I/O)

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

// convex/ai.ts
import { action } from "./_generated/server";
import { api } from "./_generated/api";
import { v } from "convex/values";

export const generateReply = action({
  args: { messageId: v.id("messages"), prompt: v.string() },
  returns: v.string(),
  handler: async (ctx, args) => {
    // External API call (allowed in actions)
    const response = await fetch("https://api.anthropic.com/v1/messages", {
      method: "POST",
      headers: {
        "x-api-key": process.env.ANTHROPIC_API_KEY!,
        "Content-Type": "application/json",
        "anthropic-version": "2023-06-01",
      },
      body: JSON.stringify({
        model: "claude-opus-4-6",
        max_tokens: 1024,
        messages: [{ role: "user", content: args.prompt }],
      }),
    });

    const data = await response.json();
    const reply = data.content[0].text;

    // Write back via mutation (not direct DB access)
    await ctx.runMutation(api.messages.sendMessage, {
      channelId: args.messageId as any,  // adjust for your schema
      text: reply,
    });

    return reply;
  },
});

Internal Functions

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

import { internalMutation, internalQuery } from "./_generated/server";

export const adminDeleteUser = internalMutation({
  args: { userId: v.string() },
  handler: async (ctx, args) => {
    // Only callable from other Convex functions, not from client
    await ctx.db.delete(args.userId as any);
  },
});

React Integration

"use client";
import { useQuery, useMutation } from "convex/react";
import { api } from "@/convex/_generated/api";

export function MessageList({ channelId }: { channelId: Id<"channels"> }) {
  // Live subscription — auto-updates when data changes
  const messages = useQuery(api.messages.listMessages, { channelId });
  const send = useMutation(api.messages.sendMessage);

  if (messages === undefined) return <div>Loading...</div>;

  return (
    <div>
      {messages.map((msg) => (
        <div key={msg._id}>{msg.text}</div>
      ))}
      <button onClick={() => send({ channelId, text: "Hello!" })}>
        Send
      </button>
    </div>
  );
}

Scheduled Functions

// convex/cron.ts
import { cronJobs } from "convex/server";
import { internal } from "./_generated/api";

const crons = cronJobs();

crons.daily(
  "clean-deleted-messages",
  { hourUTC: 2, minuteUTC: 0 },
  internal.messages.purgeDeleted,
);

crons.interval(
  "sync-analytics",
  { minutes: 15 },
  internal.analytics.sync,
);

export default crons;

Pagination

import { paginationOptsValidator } from "convex/server";

export const paginatedMessages = query({
  args: { channelId: v.id("channels"), paginationOpts: paginationOptsValidator },
  handler: async (ctx, args) => {
    return await ctx.db
      .query("messages")
      .withIndex("by_channel", (q) => q.eq("channelId", args.channelId))
      .order("desc")
      .paginate(args.paginationOpts);
  },
});

// React: usePaginatedQuery
const { results, loadMore, status } = usePaginatedQuery(
  api.messages.paginatedMessages,
  { channelId },
  { initialNumItems: 20 },
);

File Storage

// Upload from client
const sendImage = useMutation(api.messages.sendImage);
const generateUploadUrl = useMutation(api.storage.generateUploadUrl);

async function handleUpload(file: File) {
  const url = await generateUploadUrl();
  const result = await fetch(url, { method: "POST", body: file });
  const { storageId } = await result.json();
  await sendImage({ storageId, channelId });
}