Queues

When work shouldn't block the request, push it to a queue. Nwire's queue adapter runs the same wires that the HTTP adapter runs — you don't re-implement the logic for a worker.

Concept

                       handler / action
                              │
        ┌─────────────────────┼─────────────────────┐
        ▼                     ▼                     ▼
    httpKoa adapter       queueInMemory          createCronWire
    /api/orders           "orders.charge"        every 5m
    app.wire(post, ...)   app.wire(queue, ...)   defineCron({...})

The handler is the unit. Adapters route into it.

Two options for the queue runtime

Package	When to use
@nwire/queue (queueInMemory())	Dev, tests, single-process services — process-local queue with delay support
@nwire/bullmq (bullmqAdapter())	Production — Redis-backed, persistent, multi-worker, retry + DLQ baked in

Both implement the same Adapter contract; swap by changing one line at the endpoint.

In-memory queue (dev/test)

pnpm add @nwire/queue

import { createApp } from "@nwire/app";
import { endpoint } from "@nwire/endpoint";
import { queue } from "@nwire/wires/queue";
import { httpKoa } from "@nwire/koa";
import { queueInMemory } from "@nwire/queue";

const app = createApp({ appName: "orders" });

// Same handler bound to both an HTTP route and a queue topic
app.wire(queue("orders.charge"), async (input: { orderId: string }) => {
  // charge the card …
});

const q = queueInMemory();

await endpoint("orders", { port: 3000 })
  .use(httpKoa())
  .use(q)
  .mount(app)
  .run();

// From elsewhere — anywhere holding the q handle:
await q.enqueue("orders.charge", { orderId: "ord-1" });

Production queue (BullMQ)

pnpm add @nwire/bullmq bullmq ioredis

import { bullmqAdapter } from "@nwire/bullmq";

const bull = bullmqAdapter({
  connection: { host: "localhost", port: 6379 },
});

await endpoint("orders", { port: 3000 })
  .use(httpKoa())
  .use(bull)
  .mount(app)
  .run();

await bull.enqueue("orders.charge", { orderId: "ord-1" });

Pushing from an HTTP handler

The simplest pattern: resolve the queue from the container after registering it.

import { definePlugin } from "@nwire/app";

const app = createApp({
  appName: "orders",
  plugins: [
    definePlugin("queue", ({ bind }) => {
      const q = queueInMemory();
      bind("queue", q);
    }),
  ],
});

app.wire(post("/orders"), async (input, ctx) => {
  const order = await placeOrder(input);
  const queue = ctx.resolve<{ enqueue: (name: string, payload: unknown) => Promise<void> }>("queue");
  await queue.enqueue("orders.charge", { orderId: order.id });
  return { $status: 202, body: { orderId: order.id } };
});

Or expose enqueue via the forge dispatcher inside a forge handler: ctx.enqueue(actionName, input) threads the parent envelope so correlationId stays linked across the async boundary.

Wiring the worker

The queue adapter IS the worker. Mount it on the same endpoint as HTTP, or on its own endpoint in a separate process:

// worker.ts — process-only worker, no HTTP
import { endpoint } from "@nwire/endpoint";
import { bullmqAdapter } from "@nwire/bullmq";
import { buildApp } from "./build";

const app = buildApp();
await app.start();

await endpoint("orders-worker", { port: 0 })
  .use(bullmqAdapter({ connection: { host: "localhost", port: 6379 } }))
  .mount(app)
  .run();

The same wires under queue("...") bindings consume jobs from Redis.

Retry + dead-letter

For forge actions, declare retry policy on the action itself:

defineAction({
  name:   "orders.charge",
  input: z.object({ orderId: z.string() }),
  retry:  { max: 3, backoff: "exponential", baseDelayMs: 1000 },
  handler: async ({ input }) => { /* ... */ },
});

Exhausted retries land in the dead-letter sink registered via @nwire/dead-letter. The default in-memory sink is fine for dev; production binds a persistent sink (Mongo, Postgres, BullMQ failed-jobs board).

Where to next

• Multi-transport — running HTTP + queue
  • MCP side-by-side on one endpoint.
• Cron — scheduled actions via cronAdapter.
• Cross-service events — bus-based event flow between Apps.