feat: add queue auto-discovery support for BullMQ meta keys (#134)

Author: Copilot

lib/queue-factory.ts

@@ -9,10 +9,37 @@ import { Integration } from "./interfaces/integration";
 
 const chalk = require("chalk");
 
-const queueNameRegExp = new RegExp("(.*):(.*):id");
 const maxCount = 150000;
 const maxTime = 40000;
 
+const parseQueueKey = (key: string) => {
+  const suffixSeparator = key.lastIndexOf(":");
+  if (suffixSeparator === -1) {
+    return;
+  }
+
+  const keySuffix = key.slice(suffixSeparator + 1);
+  if (keySuffix !== "id" && keySuffix !== "meta") {
+    return;
+  }
+
+  const prefixAndQueueName = key.slice(0, suffixSeparator);
+  const queueNameSeparator = prefixAndQueueName.lastIndexOf(":");
+
+  if (queueNameSeparator === -1) {
+    return;
+  }
+
+  const prefix = prefixAndQueueName.slice(0, queueNameSeparator);
+  const name = prefixAndQueueName.slice(queueNameSeparator + 1);
+
+  if (!prefix || !name) {
+    return;
+  }
+
+  return { prefix, name };
+};
+
 export type RedisConnection = Redis | Cluster;
 
 // We keep a redis client that we can reuse for all the queues.
@@ -28,9 +55,9 @@ export interface FoundQueue {
 
 const scanForQueues = async (node: Redis | Cluster, startTime: number) => {
   let cursor = "0";
-  const keys = [];
+  const keys = new Set<string>();
   do {
-    const [nextCursor, scannedKeys] = await node.scan(
+    const [nextCursor, scannedIdKeys] = await node.scan(
       cursor,
       "MATCH",
       "*:*:id",
@@ -39,61 +66,78 @@ const scanForQueues = async (node: Redis | Cluster, startTime: number) => {
       "TYPE",
       "string"
     );
+    const [, scannedMetaKeys] = await node.scan(
+      cursor,
+      "MATCH",
+      "*:*:meta",
+      "COUNT",
+      maxCount,
+      "TYPE",
+      "hash"
+    );
     cursor = nextCursor;
 
-    keys.push(...scannedKeys);
+    scannedIdKeys.forEach((key) => keys.add(key));
+    scannedMetaKeys.forEach((key) => keys.add(key.replace(/:meta$/, ":id")));
   } while (Date.now() - startTime < maxTime && cursor !== "0");
 
-  return keys;
+  return [...keys];
 };
 
 const getQueueKeys = async (client: Redis | Cluster, queueNames?: string[]) => {
   let nodes = "nodes" in client ? client.nodes("master") : [client];
-  let keys = [];
+  let keys: string[] = [];
   const startTime = Date.now();
   const foundQueues = new Set<string>();
+  const queueKeys = queueNames?.map((queueName) => {
+    // Separate queue name from prefix
+    let [prefix, name] = queueName.split(":");
+    if (!name) {
+      name = prefix;
+      prefix = "bull";
+    }
+
+    // If the queue name includes a prefix use that, otherwise use the default prefix "bull"
+    return `${prefix}:${name}:id`;
+  });
 
   for await (const node of nodes) {
     // If we have proposed queue names, lets check if they exist (including prefix)
     // Basically checking if there is a id key for the queue (prefix:name:id)
-    if (queueNames) {
-      const queueKeys = queueNames.map((queueName) => {
-        // Separate queue name from prefix
-        let [prefix, name] = queueName.split(":");
-        if (!name) {
-          name = prefix;
-          prefix = "bull";
+    if (queueKeys) {
+      for (const key of queueKeys) {
+        if (foundQueues.has(key)) {
+          continue;
         }
 
-        // If the queue name includes a prefix use that, otherwise use the default prefix "bull"
-        return `${prefix}:${name}:id`;
-      });
-
-      for (const key of queueKeys) {
-        const exists = await node.exists(key);
+        const metaKey = key.replace(/:id$/, ":meta");
+        const exists = await node.exists(key, metaKey);
         if (exists) {
           foundQueues.add(key);
         }
       }
-      keys.push(...foundQueues);
-
-      // Warn for missing queues
-      for (const key of queueKeys) {
-        if (!foundQueues.has(key)) {
-          // Extract queue name from key
-          const match = queueNameRegExp.exec(key);
-          console.log(
-            chalk.yellow("Redis:") +
-              chalk.red(
-                ` Queue "${match[1]}:${match[2]}" not found in Redis. Skipping...`
-              )
-          );
-        }
-      }
     } else {
       keys.push(...(await scanForQueues(node, startTime)));
     }
   }
+
+  if (queueKeys) {
+    keys.push(...foundQueues);
+
+    // Warn for missing queues
+    for (const key of queueKeys) {
+      if (!foundQueues.has(key)) {
+        // Extract queue name from key
+        const queue = parseQueueKey(key);
+        const queueLabel = queue ? `${queue.prefix}:${queue.name}` : key;
+        console.log(
+          chalk.yellow("Redis:") +
+            chalk.red(` Queue "${queueLabel}" not found in Redis. Skipping...`)
+        );
+      }
+    }
+  }
+
   return keys;
 };
 
@@ -111,11 +155,11 @@ export async function getConnectionQueues(
       const queues = await Promise.all(
         keys
           .map(function (key) {
-            var match = queueNameRegExp.exec(key);
-            if (match) {
+            const queue = parseQueueKey(key);
+            if (queue) {
               return {
-                prefix: match[1],
-                name: match[2],
+                prefix: queue.prefix,
+                name: queue.name,
                 type: "bull", // default to bull
               };
             }

package.json

@@ -48,6 +48,7 @@
   },
   "scripts": {
     "build": "tsc",
+    "pretest": "yarn build",
     "test": "jest",
     "start": "node app.js",
     "prepare": "npm run build",

tests/queue-factory.spec.js

@@ -0,0 +1,118 @@
+const { getConnectionQueues } = require("../dist/queue-factory");
+
+describe("queue auto discovery", () => {
+  const createMockRedisClient = (idKeys, metaKeys) => ({
+    scan: jest
+      .fn()
+      .mockResolvedValueOnce(["0", idKeys])
+      .mockResolvedValueOnce(["0", metaKeys]),
+    exists: jest.fn().mockResolvedValue(1),
+    hget: jest.fn().mockImplementation((_key, field) => {
+      if (field === "version") {
+        return Promise.resolve("bullmq:5.47.0");
+      }
+      return Promise.resolve(null);
+    }),
+  });
+
+  const createMockQueueLookupClient = (exists) => ({
+    exists: jest.fn().mockImplementation(exists),
+    hget: jest.fn().mockImplementation((_key, field) => {
+      if (field === "version") {
+        return Promise.resolve("bullmq:5.47.0");
+      }
+      return Promise.resolve(null);
+    }),
+  });
+
+  it("discovers queue when only meta key exists", async () => {
+    const client = createMockRedisClient([], ["bull:emails:meta"]);
+
+    const queues = await getConnectionQueues(undefined, undefined, undefined, client);
+
+    expect(queues).toHaveLength(1);
+    expect(queues[0]).toMatchObject({
+      prefix: "bull",
+      name: "emails",
+      type: "bullmq",
+      majorVersion: 5,
+      version: "5.47.0",
+    });
+  });
+
+  it("does not duplicate queue discovered by id and meta keys", async () => {
+    const client = createMockRedisClient(
+      ["bull:notifications:id"],
+      ["bull:notifications:meta"]
+    );
+
+    const queues = await getConnectionQueues(undefined, undefined, undefined, client);
+
+    expect(queues).toHaveLength(1);
+    expect(queues[0]).toMatchObject({
+      prefix: "bull",
+      name: "notifications",
+    });
+    expect(client.exists).toHaveBeenCalledTimes(1);
+  });
+
+  it("accepts provided queue names when only the meta key exists", async () => {
+    const client = createMockQueueLookupClient((...keys) =>
+      Promise.resolve(keys.includes("bull:emails:meta") ? 1 : 0)
+    );
+    const consoleSpy = jest.spyOn(console, "log").mockImplementation(() => {});
+
+    const queues = await getConnectionQueues(
+      undefined,
+      undefined,
+      ["emails"],
+      client
+    );
+
+    expect(queues).toHaveLength(1);
+    expect(queues[0]).toMatchObject({
+      prefix: "bull",
+      name: "emails",
+      type: "bullmq",
+      majorVersion: 5,
+      version: "5.47.0",
+    });
+    expect(client.exists).toHaveBeenCalledWith(
+      "bull:emails:id",
+      "bull:emails:meta"
+    );
+    expect(consoleSpy).not.toHaveBeenCalled();
+
+    consoleSpy.mockRestore();
+  });
+
+  it("does not duplicate provided queue names across cluster nodes", async () => {
+    const node = {
+      exists: jest.fn().mockResolvedValue(1),
+    };
+    const client = {
+      nodes: jest.fn().mockReturnValue([node, node]),
+      exists: jest.fn().mockResolvedValue(1),
+      hget: jest.fn().mockImplementation((_key, field) => {
+        if (field === "version") {
+          return Promise.resolve("bullmq:5.47.0");
+        }
+        return Promise.resolve(null);
+      }),
+    };
+
+    const queues = await getConnectionQueues(
+      undefined,
+      undefined,
+      ["emails"],
+      client
+    );
+
+    expect(queues).toHaveLength(1);
+    expect(queues[0]).toMatchObject({
+      prefix: "bull",
+      name: "emails",
+    });
+    expect(node.exists).toHaveBeenCalledTimes(1);
+  });
+});