fix(medusa): EventBusService.emit using Redis mock (#3491)

* Fix eventBusService.emit using redis mock

* revert gitignore

* enqueuer

* unit test add redis_url

* fix test

.changeset/empty-dragons-invite.md

@@ -0,0 +1,5 @@
+---
+"@medusajs/medusa": patch
+---
+
+Fix eventBus.emit using redis mock

.gitignore

@@ -20,3 +20,4 @@ www/**/.yarn/*
 .idea
 .turbo
 build/**
+**/dist

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

@@ -24,11 +24,18 @@ describe("EventBusService", () => {
         find: () => Promise.resolve([]),
       })
 
-      eventBus = new EventBusService({
+      eventBus = new EventBusService(
+        {
           manager: MockManager,
           stagedJobRepository,
           logger: loggerMock,
-      })
+        },
+        {
+          projectConfig: {
+            redis_url: "localhost",
+          },
+        }
+      )
     })
 
     afterAll(async () => {
@@ -49,10 +56,17 @@ describe("EventBusService", () => {
     beforeEach(() => {
       jest.resetAllMocks()
 
-      eventBus = new EventBusService({
+      eventBus = new EventBusService(
+        {
           manager: MockManager,
           logger: loggerMock,
-      })
+        },
+        {
+          projectConfig: {
+            redis_url: "localhost",
+          },
+        }
+      )
     })
 
     afterAll(async () => {
@@ -135,11 +149,18 @@ describe("EventBusService", () => {
           create: (data) => data,
         })
 
-        eventBus = new EventBusService({
+        eventBus = new EventBusService(
+          {
             logger: loggerMock,
             manager: mockManager,
             stagedJobRepository,
-        })
+          },
+          {
+            projectConfig: {
+              redis_url: "localhost",
+            },
+          }
+        )
 
         eventBus.queue_.addBulk.mockImplementationOnce(() => "hi")
       })
@@ -195,11 +216,18 @@ describe("EventBusService", () => {
           create: (data) => data,
         })
 
-        eventBus = new EventBusService({
+        eventBus = new EventBusService(
+          {
             logger: loggerMock,
             manager: mockManager,
             stagedJobRepository,
-        })
+          },
+          {
+            projectConfig: {
+              redis_url: "localhost",
+            },
+          }
+        )
 
         eventBus.queue_.addBulk.mockImplementationOnce(() => "hi")
       })
@@ -285,7 +313,10 @@ describe("EventBusService", () => {
             stagedJobRepository,
           },
           {
-            projectConfig: { event_options: { removeOnComplete: 10 } },
+            projectConfig: {
+              event_options: { removeOnComplete: 10 },
+              redis_url: "localhost",
+            },
           }
         )
 
@@ -323,11 +354,18 @@ describe("EventBusService", () => {
           create: (data) => data,
         })
 
-        eventBus = new EventBusService({
+        eventBus = new EventBusService(
+          {
             logger: loggerMock,
             manager: mockManager,
             stagedJobRepository,
-        })
+          },
+          {
+            projectConfig: {
+              redis_url: "localhost",
+            },
+          }
+        )
 
         eventBus.queue_.addBulk.mockImplementationOnce(() => "hi")
 
@@ -370,7 +408,10 @@ describe("EventBusService", () => {
             stagedJobRepository,
           },
           {
-            projectConfig: { event_options: { removeOnComplete: 10 } },
+            projectConfig: {
+              event_options: { removeOnComplete: 10 },
+              redis_url: "localhost",
+            },
           }
         )
 
@@ -418,14 +459,11 @@ describe("EventBusService", () => {
           find: () => Promise.resolve([]),
         })
 
-        eventBus = new EventBusService(
+        eventBus = new EventBusService({
-          {
           manager: MockManager,
           stagedJobRepository,
           logger: loggerMock,
-          },
+        })
-          {}
-        )
         eventBus.subscribe("eventName", () => Promise.resolve("hi"))
         result = await eventBus.worker_({
           data: { eventName: "eventName", data: {} },

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

@@ -308,8 +308,10 @@ export default class EventBusService {
       return (!isBulkEmit ? stagedJobs[0] : stagedJobs) as unknown as TResult
     }
 
+    if (this.config_?.projectConfig?.redis_url) {
       await this.queue_.addBulk(events)
     }
+  }
 
   startEnqueuer(): void {
     this.shouldEnqueuerRun = true
@@ -346,9 +348,11 @@ export default class EventBusService {
         }
       })
 
+      if (this.config_?.projectConfig?.redis_url) {
         await this.queue_.addBulk(eventsData).then(async () => {
           return await stagedJobRepo.delete({ id: In(jobs.map((j) => j.id)) })
         })
+      }
 
       await sleep(3000)
     }