feat: make it possible to use subscribers in v2 (#6731)

packages/event-bus-redis/package.json

@@ -38,7 +38,7 @@
     "@medusajs/modules-sdk": "^1.12.8",
     "@medusajs/utils": "^1.11.5",
     "awilix": "^8.0.0",
-    "bullmq": "^3.5.6",
+    "bullmq": "^5.4.2",
     "ioredis": "^5.2.5"
   }
 }

packages/event-bus-redis/src/services/__tests__/event-bus.js

@@ -296,7 +296,7 @@ describe("RedisEventBusService", () => {
               completedSubscriberIds: ["1"],
             },
             attemptsMade: 2,
-            update: (data) => data,
+            updateData: (data) => data,
             opts: { attempts: 3 },
           })
           .catch((err) => void 0)

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

@@ -197,7 +197,7 @@ export default class RedisEventBusService extends AbstractEventBusModuleService
 
       job.data.completedSubscriberIds = updatedCompletedSubscribers
 
-      await job.update(job.data)
+      await job.updateData(job.data)
 
       const errorMessage = `One or more subscribers of ${eventName} failed. Retrying...`
 

packages/medusa/package.json

@@ -60,7 +60,7 @@
     "awilix": "^8.0.0",
     "body-parser": "^1.19.0",
     "boxen": "^5.0.1",
-    "bullmq": "^3.5.6",
+    "bullmq": "^5.4.2",
     "chokidar": "^3.4.2",
     "class-transformer": "^0.5.1",
     "class-validator": "^0.14.0",

packages/medusa/src/loaders/helpers/subscribers/index.ts

@@ -1,3 +1,4 @@
+import { ModuleRegistrationName } from "@medusajs/modules-sdk"
 import { MedusaContainer, Subscriber } from "@medusajs/types"
 import { kebabCase } from "@medusajs/utils"
 import { readdir } from "fs/promises"
@@ -15,6 +16,7 @@ type SubscriberModule<T> = {
 }
 
 export class SubscriberLoader {
+  protected isV2_: boolean
   protected container_: MedusaContainer
   protected pluginOptions_: Record<string, unknown>
   protected activityId_: string
@@ -32,12 +34,14 @@ export class SubscriberLoader {
     rootDir: string,
     container: MedusaContainer,
     options: Record<string, unknown> = {},
-    activityId: string
+    activityId: string,
+    isV2: boolean = false
   ) {
     this.rootDir_ = rootDir
     this.pluginOptions_ = options
     this.container_ = container
     this.activityId_ = activityId
+    this.isV2_ = isV2
   }
 
   private validateSubscriber(
@@ -181,8 +185,10 @@ export class SubscriberLoader {
     config: SubscriberConfig
     handler: SubscriberHandler<T>
   }) {
-    const eventBusService: EventBusService =
-      this.container_.resolve("eventBusService")
+    const resName = this.isV2_
+      ? ModuleRegistrationName.EVENT_BUS
+      : "eventBusService"
+    const eventBusService: EventBusService = this.container_.resolve(resName)
 
     const { event } = config
 

packages/medusa/src/loaders/load-medusa-project-apis.ts

@@ -177,7 +177,8 @@ async function registerSubscribers(
     path.join(pluginDetails.resolve, "subscribers"),
     container,
     pluginDetails.options,
-    activityId
+    activityId,
+    true
   ).load()
 }