@dmytromykhailiuk/message-queue

In-memory message queue with FIFO ordering, deduplication, retries, processing timeouts, delayed delivery and persistence hooks — SQS-flavoured semantics for the browser and Node.

Full documentation: open Docs in a browser — every option and behaviour, with examples, a table of contents and cross-links. This README is the short form.

Built for the work you want done eventually, in order, exactly once per payload: syncing edits to a server, sending notifications, draining user actions against an API. You enqueue messages; handlers (workers) pull them one unit at a time. Failures are retried until maxAttempts; slow handlers are cut off by maxProcessingTime; lifecycle hooks let you mirror the queue into localStorage or a DB so nothing is lost on restart.

A queue's flavour is fixed at creation:

• Plain queue — createMessageQueue(). Single-message deliveries. groupId does not exist here, at the type level and at runtime.
• Grouped queue — createMessageQueue({ grouped: true }). Every message must carry a groupId; all queued messages of a group are delivered together, as one batch, to one handler.

Install

npm i @dmytromykhailiuk/message-queue

One runtime dependency: @dmytromykhailiuk/execution-blocker — it serializes queue mutation against delivery.

Quick start

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

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

// A handler returns true to consume the delivery, false (or throws) to retry.
queue.addHandler(async ({ message, attempt }) => sendOne(message.data));

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

The grouped flavour — groupId required, handlers receive whole batches:

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

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

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

API

const queue = createMessageQueue<T>(options?);

queue.addMessage(message, delayTime?); // resolves with the enqueued message (id, createdAt)
queue.addHandler(handler);             // returns unsubscribe()
queue.size();                          // units waiting for a handler
queue.inFlight();                      // units being processed right now

message — plain queue: { data: T, deduplicationId?: string }; grouped queue: { data: T, groupId: string, deduplicationId?: string }.

handler — plain queue: ({ message, attempt }) => boolean | Promise<boolean>; grouped queue: ({ messages, attempt }) => boolean | Promise<boolean>.

options

| Option | Meaning | | ------------------- | ------------------------------------------------------------------------------------------- | | grouped | true selects the grouped flavour. Fixed at creation; changes types everywhere below. | | maxAttempts | Give up after this many failed attempts ("Max attempts exceeded"). Omit → retry forever. | | maxProcessingTime | Fail the attempt when the handler exceeds this many ms ("Max processing time exceeded"). | | onError | (reason, input) => void — called when the queue times out an attempt or gives up. | | onQueueCreated | Called synchronously with the queue right after creation — rehydrate persisted messages. | | onMessageAdded | Called when a message actually enters the queue — persist it. | | onMessageHandled | Called when a delivery is consumed — remove it from storage. |

Persistence hooks

The three hooks are the queue's storage seam: onMessageAdded writes, onMessageHandled deletes, onQueueCreated restores. Together with onError (for dead-lettering) they cover the whole lifecycle:

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
    }
  },
});

Hooks are observers: they are called synchronously, their errors are contained (reported via console.error, never thrown into the queue), and a deduplication replacement fires onMessageAdded again with the new message — keyed storage overwrites naturally.

Grouped queues

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

await queue.addMessage({ data: a, groupId: "user:42" });
await queue.addMessage({ data: b, groupId: "user:42" });
// one delivery: { messages: [a, b], attempt: 1 }

The batch keeps growing while it waits. During processing the group is locked — a message added mid-flight is never swallowed by the current batch's success; it lands in the next one. Different groups are independent units: with several handlers they are processed in parallel.

Deduplication

await queue.addMessage({ data: v1, deduplicationId: "doc:7" });
await queue.addMessage({ data: v2, deduplicationId: "doc:7" }); // replaces v1
// one delivery with v2 — and it kept v1's position in the queue

Deduplication only spans the waiting time: once the message is consumed, the same deduplicationId starts a fresh unit. In a grouped queue it replaces the batch entry.

Retries, timeouts, giving up

• false / a thrown error / a rejection → the unit returns to the back of the queue, attempt + 1.
• With maxProcessingTime, an attempt that outlives the limit fails (onError fires); the handler goes back to the worker pool and its late result is ignored.
• With maxAttempts, the unit is dropped after the last failure and onError receives "Max attempts exceeded". onMessageHandled does not fire for dropped units.

⚠️ One rule: never await queue.addMessage(...) inside a handler for the same unit it is currently processing — the queue locks a unit while it is being handled, so that await would wait for the handler itself. Fire and forget (void queue.addMessage(...)) instead.

TypeScript

The flavour picks the types end to end — addMessage, handler input, hooks and onError all agree, and mixing flavours is a compile error:

const plain = createMessageQueue<Job>();            // MessageQueue<Job>
plain.addMessage({ data, groupId: "g" });           // ✗ compile error

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

Exported types: MessageQueue, GroupedMessageQueue, Message, GroupedMessage, QueueMessage, GroupedQueueMessage, Handler, GroupedHandler, HandlerInput, GroupedHandlerInput, QueueOptions, GroupedQueueOptions, ErrorReason. The queue object is frozen — its methods cannot be reassigned.

License

MIT