message-queue

Getting started

Install#

npm i @dmytromykhailiuk/message-queue

Quick start#

Create a queue once at module scope, register a handler, enqueue messages. The handler is called with one unit at a time and answers with a boolean:

import { createMessageQueue } from "@dmytromykhailiuk/message-queue";

const queue = createMessageQueue<{ url: string }>({
  maxAttempts: 3,
  maxProcessingTime: 10_000,
  onError: (reason, input) => console.error(reason, input),
});

// true consumes the delivery, false (or a throw) retries it.
queue.addHandler(async ({ message, attempt }) => sendOne(message.data));

await queue.addMessage({ data: { url: "/sync/1" } });
await queue.addMessage({ data: { url: "/sync/2" } }, 5000); // enqueued in 5s

The grouped flavour is selected at creation — every message must name its group, and handlers receive whole batches:

const digests = createMessageQueue<AppEvent>({ grouped: true });

digests.addHandler(async ({ messages }) =>
  sendDigest(messages.map((m) => m.data)), // everything queued for the group
);

await digests.addMessage({ data: event, groupId: `user:${userId}` });

Messages are delivered in the order they were enqueued. Add more handlers to process independent units concurrently — see Concurrency.

Two flavours, one model#

The queue does not schedule raw messages — it schedules units of work. A unit is what one handler call receives, and its shape is decided once, when the queue is created:

The flavour is not a per-message choice: a plain queue's addMessage has no groupId parameter (a smuggled one is a TypeError), and a grouped queue rejects any message without a non-empty groupId. Handlers never need to ask "is this a batch?" — the queue's type already answers it.

Units line up FIFO: whoever entered the queue first is delivered first. A unit that fails goes to the back of the line. Three rules follow from the model:

• One unit — one handler call. A group of ten messages is a single delivery to a single handler, not ten.
• Units are independent. Two different groups — or two unrelated plain messages — can be processed by two handlers at the same time.
• Keys share one namespace. Inside one queue the unit id is the groupId (grouped flavour), else the deduplicationId, else the message's own id — use distinct strings for distinct things.

Declare the queue in one module and import it everywhere work is produced or consumed:

import { createMessageQueue } from "@dmytromykhailiuk/message-queue";

export const syncQueue = createMessageQueue<SyncJob>({ maxAttempts: 5 });

// anywhere else in the app
import { syncQueue } from "./queue";
await syncQueue.addMessage({ data: job });

Reference

createMessageQueue#

function createMessageQueue<T>(options: GroupedQueueOptions<T>): GroupedMessageQueue<T>;
function createMessageQueue<T>(options?: QueueOptions<T>): MessageQueue<T>;

Returns a frozen queue with no messages and no handlers. The overload is picked by the grouped option, and it decides every type downstream — addMessage, the handler input, the hooks and onError. Each call is an isolated world — two queues never see each other's messages or workers. Invalid options (maxAttempts or maxProcessingTime that is not a positive number) throw a TypeError immediately.

Options#

The queue object#

Everything below is a stable reference on a frozen object — safe to destructure, impossible to monkey-patch.

addMessage#

// plain queue
addMessage(
  message: { data: T; deduplicationId?: string },
  delayTime?: number,
): Promise<QueueMessage<T>>;

// grouped queue — groupId is required
addMessage(
  message: { data: T; groupId: string; deduplicationId?: string },
  delayTime?: number,
): Promise<GroupedQueueMessage<T>>;

Enqueues data and resolves once the message is actually in the queue — with the enqueued form, which adds two fields the queue assigned:

const message = await queue.addMessage({ data: { url: "/sync" } });

message.id;        // unique id assigned by the queue
message.createdAt; // ISO-8601 timestamp of enqueueing
message.data;      // your payload, unchanged

On a grouped queue the groupId names the batch the message joins — it is required, and an empty string is rejected. On a plain queue the parameter does not exist, and a smuggled one is a runtime TypeError. deduplicationId makes the message replaceable while it waits. All keys are plain strings — build them from your domain (`user:${id}`, `doc:${path}`).

Delayed delivery#

With delayTime the message enters the queue only after that many milliseconds. Until then it is invisible — size() does not count it, deduplication does not see it, and messages enqueued in the meantime go first:

void queue.addMessage({ data: retryJob }, 30_000); // try again in 30s
await queue.addMessage({ data: urgentJob });       // delivered first

// addMessage resolves AFTER the delay — await it only when you want to
// wait for the enqueueing itself.

Handlers#

const unsubscribe = queue.addHandler(async (input) => {
  // …process input…
  return true; // consumed; false (or a throw) retries the unit
});

unsubscribe(); // in-flight work finishes; nothing new is delivered

A handler is a worker: while it is processing one unit, the queue does not hand it another. Handlers can be added before or after messages — a handler registered late simply drains the backlog. Adding the same function twice registers it once; unsubscribing is idempotent.

