I’m working with BullMQ (v5+) and need to implement a worker to process messages for different chats, with the following requirements:

Rate Limit:

• Each queue should have a rate limit of 1 job per 3 seconds, but rate limits should be specific to each chat / group, not global.

Behavior:

• The worker should process jobs in the order they were added.
• Jobs should be processed with the rate limit in place.
• Each chat / group should have a separate processing order with its own rate limit.

The code below consumes an INCREDIBLE amount of memory (0.5 gb+). I can only have one copy of the application and trying to find a way to optimize the part with JS Maps multiple queues and workers. Seems like there should be another simpler way to implement a chat queue.

const { Queue, Worker } = require('bullmq');
const Redis = require('ioredis');
const redisConnection = {
  host: 'localhost',
  port: 6379,
  maxRetriesPerRequest: null,
};
const redis = new Redis(redisConnection);
const createQueueName = (group) => `chat_messages__${group}`;
class ChatMessageWorker {
  constructor() {
    this.queues = new Map();
    this.workers = new Map();
  }
  createWorkerForGroup(group) {
    const queueName = createQueueName(group);
    const queue = new Queue(queueName, {
      connection: redis,
      defaultJobOptions: {
        removeOnComplete: true,
        removeOnFail: true,
      },
    });
    const worker = new Worker(
      queueName,
      async (job) => {
        const processingTime = new Date().toLocaleTimeString();
        console.log(`>>> Processing job ${job.id} \
for group ${group} at ${processingTime} - message sent to chat!`);
      },
      {
        connection: redis,
        limiter: {
          max: 1,
          duration: 3000,
        },
      }
    );
    worker.on('error', (error) => {
      console.error(`Worker for group ${group} encountered an error:`, error);
    });
    this.queues.set(group, queue);
    this.workers.set(group, worker);
    return queue;
  }
  async addJobToGroup(group, data) {
    const queue = this.queues.get(group) || this.createWorkerForGroup(group);
    const addTime = new Date().toLocaleTimeString();
    const job = await queue.add(group, {
      ...data,
      addedAt: addTime,
    });
    console.log(`Job added to group ${group} at ${addTime}`);
    return job;
  }
  async close() {
    for (const worker of this.workers.values()) {
      await worker.close();
    }
    for (const queue of this.queues.values()) {
      await queue.close();
    }
  }
}

// Example usage
async function runChatMessageExample() {
  const chatMessageWorker = new ChatMessageWorker();
  const chatIds = ['chat1', 'chat2', 'chat3'];

  // Add 5 messages to each chat group
  for (const chatId of chatIds) {
    for (let i = 1; i <= 5; i++) {
      await chatMessageWorker.addJobToGroup(chatId, {
        message: `Message ${i} for ${chatId}`,
      });
    }
  }

  // Optional: Keep the process running to allow workers to process jobs
  await new Promise(resolve => setTimeout(resolve, 30000));

  // Close workers and queues
  await chatMessageWorker.close();
}
runChatMessageExample().catch(console.error);