feat(core, event-bus): Compensate emit event step utility (#13281)

* feat(core, event-bus): Compensate emit event step utility

* tests

* Update changeset to remove integration-tests-modules

Removed integration-tests-modules from changeset.

* revert test script

packages/core/core-flows/src/common/steps/emit-event.ts

@@ -5,6 +5,7 @@ import {
 import { Modules } from "@medusajs/framework/utils"
 import {
   StepExecutionContext,
+  StepResponse,
   createStep,
 } from "@medusajs/framework/workflows-sdk"
 
@@ -85,6 +86,25 @@ export const emitEventStep = createStep(
     }
 
     await eventBus.emit(message)
+
+    return new StepResponse({
+      eventGroupId: context.eventGroupId,
+      eventName: input.eventName,
+    })
   },
-  async (data: void) => {}
+  async (data, context) => {
+    if (!data || !data?.eventGroupId) {
+      return
+    }
+
+    const { container } = context
+
+    const eventBus: IEventBusModuleService = container.resolve(
+      Modules.EVENT_BUS
+    )
+
+    await eventBus.clearGroupedEvents(data!.eventGroupId, {
+      eventNames: [data!.eventName],
+    })
+  }
 )

packages/core/types/src/event-bus/event-bus-module.ts

@@ -3,14 +3,14 @@ import { Message, Subscriber, SubscriberContext } from "./common"
 export interface IEventBusModuleService {
   /**
    * This method emits one or more events. Subscribers listening to the event(s) are executed asynchronously.
-   * 
+   *
    * @param data - The details of the events to emit.
    * @param options - Additional options for the event.
-   * 
+   *
    * @example
-   * await eventModuleService.emit({ 
-   *   name: "user.created", 
-   *   data: { 
+   * await eventModuleService.emit({
+   *   name: "user.created",
+   *   data: {
    *     user_id: "user_123"
    *   }
    * })
@@ -22,12 +22,12 @@ export interface IEventBusModuleService {
 
   /**
    * This method adds a subscriber to an event. It's mainly used internally to register subscribers.
-   * 
+   *
    * @param eventName - The name of the event to subscribe to.
    * @param subscriber - The subscriber function to execute when the event is emitted.
    * @param context - The context of the subscriber.
    * @returns The instance of the Event Module
-   * 
+   *
    * @example
    * eventModuleService.subscribe("user.created", async (data) => {
    *   console.log("User created", data)
@@ -41,12 +41,12 @@ export interface IEventBusModuleService {
 
   /**
    * This method removes a subscriber from an event. It's mainly used internally to unregister subscribers.
-   * 
+   *
    * @param eventName - The name of the event to unsubscribe from.
    * @param subscriber - The subscriber function to remove.
    * @param context - The context of the subscriber.
    * @returns The instance of the Event Module
-   * 
+   *
    * @example
    * eventModuleService.unsubscribe("user.created", async (data) => {
    *   console.log("User created", data)
@@ -61,9 +61,9 @@ export interface IEventBusModuleService {
   /**
    * This method emits all events in the specified group. Grouped events are useful when you have distributed transactions
    * where you need to explicitly group, release and clear events upon lifecycle events of a transaction.
-   * 
+   *
    * @param eventGroupId - The ID of the event group.
-   * 
+   *
    * @example
    * await eventModuleService.releaseGroupedEvents("group_123")
    */
@@ -71,11 +71,19 @@ export interface IEventBusModuleService {
   /**
    * This method removes all events in the specified group. Grouped events are useful when you have distributed transactions
    * where you need to explicitly group, release and clear events upon lifecycle events of a transaction.
-   * 
+   *
    * @param eventGroupId - The ID of the event group.
-   * 
+   * @param options - Additional options for the event.
+   * @param options.eventNames - The names of the events to clear. If not provided, The group will
+   * be entirely cleared.
+   *
    * @example
    * await eventModuleService.clearGroupedEvents("group_123")
    */
-  clearGroupedEvents(eventGroupId: string): Promise<void>
+  clearGroupedEvents(
+    eventGroupId: string,
+    options?: {
+      eventNames?: string[]
+    }
+  ): Promise<void>
 }

packages/core/utils/src/event-bus/index.ts

@@ -38,8 +38,16 @@ export abstract class AbstractEventBusModuleService
   */
   // Given a eventGroupId, all the grouped events will be released
   abstract releaseGroupedEvents(eventGroupId: string): Promise<void>
-  // Given a eventGroupId, all the grouped events will be cleared
-  abstract clearGroupedEvents(eventGroupId: string): Promise<void>
+
+  // Given a eventGroupId, all the grouped events will be cleared unless eventNames are provided
+  // If eventNames are provided, only the events that match the eventNames will be cleared from the
+  // group
+  abstract clearGroupedEvents(
+    eventGroupId: string,
+    options?: {
+      eventNames?: string[]
+    }
+  ): Promise<void>
 
   protected storeSubscribers({
     event,

packages/modules/event-bus-local/integration-tests/__tests__/index.spec.ts

@@ -0,0 +1,141 @@
+import {
+  CommonEvents,
+  composeMessage,
+  Modules,
+} from "@medusajs/framework/utils"
+import { moduleIntegrationTestRunner } from "@medusajs/test-utils"
+import { IEventBusModuleService } from "@medusajs/types"
+
+moduleIntegrationTestRunner<IEventBusModuleService>({
+  moduleName: Modules.EVENT_BUS,
+  testSuite: ({ service: eventBus }) => {
+    describe("Event Bus Local Service", () => {
+      it("should emit an event", async () => {
+        const subscriber = jest.fn()
+        eventBus.subscribe("test", subscriber)
+
+        await eventBus.emit(
+          composeMessage("test", {
+            data: {
+              test: "test",
+            },
+            action: CommonEvents.CREATED,
+            source: "test",
+            object: "test",
+          })
+        )
+
+        expect(subscriber).toHaveBeenCalledWith({
+          data: {
+            test: "test",
+          },
+          metadata: {
+            source: "test",
+            object: "test",
+            action: "created",
+          },
+          name: "test",
+        })
+
+        eventBus.unsubscribe("test", subscriber)
+      })
+
+      it("should release grouped events", async () => {
+        const subscriber = jest.fn()
+        eventBus.subscribe("test", subscriber)
+
+        await eventBus.emit(
+          composeMessage("test", {
+            data: {
+              test: "test",
+            },
+            context: {
+              eventGroupId: "123",
+            },
+            action: CommonEvents.CREATED,
+            source: "test",
+            object: "test",
+          })
+        )
+
+        expect(subscriber).toHaveBeenCalledTimes(0)
+
+        await eventBus.releaseGroupedEvents("123")
+
+        expect(subscriber).toHaveBeenCalledTimes(1)
+
+        expect(subscriber).toHaveBeenCalledWith({
+          data: {
+            test: "test",
+          },
+          metadata: {
+            source: "test",
+            eventGroupId: "123",
+            object: "test",
+            action: "created",
+          },
+          name: "test",
+        })
+
+        eventBus.unsubscribe("test", subscriber)
+      })
+
+      it("should clear grouped events", async () => {
+        const subscriber = jest.fn()
+        eventBus.subscribe("test", subscriber)
+
+        await eventBus.emit(
+          composeMessage("test", {
+            data: {
+              test: "test",
+            },
+            context: {
+              eventGroupId: "123",
+            },
+            action: CommonEvents.CREATED,
+            source: "test",
+            object: "test",
+          })
+        )
+
+        expect(subscriber).toHaveBeenCalledTimes(0)
+
+        await eventBus.clearGroupedEvents("123")
+        await eventBus.releaseGroupedEvents("123")
+
+        expect(subscriber).toHaveBeenCalledTimes(0)
+
+        eventBus.unsubscribe("test", subscriber)
+      })
+
+      it("should clear grouped events with event names", async () => {
+        const subscriber = jest.fn()
+        eventBus.subscribe("test", subscriber)
+
+        await eventBus.emit(
+          composeMessage("test", {
+            data: {
+              test: "test",
+            },
+            context: {
+              eventGroupId: "123",
+            },
+            action: CommonEvents.CREATED,
+            source: "test",
+            object: "test",
+          })
+        )
+
+        await eventBus.clearGroupedEvents("123", {
+          eventNames: ["test"],
+        })
+
+        await eventBus.releaseGroupedEvents("123")
+
+        expect(subscriber).toHaveBeenCalledTimes(0)
+
+        eventBus.unsubscribe("test", subscriber)
+      })
+    })
+  },
+})

packages/modules/event-bus-local/package.json

@@ -33,7 +33,8 @@
   "scripts": {
     "watch": "tsc --build --watch",
     "build": "rimraf dist && tsc --build",
-    "test": "jest --passWithNoTests"
+    "test": "jest --passWithNoTests",
+    "test:integration": "jest --no-cache --maxWorkers=50% --bail --detectOpenHandles --forceExit --logHeapUsage -- integration-tests/__tests__/**/*.spec.ts"
   },
   "dependencies": {
     "ulid": "^2.3.0"

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

@@ -130,8 +130,19 @@ export default class LocalEventBusService extends AbstractEventBusModuleService
     await this.clearGroupedEvents(eventGroupId)
   }
 
-  async clearGroupedEvents(eventGroupId: string) {
-    this.groupedEventsMap_.delete(eventGroupId)
+  async clearGroupedEvents(
+    eventGroupId: string,
+    { eventNames }: { eventNames?: string[] } = {}
+  ) {
+    if (eventNames?.length) {
+      const groupedEvents = this.groupedEventsMap_.get(eventGroupId) || []
+      const eventsToKeep = groupedEvents.filter(
+        (event) => !eventNames!.includes(event.name)
+      )
+      this.groupedEventsMap_.set(eventGroupId, eventsToKeep)
+    } else {
+      this.groupedEventsMap_.delete(eventGroupId)
+    }
   }
 
   subscribe(

packages/modules/event-bus-redis/integration-tests/__tests__/index.spec.ts

@@ -0,0 +1,147 @@
+import {
+  CommonEvents,
+  composeMessage,
+  Modules,
+} from "@medusajs/framework/utils"
+import { moduleIntegrationTestRunner } from "@medusajs/test-utils"
+import { IEventBusModuleService } from "@medusajs/types"
+
+moduleIntegrationTestRunner<IEventBusModuleService>({
+  moduleName: Modules.EVENT_BUS,
+  moduleOptions: {
+    redis: {
+      host: "localhost",
+      port: 6379,
+    },
+  },
+  testSuite: ({ service: eventBus }) => {
+    describe("Event Bus Redis Service", () => {
+      it("should emit an event", async () => {
+        const subscriber = jest.fn()
+        eventBus.subscribe("test", subscriber)
+
+        await eventBus.emit(
+          composeMessage("test", {
+            data: {
+              test: "test",
+            },
+            action: CommonEvents.CREATED,
+            source: "test",
+            object: "test",
+          })
+        )
+
+        expect(subscriber).toHaveBeenCalledWith({
+          data: {
+            test: "test",
+          },
+          metadata: {
+            source: "test",
+            object: "test",
+            action: "created",
+          },
+          name: "test",
+        })
+
+        eventBus.unsubscribe("test", subscriber)
+      })
+
+      it("should release grouped events", async () => {
+        const subscriber = jest.fn()
+        eventBus.subscribe("test", subscriber)
+
+        await eventBus.emit(
+          composeMessage("test", {
+            data: {
+              test: "test",
+            },
+            context: {
+              eventGroupId: "123",
+            },
+            action: CommonEvents.CREATED,
+            source: "test",
+            object: "test",
+          })
+        )
+
+        expect(subscriber).toHaveBeenCalledTimes(0)
+
+        await eventBus.releaseGroupedEvents("123")
+
+        expect(subscriber).toHaveBeenCalledTimes(1)
+
+        expect(subscriber).toHaveBeenCalledWith({
+          data: {
+            test: "test",
+          },
+          metadata: {
+            source: "test",
+            eventGroupId: "123",
+            object: "test",
+            action: "created",
+          },
+          name: "test",
+        })
+
+        eventBus.unsubscribe("test", subscriber)
+      })
+
+      it("should clear grouped events", async () => {
+        const subscriber = jest.fn()
+        eventBus.subscribe("test", subscriber)
+
+        await eventBus.emit(
+          composeMessage("test", {
+            data: {
+              test: "test",
+            },
+            context: {
+              eventGroupId: "123",
+            },
+            action: CommonEvents.CREATED,
+            source: "test",
+            object: "test",
+          })
+        )
+
+        expect(subscriber).toHaveBeenCalledTimes(0)
+
+        await eventBus.clearGroupedEvents("123")
+        await eventBus.releaseGroupedEvents("123")
+
+        expect(subscriber).toHaveBeenCalledTimes(0)
+
+        eventBus.unsubscribe("test", subscriber)
+      })
+
+      it("should clear grouped events with event names", async () => {
+        const subscriber = jest.fn()
+        eventBus.subscribe("test", subscriber)
+
+        await eventBus.emit(
+          composeMessage("test", {
+            data: {
+              test: "test",
+            },
+            context: {
+              eventGroupId: "123",
+            },
+            action: CommonEvents.CREATED,
+            source: "test",
+            object: "test",
+          })
+        )
+
+        await eventBus.clearGroupedEvents("123", {
+          eventNames: ["test"],
+        })
+
+        await eventBus.releaseGroupedEvents("123")
+
+        expect(subscriber).toHaveBeenCalledTimes(0)
+
+        eventBus.unsubscribe("test", subscriber)
+      })
+    })
+  },
+})

packages/modules/event-bus-redis/package.json

@@ -33,7 +33,8 @@
   "scripts": {
     "watch": "tsc --build --watch",
     "build": "rimraf dist && tsc --build",
-    "test": "jest --silent --bail --maxWorkers=50% --forceExit"
+    "test": "jest --silent --bail --maxWorkers=50% --forceExit",
+    "test:integration": "jest --no-cache --maxWorkers=50% --bail --detectOpenHandles --forceExit --logHeapUsage -- integration-tests/__tests__/**/*.spec.ts"
   },
   "dependencies": {
     "bullmq": "5.13.0",

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

@@ -218,11 +218,49 @@ export default class RedisEventBusService extends AbstractEventBusModuleService
     await this.clearGroupedEvents(eventGroupId)
   }
 
-  async clearGroupedEvents(eventGroupId: string) {
+  async clearGroupedEvents(
+    eventGroupId: string,
+    {
+      eventNames,
+    }: {
+      eventNames?: string[]
+    } = {}
+  ) {
     if (!eventGroupId) {
       return
     }
 
+    if (eventNames?.length) {
+      /**
+       * If any event names are provided, we keep all events except the ones that match the event
+       * names. which allow to partially clear an event group.
+       */
+
+      const eventsToKeep = await this.eventBusRedisConnection_
+        .lrange(`staging:${eventGroupId}`, 0, -1)
+        .then((result) => {
+          return result
+            .map((jsonString) => JSON.parse(jsonString))
+            .filter((event) => !eventNames.includes(event.name))
+        })
+
+      // Create a pipeline
+      const pipeline = this.eventBusRedisConnection_.pipeline()
+
+      // Empty the current list
+      pipeline.del(`staging:${eventGroupId}`)
+
+      // Add the remaining events to the list
+      pipeline.rpush(
+        `staging:${eventGroupId}`,
+        ...eventsToKeep.map((event) => JSON.stringify(event))
+      )
+
+      await pipeline.exec()
+
+      return
+    }
+
     await this.eventBusRedisConnection_.unlink(`staging:${eventGroupId}`)
   }