chore(): Module Internal Events (medusajs#13296)

* chore(): Ensure the product module emits all necessary events

* chore(): Ensure the product module emits all necessary events

* Update events tests

* more events and fixes

* more tests and category fixes

* more tests and category fixes

* Add todo

* update updateProduct_ event emitting and adjust test

* Adjust update products implementation to rely on already computed events

* rm unnecessary update variants events

* Fix formatting in changeset for product events

* refactor: Manage event emitting automatically (WIP)

* refactor: Manage event emitting automatically (WIP)

* chore(api-key): Add missing emit events and refactoring

* chore(cart): Add missing emit events and refactoring

* chore(customer): Add missing emit events and refactoring

* chore(fufillment, utils): Add missing emit events and refactoring

* chore(fufillment, utils): Add missing emit events and refactoring

* chore(inventory): Add missing emit events and refactoring

* chore(notification): Add missing emit events and refactoring

* chore(utils): Remove medusa service event handling legacy

* chore(product): Add missing emit events and refactoring

* chore(order): Add missing emit events and refactoring

* chore(payment): Add missing emit events and refactoring

* chore(pricing, util): Add missing emit events and refactoring, fix internal service upsertWithReplace event dispatching

* chore(promotions): Add missing emit events and refactoring

* chore(region): Add missing emit events and refactoring

* chore(sales-channel): Add missing emit events and refactoring

* chore(settings): Add missing emit events and refactoring

* chore(stock-location): Add missing emit events and refactoring

* chore(store): Add missing emit events and refactoring

* chore(taxes): Add missing emit events and refactoring

* chore(user): Add missing emit events and refactoring

* fix unit tests

* rm changeset for regeneration

* Create changeset for Medusa.js patch updates

Add a changeset for patch updates to multiple Medusa.js modules.

* rm unused product event builders

* address feedback

* remove old changeset

* fix event action for token generated

* fix user module events

* fix import

* fix promotion events

* add new module integration tests shard

* fix medusa service

* revert shard

* fix event action

* fix pipeline

* fix pipeline

---------

Co-authored-by: Oli Juhl <[email protected]>

Author: adrien2p

.changeset/angry-bikes-perform.md

@@ -0,0 +1,15 @@
+---
+"@medusajs/api-key": patch
+"@medusajs/cart": patch
+"@medusajs/customer": patch
+"@medusajs/fulfillment": patch
+"@medusajs/inventory": patch
+"@medusajs/notification": patch
+"@medusajs/order": patch
+"@medusajs/payment": patch
+"@medusajs/pricing": patch
+"@medusajs/product": patch
+"@medusajs/utils": patch
+---
+
+chore(): Module Internal Events

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

@@ -9,12 +9,13 @@ import {
 import { composeMessage } from "./build-event-messages"
 
 export class MessageAggregator implements IMessageAggregator {
-  private messages: Message[] = []
+  #messagesHash: Set<string> = new Set()
+  #messages: Message[] = []
 
   constructor() {}
 
   count(): number {
-    return this.messages.length
+    return this.#messages.length
   }
 
   save(msg: Message | Message[]): void {
@@ -23,7 +24,19 @@ export class MessageAggregator implements IMessageAggregator {
       return
     }
 
-    this.messages.push(...messages)
+    for (const message of messages) {
+      try {
+        const hash = JSON.stringify(message)
+        if (!this.#messagesHash.has(hash)) {
+          this.#messagesHash.add(hash)
+          this.#messages.push(message)
+        }
+      } catch (e) {
+        // noop: if the message is not serializable, we don't want to deduplicate it
+        // It should not happen, but we don't want to fail the whole process
+        this.#messages.push(message)
+      }
+    }
   }
 
   saveRawMessageData<T>(
@@ -55,15 +68,15 @@ export class MessageAggregator implements IMessageAggregator {
     const { groupBy, sortBy } = format ?? {}
 
     if (sortBy) {
-      this.messages.sort((a, b) => this.compareMessages(a, b, sortBy))
+      this.#messages.sort((a, b) => this.compareMessages(a, b, sortBy))
     }
 
     let messages: { [group: string]: Message[] } = {
-      default: [...this.messages],
+      default: [...this.#messages],
     }
 
     if (groupBy) {
-      messages = this.messages.reduce<{
+      messages = this.#messages.reduce<{
         [key: string]: Message[]
       }>((acc, msg) => {
         const key = groupBy
@@ -91,7 +104,8 @@ export class MessageAggregator implements IMessageAggregator {
 
   clearMessages(): void {
     // Ensure no references are left over in case something rely on messages
-    this.messages.length = 0
+    this.#messages.length = 0
+    this.#messagesHash.clear()
   }
 
   private getValueFromPath(obj: any, path: string): any {

packages/core/utils/src/modules-sdk/__tests__/medusa-internal-service.ts

@@ -20,7 +20,7 @@ describe("Internal Module Service Factory", () => {
       findAndCount: jest.fn(),
       create: jest.fn(),
       update: jest.fn(),
-      delete: jest.fn(),
+      delete: jest.fn().mockImplementation((ids) => Promise.resolve(ids)),
       softDelete: jest.fn(),
       restore: jest.fn(),
       upsert: jest.fn(),
@@ -32,7 +32,7 @@ describe("Internal Module Service Factory", () => {
       findAndCount: jest.fn(),
       create: jest.fn(),
       update: jest.fn(),
-      delete: jest.fn(),
+      delete: jest.fn().mockImplementation((ids) => Promise.resolve(ids)),
       softDelete: jest.fn(),
       restore: jest.fn(),
       upsert: jest.fn(),

packages/core/utils/src/modules-sdk/create-medusa-mikro-orm-event-subscriber.ts

@@ -0,0 +1,71 @@
+import { Context } from "@medusajs/types"
+import { EventArgs, EventSubscriber } from "@mikro-orm/core"
+type Service = {
+  interceptEntityMutationEvents: (
+    event: "afterCreate" | "afterUpdate" | "afterUpsert" | "afterDelete",
+    args: EventArgs<any>,
+    context: Context
+  ) => void
+}
+
+export type MedusaMikroOrmEventSubscriber = {
+  new (context: Context): EventSubscriber
+}
+
+/**
+ * Build a new mikro orm event subscriber for the given models
+ * @param models
+ * @returns
+ */
+export function createMedusaMikroOrmEventSubscriber(
+  keys: string[],
+  service: Service
+): MedusaMikroOrmEventSubscriber {
+  const klass = class MikroOrmEventSubscriber implements EventSubscriber {
+    #context: Context
+    #service: Service = service
+
+    constructor(context: Context) {
+      this.#context = context
+    }
+
+    async afterCreate<T>(args: EventArgs<T>): Promise<void> {
+      this.#service.interceptEntityMutationEvents(
+        "afterCreate",
+        args,
+        this.#context
+      )
+    }
+
+    async afterUpdate<T>(args: EventArgs<T>): Promise<void> {
+      this.#service.interceptEntityMutationEvents(
+        "afterUpdate",
+        args,
+        this.#context
+      )
+    }
+
+    async afterUpsert<T>(args: EventArgs<T>): Promise<void> {
+      this.#service.interceptEntityMutationEvents(
+        "afterUpsert",
+        args,
+        this.#context
+      )
+    }
+
+    async afterDelete<T>(args: EventArgs<T>): Promise<void> {
+      this.#service.interceptEntityMutationEvents(
+        "afterDelete",
+        args,
+        this.#context
+      )
+    }
+  }
+
+  Object.defineProperty(klass, "name", {
+    value: keys.join(","),
+    writable: false,
+  })
+
+  return klass
+}

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

@@ -21,3 +21,4 @@ export * from "./types/links-config"
 export * from "./types/medusa-service"
 export * from "./module-provider-registration-key"
 export * from "./modules-to-container-types"
+export * from "./create-medusa-mikro-orm-event-subscriber"