convex-helpers

A QueryStream is an async iterable of documents ordered by indexed fields. They allow you to perform operations equivalent to SQL's UNION ALL, WHERE, JOIN, and ORDER BY directly on Convex data streams before returning the final result to the client. This is useful for merging multiple queries, filtering results based on complex predicates, or joining data from different tables.

Core Stream Operations:

• stream(ctx.db, schema): Constructs a new stream using DatabaseReader syntax.
• mergedStream(streams, fields): Combines multiple streams into one, maintaining order based on the provided fields.
• .flatMap(async (doc) => ...): Expands each document into its own stream and chains them together (useful for joins).
• .map(async (doc) => ...): Modifies each item in the stream while preserving order.
• .filterWith(async (doc) => ...): Filters documents based on a TypeScript predicate.

Finalizing a Stream: Once configured, you can treat a stream like a standard Convex query by calling methods like .first(), .collect(), .take(n), or .paginate(paginationOpts).

import { stream, mergedStream } from "convex-helpers/server/stream";
import schema from "./schema";

export const listForAuthors = query({
  args: {
    authors: v.array(v.id("users")),
    paginationOpts: paginationOptsValidator,
  },
  handler: async (ctx, { authors, paginationOpts }) => {
    const authorStreams = authors.map((author) =>
      stream(ctx.db, schema)
        .query("messages")
        .withIndex("by_author", (q) => q.eq("author", author)),
    );
    
    const allAuthorsStream = mergedStream(authorStreams, [
      "author",
      "_creationTime",
    ]);

    return await allAuthorsStream.paginate(paginationOpts);
  },
});

Triggers allow you to register functions that run automatically whenever data in a specific table changes via ctx.db.insert, ctx.db.patch, ctx.db.replace, or ctx.db.delete.

Key Characteristics:

• Atomicity: Triggers run in the same transaction as the mutation that caused them. If a trigger throws an error, the database write is aborted.
• Integration: Triggers must be paired with custom functions (like customMutation) to be executed. They will not run if you use raw Convex mutation wrappers, if you edit data in the Convex dashboard, or if you use npx convex import.
• Recursive Triggers: Triggers can trigger other triggers. These are executed in a breadth-first-search (BFS) queue. To perform writes without triggering subsequent triggers, use ctx.innerDb.
• Capabilities:
  • Denormalize computed fields.
  • Perform cascading deletes.
  • Schedule side effects (e.g., via ctx.scheduler).
  • Validate constraints or row-level security by throwing errors.

import { mutation as rawMutation } from "./_generated/server";
import { DataModel } from "./_generated/dataModel";
import { Triggers } from "convex-helpers/server/triggers";
import {
  customCtx,
  customMutation,
} from "convex-helpers/server/customFunctions";

const triggers = new Triggers<DataModel>();

triggers.register("users", async (ctx, change) => {
  if (change.operation === "insert") {
    // Logic here
  }
});

// You MUST wrap your mutation to enable triggers
export const mutation = customMutation(rawMutation, customCtx(triggers.wrapDB));

To track users even when they are not logged in, you can store a sessionId on the client and pass it to your Convex functions.

1. Client Setup: Wrap your application in a SessionProvider from convex-helpers/react/sessions inside your ConvexProvider.
2. Server Setup: Use queryWithSession (or a custom builder using SessionIdArg) to extract the sessionId and look up an anonymous user.
3. Usage: Use useSessionQuery from convex-helpers/react/sessions to automatically include the session ID in requests.

// Client: Add SessionProvider to your root
import { SessionProvider } from "convex-helpers/react/sessions";

<ConvexProvider client={convex}>
  <SessionProvider>
    <App />
  </SessionProvider>
</ConvexProvider>

// Client: Use useSessionQuery to fetch data
import { useSessionQuery } from "convex-helpers/react/sessions";
const results = useSessionQuery(api.myModule.mySessionQuery, { arg1: 1 });

// Server: Define query with session
import { customQuery } from "convex-helpers/server/customFunctions";
import { SessionIdArg } from "convex-helpers/server/sessions";

export const queryWithSession = customQuery(query, {
  args: SessionIdArg,
  input: async (ctx, { sessionId }) => {
    const anonymousUser = await getAnonUser(ctx, sessionId);
    return { ctx: { ...ctx, anonymousUser }, args: {} };
  },
});

Function Registration

• Public Functions: Use query, mutation, or action from ./_generated/server. These are exposed to the internet.
• Internal Functions: Use internalQuery, internalMutation, or internalAction from ./_generated/server. These are private and can only be called by other Convex functions.
• Requirement: ALWAYS include argument validators for all functions.

Function Calling

Use the ctx object to call functions from within other functions:

• ctx.runQuery(functionReference, args)
• ctx.runMutation(functionReference, args)
• ctx.runAction(functionReference, args)

Important Rules:

• All calls require a FunctionReference (from the api or internal objects), not the function itself.
• To avoid TypeScript circularity when calling a function in the same file, specify a type annotation on the return value.
• Avoid calling actions from actions unless crossing runtimes (e.g., V8 to Node); use shared helper functions instead.

export const f = query({
  args: { name: v.string() },
  handler: async (ctx, args) => {
    return "Hello " + args.name;
  },
});

export const g = query({
  args: {},
  handler: async (ctx, args) => {
    // Specify return type to avoid circularity
    const result: string = await ctx.runQuery(api.example.f, { name: "Bob" });
    return null;
  },
});

To schedule recurring tasks, use the cronJobs utility.

• Methods: Use crons.interval or crons.cron. Do not use the crons.hourly, crons.daily, or crons.weekly helpers.
• Arguments: These methods require a FunctionReference. Do not pass the function directly.
• Implementation: Declare a top-level crons object, call the scheduling methods, and export it as the default export of your crons.ts file.
• Internal Functions: If a cron calls an internal function, import the internal object from _generated/api.

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

const empty = internalAction({
  args: {},
  handler: async (ctx, args) => {
    console.log("empty");
  },
});

const crons = cronJobs();

// Run `internal.crons.empty` every two hours.
crons.interval("delete inactive users", { hours: 2 }, internal.crons.empty, {});

export default crons;