feat(medusa): Bulk emit events (#3407)

.changeset/serious-avocados-rush.md

@@ -0,0 +1,7 @@
+---
+"@medusajs/inventory": patch
+"medusa-test-utils": patch
+"@medusajs/medusa": patch
+---
+
+feat(medusa): Bulk emit events

integration-tests/plugins/__tests__/inventory/service.js

@@ -3,7 +3,7 @@ const path = require("path")
 const { bootstrapApp } = require("../../../helpers/bootstrap-app")
 const { initDb, useDb } = require("../../../helpers/use-db")
 
-jest.setTimeout(30000)
+jest.setTimeout(50000)
 
 describe("Inventory Module", () => {
   let appContainer

packages/inventory/src/services/reservation-item.ts

@@ -1,17 +1,16 @@
 import { EntityManager } from "typeorm"
 import { isDefined, MedusaError } from "medusa-core-utils"
 import {
-  FindConfig,
   buildQuery,
-  IEventBusService,
-  FilterableReservationItemProps,
   CreateReservationItemInput,
+  FilterableReservationItemProps,
+  FindConfig,
+  IEventBusService,
   TransactionBaseService,
   UpdateReservationItemInput,
 } from "@medusajs/medusa"
 
 import { ReservationItem } from "../models"
-import { CONNECTION_NAME } from "../config"
 import { InventoryLevelService } from "."
 
 type InjectedDependencies = {
@@ -278,10 +277,12 @@ export default class ReservationItemService extends TransactionBaseService {
             item.quantity * -1
           ),
       ])
-    })
 
-    await this.eventBusService_.emit(ReservationItemService.Events.DELETED, {
+      await this.eventBusService_
-      id: reservationItemId,
+        .withTransaction(manager)
+        .emit(ReservationItemService.Events.DELETED, {
+        id: reservationItemId,
+      })
     })
   }
 }

packages/medusa-test-utils/src/mock-repository.js

