Workers & processing

Declaring a worker#

import { Worker } from '@nextmq/sdk'

export const summarizeWorker = new Worker('ai', async (job) => {
  await job.updateProgress(20)
  const summary = await llm.summarize(job.data.docId)
  return { summary }
}, {
  concurrency: 5,
  limiter: { max: 100, duration: 1000 }, // protect your model provider's quota
})

Mounting the route#

createNextMQHandler returns GET and POST handlers that verify signatures, dispatch jobs to the right worker, and serve a /health path. It must run on the Node.js runtime.

import { createNextMQHandler } from '@nextmq/sdk/next'
import { summarizeWorker } from '@/jobs/ai'

export const runtime = 'nodejs'

export const { GET, POST } = createNextMQHandler({
  workers: [summarizeWorker],
})

Registration#

Registration publishes one deterministic webhook URL per queue. The handler registers its workers, but if producers can enqueue before the route is first hit, register explicitly from your bootstrap or producer path:

import { ensureWorkersRegistered, getWorkerRegistrationStatuses } from '@nextmq/sdk/next'
import { summarizeWorker } from '@/jobs/ai'

await ensureWorkersRegistered([summarizeWorker])

const statuses = await getWorkerRegistrationStatuses([summarizeWorker])

Liveness is inferred from webhook delivery, not an in-process heartbeat. Re-registration is idempotent: a new deploy can re-assert or update the canonical callback URL, and the live BullMQ worker restarts only when execution parameters change. Webhook secrets are provider-owned and are never sent during worker registration.

Worker options#

Most apps only need concurrency, rate limiting, and timeout:

For long-running or stalled-job tuning, lockDuration, lockRenewTime, stalledInterval, maxStalledCount, and drainDelay are also accepted — most apps never set them. Bounds are in the SDK reference.

Signalling job outcome#

NextMQ maps your processor's behavior to job state:

• Return a value → the job is completed with that value.
• Throw → the job failed; retries and backoff apply.
• Throw UnrecoverableJobError → fail immediately, skipping remaining attempts.
• Throw RateLimitError → rate-limit and re-queue without burning an attempt.

import { Worker, UnrecoverableJobError, RateLimitError } from '@nextmq/sdk'

export const aiWorker = new Worker('ai', async (job) => {
  if (!job.data.docId) throw new UnrecoverableJobError('missing docId')

  const res = await llm.summarize(job.data.docId)
  if (res.status === 429) throw new RateLimitError(res.retryAfterMs ?? 1000)

  return { summary: res.summary }
})

Non-autorun workers

Pass { autorun: false } when a worker should be mounted in the handler but not registered by default health checks. Register it explicitly with includeNonAutorun.

const backfillWorker = new Worker('backfills', processor, { autorun: false })

await ensureWorkersRegistered([backfillWorker], {
  includeNonAutorun: true,
})

Pause & resume

Pause or resume processing for a queue:

await aiQueue.pause()
await aiQueue.resume()