chore: ability to group events on redis event bus (#7655)

* chore: ability to group events on redis event bus

* chore: fix tests

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

Co-authored-by: Adrien de Peretti <adrien.deperetti@gmail.com>

* chore: change shape of input and body data

* chore: fix builds

* chore: address comments

* chore: fix unit test

---------

Co-authored-by: Adrien de Peretti <adrien.deperetti@gmail.com>

integration-tests/modules/__tests__/notification/admin/notification.spec.ts

@@ -165,7 +165,8 @@ medusaIntegrationTestRunner({
           )
           const logSpy = jest.spyOn(logger, "info")
 
-          await eventBus.emit("order.created", {
+          await eventBus.emit({
+            eventName: "order.created",
             data: {
               order: {
                 id: "1234",

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

@@ -35,7 +35,7 @@ export const emitEventStep = createStep(
       context,
     })
 
-    await eventBus.emit([message])
+    await eventBus.emit(message)
   },
   async (data: void) => {}
 )

packages/core/medusa-test-utils/src/mock-event-bus-service.ts

@@ -1,5 +1,4 @@
 import {
-  EmitData,
   EventBusTypes,
   IEventBusModuleService,
   Message,
@@ -7,24 +6,9 @@ import {
 } from "@medusajs/types"
 
 export default class EventBusService implements IEventBusModuleService {
-  emit<T>(
+  async emit<T>(
-    eventName: string,
+    data: Message<T> | Message<T>[],
-    data: T,
+    options: Record<string, unknown>
-    options?: Record<string, unknown>
-  ): Promise<void>
-  emit<T>(data: EmitData<T>[]): Promise<void>
-  emit<T>(data: Message<T>[]): Promise<void>
-
-  async emit<
-    T,
-    TInput extends
-      | string
-      | EventBusTypes.EmitData<T>[]
-      | EventBusTypes.Message<T>[] = string
-  >(
-    eventOrData: TInput,
-    data?: T,
-    options: Record<string, unknown> = {}
   ): Promise<void> {}
 
   subscribe(event: string | symbol, subscriber: Subscriber): this {

packages/core/medusa-test-utils/src/module-test-runner.ts

@@ -1,8 +1,8 @@
-import { initModules, InitModulesOptions } from "./init-modules"
 import { getDatabaseURL, getMikroOrmWrapper, TestDatabase } from "./database"
+import { initModules, InitModulesOptions } from "./init-modules"
 
-import { MockEventBusService } from "."
 import { ContainerRegistrationKeys, ModulesSdkUtils } from "@medusajs/utils"
+import { MockEventBusService } from "."
 
 export interface SuiteOptions<TService = unknown> {
   MikroOrmWrapper: TestDatabase
@@ -75,7 +75,7 @@ export function moduleIntegrationTestRunner<TService = any>({
   const moduleOptions_: InitModulesOptions = {
     injectedDependencies: {
       [ContainerRegistrationKeys.PG_CONNECTION]: connection,
-      ["eventBusModuleService"]: new MockEventBusService(),
+      eventBusModuleService: new MockEventBusService(),
       [ContainerRegistrationKeys.LOGGER]: console,
       ...injectedDependencies,
     },

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

@@ -1,7 +1,7 @@
 import { Context } from "../shared-context"
 
-export type Subscriber<T = unknown> = (
+export type Subscriber<TData = unknown> = (
-  data: T,
+  data: TData,
   eventName: string
 ) => Promise<void>
 
@@ -14,36 +14,28 @@ export type SubscriberDescriptor = {
   subscriber: Subscriber
 }
 
-export type EventHandler<T = unknown> = (
+export type EventHandler<TData = unknown> = (
-  data: T,
+  data: TData,
   eventName: string
 ) => Promise<void>
 
-export type EmitData<T = unknown> = {
+export type EventMetadata = Record<string, unknown> & {
+  eventGroupId?: string
+}
+
+export type MessageBody<TData = unknown> = {
   eventName: string
-  data: T
+  metadata?: EventMetadata
+  data: TData
+}
+
+export type Message<TData = unknown> = MessageBody<TData> & {
   options?: Record<string, unknown>
 }
 
-export type MessageBody<T = unknown> = {
+export type RawMessageFormat<TData = any> = {
-  metadata: {
-    source: string
-    action: string
-    object: string
-    eventGroupId?: string
-  }
-  data: T
-}
-
-export type Message<T = unknown> = {
   eventName: string
-  body: MessageBody<T>
+  data: TData
-  options?: Record<string, unknown>
-}
-
-export type RawMessageFormat<T = any> = {
-  eventName: string
-  data: T
   source: string
   object: string
   action?: string

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

@@ -1,13 +1,10 @@
-import { EmitData, Message, Subscriber, SubscriberContext } from "./common"
+import { Message, Subscriber, SubscriberContext } from "./common"
 
 export interface IEventBusModuleService {
   emit<T>(
-    eventName: string,
+    data: Message<T> | Message<T>[],
-    data: T,
     options?: Record<string, unknown>
   ): Promise<void>
-  emit<T>(data: EmitData<T>[]): Promise<void>
-  emit<T>(data: Message<T>[]): Promise<void>
 
   subscribe(
     eventName: string | symbol,

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

@@ -1,5 +1,5 @@
 import { ITransactionBaseService } from "../transaction-base"
-import { EmitData, Message, Subscriber, SubscriberContext } from "./common"
+import { Message, Subscriber, SubscriberContext } from "./common"
 
 export interface IEventBusService extends ITransactionBaseService {
   subscribe(
@@ -14,7 +14,5 @@ export interface IEventBusService extends ITransactionBaseService {
     context?: SubscriberContext
   ): this
 
-  emit<T>(event: string, data: T, options?: unknown): Promise<unknown | void>
+  emit<T>(data: Message<T> | Message<T>[]): Promise<unknown | void>
-  emit<T>(data: EmitData<T>[]): Promise<unknown | void>
-  emit<T>(data: Message<T>[]): Promise<unknown | void>
 }

packages/core/utils/src/event-bus/__tests__/message-aggregator.spec.ts

@@ -9,70 +9,60 @@ describe("MessageAggregator", function () {
     const aggregator = new MessageAggregator()
     aggregator.save({
       eventName: "ProductVariant.created",
-      body: {
+      metadata: {
-        metadata: {
+        source: "ProductService",
-          source: "ProductService",
+        action: "created",
-          action: "created",
+        object: "ProductVariant",
-          object: "ProductVariant",
+        eventGroupId: "1",
-          eventGroupId: "1",
-        },
-        data: { id: 999 },
       },
+      data: { id: 999 },
     })
     aggregator.save({
       eventName: "Product.created",
-      body: {
+      metadata: {
-        metadata: {
+        source: "ProductService",
-          source: "ProductService",
+        action: "created",
-          action: "created",
+        object: "Product",
-          object: "Product",
+        eventGroupId: "1",
-          eventGroupId: "1",
-        },
-        data: { id: 1 },
       },
+      data: { id: 1 },
     })
     aggregator.save({
       eventName: "ProductVariant.created",
-      body: {
+      metadata: {
-        metadata: {
+        source: "ProductService",
-          source: "ProductService",
+        action: "created",
-          action: "created",
+        object: "ProductVariant",
-          object: "ProductVariant",
+        eventGroupId: "1",
-          eventGroupId: "1",
-        },
-        data: { id: 222 },
       },
+      data: { id: 222 },
     })
     aggregator.save({
       eventName: "ProductType.detached",
-      body: {
+      metadata: {
-        metadata: {
+        source: "ProductService",
-          source: "ProductService",
+        action: "detached",
-          action: "detached",
+        object: "ProductType",
-          object: "ProductType",
+        eventGroupId: "1",
-          eventGroupId: "1",
-        },
-        data: { id: 333 },
       },
+      data: { id: 333 },
     })
     aggregator.save({
       eventName: "ProductVariant.updated",
-      body: {
+      metadata: {
-        metadata: {
+        source: "ProductService",
-          source: "ProductService",
+        action: "updated",
-          action: "updated",
+        object: "ProductVariant",
-          object: "ProductVariant",
+        eventGroupId: "1",
-          eventGroupId: "1",
-        },
-        data: { id: 123 },
       },
+      data: { id: 123 },
     })
 
     const format = {
-      groupBy: ["eventName", "body.metadata.object", "body.metadata.action"],
+      groupBy: ["eventName", "metadata.object", "metadata.action"],
       sortBy: {
-        "body.metadata.object": ["ProductType", "ProductVariant", "Product"],
+        "metadata.object": ["ProductType", "ProductVariant", "Product"],
-        "body.data.id": "asc",
+        "data.id": "asc",
       },
     }
 
@@ -85,72 +75,62 @@ describe("MessageAggregator", function () {
     expect(allGroups[0]).toEqual([
       {
         eventName: "ProductType.detached",
-        body: {
+        metadata: {
-          metadata: {
+          source: "ProductService",
-            source: "ProductService",
+          action: "detached",
-            action: "detached",
+          object: "ProductType",
-            object: "ProductType",
+          eventGroupId: "1",
-            eventGroupId: "1",
-          },
-          data: { id: 333 },
         },
+        data: { id: 333 },
       },
     ])
 
     expect(allGroups[1]).toEqual([
       {
         eventName: "ProductVariant.updated",
-        body: {
+        metadata: {
-          metadata: {
+          source: "ProductService",
-            source: "ProductService",
+          action: "updated",
-            action: "updated",
+          object: "ProductVariant",
-            object: "ProductVariant",
+          eventGroupId: "1",
-            eventGroupId: "1",
-          },
-          data: { id: 123 },
         },
+        data: { id: 123 },
       },
     ])
 
     expect(allGroups[2]).toEqual([
       {
         eventName: "ProductVariant.created",
-        body: {
+        metadata: {
-          metadata: {
+          source: "ProductService",
-            source: "ProductService",
+          action: "created",
-            action: "created",
+          object: "ProductVariant",
-            object: "ProductVariant",
+          eventGroupId: "1",
-            eventGroupId: "1",
-          },
-          data: { id: 222 },
         },
+        data: { id: 222 },
       },
       {
         eventName: "ProductVariant.created",
-        body: {
+        metadata: {
-          metadata: {
+          source: "ProductService",
-            source: "ProductService",
+          action: "created",
-            action: "created",
+          object: "ProductVariant",
-            object: "ProductVariant",
+          eventGroupId: "1",
-            eventGroupId: "1",
-          },
-          data: { id: 999 },
         },
+        data: { id: 999 },
       },
     ])
 
     expect(allGroups[3]).toEqual([
       {
         eventName: "Product.created",
-        body: {
+        metadata: {
-          metadata: {
+          source: "ProductService",
-            source: "ProductService",
+          action: "created",
-            action: "created",
+          object: "Product",
-            object: "Product",
+          eventGroupId: "1",
-            eventGroupId: "1",
-          },
-          data: { id: 1 },
         },
+        data: { id: 1 },
       },
     ])
   })

packages/core/utils/src/event-bus/build-event-messages.ts

@@ -45,10 +45,8 @@ export function composeMessage(
 
   return {
     eventName,
-    body: {
+    metadata,
-      metadata,
+    data,
-      data,
-    },
     options,
   }
 }

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

@@ -17,12 +17,9 @@ export abstract class AbstractEventBusModuleService
   }
 
   abstract emit<T>(
-    eventName: string,
+    data: EventBusTypes.Message<T> | EventBusTypes.Message<T>[],
-    data: T,
     options: Record<string, unknown>
   ): Promise<void>
-  abstract emit<T>(data: EventBusTypes.EmitData<T>[]): Promise<void>
-  abstract emit<T>(data: EventBusTypes.Message<T>[]): Promise<void>
 
   /*
     Grouped events are useful when you have distributed transactions

packages/core/utils/src/modules-sdk/abstract-module-service-factory.ts

@@ -501,6 +501,7 @@ export function abstractModuleServiceFactory<
           await this.eventBusModuleService_?.emit(
             softDeletedEntities.map(({ id }) => ({
               eventName: `${kebabCase(model.name)}.deleted`,
+              metadata: { source: "", action: "", object: "" },
               data: { id },
             }))
           )

packages/medusa/src/api/hooks/payment/[provider]/route.ts

@@ -1,6 +1,6 @@
 import { ModuleRegistrationName } from "@medusajs/modules-sdk"
-import { PaymentWebhookEvents } from "@medusajs/utils"
 import { PaymentModuleOptions } from "@medusajs/types"
+import { PaymentWebhookEvents } from "@medusajs/utils"
 
 import { MedusaRequest, MedusaResponse } from "../../../../types/routing"
 
@@ -20,10 +20,16 @@ export const POST = async (req: MedusaRequest, res: MedusaResponse) => {
     const eventBus = req.scope.resolve(ModuleRegistrationName.EVENT_BUS)
 
     // we delay the processing of the event to avoid a conflict caused by a race condition
-    await eventBus.emit(PaymentWebhookEvents.WebhookReceived, event, {
+    await eventBus.emit(
-      delay: options.webhook_delay || 5000,
+      {
-      attempts: options.webhook_retries || 3,
+        eventName: PaymentWebhookEvents.WebhookReceived,
-    })
+        data: event,
+      },
+      {
+        delay: options.webhook_delay || 5000,
+        attempts: options.webhook_retries || 3,
+      }
+    )
   } catch (err) {
     res.status(400).send(`Webhook Error: ${err.message}`)
     return

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

@@ -14,7 +14,8 @@ const moduleDeps = {
 }
 
 describe("LocalEventBusService", () => {
-  let eventBus
+  let eventBus: LocalEventBusService
+  let eventEmitter
 
   describe("emit", () => {
     describe("Successfully emits events", () => {
@@ -22,148 +23,184 @@ describe("LocalEventBusService", () => {
         jest.clearAllMocks()
 
         eventBus = new LocalEventBusService(moduleDeps as any)
+        eventEmitter = (eventBus as any).eventEmitter_
       })
 
       it("should emit an event", async () => {
-        eventBus.eventEmitter_.emit.mockImplementationOnce((data) => data)
+        eventEmitter.emit = jest.fn((data) => data)
 
-        await eventBus.emit("eventName", { hi: "1234" })
+        await eventBus.emit({
+          eventName: "eventName",
+          data: { hi: "1234" },
+        })
 
-        expect(eventBus.eventEmitter_.emit).toHaveBeenCalledTimes(1)
+        expect(eventEmitter.emit).toHaveBeenCalledTimes(1)
-        expect(eventBus.eventEmitter_.emit).toHaveBeenCalledWith("eventName", {
+        expect(eventEmitter.emit).toHaveBeenCalledWith("eventName", {
-          hi: "1234",
+          data: { hi: "1234" },
         })
       })
 
       it("should emit multiple events", async () => {
-        eventBus.eventEmitter_.emit.mockImplementationOnce((data) => data)
+        eventEmitter.emit = jest.fn((data) => data)
 
         await eventBus.emit([
           { eventName: "event-1", data: { hi: "1234" } },
           { eventName: "event-2", data: { hi: "5678" } },
         ])
 
-        expect(eventBus.eventEmitter_.emit).toHaveBeenCalledTimes(2)
+        expect(eventEmitter.emit).toHaveBeenCalledTimes(2)
-        expect(eventBus.eventEmitter_.emit).toHaveBeenCalledWith("event-1", {
+        expect(eventEmitter.emit).toHaveBeenCalledWith("event-1", {
-          hi: "1234",
+          data: { hi: "1234" },
         })
-        expect(eventBus.eventEmitter_.emit).toHaveBeenCalledWith("event-2", {
+        expect(eventEmitter.emit).toHaveBeenCalledWith("event-2", {
-          hi: "5678",
+          data: { hi: "5678" },
         })
       })
 
       it("should group an event if data consists of eventGroupId", async () => {
-        const groupEventFn = jest.spyOn(eventBus, "groupEvent")
+        let groupEventFn = jest.spyOn(eventBus, "groupEvent" as any)
+        eventEmitter.emit = jest.fn((data) => data)
 
-        eventBus.eventEmitter_.emit.mockImplementationOnce((data) => data)
+        await eventBus.emit({
-
+          eventName: "test-event",
-        await eventBus.emit("test-event", {
+          data: {
-          test: "1234",
+            test: "1234",
-          eventGroupId: "test",
+          },
+          metadata: {
+            eventGroupId: "test",
+          },
         })
 
-        expect(eventBus.eventEmitter_.emit).not.toHaveBeenCalled()
+        expect(eventEmitter.emit).not.toHaveBeenCalled()
         expect(groupEventFn).toHaveBeenCalledTimes(1)
-        expect(groupEventFn).toHaveBeenCalledWith("test", "test-event", {
+        expect(groupEventFn).toHaveBeenCalledWith("test", {
-          test: "1234",
+          data: { test: "1234" },
+          metadata: { eventGroupId: "test" },
+          eventName: "test-event",
         })
 
         jest.clearAllMocks()
 
-        eventBus.eventEmitter_.emit.mockImplementationOnce((data) => data)
+        groupEventFn = jest.spyOn(eventBus, "groupEvent" as any)
-        eventBus.emit("test-event", { test: "1234", eventGroupId: "test" })
+        eventEmitter.emit = jest.fn((data) => data)
-        eventBus.emit("test-event", { test: "test-1" })
+
+        eventBus.emit([
+          {
+            eventName: "test-event",
+            data: { test: "1234" },
+            metadata: { eventGroupId: "test" },
+          },
+          {
+            eventName: "test-event",
+            data: { test: "test-1" },
+          },
+        ])
 
-        expect(eventBus.eventEmitter_.emit).toHaveBeenCalledTimes(1)
         expect(groupEventFn).toHaveBeenCalledTimes(1)
 
-        expect(eventBus.groupedEventsMap_.get("test")).toEqual([
+        expect((eventBus as any).groupedEventsMap_.get("test")).toEqual([
           expect.objectContaining({ eventName: "test-event" }),
           expect.objectContaining({ eventName: "test-event" }),
         ])
 
-        await eventBus.emit("test-event", {
+        await eventBus.emit({
-          test: "1234",
+          eventName: "test-event",
-          eventGroupId: "test-2",
+          data: { test: "1234" },
+          metadata: { eventGroupId: "test-2" },
         })
 
-        expect(eventBus.groupedEventsMap_.get("test-2")).toEqual([
+        expect((eventBus as any).groupedEventsMap_.get("test-2")).toEqual([
           expect.objectContaining({ eventName: "test-event" }),
         ])
       })
 
       it("should release events when requested with eventGroupId", async () => {
-        eventBus.eventEmitter_.emit.mockImplementationOnce((data) => data)
+        eventEmitter.emit = jest.fn((data) => data)
 
         await eventBus.emit([
           {
             eventName: "event-1",
-            data: { test: "1", eventGroupId: "group-1" },
+            data: { test: "1" },
+            metadata: { eventGroupId: "group-1" },
           },
           {
             eventName: "event-2",
-            data: { test: "2", eventGroupId: "group-1" },
+            data: { test: "2" },
+            metadata: { eventGroupId: "group-1" },
           },
           {
             eventName: "event-1",
-            data: { test: "1", eventGroupId: "group-2" },
+            data: { test: "1" },
+            metadata: { eventGroupId: "group-2" },
           },
           {
             eventName: "event-2",
-            data: { test: "2", eventGroupId: "group-2" },
+            data: { test: "2" },
+            metadata: { eventGroupId: "group-2" },
           },
           { eventName: "event-1", data: { test: "1" } },
         ])
 
-        expect(eventBus.eventEmitter_.emit).toHaveBeenCalledTimes(1)
+        expect(eventEmitter.emit).toHaveBeenCalledTimes(1)
-        expect(eventBus.eventEmitter_.emit).toHaveBeenCalledWith("event-1", {
+        expect(eventEmitter.emit).toHaveBeenCalledWith("event-1", {
-          test: "1",
+          data: { test: "1" },
         })
 
-        expect(eventBus.groupedEventsMap_.get("group-1")).toHaveLength(2)
+        expect((eventBus as any).groupedEventsMap_.get("group-1")).toHaveLength(
-        expect(eventBus.groupedEventsMap_.get("group-2")).toHaveLength(2)
+          2
+        )
+        expect((eventBus as any).groupedEventsMap_.get("group-2")).toHaveLength(
+          2
+        )
 
         jest.clearAllMocks()
-        eventBus.eventEmitter_.emit.mockImplementationOnce((data) => data)
+        eventEmitter.emit = jest.fn((data) => data)
         eventBus.releaseGroupedEvents("group-1")
 
-        expect(eventBus.groupedEventsMap_.get("group-1")).not.toBeDefined()
+        expect(
-        expect(eventBus.groupedEventsMap_.get("group-2")).toHaveLength(2)
+          (eventBus as any).groupedEventsMap_.get("group-1")
+        ).not.toBeDefined()
+        expect((eventBus as any).groupedEventsMap_.get("group-2")).toHaveLength(
+          2
+        )
 
-        expect(eventBus.eventEmitter_.emit).toHaveBeenCalledTimes(2)
+        expect(eventEmitter.emit).toHaveBeenCalledTimes(2)
-        expect(eventBus.eventEmitter_.emit).toHaveBeenCalledWith("event-1", {
+        expect(eventEmitter.emit).toHaveBeenCalledWith("event-1", {
-          test: "1",
+          data: { test: "1" },
         })
-        expect(eventBus.eventEmitter_.emit).toHaveBeenCalledWith("event-2", {
+        expect(eventEmitter.emit).toHaveBeenCalledWith("event-2", {
-          test: "2",
+          data: { test: "2" },
         })
       })
 
       it("should clear events from grouped events when requested with eventGroupId", async () => {
-        eventBus.eventEmitter_.emit.mockImplementationOnce((data) => data)
+        eventEmitter.emit = jest.fn((data) => data)
+        const getMap = () => (eventBus as any).groupedEventsMap_
 
         await eventBus.emit([
           {
             eventName: "event-1",
-            data: { test: "1", eventGroupId: "group-1" },
+            data: { test: "1" },
+            metadata: { eventGroupId: "group-1" },
           },
           {
             eventName: "event-1",
-            data: { test: "1", eventGroupId: "group-2" },
+            data: { test: "1" },
+            metadata: { eventGroupId: "group-2" },
           },
         ])
 
-        expect(eventBus.groupedEventsMap_.get("group-1")).toHaveLength(1)
+        expect(getMap().get("group-1")).toHaveLength(1)
-        expect(eventBus.groupedEventsMap_.get("group-2")).toHaveLength(1)
+        expect(getMap().get("group-2")).toHaveLength(1)
 
         eventBus.clearGroupedEvents("group-1")
 
-        expect(eventBus.groupedEventsMap_.get("group-1")).not.toBeDefined()
+        expect(getMap().get("group-1")).not.toBeDefined()
-        expect(eventBus.groupedEventsMap_.get("group-2")).toHaveLength(1)
+        expect(getMap().get("group-2")).toHaveLength(1)
 
         eventBus.clearGroupedEvents("group-2")
 
-        expect(eventBus.groupedEventsMap_.get("group-2")).not.toBeDefined()
+        expect(getMap().get("group-2")).not.toBeDefined()
       })
     })
   })

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

@@ -1,9 +1,9 @@
 import { MedusaContainer } from "@medusajs/modules-sdk"
 import {
-  EmitData,
   EventBusTypes,
   Logger,
   Message,
+  MessageBody,
   Subscriber,
 } from "@medusajs/types"
 import { AbstractEventBusModuleService } from "@medusajs/utils"
@@ -35,47 +35,28 @@ export default class LocalEventBusService extends AbstractEventBusModuleService
     this.groupedEventsMap_ = new Map()
   }
 
-  async emit<T>(
+  async emit<T = unknown>(
-    eventName: string,
+    eventsData: Message<T> | Message<T>[],
-    data: T,
-    options: Record<string, unknown>
-  ): Promise<void>
-
-  /**
-   * Emit a number of events
-   * @param {EmitData} data - the data to send to the subscriber.
-   */
-  async emit<T>(data: EmitData<T>[]): Promise<void>
-
-  async emit<T>(data: Message<T>[]): Promise<void>
-
-  async emit<T, TInput extends string | EmitData<T>[] | Message<T>[] = string>(
-    eventOrData: TInput,
-    data?: T,
     options: Record<string, unknown> = {}
   ): Promise<void> {
-    const isBulkEmit = Array.isArray(eventOrData)
+    const normalizedEventsData = Array.isArray(eventsData)
+      ? eventsData
+      : [eventsData]
 
-    const events: EmitData[] | Message<T>[] = isBulkEmit
+    for (const eventData of normalizedEventsData) {
-      ? eventOrData
-      : [{ eventName: eventOrData, data }]
-
-    for (const event of events) {
       const eventListenersCount = this.eventEmitter_.listenerCount(
-        event.eventName
+        eventData.eventName
       )
 
       this.logger_?.info(
-        `Processing ${event.eventName} which has ${eventListenersCount} subscribers`
+        `Processing ${eventData.eventName} which has ${eventListenersCount} subscribers`
       )
 
       if (eventListenersCount === 0) {
         continue
       }
 
-      const data = (event as EmitData).data ?? (event as Message<T>).body
+      await this.groupOrEmitEvent(eventData)
-
-      await this.groupOrEmitEvent(event.eventName, data)
     }
   }
 
@@ -84,28 +65,27 @@ export default class LocalEventBusService extends AbstractEventBusModuleService
   // explicitly requested.
   // This is useful in the event of a distributed transaction where you'd want to emit
   // events only once the transaction ends.
-  private async groupOrEmitEvent(
+  private async groupOrEmitEvent<T = unknown>(eventData: Message<T>) {
-    eventName: string,
+    const { options, ...eventBody } = eventData
-    data: unknown & { eventGroupId?: string }
+    const eventGroupId = eventBody.metadata?.eventGroupId
-  ) {
-    const { eventGroupId, ...eventData } = data
 
     if (eventGroupId) {
-      await this.groupEvent(eventGroupId, eventName, eventData)
+      await this.groupEvent(eventGroupId, eventData)
     } else {
-      this.eventEmitter_.emit(eventName, data)
+      this.eventEmitter_.emit(eventData.eventName, {
+        data: eventData.data,
+      })
     }
   }
 
   // Groups an event to a queue to be emitted upon explicit release
-  private async groupEvent(
+  private async groupEvent<T = unknown>(
     eventGroupId: string,
-    eventName: string,
+    eventData: MessageBody<T>
-    data: unknown
   ) {
     const groupedEvents = this.groupedEventsMap_.get(eventGroupId) || []
 
-    groupedEvents.push({ eventName, data })
+    groupedEvents.push(eventData)
 
     this.groupedEventsMap_.set(eventGroupId, groupedEvents)
   }
@@ -116,7 +96,7 @@ export default class LocalEventBusService extends AbstractEventBusModuleService
     for (const event of groupedEvents) {
       const { eventName, data } = event
 
-      this.eventEmitter_.emit(eventName, data)
+      this.eventEmitter_.emit(eventName, { data })
     }
 
     this.clearGroupedEvents(eventGroupId)

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

@@ -1,318 +0,0 @@
-import { Queue, Worker } from "bullmq"
-import RedisEventBusService from "../event-bus-redis"
-
-jest.genMockFromModule("bullmq")
-jest.genMockFromModule("ioredis")
-jest.mock("bullmq")
-jest.mock("ioredis")
-
-const loggerMock = {
-  info: jest.fn().mockReturnValue(console.log),
-  warn: jest.fn().mockReturnValue(console.log),
-  error: jest.fn().mockReturnValue(console.log),
-}
-
-const simpleModuleOptions = { redisUrl: "test-url" }
-const moduleDeps = {
-  manager: {},
-  logger: loggerMock,
-  eventBusRedisConnection: {},
-}
-
-describe("RedisEventBusService", () => {
-  let eventBus
-
-  describe("constructor", () => {
-    beforeAll(() => {
-      jest.clearAllMocks()
-    })
-
-    it("Creates a queue + worker", () => {
-      eventBus = new RedisEventBusService(moduleDeps, simpleModuleOptions, {
-        resources: "shared",
-      })
-
-      expect(Queue).toHaveBeenCalledTimes(1)
-      expect(Queue).toHaveBeenCalledWith("events-queue", {
-        connection: expect.any(Object),
-        prefix: "RedisEventBusService",
-      })
-
-      expect(Worker).toHaveBeenCalledTimes(1)
-      expect(Worker).toHaveBeenCalledWith(
-        "events-queue",
-        expect.any(Function),
-        {
-          connection: expect.any(Object),
-          prefix: "RedisEventBusService",
-        }
-      )
-    })
-
-    it("Throws on isolated module declaration", () => {
-      try {
-        eventBus = new RedisEventBusService(moduleDeps, simpleModuleOptions, {
-          resources: "isolated",
-        })
-      } catch (error) {
-        expect(error.message).toEqual(
-          "At the moment this module can only be used with shared resources"
-        )
-      }
-    })
-  })
-
-  describe("emit", () => {
-    describe("Successfully emits events", () => {
-      beforeEach(() => {
-        jest.clearAllMocks()
-      })
-
-      it("Adds job to queue with default options", () => {
-        eventBus = new RedisEventBusService(moduleDeps, simpleModuleOptions, {
-          resources: "shared",
-        })
-
-        eventBus.queue_.addBulk.mockImplementationOnce(() => "hi")
-        eventBus.emit("eventName", { hi: "1234" })
-
-        expect(eventBus.queue_.addBulk).toHaveBeenCalledTimes(1)
-        expect(eventBus.queue_.addBulk).toHaveBeenCalledWith([
-          {
-            name: "eventName",
-            data: { eventName: "eventName", data: { hi: "1234" } },
-            opts: {
-              attempts: 1,
-              removeOnComplete: true,
-            },
-          },
-        ])
-      })
-
-      it("Adds job to queue with custom options passed directly upon emitting", () => {
-        eventBus = new RedisEventBusService(moduleDeps, simpleModuleOptions, {
-          resources: "shared",
-        })
-
-        eventBus.queue_.addBulk.mockImplementationOnce(() => "hi")
-        eventBus.emit(
-          "eventName",
-          { hi: "1234" },
-          { attempts: 3, backoff: 5000, delay: 1000 }
-        )
-
-        expect(eventBus.queue_.addBulk).toHaveBeenCalledTimes(1)
-        expect(eventBus.queue_.addBulk).toHaveBeenCalledWith([
-          {
-            name: "eventName",
-            data: { eventName: "eventName", data: { hi: "1234" } },
-            opts: {
-              attempts: 3,
-              backoff: 5000,
-              delay: 1000,
-              removeOnComplete: true,
-            },
-          },
-        ])
-      })
-
-      it("Adds job to queue with module job options", () => {
-        eventBus = new RedisEventBusService(
-          moduleDeps,
-          {
-            ...simpleModuleOptions,
-            jobOptions: {
-              removeOnComplete: {
-                age: 5,
-              },
-              attempts: 7,
-            },
-          },
-          {
-            resources: "shared",
-          }
-        )
-
-        eventBus.queue_.addBulk.mockImplementationOnce(() => "hi")
-        eventBus.emit("eventName", { hi: "1234" })
-
-        expect(eventBus.queue_.addBulk).toHaveBeenCalledTimes(1)
-        expect(eventBus.queue_.addBulk).toHaveBeenCalledWith([
-          {
-            name: "eventName",
-            data: { eventName: "eventName", data: { hi: "1234" } },
-            opts: {
-              attempts: 7,
-              removeOnComplete: {
-                age: 5,
-              },
-            },
-          },
-        ])
-      })
-
-      it("Adds job to queue with default, local, and global options merged", () => {
-        eventBus = new RedisEventBusService(
-          moduleDeps,
-          {
-            ...simpleModuleOptions,
-            jobOptions: {
-              removeOnComplete: 5,
-            },
-          },
-          {
-            resources: "shared",
-          }
-        )
-
-        eventBus.queue_.addBulk.mockImplementationOnce(() => "hi")
-        eventBus.emit("eventName", { hi: "1234" }, { delay: 1000 })
-
-        expect(eventBus.queue_.addBulk).toHaveBeenCalledTimes(1)
-        expect(eventBus.queue_.addBulk).toHaveBeenCalledWith([
-          {
-            name: "eventName",
-            data: { eventName: "eventName", data: { hi: "1234" } },
-            opts: {
-              attempts: 1,
-              removeOnComplete: 5,
-              delay: 1000,
-            },
-          },
-        ])
-      })
-    })
-  })
-
-  describe("worker_", () => {
-    let result
-
-    describe("Successfully processes the jobs", () => {
-      beforeEach(async () => {
-        jest.clearAllMocks()
-
-        eventBus = new RedisEventBusService(moduleDeps, simpleModuleOptions, {
-          resources: "shared",
-        })
-      })
-
-      it("Processes a simple event with no options", async () => {
-        eventBus.subscribe("eventName", () => Promise.resolve("hi"))
-
-        result = await eventBus.worker_({
-          data: { eventName: "eventName", data: {} },
-          opts: { attempts: 1 },
-        })
-
-        expect(loggerMock.info).toHaveBeenCalledTimes(1)
-        expect(loggerMock.info).toHaveBeenCalledWith(
-          "Processing eventName which has 1 subscribers"
-        )
-
-        expect(result).toEqual(["hi"])
-      })
-
-      it("Processes event with failing subscribers", async () => {
-        eventBus.subscribe("eventName", () => Promise.resolve("hi"))
-        eventBus.subscribe("eventName", () => Promise.reject("fail1"))
-        eventBus.subscribe("eventName", () => Promise.resolve("hi2"))
-        eventBus.subscribe("eventName", () => Promise.reject("fail2"))
-
-        result = await eventBus.worker_({
-          data: { eventName: "eventName", data: {} },
-          update: (data) => data,
-          opts: { attempts: 1 },
-        })
-
-        expect(loggerMock.info).toHaveBeenCalledTimes(1)
-        expect(loggerMock.info).toHaveBeenCalledWith(
-          "Processing eventName which has 4 subscribers"
-        )
-
-        expect(loggerMock.warn).toHaveBeenCalledTimes(3)
-        expect(loggerMock.warn).toHaveBeenCalledWith(
-          "An error occurred while processing eventName: fail1"
-        )
-        expect(loggerMock.warn).toHaveBeenCalledWith(
-          "An error occurred while processing eventName: fail2"
-        )
-
-        expect(loggerMock.warn).toHaveBeenCalledWith(
-          "One or more subscribers of eventName failed. Retrying is not configured. Use 'attempts' option when emitting events."
-        )
-
-        expect(result).toEqual(["hi", "fail1", "hi2", "fail2"])
-      })
-
-      it("Retries processing when subcribers fail, if configured - final attempt", async () => {
-        eventBus.subscribe("eventName", async () => Promise.resolve("hi"), {
-          subscriberId: "1",
-        })
-        eventBus.subscribe("eventName", async () => Promise.reject("fail1"), {
-          subscriberId: "2",
-        })
-
-        result = await eventBus
-          .worker_({
-            data: {
-              eventName: "eventName",
-              data: {},
-              completedSubscriberIds: ["1"],
-            },
-            attemptsMade: 2,
-            update: (data) => data,
-            opts: { attempts: 2 },
-          })
-          .catch((error) => void 0)
-
-        expect(loggerMock.warn).toHaveBeenCalledTimes(1)
-        expect(loggerMock.warn).toHaveBeenCalledWith(
-          "An error occurred while processing eventName: fail1"
-        )
-
-        expect(loggerMock.info).toHaveBeenCalledTimes(2)
-        expect(loggerMock.info).toHaveBeenCalledWith(
-          "Final retry attempt for eventName"
-        )
-        expect(loggerMock.info).toHaveBeenCalledWith(
-          "Retrying eventName which has 2 subscribers (1 of them failed)"
-        )
-      })
-
-      it("Retries processing when subcribers fail, if configured", async () => {
-        eventBus.subscribe("eventName", async () => Promise.resolve("hi"), {
-          subscriberId: "1",
-        })
-        eventBus.subscribe("eventName", async () => Promise.reject("fail1"), {
-          subscriberId: "2",
-        })
-
-        result = await eventBus
-          .worker_({
-            data: {
-              eventName: "eventName",
-              data: {},
-              completedSubscriberIds: ["1"],
-            },
-            attemptsMade: 2,
-            updateData: (data) => data,
-            opts: { attempts: 3 },
-          })
-          .catch((err) => void 0)
-
-        expect(loggerMock.warn).toHaveBeenCalledTimes(2)
-        expect(loggerMock.warn).toHaveBeenCalledWith(
-          "An error occurred while processing eventName: fail1"
-        )
-        expect(loggerMock.warn).toHaveBeenCalledWith(
-          "One or more subscribers of eventName failed. Retrying..."
-        )
-
-        expect(loggerMock.info).toHaveBeenCalledTimes(1)
-        expect(loggerMock.info).toHaveBeenCalledWith(
-          "Retrying eventName which has 2 subscribers (1 of them failed)"
-        )
-      })
-    })
-  })
-})

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

@@ -0,0 +1,491 @@
+import { Logger } from "@medusajs/types"
+import { Queue, Worker } from "bullmq"
+import { Redis } from "ioredis"
+import RedisEventBusService from "../event-bus-redis"
+
+// const redisURL = "redis://localhost:6379"
+// const client = new Redis(6379, redisURL, {
+//   // Lazy connect to properly handle connection errors
+//   lazyConnect: true,
+//   maxRetriesPerRequest: 0,
+// })
+
+jest.genMockFromModule("bullmq")
+jest.genMockFromModule("ioredis")
+jest.mock("bullmq")
+jest.mock("ioredis")
+
+const loggerMock = {
+  info: jest.fn().mockReturnValue(console.log),
+  warn: jest.fn().mockReturnValue(console.log),
+  error: jest.fn().mockReturnValue(console.log),
+} as unknown as Logger
+
+const redisMock = {
+  del: () => jest.fn(),
+  rpush: () => jest.fn(),
+  lrange: () => jest.fn(),
+  disconnect: () => jest.fn(),
+  expire: () => jest.fn(),
+} as unknown as Redis
+
+const simpleModuleOptions = { redisUrl: "test-url" }
+const moduleDeps = {
+  logger: loggerMock,
+  eventBusRedisConnection: redisMock,
+}
+
+describe("RedisEventBusService", () => {
+  let eventBus: RedisEventBusService
+  let queue
+  let redis
+
+  describe("constructor", () => {
+    beforeEach(async () => {
+      jest.clearAllMocks()
+
+      eventBus = new RedisEventBusService(moduleDeps, simpleModuleOptions, {
+        scope: "internal",
+        resources: "shared",
+      })
+    })
+
+    it("Creates a queue + worker", () => {
+      expect(Queue).toHaveBeenCalledTimes(1)
+      expect(Queue).toHaveBeenCalledWith("events-queue", {
+        connection: expect.any(Object),
+        prefix: "RedisEventBusService",
+      })
+
+      expect(Worker).toHaveBeenCalledTimes(1)
+      expect(Worker).toHaveBeenCalledWith(
+        "events-queue",
+        expect.any(Function),
+        {
+          connection: expect.any(Object),
+          prefix: "RedisEventBusService",
+        }
+      )
+    })
+
+    it("Throws on isolated module declaration", () => {
+      try {
+        eventBus = new RedisEventBusService(moduleDeps, simpleModuleOptions, {
+          resources: "isolated",
+          scope: "internal",
+        })
+      } catch (error) {
+        expect(error.message).toEqual(
+          "At the moment this module can only be used with shared resources"
+        )
+      }
+    })
+  })
+
+  describe("emit", () => {
+    describe("Successfully emits events", () => {
+      beforeEach(async () => {
+        jest.clearAllMocks()
+
+        eventBus = new RedisEventBusService(moduleDeps, simpleModuleOptions, {
+          scope: "internal",
+          resources: "shared",
+        })
+
+        queue = (eventBus as any).queue_
+        queue.addBulk = jest.fn()
+        redis = (eventBus as any).eventBusRedisConnection_
+        redis.rpush = jest.fn()
+      })
+
+      it("should add job to queue with default options", async () => {
+        await eventBus.emit([
+          {
+            eventName: "eventName",
+            data: {
+              hi: "1234",
+            },
+          },
+        ])
+
+        expect(queue.addBulk).toHaveBeenCalledTimes(1)
+        expect(queue.addBulk).toHaveBeenCalledWith([
+          {
+            name: "eventName",
+            data: { eventName: "eventName", data: { hi: "1234" } },
+            opts: {
+              attempts: 1,
+              removeOnComplete: true,
+            },
+          },
+        ])
+      })
+
+      it("should add job to queue with custom options passed directly upon emitting", async () => {
+        await eventBus.emit(
+          [{ eventName: "eventName", data: { hi: "1234" } }],
+          { attempts: 3, backoff: 5000, delay: 1000 }
+        )
+
+        expect(queue.addBulk).toHaveBeenCalledTimes(1)
+        expect(queue.addBulk).toHaveBeenCalledWith([
+          {
+            name: "eventName",
+            data: { eventName: "eventName", data: { hi: "1234" } },
+            opts: {
+              attempts: 3,
+              backoff: 5000,
+              delay: 1000,
+              removeOnComplete: true,
+            },
+          },
+        ])
+      })
+
+      it("should add job to queue with module job options", async () => {
+        eventBus = new RedisEventBusService(
+          moduleDeps,
+          {
+            ...simpleModuleOptions,
+            jobOptions: {
+              removeOnComplete: { age: 5 },
+              attempts: 7,
+            },
+          },
+          {
+            resources: "shared",
+            scope: "internal",
+          }
+        )
+
+        queue = (eventBus as any).queue_
+        queue.addBulk = jest.fn()
+
+        await eventBus.emit(
+          [
+            {
+              eventName: "eventName",
+              data: { hi: "1234" },
+            },
+          ],
+          { attempts: 3, backoff: 5000, delay: 1000 }
+        )
+
+        expect(queue.addBulk).toHaveBeenCalledTimes(1)
+        expect(queue.addBulk).toHaveBeenCalledWith([
+          {
+            name: "eventName",
+            data: { eventName: "eventName", data: { hi: "1234" } },
+            opts: {
+              attempts: 3,
+              backoff: 5000,
+              delay: 1000,
+              removeOnComplete: {
+                age: 5,
+              },
+            },
+          },
+        ])
+      })
+
+      it("should add job to queue with default, local, and global options merged", async () => {
+        eventBus = new RedisEventBusService(
+          moduleDeps,
+          {
+            ...simpleModuleOptions,
+            jobOptions: {
+              removeOnComplete: 5,
+            },
+          },
+          {
+            resources: "shared",
+            scope: "internal",
+          }
+        )
+
+        queue = (eventBus as any).queue_
+        queue.addBulk = jest.fn()
+
+        await eventBus.emit(
+          {
+            eventName: "eventName",
+            data: { hi: "1234" },
+          },
+          { delay: 1000 }
+        )
+
+        expect(queue.addBulk).toHaveBeenCalledTimes(1)
+        expect(queue.addBulk).toHaveBeenCalledWith([
+          {
+            name: "eventName",
+            data: { eventName: "eventName", data: { hi: "1234" } },
+            opts: {
+              attempts: 1,
+              removeOnComplete: 5,
+              delay: 1000,
+            },
+          },
+        ])
+      })
+
+      it("should successfully group events", async () => {
+        const options = { delay: 1000 }
+        const event = {
+          eventName: "eventName",
+          data: { hi: "1234" },
+          metadata: { eventGroupId: "test-group-1" },
+        }
+
+        const [builtEvent] = (eventBus as any).buildEvents([event], options)
+
+        await eventBus.emit(event, options)
+
+        expect(queue.addBulk).toHaveBeenCalledTimes(0)
+        expect(redis.rpush).toHaveBeenCalledTimes(1)
+        expect(redis.rpush).toHaveBeenCalledWith(
+          "staging:test-group-1",
+          JSON.stringify(builtEvent)
+        )
+      })
+
+      it("should successfully group, release and clear events", async () => {
+        const options = { delay: 1000 }
+        const events = [
+          {
+            eventName: "grouped-event-1",
+            data: { hi: "1234" },
+            metadata: { eventGroupId: "test-group-1" },
+          },
+          {
+            eventName: "ungrouped-event-2",
+            data: { hi: "1234" },
+          },
+          {
+            eventName: "grouped-event-2",
+            data: { hi: "1234" },
+            metadata: { eventGroupId: "test-group-2" },
+          },
+          {
+            eventName: "grouped-event-3",
+            data: { hi: "1235" },
+            metadata: { eventGroupId: "test-group-2" },
+          },
+        ]
+
+        redis.del = jest.fn()
+
+        await eventBus.emit(events, options)
+
+        // Expect 1 event to have been send
+        // Expect 2 pushes to redis as there are 2 groups of events to push
+        expect(queue.addBulk).toHaveBeenCalledTimes(1)
+        expect(redis.rpush).toHaveBeenCalledTimes(2)
+        expect(redis.del).not.toHaveBeenCalled()
+
+        const [testGroup1Event] = (eventBus as any).buildEvents(
+          [events[0]],
+          options
+        )
+        const [testGroup2Event] = (eventBus as any).buildEvents(
+          [events[2]],
+          options
+        )
+        const [testGroup2Event2] = (eventBus as any).buildEvents(
+          [events[3]],
+          options
+        )
+
+        redis.lrange = jest.fn((key) => {
+          if (key === "staging:test-group-1") {
+            return Promise.resolve([JSON.stringify(testGroup1Event)])
+          }
+
+          if (key === "staging:test-group-2") {
+            return Promise.resolve([
+              JSON.stringify(testGroup2Event),
+              JSON.stringify(testGroup2Event2),
+            ])
+          }
+        })
+
+        queue = (eventBus as any).queue_
+        queue.addBulk = jest.fn()
+
+        await eventBus.releaseGroupedEvents("test-group-1")
+
+        expect(queue.addBulk).toHaveBeenCalledTimes(1)
+        expect(queue.addBulk).toHaveBeenCalledWith([testGroup1Event])
+        expect(redis.del).toHaveBeenCalledTimes(1)
+        expect(redis.del).toHaveBeenCalledWith("staging:test-group-1")
+
+        queue = (eventBus as any).queue_
+        queue.addBulk = jest.fn()
+        redis.del = jest.fn()
+
+        await eventBus.releaseGroupedEvents("test-group-2")
+
+        expect(queue.addBulk).toHaveBeenCalledTimes(1)
+        expect(queue.addBulk).toHaveBeenCalledWith([
+          testGroup2Event,
+          testGroup2Event2,
+        ])
+        expect(redis.del).toHaveBeenCalledTimes(1)
+        expect(redis.del).toHaveBeenCalledWith("staging:test-group-2")
+      })
+    })
+  })
+
+  describe("worker_", () => {
+    let result
+
+    describe("Successfully processes the jobs", () => {
+      beforeEach(async () => {
+        jest.clearAllMocks()
+
+        eventBus = new RedisEventBusService(moduleDeps, simpleModuleOptions, {
+          resources: "shared",
+          scope: "internal",
+        })
+      })
+
+      it("should process a simple event with no options", async () => {
+        const test: string[] = []
+
+        eventBus.subscribe("eventName", () => {
+          test.push("success")
+
+          return Promise.resolve()
+        })
+
+        // TODO: The typing for this is all over the place
+        await eventBus.worker_({
+          data: { eventName: "eventName", data: { test: 1 } },
+          opts: { attempts: 1 },
+        } as any)
+
+        expect(loggerMock.info).toHaveBeenCalledTimes(1)
+        expect(loggerMock.info).toHaveBeenCalledWith(
+          "Processing eventName which has 1 subscribers"
+        )
+
+        expect(test).toEqual(["success"])
+      })
+
+      it("should process event with failing subscribers", async () => {
+        const test: string[] = []
+
+        eventBus.subscribe("eventName", () => {
+          test.push("hi")
+          return Promise.resolve()
+        })
+        eventBus.subscribe("eventName", () => {
+          test.push("fail1")
+          return Promise.reject("fail1")
+        })
+        eventBus.subscribe("eventName", () => {
+          test.push("hi2")
+          return Promise.resolve()
+        })
+        eventBus.subscribe("eventName", () => {
+          test.push("fail2")
+          return Promise.reject("fail2")
+        })
+
+        result = await eventBus.worker_({
+          data: { eventName: "eventName", data: { test: 1 } },
+          opts: { attempts: 1 },
+          update: (data) => data,
+        } as any)
+
+        expect(loggerMock.info).toHaveBeenCalledTimes(1)
+        expect(loggerMock.info).toHaveBeenCalledWith(
+          "Processing eventName which has 4 subscribers"
+        )
+
+        expect(loggerMock.warn).toHaveBeenCalledTimes(3)
+        expect(loggerMock.warn).toHaveBeenCalledWith(
+          "An error occurred while processing eventName: fail1"
+        )
+        expect(loggerMock.warn).toHaveBeenCalledWith(
+          "An error occurred while processing eventName: fail2"
+        )
+
+        expect(loggerMock.warn).toHaveBeenCalledWith(
+          "One or more subscribers of eventName failed. Retrying is not configured. Use 'attempts' option when emitting events."
+        )
+
+        expect(test.sort()).toEqual(["hi", "fail1", "hi2", "fail2"].sort())
+      })
+
+      it("should retry processing when subcribers fail, if configured - final attempt", async () => {
+        eventBus.subscribe("eventName", async () => Promise.resolve(), {
+          subscriberId: "1",
+        })
+        eventBus.subscribe("eventName", async () => Promise.reject("fail1"), {
+          subscriberId: "2",
+        })
+
+        result = await eventBus
+          .worker_({
+            data: {
+              eventName: "eventName",
+              data: {},
+              completedSubscriberIds: ["1"],
+            },
+            attemptsMade: 2,
+            update: (data) => data,
+            opts: { attempts: 2 },
+          } as any)
+          .catch((error) => void 0)
+
+        expect(loggerMock.warn).toHaveBeenCalledTimes(1)
+        expect(loggerMock.warn).toHaveBeenCalledWith(
+          "An error occurred while processing eventName: fail1"
+        )
+
+        expect(loggerMock.info).toHaveBeenCalledTimes(2)
+        expect(loggerMock.info).toHaveBeenCalledWith(
+          "Final retry attempt for eventName"
+        )
+        expect(loggerMock.info).toHaveBeenCalledWith(
+          "Retrying eventName which has 2 subscribers (1 of them failed)"
+        )
+      })
+
+      it("should retry processing when subcribers fail, if configured", async () => {
+        eventBus.subscribe("eventName", async () => Promise.resolve(), {
+          subscriberId: "1",
+        })
+        eventBus.subscribe("eventName", async () => Promise.reject("fail1"), {
+          subscriberId: "2",
+        })
+
+        result = await eventBus
+          .worker_({
+            data: {
+              eventName: "eventName",
+              data: {},
+              completedSubscriberIds: ["1"],
+            },
+            attemptsMade: 2,
+            updateData: (data) => data,
+            opts: { attempts: 3 },
+          } as any)
+          .catch((err) => void 0)
+
+        expect(loggerMock.warn).toHaveBeenCalledTimes(2)
+        expect(loggerMock.warn).toHaveBeenCalledWith(
+          "An error occurred while processing eventName: fail1"
+        )
+        expect(loggerMock.warn).toHaveBeenCalledWith(
+          "One or more subscribers of eventName failed. Retrying..."
+        )
+
+        expect(loggerMock.info).toHaveBeenCalledTimes(1)
+        expect(loggerMock.info).toHaveBeenCalledWith(
+          "Retrying eventName which has 2 subscribers (1 of them failed)"
+        )
+      })
+    })
+  })
+})

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

@@ -1,15 +1,25 @@
 import { InternalModuleDeclaration } from "@medusajs/modules-sdk"
-import { EmitData, Logger, Message } from "@medusajs/types"
+import { Logger, Message, MessageBody } from "@medusajs/types"
-import { AbstractEventBusModuleService, isString } from "@medusajs/utils"
+import {
-import { BulkJobOptions, JobsOptions, Queue, Worker } from "bullmq"
+  AbstractEventBusModuleService,
+  isPresent,
+  promiseAll,
+} from "@medusajs/utils"
+import { BulkJobOptions, Queue, Worker } from "bullmq"
 import { Redis } from "ioredis"
-import { BullJob, EmitOptions, EventBusRedisModuleOptions } from "../types"
+import { BullJob, EventBusRedisModuleOptions } from "../types"
 
 type InjectedDependencies = {
   logger: Logger
   eventBusRedisConnection: Redis
 }
 
+type IORedisEventType<T = unknown> = {
+  name: string
+  data: MessageBody<T>
+  opts: BulkJobOptions
+}
+
 /**
  * Can keep track of multiple subscribers to different events and run the
  * subscribers when events happen. Events will run asynchronously.
@@ -71,78 +81,137 @@ export default class RedisEventBusService extends AbstractEventBusModuleService
     },
   }
 
-  /**
+  private buildEvents<T>(
-   * Emit a single event
+    eventsData: Message<T>[],
-   * @param {string} eventName - the name of the event to be process.
+    options: BulkJobOptions = {}
-   * @param data - the data to send to the subscriber.
+  ): IORedisEventType<T>[] {
-   * @param options - options to add the job with
-   */
-  async emit<T>(
-    eventName: string,
-    data: T,
-    options: Record<string, unknown>
-  ): Promise<void>
-
-  /**
-   * Emit a number of events
-   * @param {EmitData} data - the data to send to the subscriber.
-   */
-  async emit<T>(data: EmitData<T>[]): Promise<void>
-
-  async emit<T>(data: Message<T>[]): Promise<void>
-
-  async emit<T, TInput extends string | EmitData<T>[] | Message<T>[] = string>(
-    eventNameOrData: TInput,
-    data?: T,
-    options: BulkJobOptions | JobsOptions = {}
-  ): Promise<void> {
-    const globalJobOptions = this.moduleOptions_.jobOptions ?? {}
-
-    const isBulkEmit = Array.isArray(eventNameOrData)
-
     const opts = {
       // default options
       removeOnComplete: true,
       attempts: 1,
       // global options
-      ...globalJobOptions,
+      ...(this.moduleOptions_.jobOptions ?? {}),
-    } as EmitOptions
+      ...options,
+    }
 
-    const dataBody = isString(eventNameOrData)
+    return eventsData.map((eventData) => {
-      ? data ?? (data as Message<T>).body
+      const { options, ...eventBody } = eventData
-      : undefined
 
-    const events = isBulkEmit
+      return {
-      ? eventNameOrData.map((event) => ({
+        name: eventData.eventName,
-          name: event.eventName,
+        data: eventBody,
-          data: {
+        opts: {
-            eventName: event.eventName,
+          // options for event group
-            data: (event as EmitData).data ?? (event as Message<T>).body,
+          ...opts,
-          },
+          // options for a particular event
-          opts: {
+          ...options,
-            ...opts,
+        },
-            // local options
+      }
-            ...event.options,
+    })
-          },
-        }))
-      : [
-          {
-            name: eventNameOrData as string,
-            data: { eventName: eventNameOrData, data: dataBody },
-            opts: {
-              ...opts,
-              // local options
-              ...options,
-            },
-          },
-        ]
-
-    await this.queue_.addBulk(events)
   }
 
-  // TODO: Implement redis based staging + release
+  /**
-  async releaseGroupedEvents(eventGroupId: string) {}
+   * Emit a single or number of events
-  async clearGroupedEvents(eventGroupId: string) {}
+   * @param {Message} data - the data to send to the subscriber.
+   * @param {BulkJobOptions} data - the options to add to bull mq
+   */
+  async emit<T = unknown>(
+    eventsData: Message<T> | Message<T>[],
+    options: BulkJobOptions & { groupedEventsTTL?: number } = {}
+  ): Promise<void> {
+    let eventsDataArray = Array.isArray(eventsData) ? eventsData : [eventsData]
+
+    const { groupedEventsTTL = 600 } = options
+    delete options.groupedEventsTTL
+
+    const eventsToEmit = eventsDataArray.filter(
+      (eventData) => !isPresent(eventData.metadata?.eventGroupId)
+    )
+
+    const eventsToGroup = eventsDataArray.filter((eventData) =>
+      isPresent(eventData.metadata?.eventGroupId)
+    )
+
+    const groupEventsMap = new Map<string, Message<T>[]>()
+
+    for (const event of eventsToGroup) {
+      const groupId = event.metadata?.eventGroupId!
+      const array = groupEventsMap.get(groupId) ?? []
+
+      array.push(event)
+      groupEventsMap.set(groupId, array)
+    }
+
+    const promises: Promise<unknown>[] = []
+
+    if (eventsToEmit.length) {
+      const emitData = this.buildEvents(eventsToEmit, options)
+
+      promises.push(this.queue_.addBulk(emitData))
+    }
+
+    for (const [groupId, events] of groupEventsMap.entries()) {
+      if (!events?.length) {
+        continue
+      }
+
+      // Set a TTL for the key of the list that is scoped to a group
+      // This will be helpful in preventing stale data from staying in redis for too long
+      // in the event the module fails to cleanup events. For long running workflows, setting a much higher
+      // TTL or even skipping the TTL would be required
+      this.setExpire(groupId, groupedEventsTTL)
+
+      const eventsData = this.buildEvents(events, options)
+
+      promises.push(this.groupEvents(groupId, eventsData))
+    }
+
+    await promiseAll(promises)
+  }
+
+  private async setExpire(eventGroupId: string, ttl: number) {
+    if (!eventGroupId) {
+      return
+    }
+
+    await this.eventBusRedisConnection_.expire(`staging:${eventGroupId}`, ttl)
+  }
+
+  private async groupEvents<T = unknown>(
+    eventGroupId: string,
+    events: IORedisEventType<T>[]
+  ) {
+    await this.eventBusRedisConnection_.rpush(
+      `staging:${eventGroupId}`,
+      ...events.map((event) => JSON.stringify(event))
+    )
+  }
+
+  private async getGroupedEvents(
+    eventGroupId: string
+  ): Promise<IORedisEventType[]> {
+    return await this.eventBusRedisConnection_
+      .lrange(`staging:${eventGroupId}`, 0, -1)
+      .then((result) => {
+        return result.map((jsonString) => JSON.parse(jsonString))
+      })
+  }
+
+  async releaseGroupedEvents(eventGroupId: string) {
+    const groupedEvents = await this.getGroupedEvents(eventGroupId)
+
+    await this.queue_.addBulk(groupedEvents)
+
+    await this.clearGroupedEvents(eventGroupId)
+  }
+
+  async clearGroupedEvents(eventGroupId: string) {
+    if (!eventGroupId) {
+      return
+    }
+
+    await this.eventBusRedisConnection_.del(`staging:${eventGroupId}`)
+  }
 
   /**
    * Handles incoming jobs.

packages/modules/fulfillment/integration-tests/__fixtures__/events.ts

@@ -10,15 +10,13 @@ export function buildExpectedEventMessageShape(options: {
 }): EventBusTypes.Message {
   return {
     eventName: options.eventName,
-    body: {
+    metadata: {
-      metadata: {
+      action: options.action,
-        action: options.action,
+      eventGroupId: options.eventGroupId,
-        eventGroupId: options.eventGroupId,
+      source: "fulfillment",
-        source: "fulfillment",
+      object: options.object,
-        object: options.object,
-      },
-      data: options.data,
     },
+    data: options.data,
     options: options.options,
   }
 }

packages/modules/link-modules/src/services/link-module-service.ts

@@ -208,15 +208,13 @@ export default class LinkModuleService<TLink> implements ILinkModule {
     await this.eventBusModuleService_?.emit<Record<string, unknown>>(
       (data as { id: unknown }[]).map(({ id }) => ({
         eventName: this.entityName_ + "." + CommonEvents.ATTACHED,
-        body: {
+        metadata: {
-          metadata: {
+          source: this.serviceName_,
-            source: this.serviceName_,
+          action: CommonEvents.ATTACHED,
-            action: CommonEvents.ATTACHED,
+          object: this.entityName_,
-            object: this.entityName_,
+          eventGroupId: sharedContext.eventGroupId,
-            eventGroupId: sharedContext.eventGroupId,
-          },
-          data: { id },
         },
+        data: { id },
       }))
     )
 
@@ -261,15 +259,13 @@ export default class LinkModuleService<TLink> implements ILinkModule {
     await this.eventBusModuleService_?.emit<Record<string, unknown>>(
       allData.map(({ id }) => ({
         eventName: this.entityName_ + "." + CommonEvents.DETACHED,
-        body: {
+        metadata: {
-          metadata: {
+          source: this.serviceName_,
-            source: this.serviceName_,
+          action: CommonEvents.DETACHED,
-            action: CommonEvents.DETACHED,
+          object: this.entityName_,
-            object: this.entityName_,
+          eventGroupId: sharedContext.eventGroupId,
-            eventGroupId: sharedContext.eventGroupId,
-          },
-          data: { id },
         },
+        data: { id },
       }))
     )
   }
@@ -312,15 +308,13 @@ export default class LinkModuleService<TLink> implements ILinkModule {
     await this.eventBusModuleService_?.emit<Record<string, unknown>>(
       (deletedEntities as { id: string }[]).map(({ id }) => ({
         eventName: this.entityName_ + "." + CommonEvents.DETACHED,
-        body: {
+        metadata: {
-          metadata: {
+          source: this.serviceName_,
-            source: this.serviceName_,
+          action: CommonEvents.DETACHED,
-            action: CommonEvents.DETACHED,
+          object: this.entityName_,
-            object: this.entityName_,
+          eventGroupId: sharedContext.eventGroupId,
-            eventGroupId: sharedContext.eventGroupId,
-          },
-          data: { id },
         },
+        data: { id },
       }))
     )
 
@@ -372,15 +366,13 @@ export default class LinkModuleService<TLink> implements ILinkModule {
     await this.eventBusModuleService_?.emit<Record<string, unknown>>(
       (restoredEntities as { id: string }[]).map(({ id }) => ({
         eventName: this.entityName_ + "." + CommonEvents.ATTACHED,
-        body: {
+        metadata: {
-          metadata: {
+          source: this.serviceName_,
-            source: this.serviceName_,
+          action: CommonEvents.ATTACHED,
-            action: CommonEvents.ATTACHED,
+          object: this.entityName_,
-            object: this.entityName_,
+          eventGroupId: sharedContext.eventGroupId,
-            eventGroupId: sharedContext.eventGroupId,
-          },
-          data: { id },
         },
+        data: { id },
       }))
     )
 

packages/modules/product/integration-tests/__tests__/product-module-service/product-categories.spec.ts

@@ -291,8 +291,9 @@ moduleIntegrationTestRunner<IProductModuleService>({
           })
 
           expect(eventBusSpy).toHaveBeenCalledTimes(1)
-          expect(eventBusSpy).toHaveBeenCalledWith("product-category.created", {
+          expect(eventBusSpy).toHaveBeenCalledWith({
-            id: category.id,
+            data: { id: category.id },
+            eventName: "product-category.created",
           })
         })
 
@@ -380,8 +381,9 @@ moduleIntegrationTestRunner<IProductModuleService>({
           })
 
           expect(eventBusSpy).toHaveBeenCalledTimes(1)
-          expect(eventBusSpy).toHaveBeenCalledWith("product-category.updated", {
+          expect(eventBusSpy).toHaveBeenCalledWith({
-            id: productCategoryZero.id,
+            data: { id: productCategoryZero.id },
+            eventName: "product-category.updated",
           })
         })
 
@@ -547,8 +549,9 @@ moduleIntegrationTestRunner<IProductModuleService>({
           await service.deleteCategory(productCategoryOne.id)
 
           expect(eventBusSpy).toHaveBeenCalledTimes(1)
-          expect(eventBusSpy).toHaveBeenCalledWith("product-category.deleted", {
+          expect(eventBusSpy).toHaveBeenCalledWith({
-            id: productCategoryOne.id,
+            data: { id: productCategoryOne.id },
+            eventName: "product-category.deleted",
           })
         })
 

packages/modules/product/integration-tests/__tests__/product-module-service/product-collections.spec.ts

@@ -298,8 +298,8 @@ moduleIntegrationTestRunner<IProductModuleService>({
           expect(eventBusSpy).toHaveBeenCalledTimes(1)
           expect(eventBusSpy).toHaveBeenCalledWith([
             {
-              eventName: "product-collection.updated",
               data: { id: collectionId },
+              eventName: "product-collection.updated",
             },
           ])
         })
@@ -488,16 +488,14 @@ moduleIntegrationTestRunner<IProductModuleService>({
           const eventBusSpy = jest.spyOn(MockEventBusService.prototype, "emit")
 
           const collections = await service.createCollections([
-            {
+            { title: "New Collection" },
-              title: "New Collection",
-            },
           ])
 
           expect(eventBusSpy).toHaveBeenCalledTimes(1)
           expect(eventBusSpy).toHaveBeenCalledWith([
             {
-              eventName: "product-collection.created",
               data: { id: collections[0].id },
+              eventName: "product-collection.created",
             },
           ])
         })

packages/modules/product/src/services/product-module-service.ts

@@ -51,8 +51,8 @@ import {
   UpdateTagInput,
   UpdateTypeInput,
 } from "../types"
-import { entityNameToLinkableKeysMap, joinerConfig } from "./../joiner-config"
 import { eventBuilders } from "../utils"
+import { entityNameToLinkableKeysMap, joinerConfig } from "./../joiner-config"
 
 type InjectedDependencies = {
   baseRepository: DAL.RepositoryService
@@ -1134,10 +1134,10 @@ export default class ProductModuleService<
       sharedContext
     )
 
-    await this.eventBusModuleService_?.emit<ProductCategoryEventData>(
+    await this.eventBusModuleService_?.emit<ProductCategoryEventData>({
-      ProductCategoryEvents.CATEGORY_CREATED,
+      eventName: ProductCategoryEvents.CATEGORY_CREATED,
-      { id: productCategory.id }
+      data: { id: productCategory.id },
-    )
+    })
 
     return productCategory
   }
@@ -1154,10 +1154,10 @@ export default class ProductModuleService<
       sharedContext
     )
 
-    await this.eventBusModuleService_?.emit<ProductCategoryEventData>(
+    await this.eventBusModuleService_?.emit<ProductCategoryEventData>({
-      ProductCategoryEvents.CATEGORY_UPDATED,
+      eventName: ProductCategoryEvents.CATEGORY_UPDATED,
-      { id: productCategory.id }
+      data: { id: productCategory.id },
-    )
+    })
 
     return await this.baseRepository_.serialize(productCategory, {
       populate: true,
@@ -1171,10 +1171,10 @@ export default class ProductModuleService<
   ): Promise<void> {
     await this.productCategoryService_.delete(categoryId, sharedContext)
 
-    await this.eventBusModuleService_?.emit<ProductCategoryEventData>(
+    await this.eventBusModuleService_?.emit<ProductCategoryEventData>({
-      ProductCategoryEvents.CATEGORY_DELETED,
+      eventName: ProductCategoryEvents.CATEGORY_DELETED,
-      { id: categoryId }
+      data: { id: categoryId },
-    )
+    })
   }
 
   create(

packages/modules/user/integration-tests/__tests__/services/module/invite.spec.ts

@@ -1,9 +1,12 @@
-import { IUserModuleService } from "@medusajs/types/dist/user"
-import { MockEventBusService } from "medusa-test-utils"
 import { Modules } from "@medusajs/modules-sdk"
+import { IUserModuleService } from "@medusajs/types/dist/user"
 import { UserEvents } from "@medusajs/utils"
+import {
+  MockEventBusService,
+  moduleIntegrationTestRunner,
+  SuiteOptions,
+} from "medusa-test-utils"
 import { createInvites } from "../../../__fixtures__/invite"
-import { moduleIntegrationTestRunner, SuiteOptions } from "medusa-test-utils"
 
 jest.setTimeout(30000)
 
@@ -178,9 +181,7 @@ moduleIntegrationTestRunner({
           expect(eventBusSpy).toHaveBeenCalledTimes(1)
           expect(eventBusSpy).toHaveBeenCalledWith([
             expect.objectContaining({
-              body: expect.objectContaining({
+              data: { id: "1" },
-                data: { id: "1" },
-              }),
               eventName: UserEvents.invite_updated,
             }),
           ])
@@ -197,9 +198,7 @@ moduleIntegrationTestRunner({
           expect(eventBusSpy).toHaveBeenCalledTimes(1)
           expect(eventBusSpy).toHaveBeenCalledWith([
             expect.objectContaining({
-              body: expect.objectContaining({
+              data: { id: "1" },
-                data: { id: "1" },
-              }),
               eventName: UserEvents.invite_token_generated,
             }),
           ])
@@ -228,27 +227,19 @@ moduleIntegrationTestRunner({
           expect(eventBusSpy).toHaveBeenCalledTimes(1)
           expect(eventBusSpy).toHaveBeenCalledWith([
             expect.objectContaining({
-              body: expect.objectContaining({
+              data: { id: "1" },
-                data: { id: "1" },
-              }),
               eventName: UserEvents.invite_created,
             }),
             expect.objectContaining({
-              body: expect.objectContaining({
+              data: { id: "2" },
-                data: { id: "2" },
-              }),
               eventName: UserEvents.invite_created,
             }),
             expect.objectContaining({
-              body: expect.objectContaining({
+              data: { id: "1" },
-                data: { id: "1" },
-              }),
               eventName: UserEvents.invite_token_generated,
             }),
             expect.objectContaining({
-              body: expect.objectContaining({
+              data: { id: "2" },
-                data: { id: "2" },
-              }),
               eventName: UserEvents.invite_token_generated,
             }),
           ])

packages/modules/user/integration-tests/__tests__/services/module/user.spec.ts

@@ -1,9 +1,12 @@
-import { IUserModuleService } from "@medusajs/types/dist/user"
-import { MockEventBusService } from "medusa-test-utils"
 import { Modules } from "@medusajs/modules-sdk"
+import { IUserModuleService } from "@medusajs/types/dist/user"
 import { UserEvents } from "@medusajs/utils"
+import {
+  MockEventBusService,
+  moduleIntegrationTestRunner,
+  SuiteOptions,
+} from "medusa-test-utils"
 import { createUsers } from "../../../__fixtures__/user"
-import { moduleIntegrationTestRunner, SuiteOptions } from "medusa-test-utils"
 
 jest.setTimeout(30000)
 
@@ -190,9 +193,7 @@ moduleIntegrationTestRunner({
           expect(eventBusSpy).toHaveBeenCalledTimes(1)
           expect(eventBusSpy).toHaveBeenCalledWith([
             expect.objectContaining({
-              body: expect.objectContaining({
+              data: { id: "1" },
-                data: { id: "1" },
-              }),
               eventName: UserEvents.updated,
             }),
           ])
@@ -222,15 +223,11 @@ moduleIntegrationTestRunner({
           expect(eventBusSpy).toHaveBeenCalledTimes(1)
           expect(eventBusSpy).toHaveBeenCalledWith([
             expect.objectContaining({
-              body: expect.objectContaining({
+              data: { id: "1" },
-                data: { id: "1" },
-              }),
               eventName: UserEvents.created,
             }),
             expect.objectContaining({
-              body: expect.objectContaining({
+              data: { id: "2" },
-                data: { id: "2" },
-              }),
               eventName: UserEvents.created,
             }),
           ])