# convex-mq

A typed message queue component for Convex with reactive consumers, visibility timeouts, and automatic retries.

This is an upstream package. It has not been adapted or qualified for Bijection.

- Package: convex-mq
- Source: https://github.com/gergesh/convex-mq
- npm: https://www.npmjs.com/package/convex-mq
- Directory: https://bijection.com/components/convex-mq

Read the upstream README for setup, license, and compatibility before using it.

## Upstream package documentation

Recorded 2026-10-23. This describes convex-mq; check the Bijection adaptation separately.

ConvexMQ is a typed message queue component that enables reliable background job processing with reactive consumption, visibility timeouts, and automatic retries. It offers two deployment modes: component mode for isolated queue tables, or library mode for custom schemas with filtered consumption via custom indexes. The component generates typed Convex functions for publishing, claiming, and acknowledging messages while handling lease tokens to prevent race conditions.

### Use cases

• **Background email processing** - Queue email jobs from user signup mutations and process them in external services with automatic retries if the SMTP provider fails
• **Image processing pipelines** - Enqueue image resize tasks with metadata like file size, then use filtered consumption to route large files to high-memory workers
• **Multi-tenant job routing** - Use custom indexes to consume only messages for specific tenants or worker types, enabling dedicated processing lanes
• **Dead letter handling** - Automatically retry failed jobs with configurable max attempts, then handle permanently failed messages in dead letter queues
• **Batch operations** - Publish multiple messages at once for bulk operations like sending welcome emails to new user cohorts

### How it works

The component operates in two modes: component mode uses isolated tables via `app.use(messageQueue)` in your Convex config, while library mode integrates with your schema using `messageQueueTable()`. You define message schemas with Convex validators, then call `messageQueue.api(query, mutation)` to generate typed functions like `publish`, `claim`, `ack`, and `nack`.

Messages follow a claim-ack pattern where consumers call `claim()` to reserve messages with visibility timeouts, process them externally, then call `ack()` or `nack()` to complete or retry. The component uses lease tokens (`claimId`) to prevent stale consumers from acknowledging messages they no longer own. Library mode supports filtered consumption via custom indexes, letting you consume subsets of messages with additional query parameters.

Consumers can operate reactively via WebSocket subscriptions using `consume(client, api.queue, handler)` or poll with `consumePolling()`. Authentication typically uses internal functions with deploy keys for consumers and authenticated mutations for publishers. The component automatically handles retries up to `maxAttempts` and returns dead letter messages when retries are exhausted.

### Upstream README

# ConvexMQ

