feat: Add emitEvent step + cleanup (#7643)

* feat: Add emitEvent step + cleanup

* fix typo

* fix typo

packages/core/core-flows/src/common/index.ts

@@ -1,6 +1,7 @@
 export * from "./steps/create-remote-links"
 export * from "./steps/dismiss-remote-links"
 export * from "./steps/remove-remote-links"
+export * from "./steps/emit-event"
 export * from "./steps/use-remote-query"
 export * from "./workflows/batch-links"
 export * from "./workflows/create-links"

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

@@ -0,0 +1,41 @@
+import { composeMessage, ModuleRegistrationName } from "@medusajs/utils"
+import { createStep, StepExecutionContext } from "@medusajs/workflows-sdk"
+
+type Input = {
+  eventName: string
+  source: string
+  object: string
+  action?: string
+  options?: Record<string, any>
+  data: (
+    context: StepExecutionContext
+  ) => Promise<Record<any, any>> | Record<any, any>
+}
+
+export const emitEventStepId = "emit-event-step"
+export const emitEventStep = createStep(
+  emitEventStepId,
+  async (input: Input, context) => {
+    if (!input) {
+      return
+    }
+
+    const { container } = context
+
+    const eventBus = container.resolve(ModuleRegistrationName.EVENT_BUS)
+
+    const data_ =
+      typeof input.data === "function" ? await input.data(context) : input.data
+    const message = composeMessage(input.eventName, {
+      data: data_,
+      action: input.action ?? "",
+      object: input.object,
+      source: input.source,
+      options: input.options,
+      context,
+    })
+
+    await eventBus.emit([message])
+  },
+  async (data: void) => {}
+)

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

@@ -0,0 +1,20 @@
+import { ModuleRegistrationName } from "@medusajs/utils"
+import { createStep } from "@medusajs/workflows-sdk"
+
+export const releaseEventsStepId = "release-events-step"
+export const releaseEventsStep = createStep(
+  releaseEventsStepId,
+  async (
+    input: void,
+    {
+      container,
+      metadata: {
+        /* eventGroupId */
+      },
+    }
+  ) => {
+    const eventBus = container.resolve(ModuleRegistrationName.EVENT_BUS)
+    // await eventBus.release
+  },
+  async (data: void) => {}
+)

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

@@ -27,7 +27,7 @@ export type EmitData<T = unknown> = {
 
 export type MessageBody<T = unknown> = {
   metadata: {
-    service: string
+    source: string
     action: string
     object: string
     eventGroupId?: string
@@ -44,7 +44,7 @@ export type Message<T = unknown> = {
 export type RawMessageFormat<T = any> = {
   eventName: string
   data: T
-  service: string
+  source: string
   object: string
   action?: string
   context?: Pick<Context, "eventGroupId">

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

@@ -11,7 +11,7 @@ describe("MessageAggregator", function () {
       eventName: "ProductVariant.created",
       body: {
         metadata: {
-          service: "ProductService",
+          source: "ProductService",
           action: "created",
           object: "ProductVariant",
           eventGroupId: "1",
@@ -23,7 +23,7 @@ describe("MessageAggregator", function () {
       eventName: "Product.created",
       body: {
         metadata: {
-          service: "ProductService",
+          source: "ProductService",
           action: "created",
           object: "Product",
           eventGroupId: "1",
@@ -35,7 +35,7 @@ describe("MessageAggregator", function () {
       eventName: "ProductVariant.created",
       body: {
         metadata: {
-          service: "ProductService",
+          source: "ProductService",
           action: "created",
           object: "ProductVariant",
           eventGroupId: "1",
@@ -47,7 +47,7 @@ describe("MessageAggregator", function () {
       eventName: "ProductType.detached",
       body: {
         metadata: {
-          service: "ProductService",
+          source: "ProductService",
           action: "detached",
           object: "ProductType",
           eventGroupId: "1",
@@ -59,7 +59,7 @@ describe("MessageAggregator", function () {
       eventName: "ProductVariant.updated",
       body: {
         metadata: {
-          service: "ProductService",
+          source: "ProductService",
           action: "updated",
           object: "ProductVariant",
           eventGroupId: "1",
@@ -87,7 +87,7 @@ describe("MessageAggregator", function () {
         eventName: "ProductType.detached",
         body: {
           metadata: {
-            service: "ProductService",
+            source: "ProductService",
             action: "detached",
             object: "ProductType",
             eventGroupId: "1",
@@ -102,7 +102,7 @@ describe("MessageAggregator", function () {
         eventName: "ProductVariant.updated",
         body: {
           metadata: {
-            service: "ProductService",
+            source: "ProductService",
             action: "updated",
             object: "ProductVariant",
             eventGroupId: "1",
@@ -117,7 +117,7 @@ describe("MessageAggregator", function () {
         eventName: "ProductVariant.created",
         body: {
           metadata: {
-            service: "ProductService",
+            source: "ProductService",
             action: "created",
             object: "ProductVariant",
             eventGroupId: "1",
@@ -129,7 +129,7 @@ describe("MessageAggregator", function () {
         eventName: "ProductVariant.created",
         body: {
           metadata: {
-            service: "ProductService",
+            source: "ProductService",
             action: "created",
             object: "ProductVariant",
             eventGroupId: "1",
@@ -144,7 +144,7 @@ describe("MessageAggregator", function () {
         eventName: "Product.created",
         body: {
           metadata: {
-            service: "ProductService",
+            source: "ProductService",
             action: "created",
             object: "Product",
             eventGroupId: "1",

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

@@ -12,14 +12,14 @@ export function composeMessage(
   eventName: string,
   {
     data,
-    service,
+    source,
     object,
     action,
     context = {},
     options,
   }: {
     data: any
-    service: string
+    source: string
     object: string
     action?: string
     context?: Context
@@ -34,7 +34,7 @@ export function composeMessage(
   }
 
   const metadata: EventBusTypes.MessageBody["metadata"] = {
-    service,
+    source,
     object,
     action: act!,
   }

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

@@ -40,7 +40,7 @@ export class MessageAggregator implements IMessageAggregator {
     const composedMessages = messages.map((message) => {
       return composeMessage(message.eventName, {
         data: message.data,
-        service: message.service,
+        source: message.source,
         object: message.object,
         action: message.action,
         options,

packages/core/utils/src/modules-sdk/event-builder-factory.ts

@@ -5,7 +5,7 @@ import { Context, EventBusTypes } from "@medusajs/types"
  *
  * @example
  * const createdFulfillment = eventBuilderFactory({
- *   service: Modules.FULFILLMENT,
+ *   source: Modules.FULFILLMENT,
  *   action: CommonEvents.CREATED,
  *   object: "fulfillment",
  *   eventsEnum: FulfillmentEvents,
@@ -27,13 +27,13 @@ export function eventBuilderFactory({
   action,
   object,
   eventsEnum,
-  service,
+  source,
 }: {
   isMainEntity?: boolean
   action: string
   object: string
   eventsEnum: Record<string, string>
-  service: string
+  source: string
 }) {
   return function ({
     data,
@@ -51,7 +51,7 @@ export function eventBuilderFactory({
 
     data.forEach((dataItem) => {
       messages.push({
-        service,
+        source,
         action,
         context: sharedContext,
         data: { id: dataItem.id },