Run a worker

A queue needs something to drain it. better-push gives you two shapes: a long-lived worker, and a bounded drain you can call from a cron job or a serverless function. Both run the same code and both are safe to run at once.

This page assumes you have async delivery configured.

The long-lived worker

import { push } from "./src/push";

const stop = push.startWorker();
console.log("worker started");

for (const signal of ["SIGTERM", "SIGINT"] as const) {
  process.on(signal, () => {
    // Awaiting stop() is the point: it stops claiming and resolves once the
    // jobs already in flight have settled.
    void stop().then(() => process.exit(0));
  });
}

startWorker() returns synchronously and loops in the background: claim due jobs, send them, settle them, repeat. When a claim comes back full it goes straight round again; when it comes back short it sleeps pollIntervalMs.

It never dies from a bad job. A handler that throws is treated as a retryable failure, and a claim that fails because the database blipped is logged and retried on the next tick.

Tuning one worker

push.startWorker({
  batchSize: 25,      // jobs claimed per poll
  pollIntervalMs: 500, // sleep between polls that found nothing
});

Both default to the values you gave the backend, which default to 10 and 1000ms on dbQueue. On bullmq there is no poll loop, so pollIntervalMs has nothing to do and batchSize sets the worker's concurrency instead.

The worker itself is backend-independent: it generates the worker id, writes the bp_worker heartbeat the studio reads, sweeps overdue digest windows, and schedules rollups and pruning on a 15 second tick - all of that lives above the queue adapter on purpose, so a BullMQ deployment gets studio worker rows for free.

Running it as a container

The worker is a plain Node process with no HTTP port, so it wants no health check and no domain. In a Dockerfile that already builds your app, add a second entrypoint and override the start command for the worker service:

# one image, two roles
CMD ["node", "server.js"]              # web service
# worker service start command: node worker.js

Give the worker the same DATABASE_URL and the same provider credentials as the web service - it is the process that actually talks to APNs, FCM, and the browser push services.

Graceful shutdown

Platforms send SIGTERM before replacing a container. Without a handler, the process dies mid-send: the push may or may not have reached the service, and the delivery row is left at queued.

stop() fixes both halves. It stops claiming immediately and resolves only once the in-flight batch has settled, so the delivery rows are written before the process exits. A job that was claimed but never settled - because the container was killed outright - is not lost either: after visibilityTimeoutMs another worker reclaims it.

Multiple workers

Run as many as you like. Claims use FOR UPDATE SKIP LOCKED, so concurrent workers never block each other and never receive the same row. Scaling the worker service to three replicas needs no configuration and no coordination.

The same guarantee is what makes a crashed worker recoverable rather than fatal: its claim ages out and someone else takes the job.

Cron and serverless

If you cannot run a process that never returns, drain on a schedule instead. runPending() does bounded work and resolves with what it did:

const stats = await push.runPending({ maxJobs: 100, maxMs: 20_000 });
// { claimed: 12, completed: 11, retried: 1, deadLettered: 0, budgetExhausted: false }

Both budgets are optional; either one stopping the drain sets budgetExhausted when work is still due, which is your signal to invoke again immediately rather than waiting for the next tick.

The HTTP entrypoint

For schedulers that can only make an HTTP request, mount the built-in route by configuring a secret:

export const push = betterPush({
  // ...
  queue: dbQueue(),
  runPendingSecret: process.env.BETTER_PUSH_RUN_PENDING_SECRET,
});

curl -X POST https://your.app/api/push/_internal/run-pending \
  -H "authorization: Bearer $BETTER_PUSH_RUN_PENDING_SECRET" \
  -H "content-type: application/json" \
  -d '{"maxJobs": 100, "maxMs": 20000}'

• Without runPendingSecret the route answers 404: it does not exist unless you enable it.
• It is the only route outside the session gate - a scheduler has no user to be - and authenticates with the shared secret alone, compared in constant time.
• A wrong or missing secret is 401; anything but POST is 405.
• The JSON body is optional and validated; { maxJobs?, maxMs? } only.
• Success is 200 with the same stats object runPending() returns, including a digests block:

{
  "claimed": 12, "completed": 12, "retried": 0, "deadLettered": 0,
  "budgetExhausted": false,
  "digests": { "flushed": 2, "rearmed": 0, "failed": 0, "remaining": 0 }
}

Digest windows are flushed before jobs are drained, so the notifications they produce are delivered by the same call rather than waiting for the next tick. That also means the route is useful with no queue at all: configure runPendingSecret alongside a type that declares a digest and cron becomes the only trigger a window needs.

Use a long random value, keep it out of your client bundle, and rotate it like any other credential.

Mixing the two

A long-lived worker and a cron drain can run against the same queue at the same time - they claim through the same statement. A common shape is one worker for normal throughput plus an hourly runPending() as a safety net, so a queue never sits still because a worker died quietly.

Worker options