The handler input#

The input's shape follows the queue's flavour — there is no runtime discriminator to check, because a queue only ever delivers one shape:

// plain queue
interface HandlerInput<T> {
  message: QueueMessage<T>;
  attempt: number;
}

// grouped queue — the whole batch, in arrival order
interface GroupedHandlerInput<T> {
  messages: GroupedQueueMessage<T>[];
  attempt: number;
}

queue.addHandler(async ({ message, attempt }) => sendOne(message));
digests.addHandler(async ({ messages }) => sendBatch(messages));

attempt is 1-based and counts deliveries of the unit: the first delivery is attempt 1, the first retry attempt 2, and so on. It resets when the unit is consumed or dropped.

Concurrency#

The number of handlers is the queue's concurrency. One handler means strictly sequential processing; N handlers mean up to N independent units in flight at once — messages of one group still never split across workers:

const stopAll = Array.from({ length: 4 }, () =>
  queue.addHandler(processJob),
);

// later: wind the pool down
for (const stop of stopAll) stop();

Grouped queues#

A queue created with { grouped: true } batches by groupId — which every message must carry. Messages that share a groupId travel as one unit: the queue delivers every queued message of the group together, as a single batch, to a single handler — in arrival order:

const queue = createMessageQueue<Email>({ grouped: true });

await queue.addMessage({ data: emailA, groupId: "user:42" });
await queue.addMessage({ data: emailB, groupId: "user:42" });

// one delivery: { messages: [emailA, emailB], attempt: 1 }
await queue.addMessage({ data: emailC }); // ✗ rejects — groupId is required

The batch keeps growing while it waits — every addMessage with the same groupId merges in. Success (true) consumes the whole batch; failure sends the whole batch — including anything that joined in the meantime — to the back of the queue for another attempt.

Different groups are different units: with two handlers, "user:42" and "user:7" are processed in parallel, while messages inside each group stay batched.

Locked while processed#

While a batch is in a handler, its group is locked: addMessage for the same group waits until the attempt settles, then merges. A message added mid-flight can therefore never be swallowed by the current batch's success — it always lands in the next batch:

// batch ["a", "b"] is being processed…
void queue.addMessage({ data: "c", groupId: "g" }); // waits at the lock

// …the batch succeeds and is consumed. Only then does "c" enter:
// next delivery is { messages: ["c"], attempt: 1 }

Deduplication#

A message with a deduplicationId occupies one slot in the queue. Enqueue again with the same id while the first is still waiting, and the new payload replaces the old one — same slot, same position in line, one delivery:

await queue.addMessage({ data: draftV1, deduplicationId: "doc:7" });
await queue.addMessage({ data: draftV2, deduplicationId: "doc:7" });
await queue.addMessage({ data: draftV3, deduplicationId: "doc:7" });

queue.size(); // 1 — one slot, latest payload
// the handler receives draftV3, once

Deduplication spans exactly the waiting time. Once the message is consumed (or dropped after maxAttempts), the id is free again — the next addMessage with it starts a fresh unit with attempt: 1.

Inside a grouped queue, deduplication works per batch entry: the newer message replaces the older one in the batch and moves to the batch's end, while other entries keep their places.

Retries & giving up#

An attempt fails when the handler returns false, throws, rejects — or exceeds maxProcessingTime. A failed unit goes to the back of the queue and is redelivered with attempt + 1; other units are not blocked by it:

queue.addHandler(async ({ attempt, ...input }) => {
  console.log(attempt); // 1, then 2, then 3…
  return trySend(input);
});

// delivery order with a failing "a": a#1, b#1, c#1, a#2, a#3, …

With maxAttempts the queue eventually gives up: the unit is dropped after the last failed attempt and onError receives "Max attempts exceeded" together with the exact input of that final delivery:

const queue = createMessageQueue<Job>({
  maxAttempts: 3,
  onError: (reason, input) => {
    if (reason === "Max attempts exceeded") {
      deadLetter.push(input); // your dead-letter strategy
    }
  },
});

Processing timeouts#

maxProcessingTime puts an upper bound on one attempt. When a handler has not settled within the limit, the attempt fails: onError fires with "Max processing time exceeded", the unit rejoins the queue by the usual retry rules, and the worker slot is freed so the queue keeps moving:

const queue = createMessageQueue<Job>({
  maxProcessingTime: 10_000, // an attempt may take at most 10s
  maxAttempts: 3,            // timeouts count as failed attempts
  onError: (reason) => console.warn(reason),
});

Two consequences worth knowing:

• The late result is ignored. The queue cannot cancel your function — a timed-out handler keeps running, but whatever it eventually returns (or throws) changes nothing: the attempt already failed, and a late rejection is swallowed, never an unhandled rejection.
• The worker returns to the pool immediately. The next delivery may reach that handler while its timed-out body is still running. If the handler touches shared state, make it safe to overlap with itself — or don't set maxProcessingTime and keep the strict one-at-a-time guarantee.

