feat(modules-sdk, types, user, utils):init user module events (#6431)

* init user module events

* refactor utils

* undo test script update

* fix feedback

* add eventbus service to module test-runner

* add injected dependencies

* move events to utils

* use const eventname in tests

* rm withTransaction

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

@@ -4,4 +4,5 @@ export * as JestUtils from "./jest"
 export { default as MockManager } from "./mock-manager"
 export { default as MockRepository } from "./mock-repository"
 export * from "./init-modules"
+export { default as MockEventBusService } from "./mock-event-bus-service"
 export * from "./module-test-runner"

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

@@ -0,0 +1,41 @@
+import {
+  EmitData,
+  EventBusTypes,
+  IEventBusModuleService,
+  Message,
+  Subscriber,
+} from "@medusajs/types"
+
+export default class EventBusService implements IEventBusModuleService {
+  emit<T>(
+    eventName: string,
+    data: T,
+    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 {
+    return this
+  }
+
+  unsubscribe(
+    event: string | symbol,
+    subscriber: Subscriber,
+    context?: EventBusTypes.SubscriberContext
+  ): this {
+    return this
+  }
+}

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

@@ -1,7 +1,9 @@
-import { getDatabaseURL, getMikroOrmWrapper, TestDatabase } from "./database"
-import { MedusaAppOutput, ModulesDefinition } from "@medusajs/modules-sdk"
-import { initModules, InitModulesOptions } from "./init-modules"
 import { ContainerRegistrationKeys, ModulesSdkUtils } from "@medusajs/utils"
+import { InitModulesOptions, initModules } from "./init-modules"
+import { MedusaAppOutput, ModulesDefinition } from "@medusajs/modules-sdk"
+import { TestDatabase, getDatabaseURL, getMikroOrmWrapper } from "./database"
+
+import { MockEventBusService } from "."
 
 export interface SuiteOptions<TService = unknown> {
   MikroOrmWrapper: TestDatabase
@@ -20,12 +22,14 @@ export function moduleIntegrationTestRunner({
   schema = "public",
   debug = false,
   testSuite,
+  injectedDependencies = {},
 }: {
   moduleName: string
   moduleModels?: any[]
   joinerConfig?: any[]
   schema?: string
   dbName?: string
+  injectedDependencies?: Record<string, any>
   debug?: boolean
   testSuite: <TService = unknown>(options: SuiteOptions<TService>) => () => void
 }) {
@@ -65,6 +69,8 @@ export function moduleIntegrationTestRunner({
   const moduleOptions: InitModulesOptions = {
     injectedDependencies: {
       [ContainerRegistrationKeys.PG_CONNECTION]: connection,
+      eventBusService: new MockEventBusService(),
+      ...injectedDependencies,
     },
     modulesConfig: modulesConfig_,
     databaseConfig: dbConfig,

packages/modules-sdk/src/definitions.ts

@@ -266,7 +266,7 @@ export const ModulesDefinition: { [key: string | Modules]: ModuleDefinition } =
       label: upperCaseFirst(ModuleRegistrationName.USER),
       isRequired: false,
       isQueryable: true,
-      dependencies: ["logger"],
+      dependencies: [ModuleRegistrationName.EVENT_BUS, "logger"],
       defaultModuleDeclaration: {
         scope: MODULE_SCOPE.INTERNAL,
         resources: MODULE_RESOURCE_TYPE.SHARED,

packages/types/src/shared-context.ts

@@ -1,4 +1,5 @@
 import { EntityManager } from "typeorm"
+import { EventBusTypes } from "./bundles"
 import { Message } from "./event-bus"
 
 /**
@@ -27,6 +28,12 @@ export interface IMessageAggregator {
   save(msg: Message | Message[]): void
   getMessages(format?: MessageAggregatorFormat): Record<string, Message[]>
   clearMessages(): void
+  saveRawMessageData<T>(
+    messageData:
+      | EventBusTypes.MessageFormat<T>
+      | EventBusTypes.MessageFormat<T>[],
+    options?: Record<string, unknown>
+  ): void
 }
 
 /**

packages/types/src/user/mutations.ts

@@ -5,6 +5,7 @@ export interface CreateUserDTO {
   avatar_url?: string | null
   metadata?: Record<string, unknown> | null
 }
+
 export interface UpdateUserDTO extends Partial<Omit<CreateUserDTO, "email">> {
   id: string
 }

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

@@ -1,7 +1,9 @@
 import { IUserModuleService } from "@medusajs/types/dist/user"
 import { MikroOrmWrapper } from "../../../utils"
+import { MockEventBusService } from "medusa-test-utils"
 import { Modules } from "@medusajs/modules-sdk"
 import { SqlEntityManager } from "@mikro-orm/postgresql"
+import { UserEvents } from "@medusajs/utils"
 import { createInvites } from "../../../__fixtures__/invite"
 import { getInitModuleConfig } from "../../../utils/get-init-module-config"
 import { initModules } from "medusa-test-utils"
@@ -44,6 +46,7 @@ describe("UserModuleService - Invite", () => {
   beforeEach(async () => {
     await MikroOrmWrapper.setupDatabase()
     testManager = MikroOrmWrapper.forkManager()
+    jest.clearAllMocks()
   })
 
   afterEach(async () => {
@@ -171,6 +174,30 @@ describe("UserModuleService - Invite", () => {
 
       expect(error.message).toEqual('Invite with id "does-not-exist" not found')
     })
+
+    it("should emit invite updated events", async () => {
+      await createInvites(testManager, defaultInviteData)
+
+      jest.clearAllMocks()
+
+      const eventBusSpy = jest.spyOn(MockEventBusService.prototype, "emit")
+      await service.updateInvites([
+        {
+          id: "1",
+          accepted: true,
+        },
+      ])
+
+      expect(eventBusSpy).toHaveBeenCalledTimes(1)
+      expect(eventBusSpy).toHaveBeenCalledWith([
+        expect.objectContaining({
+          body: expect.objectContaining({
+            data: { id: "1" },
+          }),
+          eventName: UserEvents.invite_updated,
+        }),
+      ])
+    })
   })
 
   describe("createInvitie", () => {
@@ -188,5 +215,26 @@ describe("UserModuleService - Invite", () => {
         })
       )
     })
+
+    it("should emit invite created events", async () => {
+      const eventBusSpy = jest.spyOn(MockEventBusService.prototype, "emit")
+      await service.createInvites(defaultInviteData)
+
+      expect(eventBusSpy).toHaveBeenCalledTimes(1)
+      expect(eventBusSpy).toHaveBeenCalledWith([
+        expect.objectContaining({
+          body: expect.objectContaining({
+            data: { id: "1" },
+          }),
+          eventName: UserEvents.invite_created,
+        }),
+        expect.objectContaining({
+          body: expect.objectContaining({
+            data: { id: "2" },
+          }),
+          eventName: UserEvents.invite_created,
+        }),
+      ])
+    })
   })
 })

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

@@ -1,7 +1,9 @@
 import { IUserModuleService } from "@medusajs/types/dist/user"
 import { MikroOrmWrapper } from "../../../utils"
+import { MockEventBusService } from "medusa-test-utils"
 import { Modules } from "@medusajs/modules-sdk"
 import { SqlEntityManager } from "@mikro-orm/postgresql"
+import { UserEvents } from "@medusajs/utils"
 import { createUsers } from "../../../__fixtures__/user"
 import { getInitModuleConfig } from "../../../utils/get-init-module-config"
 import { initModules } from "medusa-test-utils"
@@ -41,6 +43,7 @@ describe("UserModuleService - User", () => {
 
   afterEach(async () => {
     await MikroOrmWrapper.clearDatabase()
+    jest.clearAllMocks()
   })
 
   afterAll(async () => {
@@ -182,6 +185,30 @@ describe("UserModuleService - User", () => {
 
       expect(error.message).toEqual('User with id "does-not-exist" not found')
     })
+
+    it("should emit user created events", async () => {
+      const eventBusSpy = jest.spyOn(MockEventBusService.prototype, "emit")
+      await service.create(defaultUserData)
+
+      jest.clearAllMocks()
+
+      await service.update([
+        {
+          id: "1",
+          first_name: "John",
+        },
+      ])
+
+      expect(eventBusSpy).toHaveBeenCalledTimes(1)
+      expect(eventBusSpy).toHaveBeenCalledWith([
+        expect.objectContaining({
+          body: expect.objectContaining({
+            data: { id: "1" },
+          }),
+          eventName: UserEvents.updated,
+        }),
+      ])
+    })
   })
 
   describe("create", () => {
@@ -199,5 +226,26 @@ describe("UserModuleService - User", () => {
         })
       )
     })
+
+    it("should emit user created events", async () => {
+      const eventBusSpy = jest.spyOn(MockEventBusService.prototype, "emit")
+      await service.create(defaultUserData)
+
+      expect(eventBusSpy).toHaveBeenCalledTimes(1)
+      expect(eventBusSpy).toHaveBeenCalledWith([
+        expect.objectContaining({
+          body: expect.objectContaining({
+            data: { id: "1" },
+          }),
+          eventName: UserEvents.created,
+        }),
+        expect.objectContaining({
+          body: expect.objectContaining({
+            data: { id: "2" },
+          }),
+          eventName: UserEvents.created,
+        }),
+      ])
+    })
   })
 })

packages/user/integration-tests/utils/get-init-module-config.ts

@@ -1,6 +1,7 @@
 import { Modules, ModulesDefinition } from "@medusajs/modules-sdk"
 
 import { DB_URL } from "./database"
+import { MockEventBusService } from "medusa-test-utils"
 
 export function getInitModuleConfig() {
   const moduleOptions = {
@@ -13,7 +14,9 @@ export function getInitModuleConfig() {
     jwt_secret: "test",
   }
 
-  const injectedDependencies = {}
+  const injectedDependencies = {
+    eventBusModuleService: new MockEventBusService(),
+  }
 
   const modulesConfig_ = {
     [Modules.USER]: {

packages/user/src/services/user-module.ts

@@ -5,13 +5,17 @@ import {
   ModuleJoinerConfig,
   UserTypes,
   ModulesSdkTypes,
+  IEventBusModuleService,
 } from "@medusajs/types"
 import {
-  InjectManager,
+  EmitEvents,
   InjectTransactionManager,
   MedusaContext,
-  MedusaError,
   ModulesSdkUtils,
+  InjectManager,
+  buildEventMessages,
+  CommonEvents,
+  UserEvents,
 } from "@medusajs/utils"
 import { entityNameToLinkableKeysMap, joinerConfig } from "../joiner-config"
 
@@ -22,6 +26,7 @@ type InjectedDependencies = {
   baseRepository: DAL.RepositoryService
   userService: ModulesSdkTypes.InternalModuleService<any>
   inviteService: InviteService<any>
+  eventBusModuleService: IEventBusModuleService
 }
 
 const generateMethodForModels = [Invite]
@@ -81,7 +86,8 @@ export default class UserModuleService<
     sharedContext?: Context
   ): Promise<UserTypes.UserDTO>
 
-  @InjectTransactionManager("baseRepository_")
+  @InjectManager("baseRepository_")
+  @EmitEvents()
   async create(
     data: UserTypes.CreateUserDTO[] | UserTypes.CreateUserDTO,
     @MedusaContext() sharedContext: Context = {}
@@ -96,6 +102,18 @@ export default class UserModuleService<
       populate: true,
     })
 
+    sharedContext.messageAggregator?.saveRawMessageData(
+      users.map((user) => ({
+        eventName: UserEvents.created,
+        metadata: {
+          service: this.constructor.name,
+          action: CommonEvents.CREATED,
+          object: "user",
+        },
+        data: { id: user.id },
+      }))
+    )
+
     return Array.isArray(data) ? serializedUsers : serializedUsers[0]
   }
 
@@ -108,7 +126,8 @@ export default class UserModuleService<
     sharedContext?: Context
   ): Promise<UserTypes.UserDTO>
 
-  @InjectTransactionManager("baseRepository_")
+  @InjectManager("baseRepository_")
+  @EmitEvents()
   async update(
     data: UserTypes.UpdateUserDTO | UserTypes.UpdateUserDTO[],
     @MedusaContext() sharedContext: Context = {}
@@ -123,6 +142,18 @@ export default class UserModuleService<
       populate: true,
     })
 
+    sharedContext.messageAggregator?.saveRawMessageData(
+      updatedUsers.map((user) => ({
+        eventName: UserEvents.updated,
+        metadata: {
+          service: this.constructor.name,
+          action: CommonEvents.UPDATED,
+          object: "user",
+        },
+        data: { id: user.id },
+      }))
+    )
+
     return Array.isArray(data) ? serializedUsers : serializedUsers[0]
   }
 
@@ -135,7 +166,8 @@ export default class UserModuleService<
     sharedContext?: Context
   ): Promise<UserTypes.InviteDTO>
 
-  @InjectTransactionManager("baseRepository_")
+  @InjectManager("baseRepository_")
+  @EmitEvents()
   async createInvites(
     data: UserTypes.CreateInviteDTO[] | UserTypes.CreateInviteDTO,
     @MedusaContext() sharedContext: Context = {}
@@ -150,6 +182,18 @@ export default class UserModuleService<
       populate: true,
     })
 
+    sharedContext.messageAggregator?.saveRawMessageData(
+      invites.map((invite) => ({
+        eventName: UserEvents.invite_created,
+        metadata: {
+          service: this.constructor.name,
+          action: CommonEvents.CREATED,
+          object: "invite",
+        },
+        data: { id: invite.id },
+      }))
+    )
+
     return Array.isArray(data) ? serializedInvites : serializedInvites[0]
   }
 
@@ -178,7 +222,8 @@ export default class UserModuleService<
     sharedContext?: Context
   ): Promise<UserTypes.InviteDTO>
 
-  @InjectTransactionManager("baseRepository_")
+  @InjectManager("baseRepository_")
+  @EmitEvents()
   async updateInvites(
     data: UserTypes.UpdateInviteDTO | UserTypes.UpdateInviteDTO[],
     @MedusaContext() sharedContext: Context = {}
@@ -196,6 +241,18 @@ export default class UserModuleService<
       populate: true,
     })
 
+    sharedContext.messageAggregator?.saveRawMessageData(
+      serializedInvites.map((invite) => ({
+        eventName: UserEvents.invite_updated,
+        metadata: {
+          service: this.constructor.name,
+          action: CommonEvents.UPDATED,
+          object: "invite",
+        },
+        data: { id: invite.id },
+      }))
+    )
+
     return Array.isArray(data) ? serializedInvites : serializedInvites[0]
   }
 }

packages/utils/src/bundles.ts

@@ -11,4 +11,5 @@ export * as ProductUtils from "./product"
 export * as PromotionUtils from "./promotion"
 export * as SearchUtils from "./search"
 export * as ShippingProfileUtils from "./shipping"
+export * as UserUtils from "./user"
 export * as ApiKeyUtils from "./api-key"

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

@@ -1,9 +1,12 @@
 import {
+  EventBusTypes,
   IMessageAggregator,
   Message,
   MessageAggregatorFormat,
 } from "@medusajs/types"
 
+import { buildEventMessages } from "./build-event-messages"
+
 export class MessageAggregator implements IMessageAggregator {
   private messages: Message[]
 
@@ -23,6 +26,15 @@ export class MessageAggregator implements IMessageAggregator {
     }
   }
 
+  saveRawMessageData<T>(
+    messageData:
+      | EventBusTypes.MessageFormat<T>
+      | EventBusTypes.MessageFormat<T>[],
+    options?: Record<string, unknown>
+  ): void {
+    this.save(buildEventMessages(messageData, options))
+  }
+
   getMessages(format?: MessageAggregatorFormat): {
     [group: string]: Message[]
   } {

packages/utils/src/index.ts

@@ -19,6 +19,7 @@ export * from "./search"
 export * from "./shipping"
 export * from "./totals"
 export * from "./totals/big-number"
+export * from "./user"
 export * from "./api-key"
 
 export const MedusaModuleType = Symbol.for("MedusaModule")

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

@@ -474,6 +474,19 @@ export function abstractModuleServiceFactory<
         /* ignore */
       }
     }
+
+    protected async emitEvents_(groupedEvents) {
+      if (!this.eventBusModuleService_ || !groupedEvents) {
+        return
+      }
+
+      const promises: Promise<void>[] = []
+      for (const group of Object.keys(groupedEvents)) {
+        promises.push(this.eventBusModuleService_?.emit(groupedEvents[group]))
+      }
+
+      await Promise.all(promises)
+    }
   }
 
   const mainModelMethods = buildMethodNamesFromModel(mainModel, false)

packages/utils/src/modules-sdk/decorators/emit-events.ts

@@ -0,0 +1,26 @@
+import { MessageAggregator } from "../../event-bus"
+import { InjectIntoContext } from "./inject-into-context"
+
+export function EmitEvents() {
+  return function (
+    target: any,
+    propertyKey: string | symbol,
+    descriptor: any
+  ): void {
+    const aggregator = new MessageAggregator()
+    InjectIntoContext({
+      messageAggregator: () => aggregator,
+    })(target, propertyKey, descriptor)
+
+    const original = descriptor.value
+
+    descriptor.value = async function (...args: any[]) {
+      const result = await original.apply(this, args)
+
+      await target.emitEvents_.apply(this, [aggregator.getMessages()])
+
+      aggregator.clearMessages()
+      return result
+    }
+  }
+}

packages/utils/src/modules-sdk/decorators/index.ts

@@ -2,3 +2,5 @@ export * from "./context-parameter"
 export * from "./inject-manager"
 export * from "./inject-shared-context"
 export * from "./inject-transaction-manager"
+export * from "./inject-into-context"
+export * from "./emit-events"

packages/utils/src/modules-sdk/decorators/inject-into-context.ts

@@ -1,3 +1,5 @@
+import { MessageAggregator } from "../../event-bus"
+
 export function InjectIntoContext(
   properties: Record<string, unknown | Function>
 ): MethodDecorator {

packages/utils/src/user/events.ts

@@ -0,0 +1,8 @@
+import { CommonEvents } from "../event-bus"
+
+export const UserEvents = {
+  created: "user." + CommonEvents.CREATED,
+  updated: "user." + CommonEvents.UPDATED,
+  invite_created: "invite." + CommonEvents.CREATED,
+  invite_updated: "invite." + CommonEvents.UPDATED,
+}

packages/utils/src/user/index.ts

@@ -0,0 +1 @@
+export * from "./events"