@@ -12,6 +12,8 @@ class MockRepo {
     save,
     findAndCount,
     del,
+    count,
+    insertBulk
   }) {
     this.create_ = create;
     this.update_ = update;
@@ -25,12 +27,19 @@ class MockRepo {
     this.save_ = save;
     this.findAndCount_ = findAndCount;
     this.findOneWithRelations_ = findOneWithRelations;
+    this.insertBulk_ = insertBulk;
   }
 
   setFindOne(fn) {
     this.findOne_ = fn;
   }
 
+  insertBulk = jest.fn().mockImplementation((...args) => {
+    if (this.insertBulk_) {
+      return this.insertBulk_(...args)
+    }
+    return {}
+  })
   create = jest.fn().mockImplementation((...args) => {
     if (this.create_) {
       return this.create_(...args);

packages/medusa/src/repositories/staged-job.ts

@@ -1,5 +1,28 @@
 import { EntityRepository, Repository } from "typeorm"
-import { StagedJob } from "../models/staged-job"
+import { QueryDeepPartialEntity } from "typeorm/query-builder/QueryPartialEntity"
+import { StagedJob } from "../models"
+import { rowSqlResultsToEntityTransformer } from "../utils"
 
 @EntityRepository(StagedJob)
-export class StagedJobRepository extends Repository<StagedJob> {}
+export class StagedJobRepository extends Repository<StagedJob> {
+  async insertBulk(jobToCreates: QueryDeepPartialEntity<StagedJob>[]) {
+    const queryBuilder = this.createQueryBuilder()
+      .insert()
+      .into(StagedJob)
+      .values(jobToCreates)
+
+    // TODO: remove if statement once this issue is resolved https://github.com/typeorm/typeorm/issues/9850
+    if (!queryBuilder.connection.driver.isReturningSqlSupported("insert")) {
+      const rawStagedJobs = await queryBuilder.execute()
+      return rawStagedJobs.generatedMaps
+    }
+
+    const rawStagedJobs = await queryBuilder.returning("*").execute()
+
+    return rowSqlResultsToEntityTransformer(
+      rawStagedJobs.raw,
+      queryBuilder,
+      this.queryRunner!
+    )
+  }
+}

packages/medusa/src/services/__tests__/event-bus.js

@@ -114,79 +114,174 @@ describe("EventBusService", () => {
   })
 
   describe("emit", () => {
-    let eventBus
+    const eventName = "eventName"
-    let job
+    const defaultOptions = {
+      attempts: 1,
+      removeOnComplete: true,
+    }
+
+    const data = { hi: "1234" }
+    const bulkData = [{ hi: "1234" }, { hi: "12345" }]
+
+    const mockManager = MockManager
+
     describe("successfully adds job to queue", () => {
-      beforeAll(() => {
+      let eventBus
-        jest.resetAllMocks()
+      let stagedJobRepository
-        const stagedJobRepository = MockRepository({
+
-          find: () => Promise.resolve([]),
+      beforeEach(() => {
+        stagedJobRepository = MockRepository({
+          insertBulk: async (data) => data,
+          create: (data) => data,
         })
 
         eventBus = new EventBusService({
           logger: loggerMock,
-          manager: MockManager,
+          manager: mockManager,
           stagedJobRepository,
         })
 
-        eventBus.queue_.add.mockImplementationOnce(() => "hi")
+        eventBus.queue_.addBulk.mockImplementationOnce(() => "hi")
-
-        job = eventBus.emit("eventName", { hi: "1234" })
       })
-      afterAll(async () => {
+
+      afterEach(async () => {
         await eventBus.stopEnqueuer()
+        jest.clearAllMocks()
       })
 
-      it("calls queue.add", () => {
+      it("calls queue.addBulk", async () => {
-        expect(eventBus.queue_.add).toHaveBeenCalled()
+        await eventBus.emit(eventName, data)
+
+        expect(eventBus.queue_.addBulk).toHaveBeenCalled()
+        expect(eventBus.queue_.addBulk).toHaveBeenCalledWith([
+          {
+            data: {
+              data,
+              eventName,
+            },
+            opts: defaultOptions,
+          },
+        ])
+      })
+
+      it("calls stagedJob repository insertBulk", async () => {
+        await eventBus.withTransaction(mockManager).emit(eventName, data)
+
+        expect(stagedJobRepository.create).toHaveBeenCalled()
+        expect(stagedJobRepository.create).toHaveBeenCalledWith({
+          event_name: eventName,
+          data: data,
+          options: defaultOptions,
+        })
+
+        expect(stagedJobRepository.insertBulk).toHaveBeenCalled()
+        expect(stagedJobRepository.insertBulk).toHaveBeenCalledWith([
+          {
+            event_name: eventName,
+            data,
+            options: defaultOptions,
+          },
+        ])
       })
     })
 
-    describe("successfully adds job to queue with local options", () => {
+    describe("successfully adds jobs in bulk to queue", () => {
-      beforeAll(() => {
+      let eventBus
-        jest.resetAllMocks()
+      let stagedJobRepository
-        const stagedJobRepository = MockRepository({
+
-          find: () => Promise.resolve([]),
+      beforeEach(() => {
+        stagedJobRepository = MockRepository({
+          insertBulk: async (data) => data,
+          create: (data) => data,
         })
 
         eventBus = new EventBusService({
           logger: loggerMock,
-          manager: MockManager,
+          manager: mockManager,
           stagedJobRepository,
         })
 
-        eventBus.queue_.add.mockImplementationOnce(() => "hi")
+        eventBus.queue_.addBulk.mockImplementationOnce(() => "hi")
-
-        job = eventBus.emit(
-          "eventName",
-          { hi: "1234" },
-          { removeOnComplete: 100 }
-        )
       })
-      afterAll(async () => {
+
+      afterEach(async () => {
+        jest.clearAllMocks()
         await eventBus.stopEnqueuer()
       })
 
-      it("calls queue.add", () => {
+      it("calls queue.addBulk", async () => {
-        expect(eventBus.queue_.add).toHaveBeenCalled()
+        await eventBus.emit([
-        expect(eventBus.queue_.add).toHaveBeenCalledWith(
+          { eventName, data: bulkData[0] },
-          { eventName: "eventName", data: { hi: "1234" } },
+          { eventName, data: bulkData[1] },
-          { removeOnComplete: 100 }
+        ])
-        )
+
+        expect(eventBus.queue_.addBulk).toHaveBeenCalledTimes(1)
+        expect(eventBus.queue_.addBulk).toHaveBeenCalledWith([
+          {
+            data: {
+              data: bulkData[0],
+              eventName,
+            },
+            opts: defaultOptions,
+          },
+          {
+            data: {
+              data: bulkData[1],
+              eventName,
+            },
+            opts: defaultOptions,
+          },
+        ])
+      })
+
+      it("calls stagedJob repository insertBulk", async () => {
+        await eventBus.withTransaction(mockManager).emit([
+          { eventName, data: bulkData[0] },
+          { eventName, data: bulkData[1] },
+        ])
+
+        expect(stagedJobRepository.create).toHaveBeenCalledTimes(2)
+        expect(stagedJobRepository.create).toHaveBeenNthCalledWith(1, {
+          data: bulkData[0],
+          event_name: eventName,
+          options: defaultOptions,
+        })
+        expect(stagedJobRepository.create).toHaveBeenNthCalledWith(2, {
+          data: bulkData[1],
+          event_name: eventName,
+          options: defaultOptions,
+        })
+
+        expect(stagedJobRepository.insertBulk).toHaveBeenCalledTimes(1)
+        expect(stagedJobRepository.insertBulk).toHaveBeenCalledWith([
+          {
+            data: bulkData[0],
+            event_name: eventName,
+            options: defaultOptions,
+          },
+          {
+            data: bulkData[1],
+            event_name: eventName,
+            options: defaultOptions,
+          },
+        ])
       })
     })
 
     describe("successfully adds job to queue with global options", () => {
-      beforeAll(() => {
+      let eventBus
-        jest.resetAllMocks()
+      let stagedJobRepository
-        const stagedJobRepository = MockRepository({
+
-          find: () => Promise.resolve([]),
+      beforeEach(() => {
+        stagedJobRepository = MockRepository({
+          insertBulk: async (data) => data,
+          create: (data) => data,
         })
 
         eventBus = new EventBusService(
           {
             logger: loggerMock,
-            manager: MockManager,
+            manager: mockManager,
             stagedJobRepository,
           },
           {
@@ -194,58 +289,78 @@ describe("EventBusService", () => {
           }
         )
 
-        eventBus.queue_.add.mockImplementationOnce(() => "hi")
+        eventBus.queue_.addBulk.mockImplementationOnce(() => "hi")
 
-        job = eventBus.emit("eventName", { hi: "1234" })
+        eventBus.emit(eventName, data)
       })
-      afterAll(async () => {
+
+      afterEach(async () => {
+        jest.clearAllMocks()
         await eventBus.stopEnqueuer()
       })
 
-      it("calls queue.add", () => {
+      it("calls queue.addBulk", () => {
-        expect(eventBus.queue_.add).toHaveBeenCalled()
+        expect(eventBus.queue_.addBulk).toHaveBeenCalled()
-        expect(eventBus.queue_.add).toHaveBeenCalledWith(
+        expect(eventBus.queue_.addBulk).toHaveBeenCalledWith([
-          { eventName: "eventName", data: { hi: "1234" } },
+          {
-          { removeOnComplete: 10, attempts: 1 }
+            data: {
-        )
+              data,
+              eventName,
+            },
+            opts: { removeOnComplete: 10, attempts: 1 },
+          },
+        ])
       })
     })
 
     describe("successfully adds job to queue with default options", () => {
-      beforeAll(() => {
+      let eventBus
-        jest.resetAllMocks()
+      let stagedJobRepository
-        const stagedJobRepository = MockRepository({
+
-          find: () => Promise.resolve([]),
+      beforeEach(() => {
+        stagedJobRepository = MockRepository({
+          insertBulk: async (data) => data,
+          create: (data) => data,
         })
 
         eventBus = new EventBusService({
           logger: loggerMock,
-          manager: MockManager,
+          manager: mockManager,
           stagedJobRepository,
         })
 
-        eventBus.queue_.add.mockImplementationOnce(() => "hi")
+        eventBus.queue_.addBulk.mockImplementationOnce(() => "hi")
 
-        job = eventBus.emit("eventName", { hi: "1234" })
+        eventBus.emit(eventName, data)
       })
-      afterAll(async () => {
+
+      afterEach(async () => {
+        jest.clearAllMocks()
         await eventBus.stopEnqueuer()
       })
 
-      it("calls queue.add", () => {
+      it("calls queue.addBulk", () => {
-        expect(eventBus.queue_.add).toHaveBeenCalled()
+        expect(eventBus.queue_.addBulk).toHaveBeenCalled()
-        expect(eventBus.queue_.add).toHaveBeenCalledWith(
+        expect(eventBus.queue_.addBulk).toHaveBeenCalledWith([
-          { eventName: "eventName", data: { hi: "1234" } },
+          {
-          { removeOnComplete: true, attempts: 1 }
+            data: {
-        )
+              data,
+              eventName,
+            },
+            opts: { removeOnComplete: true, attempts: 1 },
+          },
+        ])
       })
     })
 
     describe("successfully adds job to queue with local options and global options merged", () => {
-      beforeAll(() => {
+      let eventBus
-        jest.resetAllMocks()
+      let stagedJobRepository
-        const stagedJobRepository = MockRepository({
+
-          find: () => Promise.resolve([]),
+      beforeEach(() => {
+        stagedJobRepository = MockRepository({
+          insertBulk: async (data) => data,
+          create: (data) => data,
         })
 
         eventBus = new EventBusService(
@@ -259,29 +374,36 @@ describe("EventBusService", () => {
           }
         )
 
-        eventBus.queue_.add.mockImplementationOnce(() => "hi")
+        eventBus.queue_.addBulk.mockImplementationOnce(() => "hi")
 
-        job = eventBus.emit(
+        eventBus.emit(eventName, data, {
-          "eventName",
+          attempts: 10,
-          { hi: "1234" },
+          delay: 1000,
-          { attempts: 10, delay: 1000, backoff: { type: "exponential" } }
+          backoff: { type: "exponential" },
-        )
+        })
       })
-      afterAll(async () => {
+
+      afterEach(async () => {
+        jest.clearAllMocks()
         await eventBus.stopEnqueuer()
       })
 
       it("calls queue.add", () => {
-        expect(eventBus.queue_.add).toHaveBeenCalled()
+        expect(eventBus.queue_.addBulk).toHaveBeenCalled()
-        expect(eventBus.queue_.add).toHaveBeenCalledWith(
+        expect(eventBus.queue_.addBulk).toHaveBeenCalledWith([
-          { eventName: "eventName", data: { hi: "1234" } },
           {
-            removeOnComplete: 10, // global option
+            data: {
-            attempts: 10, // local option
+              data,
-            delay: 1000, // local option
+              eventName,
-            backoff: { type: "exponential" }, // local option
+            },
-          }
+            opts: {
-        )
+              removeOnComplete: 10, // global option
+              attempts: 10, // local option
+              delay: 1000, // local option
+              backoff: { type: "exponential" }, // local option
+            },
+          },
+        ])
       })
     })
   })

packages/medusa/src/services/__tests__/order.js

@@ -1,8 +1,8 @@
 import { IdMap, MockManager, MockRepository } from "medusa-test-utils"
 import OrderService from "../order"
-import { ProductVariantInventoryServiceMock } from "../__mocks__/product-variant-inventory"
 import { LineItemServiceMock } from "../__mocks__/line-item"
 import { newTotalsServiceMock } from "../__mocks__/new-totals"
+import { ProductVariantInventoryServiceMock } from "../__mocks__/product-variant-inventory"
 import { taxProviderServiceMock } from "../__mocks__/tax-provider"
 
 describe("OrderService", () => {
@@ -1072,10 +1072,15 @@ describe("OrderService", () => {
           { no_notification: input }
         )
 
-        expect(eventBusService.emit).toHaveBeenCalledWith(expect.any(String), {
+        expect(eventBusService.emit).toHaveBeenCalledWith([
-          id: expect.any(String),
+          {
-          no_notification: expected,
+            eventName: expect.any(String),
-        })
+            data: {
+              id: expect.any(String),
+              no_notification: expected,
+            },
+          },
+        ])
       }
     )
   })
@@ -1246,17 +1251,10 @@ describe("OrderService", () => {
       save: jest.fn().mockImplementation((f) => f),
     })
 
-    const eventBus = {
-      emit: () =>
-        Promise.resolve({
-          finished: () => Promise.resolve({}),
-        }),
-    }
-
     const orderService = new OrderService({
       manager: MockManager,
       orderRepository: orderRepo,
-      eventBusService: eventBus,
+      eventBusService: eventBusService,
     })
 
     beforeEach(async () => {

packages/medusa/src/services/claim.ts

@@ -638,15 +638,17 @@ export default class ClaimService extends TransactionBaseService {
         )
         const claimOrder = await claimRepo.save(claim)
 
-        const eventBusTx = this.eventBus_.withTransaction(transactionManager)
+        const eventsToEmit = fulfillments.map((fulfillment) => ({
-
+          eventName: ClaimService.Events.FULFILLMENT_CREATED,
-        for (const fulfillment of fulfillments) {
+          data: {
-          await eventBusTx.emit(ClaimService.Events.FULFILLMENT_CREATED, {
             id: id,
             fulfillment_id: fulfillment.id,
             no_notification: claim.no_notification,
-          })
+          },
-        }
+        }))
+        await this.eventBus_
+          .withTransaction(transactionManager)
+          .emit(eventsToEmit)
 
         return claimOrder
       }

packages/medusa/src/services/currency.ts

@@ -122,9 +122,11 @@ export default class CurrencyService extends TransactionBaseService {
       )
       await currencyRepo.save(currency)
 
-      await this.eventBusService_.emit(CurrencyService.Events.UPDATED, {
+      await this.eventBusService_
-        code,
+        .withTransaction(transactionManager)
-      })
+        .emit(CurrencyService.Events.UPDATED, {
+          code,
+        })
 
       return currency
     })

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

@@ -1,11 +1,12 @@
 import Bull, { JobOptions } from "bull"
 import Redis from "ioredis"
-import { isDefined } from "medusa-core-utils"
+import { DeepPartial, EntityManager, In } from "typeorm"
-import { EntityManager } from "typeorm"
+import { QueryDeepPartialEntity } from "typeorm/query-builder/QueryPartialEntity"
 import { ulid } from "ulid"
 import { StagedJob } from "../models"
 import { StagedJobRepository } from "../repositories/staged-job"
 import { ConfigModule, Logger } from "../types/global"
+import { isString } from "../utils"
 import { sleep } from "../utils/sleep"
 import JobSchedulerService, { CreateJobOptions } from "./job-scheduler"
 
@@ -49,6 +50,12 @@ export type EmitOptions = {
   }
 } & JobOptions
 
+export type EmitData<T = unknown> = {
+  eventName: string
+  data: T
+  opts?: Record<string, unknown> & EmitOptions
+}
+
 /**
  * Can keep track of multiple subscribers to different events and run the
  * subscribers when events happen. Events will run asynchronously.
@@ -213,6 +220,13 @@ export default class EventBusService {
     return this
   }
 
+  /**
+   * Calls all subscribers when an event occurs.
+   * @param data - The data to use to process the events
+   * @return the jobs from our queue
+   */
+  async emit<T>(data: EmitData<T>[]): Promise<StagedJob[] | void>
+
   /**
    * Calls all subscribers when an event occurs.
    * @param {string} eventName - the name of the event to be process.
@@ -223,29 +237,49 @@ export default class EventBusService {
   async emit<T>(
     eventName: string,
     data: T,
-    options: Record<string, unknown> & EmitOptions = { attempts: 1 }
+    options?: Record<string, unknown> & EmitOptions
-  ): Promise<StagedJob | void> {
+  ): Promise<StagedJob | void>
+
+  async emit<
+    T,
+    TInput extends string | EmitData<T>[] = string,
+    TResult = TInput extends EmitData<T>[] ? StagedJob[] : StagedJob
+  >(
+    eventNameOrData: TInput,
+    data?: T,
+    options: Record<string, unknown> & EmitOptions = {}
+  ): Promise<TResult | void> {
     const globalEventOptions = this.config_?.projectConfig?.event_options ?? {}
 
+    const isBulkEmit = !isString(eventNameOrData)
+    const events = isBulkEmit
+      ? eventNameOrData.map((event) => ({
+          data: { eventName: event.eventName, data: event.data },
+          opts: event.opts,
+        }))
+      : [
+          {
+            data: { eventName: eventNameOrData, data },
+            opts: options,
+          },
+        ]
+
     // The order of precedence for job options is:
     // 1. local options
     // 2. global options
     // 3. default options
-    const opts: EmitOptions = {
+    const defaultOptions: EmitOptions = {
-      removeOnComplete: true,
+      attempts: 1, // default
-      ...globalEventOptions,
+      removeOnComplete: true, // default
-      ...options,
+      ...globalEventOptions, // global
     }
 
-    if (typeof options.attempts === "number") {
+    for (const event of events) {
-      opts.attempts = options.attempts
+      event.opts = {
-      if (isDefined(options.backoff)) {
+        ...defaultOptions,
-        opts.backoff = options.backoff
+        ...(event.opts ?? {}), // local
       }
     }
-    if (typeof options.delay === "number") {
-      opts.delay = options.delay
-    }
 
     /**
      * If we are in an ongoing transaction, we store the jobs in the database
@@ -261,18 +295,20 @@ export default class EventBusService {
         this.stagedJobRepository_
       )
 
-      const jobToCreate = {
+      const jobsToCreate = events.map((event) => {
-        event_name: eventName,
+        return stagedJobRepository.create({
-        data: data as unknown as Record<string, unknown>,
+          event_name: event.data.eventName,
-        options: opts,
+          data: event.data.data,
-      } as Partial<StagedJob>
+          options: event.opts,
+        } as DeepPartial<StagedJob>) as QueryDeepPartialEntity<StagedJob>
+      })
 
-      const stagedJobInstance = stagedJobRepository.create(jobToCreate)
+      const stagedJobs = await stagedJobRepository.insertBulk(jobsToCreate)
 
-      return await stagedJobRepository.save(stagedJobInstance)
+      return (!isBulkEmit ? stagedJobs[0] : stagedJobs) as unknown as TResult
     }
 
-    this.queue_.add({ eventName, data }, opts)
+    await this.queue_.addBulk(events)
   }
 
   startEnqueuer(): void {
@@ -298,18 +334,21 @@ export default class EventBusService {
       )
       const jobs = await stagedJobRepo.find(listConfig)
 
-      await Promise.all(
+      if (!jobs.length) {
-        jobs.map((job) => {
+        await sleep(3000)
-          this.queue_
+        continue
-            .add(
+      }
-              { eventName: job.event_name, data: job.data },
+
-              { jobId: job.id, ...job.options }
+      const eventsData = jobs.map((job) => {
-            )
+        return {
-            .then(async () => {
+          data: { eventName: job.event_name, data: job.data },
-              await stagedJobRepo.remove(job)
+          opts: { jobId: job.id, ...job.options },
-            })
+        }
-        })
+      })
-      )
+
+      await this.queue_.addBulk(eventsData).then(async () => {
+        return await stagedJobRepo.delete({ id: In(jobs.map((j) => j.id)) })
+      })
 
       await sleep(3000)
     }

packages/medusa/src/services/order.ts

@@ -17,14 +17,14 @@ import {
   PaymentStatus,
   Return,
   Swap,
-  TrackingLink,
+  TrackingLink
 } from "../models"
 import { AddressRepository } from "../repositories/address"
 import { OrderRepository } from "../repositories/order"
 import { FindConfig, QuerySelector, Selector } from "../types/common"
 import {
   CreateFulfillmentOrder,
-  FulFillmentItemType,
+  FulFillmentItemType
 } from "../types/fulfillment"
 import { TotalsContext, UpdateOrderInput } from "../types/orders"
 import { CreateShippingMethodDto } from "../types/shipping-options"
@@ -48,7 +48,7 @@ import {
   ShippingOptionService,
   ShippingProfileService,
   TaxProviderService,
-  TotalsService,
+  TotalsService
 } from "."
 
 export const ORDER_CART_ALREADY_EXISTS_ERROR = "Order from cart already exists"
@@ -521,10 +521,12 @@ class OrderService extends TransactionBaseService {
         )
       }
 
-      await this.eventBus_.emit(OrderService.Events.COMPLETED, {
+      await this.eventBus_
-        id: orderId,
+        .withTransaction(manager)
-        no_notification: order.no_notification,
+        .emit(OrderService.Events.COMPLETED, {
-      })
+          id: orderId,
+          no_notification: order.no_notification,
+        })
 
       order.status = OrderStatus.COMPLETED
 
@@ -1396,14 +1398,15 @@ class OrderService extends TransactionBaseService {
       const evaluatedNoNotification =
         no_notification !== undefined ? no_notification : order.no_notification
 
-      const eventBusTx = this.eventBus_.withTransaction(manager)
+      const eventsToEmit = fulfillments.map((fulfillment) => ({
-      for (const fulfillment of fulfillments) {
+        eventName: OrderService.Events.FULFILLMENT_CREATED,
-        await eventBusTx.emit(OrderService.Events.FULFILLMENT_CREATED, {
+        data: {
           id: orderId,
           fulfillment_id: fulfillment.id,
           no_notification: evaluatedNoNotification,
-        })
+        },
-      }
+      }))
+      await this.eventBus_.withTransaction(manager).emit(eventsToEmit)
 
       return result
     })
@@ -1560,11 +1563,13 @@ class OrderService extends TransactionBaseService {
       const evaluatedNoNotification =
         no_notification !== undefined ? no_notification : order.no_notification
 
-      await this.eventBus_.emit(OrderService.Events.REFUND_CREATED, {
+      await this.eventBus_
-        id: result.id,
+        .withTransaction(manager)
-        refund_id: refund.id,
+        .emit(OrderService.Events.REFUND_CREATED, {
-        no_notification: evaluatedNoNotification,
+          id: result.id,
-      })
+          refund_id: refund.id,
+          no_notification: evaluatedNoNotification,
+        })
       return result
     })
   }

packages/medusa/src/services/user.ts

@@ -275,7 +275,9 @@ class UserService extends TransactionBaseService {
 
       await userRepo.softRemove(user)
 
-      await this.eventBus_.emit(UserService.Events.DELETED, { id: user.id })
+      await this.eventBus_
+        .withTransaction(manager)
+        .emit(UserService.Events.DELETED, { id: user.id })
 
       return Promise.resolve()
     })

packages/medusa/src/utils/index.ts

@@ -8,3 +8,4 @@ export * from "./calculate-price-tax-amount"
 export * from "./csv-cell-content-formatter"
 export * from "./exception-formatter"
 export * from "./db-aware-column"
+export * from "./row-sql-results-to-entity-transformer"

packages/medusa/src/utils/row-sql-results-to-entity-transformer.ts

@@ -0,0 +1,27 @@
+import { RelationIdLoader } from "typeorm/query-builder/relation-id/RelationIdLoader"
+import { RawSqlResultsToEntityTransformer } from "typeorm/query-builder/transformer/RawSqlResultsToEntityTransformer"
+import { QueryBuilder, QueryRunner } from "typeorm"
+
+export async function rowSqlResultsToEntityTransformer<T>(
+  rows: any[],
+  queryBuilder: QueryBuilder<T>,
+  queryRunner: QueryRunner
+): Promise<T[]> {
+  const relationIdLoader = new RelationIdLoader(
+    queryBuilder.connection,
+    queryRunner,
+    queryBuilder.expressionMap.relationIdAttributes
+  )
+  const transformer = new RawSqlResultsToEntityTransformer(
+    queryBuilder.expressionMap,
+    queryBuilder.connection.driver,
+    [],
+    [],
+    queryRunner
+  )
+
+  return transformer.transform(
+    rows,
+    queryBuilder.expressionMap.mainAlias!
+  ) as T[]
+}