feat: event aggregator (#6218)

What:
- Event Aggregator Util
- Preparation for normalizing event in a new format (backward compatible with the current format)
- GQL Schema to joiner config and some Entities configured
- Link modules emmiting events

packages/medusa/src/interfaces/__tests__/event-bus-service.spec.ts

@@ -16,8 +16,15 @@ class EventBus extends EventBusUtils.AbstractEventBusModuleService {
     options: Record<string, unknown>
   ): Promise<void>
   async emit<T>(data: EventBusTypes.EmitData<T>[]): Promise<void>
+  async emit<T>(data: EventBusTypes.Message<T>[]): Promise<void>
 
-  async emit<T, TInput extends string | EventBusTypes.EmitData<T>[] = string>(
+  async emit<
+    T,
+    TInput extends
+      | string
+      | EventBusTypes.EmitData<T>[]
+      | EventBusTypes.Message<T>[] = string
+  >(
     eventOrData: TInput,
     data?: T,
     options: Record<string, unknown> = {}

packages/medusa/src/joiner-configs/publishable-api-key-service.ts

@@ -20,7 +20,9 @@ export default {
   alias: [
     {
       name: ["publishable_api_key", "publishable_api_keys"],
-      args: { entity: "PublishableApiKey" },
+      args: {
+        entity: "PublishableApiKey",
+      },
     },
   ],
 } as ModuleJoinerConfig

packages/medusa/src/joiner-configs/shipping-profile-service.ts

@@ -5,25 +5,25 @@ export default {
   primaryKeys: ["id"],
   linkableKeys: { profile_id: "ShippingProfile" },
   schema: `
-      scalar Date
-      scalar JSON
-      
-      type ShippingProfile {
-        id: ID!
-        name: String!
-        type: String!
-        created_at: Date!
-        updated_at: Date!  
-        deleted_at: Date
-        metadata: JSON
-      }
-    `,
+    scalar Date
+    scalar JSON
+    
+    type ShippingProfile {
+      id: ID!
+      name: String!
+      type: String!
+      created_at: Date!
+      updated_at: Date!  
+      deleted_at: Date
+      metadata: JSON
+    }
+  `,
   alias: [
     {
-      name: "shipping_profile",
-    },
-    {
-      name: "shipping_profiles",
+      name: ["shipping_profile", "shipping_profiles"],
+      args: {
+        entity: "ShippingProfile",
+      },
     },
   ],
 } as ModuleJoinerConfig

packages/medusa/src/loaders/helpers/subscribers/index.ts

@@ -188,7 +188,7 @@ export class SubscriberLoader {
 
     const events = Array.isArray(event) ? event : [event]
 
-    const subscriber: Subscriber<T> = async (data: T, eventName: string) => {
+    const subscriber = async (data: T, eventName: string) => {
       return handler({
         eventName,
         data,
@@ -200,7 +200,7 @@ export class SubscriberLoader {
     const subscriberId = this.inferIdentifier(fileName, config, handler)
 
     for (const e of events) {
-      eventBusService.subscribe(e, subscriber as Subscriber<unknown>, {
+      eventBusService.subscribe(e, subscriber as Subscriber, {
         ...(config.context ?? {}),
         subscriberId,
       })

packages/medusa/src/loaders/medusa-app.ts

@@ -128,7 +128,10 @@ export const loadMedusaApp = async (
     return medusaApp
   }
 
-  container.register("remoteLink", asValue(medusaApp.link))
+  container.register(
+    ContainerRegistrationKeys.REMOTE_LINK,
+    asValue(medusaApp.link)
+  )
   container.register(
     ContainerRegistrationKeys.REMOTE_QUERY,
     asValue(medusaApp.query)
@@ -143,7 +146,7 @@ export const loadMedusaApp = async (
 
   // Register all unresolved modules as undefined to be present in the container with undefined value by defaul
   // but still resolvable
-  for (const [, moduleDefinition] of Object.entries(ModulesDefinition)) {
+  for (const moduleDefinition of Object.values(ModulesDefinition)) {
     if (!container.hasRegistration(moduleDefinition.registrationName)) {
       container.register(moduleDefinition.registrationName, asValue(undefined))
     }

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

@@ -1,14 +1,14 @@
-import { EventBusTypes, Logger } from "@medusajs/types"
+import { EmitData, EventBusTypes, Logger, Message } from "@medusajs/types"
 import { DatabaseErrorCode, EventBusUtils } from "@medusajs/utils"
+import { EOL } from "os"
 import { EntityManager } from "typeorm"
 import { TransactionBaseService } from "../interfaces"
 import { StagedJob } from "../models"
+import { FindConfig } from "../types/common"
 import { ConfigModule } from "../types/global"
 import { isString } from "../utils"
 import { sleep } from "../utils/sleep"
 import StagedJobService from "./staged-job"
-import { FindConfig } from "../types/common"
-import { EOL } from "os"
 
 type InjectedDependencies = {
   stagedJobService: StagedJobService
@@ -118,6 +118,8 @@ export default class EventBusService
    */
   async emit<T>(data: EventBusTypes.EmitData<T>[]): Promise<StagedJob[] | void>
 
+  async emit<T>(data: EventBusTypes.Message<T>[]): Promise<StagedJob[] | void>
+
   /**
    * Calls all subscribers when an event occurs.
    * @param {string} eventName - the name of the event to be process.
@@ -133,7 +135,10 @@ export default class EventBusService
 
   async emit<
     T,
-    TInput extends string | EventBusTypes.EmitData<T>[] = string,
+    TInput extends
+      | string
+      | EventBusTypes.EmitData<T>[]
+      | EventBusTypes.Message<T>[] = string,
     TResult = TInput extends EventBusTypes.EmitData<T>[]
       ? StagedJob[]
       : StagedJob
@@ -144,16 +149,19 @@ export default class EventBusService
   ): Promise<TResult | void> {
     const manager = this.activeManager_
     const isBulkEmit = !isString(eventNameOrData)
+    const dataBody = isString(eventNameOrData)
+      ? data ?? (data as Message<T>).body
+      : undefined
     const events: EventBusTypes.EmitData[] = isBulkEmit
       ? eventNameOrData.map((event) => ({
           eventName: event.eventName,
-          data: event.data,
+          data: (event as EmitData).data ?? (event as Message<T>).body.data,
           options: event.options,
         }))
       : [
           {
             eventName: eventNameOrData,
-            data: data,
+            data: dataBody,
             options: options,
           },
         ]

packages/medusa/src/services/product-variant-inventory.ts

@@ -1,32 +1,40 @@
-import { EntityManager, In } from "typeorm"
 import {
   IEventBusService,
   IInventoryService,
+  IStockLocationService,
   InventoryItemDTO,
   InventoryLevelDTO,
-  IStockLocationService,
+  RemoteQueryFunction,
   ReservationItemDTO,
   ReserveQuantityContext,
 } from "@medusajs/types"
+import {
+  FlagRouter,
+  MedusaError,
+  MedusaV2Flag,
+  isDefined,
+  promiseAll,
+  remoteQueryObjectFromString,
+} from "@medusajs/utils"
+import { EntityManager, In } from "typeorm"
 import { LineItem, Product, ProductVariant } from "../models"
-import { isDefined, MedusaError, promiseAll } from "@medusajs/utils"
 import { PricedProduct, PricedVariant } from "../types/pricing"
 
+import { TransactionBaseService } from "../interfaces"
 import { ProductVariantInventoryItem } from "../models/product-variant-inventory-item"
+import { getSetDifference } from "../utils/diff-set"
 import ProductVariantService from "./product-variant"
 import SalesChannelInventoryService from "./sales-channel-inventory"
 import SalesChannelLocationService from "./sales-channel-location"
-import { TransactionBaseService } from "../interfaces"
-import { getSetDifference } from "../utils/diff-set"
 
 type InjectedDependencies = {
   manager: EntityManager
   salesChannelLocationService: SalesChannelLocationService
   salesChannelInventoryService: SalesChannelInventoryService
   productVariantService: ProductVariantService
-  stockLocationService: IStockLocationService
-  inventoryService: IInventoryService
   eventBusService: IEventBusService
+  featureFlagRouter: FlagRouter
+  remoteQuery: RemoteQueryFunction
 }
 
 type AvailabilityContext = {
@@ -42,6 +50,8 @@ class ProductVariantInventoryService extends TransactionBaseService {
   protected readonly salesChannelInventoryService_: SalesChannelInventoryService
   protected readonly productVariantService_: ProductVariantService
   protected readonly eventBusService_: IEventBusService
+  protected readonly featureFlagRouter_: FlagRouter
+  protected readonly remoteQuery_: RemoteQueryFunction
 
   protected get inventoryService_(): IInventoryService {
     return this.__container__.inventoryService
@@ -56,6 +66,8 @@ class ProductVariantInventoryService extends TransactionBaseService {
     salesChannelInventoryService,
     productVariantService,
     eventBusService,
+    featureFlagRouter,
+    remoteQuery,
   }: InjectedDependencies) {
     // eslint-disable-next-line prefer-rest-params
     super(arguments[0])
@@ -64,6 +76,8 @@ class ProductVariantInventoryService extends TransactionBaseService {
     this.salesChannelInventoryService_ = salesChannelInventoryService
     this.productVariantService_ = productVariantService
     this.eventBusService_ = eventBusService
+    this.featureFlagRouter_ = featureFlagRouter
+    this.remoteQuery_ = remoteQuery
   }
 
   /**
@@ -307,16 +321,29 @@ class ProductVariantInventoryService extends TransactionBaseService {
     }
 
     // Verify that variant exists
-    const variants = await this.productVariantService_
-      .withTransaction(this.activeManager_)
-      .list(
-        {
-          id: data.map((d) => d.variantId),
-        },
-        {
-          select: ["id"],
-        }
+    let variants
+    if (this.featureFlagRouter_.isFeatureEnabled(MedusaV2Flag.key)) {
+      variants = await this.remoteQuery_(
+        remoteQueryObjectFromString({
+          entryPoint: "variants",
+          variables: {
+            id: data.map((d) => d.variantId),
+          },
+          fields: ["id"],
+        })
       )
+    } else {
+      variants = await this.productVariantService_
+        .withTransaction(this.activeManager_)
+        .list(
+          {
+            id: data.map((d) => d.variantId),
+          },
+          {
+            select: ["id"],
+          }
+        )
+    }
 
     const foundVariantIds = new Set(variants.map((v) => v.id))
     const requestedVariantIds = new Set(data.map((v) => v.variantId))
@@ -404,7 +431,11 @@ class ProductVariantInventoryService extends TransactionBaseService {
       ): tc is ProductVariantInventoryItem => !!tc
     )
 
-    return await variantInventoryRepo.save(toCreate)
+    const createdVariantInventoryItems = await variantInventoryRepo.save(
+      toCreate
+    )
+
+    return createdVariantInventoryItems
   }
 
   /**