@therms/future-queue

FutureQueue

FutureQueue is a lightweight scheduling service built on Node.js and MongoDB that allows you to enqueue messages for future processing. It’s ideal for scenarios where you need to defer tasks—such as sending reminder emails or processing deferred jobs—while also handling retries, cancellations, and failures via a dead-letter queue.

Note: FutureQueue requires Node.js 20+ and a MongoDB instance.

Table of Contents

• Features
• Requirements
• Installation
• Usage
  • Producer Example
  • Consumer Example
• API Reference
  • Producer
  • Consumer
• Use Cases
• Testing
• Contributing
• License

Features

• Schedule Future Messages: Enqueue messages to be processed at a specified future time.
• Multiple Consumers: Process messages concurrently with distributed consumer instances.
• Retry Mechanism: Automatically retry processing failed messages.
• Dead Letter Queue: Move messages that exceed retry limits to a dedicated dead-letter queue.
• Update & Cancellation: Retrieve, update, or cancel scheduled messages using unique keys.
• Progress Tracking: Use helper functions to update and monitor the progress of message processing.
• Message Locking: Prevent duplicate processing by locking messages during handling.
• Idle Queue Notifications: Optionally receive notifications when the queue is idle.

Requirements

• Node.js: Version 20 or later.
• MongoDB: A running MongoDB instance (local or remote).

Installation

Install FutureQueue via npm:

npm install future-queue

Usage

FutureQueue provides two main classes:

• Producer: For adding, updating, retrieving, and canceling messages.
• Consumer: For processing messages when they become ready.

Producer Example

The Producer is used to schedule messages. Each message is defined by a unique key and contains custom data. You can specify when the message should be processed, set an expiration, or configure retry options.

import { Producer } from 'futurequeue';

const queueName = 'email-reminders';
const producer = new Producer(queueName, {
  mongo: {
    uri: 'mongodb://localhost:27017/your-db-name', // Or pass a connected MongoClient instance
  },
});

// Schedule a reminder email to be sent in 3 days
const threeDaysFromNow = new Date();
threeDaysFromNow.setDate(threeDaysFromNow.getDate() + 3);

await producer.add(
  {
    key: 'reminder-email-user-123',
    data: { email: '[e-mail]', subject: 'Reminder', body: 'Your subscription expires soon!' },
  },
  {
    readyTime: threeDaysFromNow,
    retries: 2, // Maximum retry attempts before moving to dead-letter queue
  },
);

// Retrieve the scheduled message by its key
const message = await producer.getByKey('reminder-email-user-123');
console.log('Retrieved message:', message);

// Update message data if needed
await producer.updateByKey('reminder-email-user-123', {
  data: { email: '[e-mail]', subject: 'Updated Reminder', body: 'Please check your subscription!' },
});

// Cancel the scheduled message
await producer.cancelByKey('reminder-email-user-123');

// Gracefully shut down the producer
await producer.shutdown();

Consumer Example

The Consumer listens for messages that are ready to be processed. You define a handler function to process each message. The handler receives helper functions to extend the processing lock and update progress.

import { Consumer } from 'futurequeue';

const queueName = 'email-reminders';
const consumer = new Consumer(queueName, {
  mongo: {
    uri: 'mongodb://localhost:27017/your-db-name',
  },
});

// Define the message processing handler
consumer.processMessages(async (message, { extendMsgHandlingLock, updateMsgHandlerProgress }) => {
  // Simulate processing by updating progress
  await updateMsgHandlerProgress({ percentComplete: 50 });
  
  // Process the message (e.g., send an email)
  console.log(`Processing message with key: ${message.key}`, message.data);
  
  // Optionally extend the processing lock if processing takes longer than expected
  // await extendMsgHandlingLock();
  
  // Mark the processing as complete
  await updateMsgHandlerProgress({ percentComplete: 100 });
});

// Optionally, register an idle callback
consumer.processMessages(
  async (message) => {
    console.log(`Processed: ${message.key}`);
  },
  {
    onQueueIdle: () => {
      console.log('The queue is currently idle.');
    },
  },
);

// When your application is shutting down, gracefully close the consumer
await consumer.shutdown();

API Reference

Producer

new Producer(queueName: string, options: ProducerOptions)

• queueName: Unique name for your queue.
• options: Configuration options, including MongoDB connection settings.

Methods

add(message: QueueMessage | QueueMessage[], options?: { readyTime?: Date; expireTime?: Date; retries?: number; })

Adds one or more messages to the queue.

• readyTime (optional): When the message should be available for processing.
• expireTime (optional): When the message expires.
• retries (optional): Number of retry attempts allowed.

getByKey(key: string)

Retrieves a scheduled message by its unique key.

cancelByKey(key: string)

Cancels (deletes) a scheduled message from the queue.

updateByKey(key: string, update: Partial<QueueMessage>)

Updates properties of a scheduled message.

totalReadyMessages()

Returns the count of messages that are ready to be processed immediately.

totalFutureMessages()

Returns the count of messages scheduled for future processing.

shutdown()

Gracefully shuts down the producer and releases any held resources.

Consumer

new Consumer(queueName: string, options: ConsumerOptions)

• queueName: The name of the queue to process.
• options: Configuration options, including MongoDB settings, logging, and message-handling configurations.

Methods

processMessages(handler: (message: ReadyQueueMessage, helpers?: MessageProcessingHelpers) => Promise<void>, options?: { onQueueIdle?: () => void; })

Registers a message processing handler.

The handler receives:

• message: The message object.
• helpers: An object containing:
  • extendMsgHandlingLock(): Function to extend the lock on the message.
  • updateMsgHandlerProgress(progress: any): Function to update the progress of message processing.
• onQueueIdle (optional): Callback invoked when there are no messages left to process.

totalFutureMessages()

Returns the count of messages still scheduled for future processing.

shutdown()

Gracefully shuts down the consumer.

Use Cases

Scheduled Email Reminders

Schedule reminder emails to be sent at a future date. For example, schedule an email to be sent 3 days before a user's subscription expires.

Deferred Task Processing

Enqueue background tasks—such as data processing, notifications, or report generation—to run during off-peak hours.

Retry and Error Handling

Automatically retry tasks that fail during processing. If a task fails after the configured number of retries, it is moved to a dead-letter queue for further investigation.

Distributed Workload Processing

Scale your application by deploying multiple consumer instances. The locking mechanism ensures that each message is processed only once, even when multiple consumers are polling the queue.