chore(event-bus, workflow-engine): Enable more granualar queues configuration (medusajs#14201)

Summary

  This PR adds BullMQ queue and worker configuration options to the workflow-engine-redis module, bringing feature parity with the event-bus-redis module. It also introduces per-queue
  configuration options for fine-grained control over the three internal queues (main, job, and cleaner).

  Key changes:
  - Added per-queue BullMQ configuration options (mainQueueOptions, jobQueueOptions, cleanerQueueOptions and their worker counterparts) with shared defaults
  - Unified Redis option naming across modules: deprecated url → redisUrl, options → redisOptions (with backward compatibility)
  - Moved configuration resolution to the loader and registered options in the DI container
  - Added comprehensive JSDoc documentation for all configuration options
  - Added unit tests for option merging and queue/worker configuration

  Configuration Example

```ts
  // Simple configuration - same options for all queues
  {
    redisUrl: "redis://localhost:6379",
    queueOptions: { defaultJobOptions: { removeOnComplete: 1000 } },
    workerOptions: { concurrency: 10 }
  }
```

```ts
  // Advanced configuration - per-queue overrides
  {
    redisUrl: "redis://localhost:6379",
    workerOptions: { concurrency: 10 },        // shared default
    jobWorkerOptions: { concurrency: 5 },      // override for scheduled workflows
    cleanerWorkerOptions: { concurrency: 1 }   // override for cleanup (low priority)
  }
```

Author: adrien2p

.changeset/slimy-avocados-film.md

@@ -0,0 +1,6 @@
+---
+"@medusajs/event-bus-redis": patch
+"@medusajs/workflow-engine-redis": patch
+---
+
+chore(): Enable more granualar queue configuration

packages/modules/event-bus-redis/src/loaders/index.ts

@@ -1,5 +1,5 @@
-import { LoaderOptions } from "@medusajs/framework/types"
 import { asValue } from "@medusajs/framework/awilix"
+import { LoaderOptions } from "@medusajs/framework/types"
 import Redis from "ioredis"
 import { EOL } from "os"
 import { EventBusRedisModuleOptions } from "../types"
@@ -9,7 +9,14 @@ export default async ({
   logger,
   options,
 }: LoaderOptions): Promise<void> => {
-  const { redisUrl, redisOptions } = options as EventBusRedisModuleOptions
+  const {
+    redisUrl,
+    redisOptions,
+    queueName,
+    queueOptions,
+    workerOptions,
+    jobOptions,
+  } = options as EventBusRedisModuleOptions
 
   if (!redisUrl) {
     throw Error(
@@ -39,5 +46,9 @@ export default async ({
 
   container.register({
     eventBusRedisConnection: asValue(connection),
+    eventBusRedisQueueName: asValue(queueName ?? "events-queue"),
+    eventBusRedisQueueOptions: asValue(queueOptions ?? {}),
+    eventBusRedisWorkerOptions: asValue(workerOptions ?? {}),
+    eventBusRedisJobOptions: asValue(jobOptions ?? {}),
   })
 }

packages/modules/event-bus-redis/src/services/__tests__/event-bus.ts

@@ -21,12 +21,17 @@ const redisMock = {
   unlink: () => jest.fn(),
 } as unknown as Redis
 
-const simpleModuleOptions = { redisUrl: "test-url" }
 const moduleDeps = {
   logger: loggerMock,
   eventBusRedisConnection: redisMock,
+  eventBusRedisQueueName: "events-queue",
+  eventBusRedisQueueOptions: {},
+  eventBusRedisWorkerOptions: {},
+  eventBusRedisJobOptions: {},
 }
 
+const moduleDeclaration = { scope: "internal" } as any
+
 describe("RedisEventBusService", () => {
   let eventBus: RedisEventBusService
   let queue
@@ -36,9 +41,7 @@ describe("RedisEventBusService", () => {
     beforeEach(async () => {
       jest.clearAllMocks()
 
-      eventBus = new RedisEventBusService(moduleDeps, simpleModuleOptions, {
-        scope: "internal",
-      })
+      eventBus = new RedisEventBusService(moduleDeps, {}, moduleDeclaration)
     })
 
     it("Creates a queue + worker", () => {
@@ -62,9 +65,7 @@ describe("RedisEventBusService", () => {
 
     it("Throws on isolated module declaration", () => {
       try {
-        eventBus = new RedisEventBusService(moduleDeps, simpleModuleOptions, {
-          scope: "internal",
-        })
+        eventBus = new RedisEventBusService(moduleDeps, {}, moduleDeclaration)
       } catch (error) {
         expect(error.message).toEqual(
           "At the moment this module can only be used with shared resources"
@@ -78,9 +79,7 @@ describe("RedisEventBusService", () => {
       beforeEach(async () => {
         jest.clearAllMocks()
 
-        eventBus = new RedisEventBusService(moduleDeps, simpleModuleOptions, {
-          scope: "internal",
-        })
+        eventBus = new RedisEventBusService(moduleDeps, {}, moduleDeclaration)
 
         queue = (eventBus as any).queue_
         queue.addBulk = jest.fn()
@@ -139,17 +138,15 @@ describe("RedisEventBusService", () => {
 
       it("should add job to queue with module job options", async () => {
         eventBus = new RedisEventBusService(
-          moduleDeps,
           {
-            ...simpleModuleOptions,
-            jobOptions: {
+            ...moduleDeps,
+            eventBusRedisJobOptions: {
               removeOnComplete: { age: 5 },
               attempts: 7,
             },
           },
-          {
-            scope: "internal",
-          }
+          {},
+          moduleDeclaration
         )
 
         queue = (eventBus as any).queue_
@@ -186,16 +183,14 @@ describe("RedisEventBusService", () => {
 
       it("should add job to queue with default, local, and global options merged", async () => {
         eventBus = new RedisEventBusService(
-          moduleDeps,
           {
-            ...simpleModuleOptions,
-            jobOptions: {
+            ...moduleDeps,
+            eventBusRedisJobOptions: {
               removeOnComplete: 5,
             },
           },
-          {
-            scope: "internal",
-          }
+          {},
+          moduleDeclaration
         )
 
         queue = (eventBus as any).queue_
@@ -340,9 +335,7 @@ describe("RedisEventBusService", () => {
       beforeEach(async () => {
         jest.clearAllMocks()
 
-        eventBus = new RedisEventBusService(moduleDeps, simpleModuleOptions, {
-          scope: "internal",
-        })
+        eventBus = new RedisEventBusService(moduleDeps, {}, moduleDeclaration)
 
         queue = (eventBus as any).queue_
         queue.addBulk = jest.fn()
@@ -485,9 +478,7 @@ describe("RedisEventBusService", () => {
       beforeEach(async () => {
         jest.clearAllMocks()
 
-        eventBus = new RedisEventBusService(moduleDeps, simpleModuleOptions, {
-          scope: "internal",
-        })
+        eventBus = new RedisEventBusService(moduleDeps, {}, moduleDeclaration)
       })
 
       it("should process a simple event with no options", async () => {

packages/modules/event-bus-redis/src/services/event-bus-redis.ts

@@ -9,13 +9,28 @@ import {
   isPresent,
   promiseAll,
 } from "@medusajs/framework/utils"
-import { BulkJobOptions, Queue, Worker } from "bullmq"
+import {
+  BulkJobOptions,
+  Queue,
+  QueueOptions,
+  Worker,
+  WorkerOptions,
+} from "bullmq"
 import { Redis } from "ioredis"
-import { BullJob, EventBusRedisModuleOptions, Options } from "../types"
+import {
+  BullJob,
+  EmitOptions,
+  EventBusRedisModuleOptions,
+  Options,
+} from "../types"
 
 type InjectedDependencies = {
   logger: Logger
   eventBusRedisConnection: Redis
+  eventBusRedisQueueName: string
+  eventBusRedisQueueOptions: Omit<QueueOptions, "connection">
+  eventBusRedisWorkerOptions: Omit<WorkerOptions, "connection">
+  eventBusRedisJobOptions: EmitOptions
 }
 
 type IORedisEventType<T = unknown> = {
@@ -31,46 +46,53 @@ type IORedisEventType<T = unknown> = {
 // eslint-disable-next-line max-len
 export default class RedisEventBusService extends AbstractEventBusModuleService {
   protected readonly logger_: Logger
-  protected readonly moduleOptions_: EventBusRedisModuleOptions
-  // eslint-disable-next-line max-len
-  protected readonly moduleDeclaration_: InternalModuleDeclaration
   protected readonly eventBusRedisConnection_: Redis
 
+  protected readonly queueName_: string
+  protected readonly queueOptions_: Omit<QueueOptions, "connection">
+  protected readonly workerOptions_: Omit<WorkerOptions, "connection">
+  protected readonly jobOptions_: EmitOptions
+
   protected queue_: Queue
   protected bullWorker_: Worker
 
   constructor(
-    { logger, eventBusRedisConnection }: InjectedDependencies,
-    moduleOptions: EventBusRedisModuleOptions = {},
-    moduleDeclaration: InternalModuleDeclaration
+    {
+      logger,
+      eventBusRedisConnection,
+      eventBusRedisQueueName,
+      eventBusRedisQueueOptions,
+      eventBusRedisWorkerOptions,
+      eventBusRedisJobOptions,
+    }: InjectedDependencies,
+    _moduleOptions: EventBusRedisModuleOptions = {},
+    _moduleDeclaration: InternalModuleDeclaration
   ) {
     // @ts-ignore
-    // eslint-disable-next-line prefer-rest-params
     super(...arguments)
 
     this.eventBusRedisConnection_ = eventBusRedisConnection
-
-    this.moduleOptions_ = moduleOptions
     this.logger_ = logger
 
-    this.queue_ = new Queue(moduleOptions.queueName ?? `events-queue`, {
+    this.queueName_ = eventBusRedisQueueName ?? "events-queue"
+    this.queueOptions_ = eventBusRedisQueueOptions ?? {}
+    this.workerOptions_ = eventBusRedisWorkerOptions ?? {}
+    this.jobOptions_ = eventBusRedisJobOptions ?? {}
+
+    this.queue_ = new Queue(this.queueName_, {
       prefix: `${this.constructor.name}`,
-      ...(moduleOptions.queueOptions ?? {}),
+      ...this.queueOptions_,
       connection: eventBusRedisConnection,
     })
 
     // Register our worker to handle emit calls
     if (this.isWorkerMode) {
-      this.bullWorker_ = new Worker(
-        moduleOptions.queueName ?? "events-queue",
-        this.worker_,
-        {
-          prefix: `${this.constructor.name}`,
-          ...(moduleOptions.workerOptions ?? {}),
-          connection: eventBusRedisConnection,
-          autorun: false,
-        }
-      )
+      this.bullWorker_ = new Worker(this.queueName_, this.worker_, {
+        prefix: `${this.constructor.name}`,
+        ...this.workerOptions_,
+        connection: eventBusRedisConnection,
+        autorun: false,
+      })
     }
   }
 
@@ -97,7 +119,7 @@ export default class RedisEventBusService extends AbstractEventBusModuleService
       removeOnComplete: true,
       attempts: 1,
       // global options
-      ...(this.moduleOptions_.jobOptions ?? {}),
+      ...this.jobOptions_,
       ...options,
     }
 

packages/modules/event-bus-redis/src/types/index.ts

@@ -26,12 +26,31 @@ export type BullJob<T> = {
 export type EmitOptions = JobsOptions
 
 export type EventBusRedisModuleOptions = {
+  /**
+   * Queue name for the event bus
+   */
   queueName?: string
-  queueOptions?: QueueOptions
 
-  workerOptions?: WorkerOptions
+  /**
+   * Options for BullMQ Queue instance
+   * @see https://api.docs.bullmq.io/interfaces/v5.QueueOptions.html
+   */
+  queueOptions?: Omit<QueueOptions, "connection">
+
+  /**
+   * Options for BullMQ Worker instance
+   * @see https://api.docs.bullmq.io/interfaces/v5.WorkerOptions.html
+   */
+  workerOptions?: Omit<WorkerOptions, "connection">
 
+  /**
+   * Redis connection string
+   */
   redisUrl?: string
+
+  /**
+   * Redis client options
+   */
   redisOptions?: RedisOptions
 
   /**