A typed message queue for [Convex](https://convex.dev). Publish messages from your Convex mutations, consume them from any external service — reactively via WebSocket or by polling.

- **Two modes** — component mode (isolated tables) or library mode (your schema, your indexes)
- **Typed payloads** — define your message schema once, get type safety everywhere
- **Reactive consumption** — subscribe to new messages in real-time, no polling needed
- **Visibility timeouts** — claimed messages auto-return to the queue if not acked
- **Lease tokens** — prevents stale consumers from acking messages they no longer own
- **Automatic retries** — configurable max attempts with dead-letter handling
- **Batch operations** — publish and claim messages in batches
- **Filtered consumption** — consume subsets of messages using custom indexes (library mode)
- **Zero boilerplate** — `api()` generates all your Convex function exports in one line

## Installation

```bash
npm install convex-mq
```

## Choose Your Mode

ConvexMQ offers two ways to use it:

|                          | Component Mode                    | Library Mode                     |
| ------------------------ | --------------------------------- | -------------------------------- |
| **Table ownership**      | Component owns the table          | You own the table in your schema |
| **Custom indexes**       | Not supported                     | Full index support               |
| **Filtered consumption** | Not supported                     | Filter by any indexed field      |
| **Setup**                | `app.use(messageQueue)` in config | `messageQueueTable()` in schema  |
| **Import**               | `convex-mq`                       | `convex-mq/table`                |

**Use component mode** when you want a simple, isolated queue with no custom queries.

**Use library mode** when you need custom indexes, filtered consumption, or want to query messages alongside your other data.

---

## Component Mode

### 1. Add the component to your Convex app

```ts
// convex/convex.config.ts
import { defineApp } from "convex/server";
import messageQueue from "convex-mq/convex.config";

const app = defineApp();
app.use(messageQueue, { name: "emailQueue" });
export default app;
```

Each queue instance gets an isolated table. For multiple message types, add multiple instances:

```ts
app.use(messageQueue, { name: "emailQueue" });
app.use(messageQueue, { name: "taskQueue" });
```

### 2. Define your queue

```ts
// convex/emailQueue.ts
import { MessageQueue } from "convex-mq";
import { components } from "./_generated/api";
import { query, mutation } from "./_generated/server";
import { v } from "convex/values";

const emailQueue = new MessageQueue(components.emailQueue, {
  message: v.object({
    to: v.string(),
    subject: v.string(),
    body: v.string(),
  }),
  defaultMaxAttempts: 3, // optional, default: 3
  defaultVisibilityTimeoutMs: 30_000, // optional, default: 30s
});

// Generates typed peek/claim/ack/nack/publish/publishBatch exports
export const { peek, claim, ack, nack, publish, publishBatch } = emailQueue.api(query, mutation);
```

That's it. No boilerplate wrappers needed.

### Authentication

By default, `api()` generates public functions. In production, you'll typically want:

- **Publishing** — requires user authentication (only logged-in users can enqueue messages)
- **Consuming** — requires a [deploy key](https://docs.convex.dev/cli/deploy-key-types) (only your server-side consumers can claim/ack messages)

Convex has built-in support for this via [`internalQuery`/`internalMutation`](https://docs.convex.dev/functions/internal-functions), which can only be called from other Convex functions or with a deploy key:

```ts
// convex/emailQueue.ts
import { MessageQueue } from "convex-mq";
import { components } from "./_generated/api";
import { internalQuery, internalMutation } from "./_generated/server";
import { v } from "convex/values";

const emailQueue = new MessageQueue(components.emailQueue, {
  message: v.object({ to: v.string(), subject: v.string(), body: v.string() }),
});

// Consumer ops — deploy key only (internal functions)
export const { peek, claim, ack, nack } = emailQueue.api(internalQuery, internalMutation);
```

For publishing, use [custom function builders](https://docs.convex.dev/auth/functions-auth) from `convex-helpers` to require user auth:

```ts
// convex/auth.ts
import { customMutation } from "convex-helpers/server/customFunctions";
import { mutation } from "./_generated/server";

export const authenticatedMutation = customMutation(mutation, {
  args: {},
  input: async (ctx) => {
    const identity = await ctx.auth.getUserIdentity();
    if (!identity) throw new Error("Unauthenticated");
    return { ctx: { userId: identity.subject }, args: {} };
  },
});
```

```ts
// convex/emailQueue.ts (continued)
import { authenticatedMutation } from "./auth";

// Publishing — user auth required
export const publish = authenticatedMutation({
  args: { to: v.string(), subject: v.string(), body: v.string() },
  handler: async (ctx, args) => emailQueue.publish(ctx, args),
});
```

Server-side consumers connect with a deploy key using `setAdminAuth`:

```ts
// ConvexHttpClient
const http = new ConvexHttpClient(CONVEX_URL);
(http as any).setAdminAuth(process.env.CONVEX_DEPLOY_KEY!);

// ConvexClient (WebSocket)
const ws = new ConvexClient(CONVEX_URL);
(ws as any).setAdminAuth(process.env.CONVEX_DEPLOY_KEY!);
```

> **Note:** `setAdminAuth` exists on both clients but is not in the published type
> definitions — the `as any` cast is needed. Do not use `setAuth` for deploy keys;
> that method is for user JWT tokens.

See [`examples/authenticated-queue.ts`](examples/authenticated-queue.ts) for a complete setup.

### 3. Publish messages

From any Convex mutation:

```ts
// From a Convex mutation in your app
await ctx.runMutation(api.emailQueue.publish, {
  to: "alice@example.com",
  subject: "Welcome!",
  body: "Thanks for signing up.",
});
```

Or publish a batch:

```ts
await ctx.runMutation(api.emailQueue.publishBatch, {
  messages: [
    { to: "alice@example.com", subject: "Hello", body: "..." },
    { to: "bob@example.com", subject: "Hello", body: "..." },
  ],
});
```

---

## Library Mode

Library mode lets you define the queue table in your own schema, enabling custom indexes and filtered consumption.

### 1. Define the table in your schema

```ts
// convex/schema.ts
import { defineSchema } from "convex/server";
import { messageQueueTable } from "convex-mq/table";
import { v } from "convex/values";

export default defineSchema({
  tasks: messageQueueTable({
    worker: v.string(),
    task: v.string(),
    fileSize: v.number(),
  })
    .index("by_worker", ["status", "worker"])
    .index("by_size", ["status", "fileSize"]),
});
```

`messageQueueTable()` adds the system fields (`status`, `attempts`, `maxAttempts`, `visibilityTimeoutMs`, `claimId`) and a default `by_status` index. Your fields are stored at the top level, so they're indexable.

### 2. Define your queue

```ts
// convex/tasks.ts
import { MessageQueue } from "convex-mq/table";
import { internalQuery, internalMutation } from "./_generated/server";
import { v } from "convex/values";

const taskQueue = new MessageQueue("tasks", {
  worker: v.string(),
  task: v.string(),
  fileSize: v.number(),
});

// Unfiltered — processes all pending messages
export const { peek, claim, ack, nack, publish, publishBatch, reclaimStale } = taskQueue.api(
  internalQuery,
  internalMutation,
);
```

### 2. Add the table to your schema

```ts
// convex/schema.ts
import { defineSchema } from "convex/server";
import { taskQueue } from "./tasks";

export default defineSchema({
  tasks: taskQueue
    .table()
    .index("by_worker", ["status", "worker"])
    .index("by_size", ["status", "fileSize"]),
});
```

`.table()` adds the system fields (`status`, `attempts`, `maxAttempts`, `visibilityTimeoutMs`, `claimId`) and a default `by_status` index. Your fields are stored at the top level, so they're indexable. Chain `.index()` to add custom indexes.

The module path defaults to the table name (`"tasks"` → scheduler calls `"tasks:reclaimStale"`). If your file name differs, pass `modulePath`:

```ts
export const taskQueue = new MessageQueue("tasks", { ... }, {
  modulePath: "myTasks",  // for convex/myTasks.ts
});
```

### 3. Filtered consumption

Define additional exports that consume a subset of messages using a custom index:

```ts
// convex/tasks.ts (continued)

// Filtered — consume only tasks for a specific worker
export const { peek: peekByWorker, claim: claimByWorker } = taskQueue.api(
  internalQuery,
  internalMutation,
  {
    filterArgs: { worker: v.string() },
    filter: (q, { worker }) =>
      q.withIndex("by_worker", (idx) => idx.eq("status", "pending").eq("worker", worker)),
  },
);
```

Consumers pass filter args directly in the options:

```ts
consume(client, internal.tasks, handler, { worker: "worker-1" });
```

### 4. Publish messages

```ts
// From any Convex mutation
await ctx.runMutation(internal.tasks.publish, {
  worker: "worker-1",
  task: "resize-image",
  fileSize: 1024,
});
```

In library mode, message fields are top-level args (not nested under `payload`).

---

## Consuming Messages

### Reactive consumer (recommended)

Uses a WebSocket subscription — processes messages as soon as they're published:

```ts
import { ConvexClient } from "convex/browser";
import { consume } from "convex-mq/client";
import { api } from "../convex/_generated/api.js";

const client = new ConvexClient(process.env.CONVEX_URL!);

const stop = consume(client, api.emailQueue, async (messages) => {
  for (const msg of messages) {
    await sendEmail(msg.payload.to, msg.payload.subject, msg.payload.body);
  }
});

// Graceful shutdown
process.on("SIGINT", () => {
  stop();
  process.exit(0);
});
```

### Polling consumer

For environments without WebSocket support:

```ts
import { ConvexHttpClient } from "convex/browser";
import { consumePolling } from "convex-mq/client";
import { api } from "../convex/_generated/api.js";

const client = new ConvexHttpClient(process.env.CONVEX_URL!);

const controller = consumePolling(
  client,
  api.emailQueue,
  async (messages) => {
    for (const msg of messages) {
      await sendEmail(msg.payload.to, msg.payload.subject, msg.payload.body);
    }
  },
  { pollIntervalMs: 2000 },
);

// Stop: controller.abort();
```

### Manual consumption

For full control, use the raw functions directly:

```ts
const hasPending = await client.query(api.emailQueue.peek, {});

if (hasPending) {
  const messages = await client.mutation(api.emailQueue.claim, { limit: 5 });

  for (const msg of messages) {
    try {
      await processMessage(msg.payload);
      await client.mutation(api.emailQueue.ack, {
        messageId: msg.id,
        claimId: msg.claimId,
      });
    } catch (err) {
      const result = await client.mutation(api.emailQueue.nack, {
        messageId: msg.id,
        claimId: msg.claimId,
        error: err.message,
      });

      if (result) {
        // Retries exhausted — handle dead letter
        console.error("Message failed permanently:", result.payload);
      }
    }
  }
}
```

---

## Authentication

By default, `api()` generates public functions. In production, you'll typically want:

- **Publishing** — requires user authentication (only logged-in users can enqueue messages)
- **Consuming** — requires a [deploy key](https://docs.convex.dev/cli/deploy-key-types) (only your server-side consumers can claim/ack messages)

Use [`internalQuery`/`internalMutation`](https://docs.convex.dev/functions/internal-functions) for deploy-key-only access:

```ts
export const { peek, claim, ack, nack } = emailQueue.api(internalQuery, internalMutation);
```

For publishing with user auth, use [custom function builders](https://docs.convex.dev/auth/functions-auth) from `convex-helpers`:

```ts
// convex/auth.ts
i

[README truncated for prompt length]