Skip to content
Moku
@moku-labs/worker pages

queues

Cloudflare Queues from one plugin: send from a request, consume with a handler declared in config.

On this page

What it does

queues gives a Worker both sides of a queue. send and sendBatch produce messages. consume is what your queue export calls to deliver a batch to the instance's onMessage.

How it works

The consumer is config: each instance can declare onMessage. consume picks the instance in this order:

  1. One instance. It always matches.
  2. By name. Its name equals batch.queue, or batch.queue starts with name-. That is the stage suffix: orders-dev finds orders.
  3. The default. With no match, the default instance gets the batch. With no default, consume throws.

Then it walks the messages one at a time: it awaits onMessage(message, env), then emits queue:message. An instance with no onMessage takes the batch and does nothing.

Config

pluginConfigs.queues, keyed by your logical name. The default is {}.

OptionTypeDefaultDescription
[key].namestringrequiredThe Cloudflare queue. A stage other than production suffixes it at deploy.
[key].bindingstringrequiredThe producer key on env.
[key].onMessage(message: Message, env: WorkerEnv) => Promise<void>noneThe consumer. With it, deploy registers this Worker as a consumer of the queue.
[key].maxBatchTimeoutnumberCloudflare'sSeconds to wait to fill a batch, 0 to 60. Written as max_batch_timeout, so it needs onMessage.
[key].defaultbooleannoneMarks the default when there are several.

API

Mounted at app.queues, reachable with require(queuesPlugin) in a handler.

MethodSignatureDescription
send(env, body: unknown) => Promise<void>One message.
sendBatch(env, bodies: unknown[]) => Promise<void>One message per element.
use(key: string) => QueueProducerApisend and sendBatch on another queue.
consume(batch: MessageBatch, env, ctx: ExecutionContext) => Promise<void>The delivery. Resolves after every message settled.
deployManifest() => { kind: "queue"; name: string; binding: string; consumer?: boolean; maxBatchTimeout?: number }[]Build-time metadata. consumer is true with onMessage.

Events

Emits

EventPayloadWhen
queue:message{ queue: string; messageId: string }After onMessage settled for a message.

A hook on it needs depends: [queuesPlugin].

Listens to

None.

Depends on

  • bindings: resolves the producer on every send.

Example

src/app.ts
import type { WorkerEnv } from "@moku-labs/worker";
import { createApp, endpoint, kvPlugin, queuesPlugin } from "@moku-labs/worker";

type Order = { id: string };

export const app = createApp({
  plugins: [queuesPlugin, kvPlugin],
  pluginConfigs: {
    kv: { orders: { name: "orders-state", binding: "ORDERS_STATE" } },
    queues: {
      orders: {
        name: "orders",
        binding: "ORDERS",
        maxBatchTimeout: 5,
        onMessage: async (message: { body: unknown }, env: WorkerEnv) => {
          const order = message.body as Order;
          await app.kv.put(env, order.id, "received");
        }
      }
    },
    server: {
      endpoints: [
        endpoint("/orders").post(async ({ request, env, require }) => {
          await require(queuesPlugin).send(env, await request.json());
          return new Response(null, { status: 202 });
        })
      ]
    }
  }
});
  1. A throw here rejects consume.
src/worker.ts
import { app } from "./app";

export default {
  fetch: (request: Request, env: Record<string, unknown>, ctx: ExecutionContext) =>
    app.server.handle(request, env, ctx),
  queue: (batch: MessageBatch, env: Record<string, unknown>, ctx: ExecutionContext) =>
    app.queues.consume(batch, env, ctx)
} satisfies ExportedHandler;
  • Wire the queue export. onMessage runs only through consume. The deploy plugin registers the consumer, your entry has to deliver.
  • A throw stops the batch. consume rejects at the first failing message, and the rest of the loop does not run. Cloudflare then applies its retry policy to what was not acknowledged. Call message.ack() for finished work if a partial retry matters.
  • Messages run one by one. A slow onMessage holds the batch.
  • send takes the body only. There is no delaySeconds or contentType. Take the raw queue with app.bindings.require for those.
  • A name that prefixes another. With orders and orders-eu configured, the first declared match wins for orders-eu. Use names where neither is a prefix of the other.