chore(medusa-test-utils): Prevent waiting for event indefinately (#12137)

**What**
Currently the util await for event infinitely, this can lead to chain crashes in the jest tests suites leading to too much noise to investigate proper issues.
We now have a default time out raced against the promise that is configurable to prevent from waiting for an excessive amount of time

.changeset/wild-paws-learn.md

@@ -0,0 +1,5 @@
+---
+"@medusajs/test-utils": patch
+---
+
+chore(medusa-test-utils): Prevent waiting for event indefinately

packages/medusa-test-utils/jest.config.js

@@ -0,0 +1,2 @@
+const defineJestConfig = require("../../define_jest_config")
+module.exports = defineJestConfig({})

packages/medusa-test-utils/src/__tests__/events.spec.ts

@@ -0,0 +1,253 @@
+import { EventEmitter } from "events"
+import { waitSubscribersExecution } from "../events"
+
+// Mock the IEventBusModuleService
+class MockEventBus {
+  public eventEmitter_: EventEmitter
+
+  constructor() {
+    this.eventEmitter_ = new EventEmitter()
+  }
+
+  emit(eventName: string, data?: any) {
+    this.eventEmitter_.emit(eventName, data)
+    return Promise.resolve()
+  }
+}
+
+describe("waitSubscribersExecution", () => {
+  let eventBus: MockEventBus
+  const TEST_EVENT = "test-event"
+
+  beforeEach(() => {
+    eventBus = new MockEventBus()
+    jest.useFakeTimers()
+  })
+
+  afterEach(() => {
+    jest.useRealTimers()
+  })
+
+  describe("with no existing listeners", () => {
+    it("should resolve when event is fired before timeout", async () => {
+      const waitPromise = waitSubscribersExecution(TEST_EVENT, eventBus as any)
+      setTimeout(() => eventBus.emit(TEST_EVENT, "test-data"), 100)
+
+      jest.advanceTimersByTime(100)
+
+      await expect(waitPromise).resolves.toEqual(["test-data"])
+    })
+
+    it("should reject when timeout is reached before event is fired", async () => {
+      const waitPromise = waitSubscribersExecution(TEST_EVENT, eventBus as any)
+
+      jest.advanceTimersByTime(5100)
+
+      await expect(waitPromise).rejects.toThrow(
+        `Timeout of 5000ms exceeded while waiting for event "${TEST_EVENT}"`
+      )
+    })
+
+    it("should respect custom timeout value", async () => {
+      const customTimeout = 2000
+      const waitPromise = waitSubscribersExecution(
+        TEST_EVENT,
+        eventBus as any,
+        {
+          timeout: customTimeout,
+        }
+      )
+
+      jest.advanceTimersByTime(customTimeout + 100)
+
+      await expect(waitPromise).rejects.toThrow(
+        `Timeout of ${customTimeout}ms exceeded while waiting for event "${TEST_EVENT}"`
+      )
+    })
+  })
+
+  describe("with existing listeners", () => {
+    it("should resolve when all listeners complete successfully", async () => {
+      const listener = jest.fn().mockImplementation(() => {
+        return new Promise((resolve) => setTimeout(resolve, 200))
+      })
+
+      eventBus.eventEmitter_.on(TEST_EVENT, listener)
+
+      // Setup the promise
+      const waitPromise = waitSubscribersExecution(TEST_EVENT, eventBus as any)
+
+      // Emit the event
+      eventBus.emit(TEST_EVENT, "test-data")
+
+      // Fast forward to let the listener complete
+      jest.advanceTimersByTime(300)
+
+      // Await the promise - it should resolve
+      await expect(waitPromise).resolves.not.toThrow()
+
+      // Ensure the listener was called
+      expect(listener).toHaveBeenCalledWith("test-data")
+    })
+
+    it("should reject when a listener throws an error", async () => {
+      const errorMessage = "Test error from listener"
+
+      const listener = jest.fn().mockImplementation(() => {
+        return Promise.reject(new Error(errorMessage))
+      })
+
+      eventBus.eventEmitter_.on(TEST_EVENT, listener)
+
+      const waitPromise = waitSubscribersExecution(TEST_EVENT, eventBus as any)
+
+      eventBus.emit(TEST_EVENT, "test-data")
+
+      await expect(waitPromise).rejects.toThrow(errorMessage)
+    })
+
+    it("should reject with timeout if event is not fired in time", async () => {
+      const listener = jest.fn()
+      eventBus.eventEmitter_.on(TEST_EVENT, listener)
+
+      const waitPromise = waitSubscribersExecution(
+        TEST_EVENT,
+        eventBus as any,
+        {
+          timeout: 1000,
+        }
+      )
+
+      jest.advanceTimersByTime(1100)
+
+      await expect(waitPromise).rejects.toThrow(
+        `Timeout of 1000ms exceeded while waiting for event "${TEST_EVENT}"`
+      )
+
+      expect(listener).not.toHaveBeenCalled()
+    })
+  })
+
+  describe("with multiple listeners", () => {
+    it("should resolve when all listeners complete", async () => {
+      const listener1 = jest.fn().mockImplementation(() => {
+        return new Promise((resolve) => setTimeout(resolve, 100))
+      })
+
+      const listener2 = jest.fn().mockImplementation(() => {
+        return new Promise((resolve) => setTimeout(resolve, 200))
+      })
+
+      const listener3 = jest.fn().mockImplementation(() => {
+        return new Promise((resolve) => setTimeout(resolve, 300))
+      })
+
+      eventBus.eventEmitter_.on(TEST_EVENT, listener1)
+      eventBus.eventEmitter_.on(TEST_EVENT, listener2)
+      eventBus.eventEmitter_.on(TEST_EVENT, listener3)
+
+      const waitPromise = waitSubscribersExecution(TEST_EVENT, eventBus as any)
+
+      eventBus.emit(TEST_EVENT, "test-data")
+
+      jest.advanceTimersByTime(400)
+
+      await expect(waitPromise).resolves.not.toThrow()
+
+      expect(listener1).toHaveBeenCalledWith("test-data")
+      expect(listener2).toHaveBeenCalledWith("test-data")
+      expect(listener3).toHaveBeenCalledWith("test-data")
+    })
+
+    it("should reject if any listener throws an error", async () => {
+      const errorMessage = "Test error from listener 2"
+
+      const listener1 = jest.fn().mockImplementation(() => {
+        return Promise.resolve()
+      })
+
+      const listener2 = jest.fn().mockImplementation(() => {
+        return Promise.reject(new Error(errorMessage))
+      })
+
+      const listener3 = jest.fn().mockImplementation(() => {
+        return Promise.resolve()
+      })
+
+      eventBus.eventEmitter_.on(TEST_EVENT, listener1)
+      eventBus.eventEmitter_.on(TEST_EVENT, listener2)
+      eventBus.eventEmitter_.on(TEST_EVENT, listener3)
+
+      const waitPromise = waitSubscribersExecution(TEST_EVENT, eventBus as any)
+
+      eventBus.emit(TEST_EVENT, "test-data")
+
+      await expect(waitPromise).rejects.toThrow(errorMessage)
+    })
+  })
+
+  describe("cleanup", () => {
+    it("should restore original listeners after completion", async () => {
+      const originalListener = jest.fn()
+      eventBus.eventEmitter_.on(TEST_EVENT, originalListener)
+
+      const listenersBefore =
+        eventBus.eventEmitter_.listeners(TEST_EVENT).length
+
+      const waitPromise = waitSubscribersExecution(TEST_EVENT, eventBus as any)
+
+      eventBus.emit(TEST_EVENT, "test-data")
+
+      await waitPromise
+
+      const listenersAfter = eventBus.eventEmitter_.listeners(TEST_EVENT).length
+      expect(listenersAfter).toBe(listenersBefore)
+
+      eventBus.emit(TEST_EVENT, "after-test-data")
+      expect(originalListener).toHaveBeenCalledWith("after-test-data")
+    })
+
+    it("should restore original listeners after timeout", async () => {
+      const originalListener = jest.fn()
+      eventBus.eventEmitter_.on(TEST_EVENT, originalListener)
+
+      const listenersBefore =
+        eventBus.eventEmitter_.listeners(TEST_EVENT).length
+
+      const waitPromise = waitSubscribersExecution(
+        TEST_EVENT,
+        eventBus as any,
+        {
+          timeout: 500,
+        }
+      )
+
+      jest.advanceTimersByTime(600)
+
+      await waitPromise.catch(() => {})
+
+      const listenersAfter = eventBus.eventEmitter_.listeners(TEST_EVENT).length
+      expect(listenersAfter).toBe(listenersBefore)
+
+      eventBus.emit(TEST_EVENT, "after-timeout-data")
+      expect(originalListener).toHaveBeenCalledWith("after-timeout-data")
+    })
+  })
+
+  describe("timeout clearing", () => {
+    it("should clear timeout when events fire", async () => {
+      const clearTimeoutSpy = jest.spyOn(global, "clearTimeout")
+
+      const waitPromise = waitSubscribersExecution(TEST_EVENT, eventBus as any)
+
+      eventBus.emit(TEST_EVENT, "test-data")
+
+      await waitPromise
+
+      expect(clearTimeoutSpy).toHaveBeenCalled()
+      expect(clearTimeoutSpy).toHaveBeenCalled()
+
+      clearTimeoutSpy.mockRestore()
+    })
+  })
+})

packages/medusa-test-utils/src/events.ts

@@ -4,11 +4,28 @@ import { EventEmitter } from "events"
 // Allows you to wait for all subscribers to execute for a given event. Only works with the local event bus.
 export const waitSubscribersExecution = (
   eventName: string,
-  eventBus: IEventBusModuleService
+  eventBus: IEventBusModuleService,
+  {
+    timeout = 5000,
+  }: {
+    timeout?: number
+  } = {}
 ) => {
   const eventEmitter: EventEmitter = (eventBus as any).eventEmitter_
   const subscriberPromises: Promise<any>[] = []
   const originalListeners = eventEmitter.listeners(eventName)
+  let timeoutId: NodeJS.Timeout | null = null
+
+  // Create a promise that rejects after the timeout
+  const timeoutPromise = new Promise((_, reject) => {
+    timeoutId = setTimeout(() => {
+      reject(
+        new Error(
+          `Timeout of ${timeout}ms exceeded while waiting for event "${eventName}"`
+        )
+      )
+    }, timeout)
+  })
 
   // If there are no existing listeners, resolve once the event happens. Otherwise, wrap the existing subscribers in a promise and resolve once they are done.
   if (!eventEmitter.listeners(eventName).length) {
@@ -31,17 +48,34 @@ export const waitSubscribersExecution = (
       subscriberPromises.push(promise)
 
       const newListener = async (...args2) => {
-        return await listener.apply(eventBus, args2).then(ok).catch(nok)
+        try {
+          const res = await listener.apply(eventBus, args2)
+
+          ok(res)
+
+          return res
+        } catch (error) {
+          nok(error)
+        }
       }
 
       eventEmitter.on(eventName, newListener)
     })
   }
 
-  return Promise.all(subscriberPromises).finally(() => {
+  const subscribersPromise = Promise.all(subscriberPromises).finally(() => {
+    // Clear the timeout since events have been fired and handled
+    if (timeoutId !== null) {
+      clearTimeout(timeoutId)
+    }
+
+    // Restore original event listeners
     eventEmitter.removeAllListeners(eventName)
     originalListeners.forEach((listener) => {
       eventEmitter.on(eventName, listener as (...args: any) => void)
     })
   })
+
+  // Race between the subscribers and the timeout
+  return Promise.race([subscribersPromise, timeoutPromise])
 }