Persistence hooks#

An in-memory queue forgets everything on reload — the hooks are its storage seam. Three callbacks cover the lifecycle of a message, and together with onError they are enough to mirror the queue into localStorage, IndexedDB or a server:

Hooks are observers, not middleware: they are called synchronously, they cannot veto or transform anything, and a hook that throws is contained — reported via console.error, never allowed to break the queue. Their argument types follow the queue's flavour, like everything else.

• onMessageAdded waits for the delay. A message enqueued with delayTime is persisted when it actually enters the queue, not at the call.
• Replacement fires it again. A deduplication replacement is a new onMessageAdded with the new message — storage keyed by deduplicationId overwrites naturally.
• Failed attempts fire nothing. onMessageHandled only reports consumption; a retrying unit stays in storage until it is handled or dead-lettered.

A storage recipe#

Key entries by deduplicationId || id — the same key the queue deduplicates by — and the four callbacks keep storage exactly in sync with the queue:

const key = (m: { deduplicationId?: string; id: string }) =>
  m.deduplicationId || m.id;

const queue = createMessageQueue<Job>({
  maxAttempts: 5,

  onQueueCreated: (q) => {
    for (const saved of db.readAll()) {
      void q.addMessage({ data: saved.data, deduplicationId: saved.deduplicationId });
    }
    db.clear(); // onMessageAdded re-persists them under fresh ids
  },

  onMessageAdded: (message) => db.put(key(message), message),
  onMessageHandled: (input) => db.delete(key(input.message)),

  onError: (reason, input) => {
    if (reason === "Max attempts exceeded") {
      db.delete(key(input.message)); // dead-letter instead of retrying forever
    }
  },
});

Patterns#

Autosave that never floods#

Fire an autosave on every keystroke; deduplication collapses the backlog so the server sees at most one in-flight save plus one queued — always the newest:

const queue = createMessageQueue<Doc>();

const autosave = (doc: Doc) =>
  void queue.addMessage({
    data: doc,
    deduplicationId: `save:${doc.id}`, // newer drafts replace queued ones
  });

queue.addHandler(async ({ message }) => api.save(message.data));

Batch per entity#

Group by entity id and each delivery is "everything pending for this entity" — one network call instead of N, while different entities still process in parallel:

const queue = createMessageQueue<AppEvent>({ grouped: true });

const notify = (userId: string, event: AppEvent) =>
  void queue.addMessage({ data: event, groupId: `notify:${userId}` });

queue.addHandler(async ({ messages }) => {
  const events = messages.map((m) => m.data);
  return api.sendDigest(events); // one digest email per user
});

Scheduled retry with backoff#

The built-in retry redelivers as soon as a worker is free. For spaced-out retries, consume the failure yourself and re-enqueue with a delay:

queue.addHandler(async ({ message }) => {
  try {
    await api.send(message.data);
  } catch {
    const wait = 2 ** message.data.tries * 1000; // 1s, 2s, 4s…
    void queue.addMessage(
      { data: { ...message.data, tries: message.data.tries + 1 } },
      wait,
    );
  }
  return true; // always consume — the re-enqueue IS the retry
});

A bounded worker pool#

N wrapped handlers give you "at most N requests in flight" for free — enqueue as fast as you like, the pool drains at its own pace:

const stops = Array.from({ length: 3 }, () =>
  queue.addHandler((input) => upload(input)), // wrapper per worker
);

for (const file of files) void queue.addMessage({ data: file });

TypeScript#

The queue is generic over the payload, and the grouped option picks the flavour end to end — addMessage, the handler input, the hooks and onError all agree, and mixing flavours is a compile error:

interface Job {
  url: string;
  tries: number;
}

const plain = createMessageQueue<Job>();          // MessageQueue<Job>
const message = await plain.addMessage({ data }); // QueueMessage<Job>
plain.addMessage({ data, groupId: "g" });         // ✗ no groupId on a plain queue

const grouped = createMessageQueue<Job>({ grouped: true }); // GroupedMessageQueue<Job>
grouped.addMessage({ data });                     // ✗ groupId is required
grouped.addHandler(({ messages }) => {
  messages;                                       // GroupedQueueMessage<Job>[]
  return true;
});

The queue object is frozen (Object.freeze), so its methods cannot be reassigned — what you export from queue.ts is exactly what every consumer gets.

Exports#

Values

createMessageQueue

Types

MessageQueue · GroupedMessageQueue · Message · GroupedMessage · QueueMessage · GroupedQueueMessage · Handler · GroupedHandler · HandlerInput · GroupedHandlerInput · QueueOptions · GroupedQueueOptions · ErrorReason