fix(medusa): Bulk create variant + pass transaction to the inventory service context methods (#3835)

* fix(medusa): Bulk create variant and pass transaction where needed

* Create fair-penguins-stare.md

* fix unit tests

* event

* transaction orchestration

* revert options

* Prevent isolated module to use the given transaction if any in the exposed service

* Use enum

* remove changeset to re do it

* Create thick-ants-tickle.md

* update event bus local

* remove changeset to re do it

* Create thick-kings-wonder.md

* remove changeset to re do it

* Create slimy-bees-eat.md

* Update packages/utils/src/event-bus/index.ts

Co-authored-by: Oliver Windall Juhl <59018053+olivermrbl@users.noreply.github.com>

---------

Co-authored-by: Philip Korsholm <88927411+pKorsholm@users.noreply.github.com>
Co-authored-by: Oliver Windall Juhl <59018053+olivermrbl@users.noreply.github.com>

.changeset/slimy-bees-eat.md

@@ -0,0 +1,10 @@
+---
+"@medusajs/event-bus-local": patch
+"@medusajs/inventory": patch
+"@medusajs/medusa": patch
+"@medusajs/stock-location": patch
+"@medusajs/types": patch
+"@medusajs/utils": patch
+---
+
+fix(medusa): Bulk create variant and pass transaction to the inventory service context methods

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

@@ -1,7 +1,8 @@
 import { Logger, MedusaContainer } from "@medusajs/modules-sdk"
-import { EmitData, Subscriber } from "@medusajs/types"
+import { EmitData, EventBusTypes, Subscriber } from "@medusajs/types"
 import { AbstractEventBusModuleService } from "@medusajs/utils"
 import { EventEmitter } from "events"
+import { ulid } from "ulid"
 
 type InjectedDependencies = {
   logger: Logger
@@ -65,6 +66,8 @@ export default class LocalEventBusService extends AbstractEventBusModuleService
   }
 
   subscribe(event: string | symbol, subscriber: Subscriber): this {
+    const randId = ulid()
+    this.storeSubscribers({ event, subscriberId: randId, subscriber })
     this.eventEmitter_.on(event, async (...args) => {
       try {
         // @ts-ignore
@@ -78,7 +81,23 @@ export default class LocalEventBusService extends AbstractEventBusModuleService
     return this
   }
 
-  unsubscribe(event: string | symbol, subscriber: Subscriber): this {
+  unsubscribe(
+    event: string | symbol,
+    subscriber: Subscriber,
+    context?: EventBusTypes.SubscriberContext
+  ): this {
+    const existingSubscribers = this.retrieveSubscribers(event)
+
+    if (existingSubscribers?.length) {
+      const subIndex = existingSubscribers?.findIndex(
+        (sub) => sub.id === context?.subscriberId
+      )
+
+      if (subIndex !== -1) {
+        this.eventToSubscribersMap_.get(event)?.splice(subIndex as number, 1)
+      }
+    }
+
     this.eventEmitter_.off(event, subscriber)
     return this
   }

packages/inventory/src/services/inventory.ts

@@ -10,6 +10,7 @@ import {
   IInventoryService,
   InventoryItemDTO,
   InventoryLevelDTO,
+  MODULE_RESOURCE_TYPE,
   ReservationItemDTO,
   SharedContext,
   UpdateInventoryLevelInput,
@@ -46,7 +47,7 @@ export default class InventoryService implements IInventoryService {
       reservationItemService,
     }: InjectedDependencies,
     options?: unknown,
-    moduleDeclaration?: InternalModuleDeclaration
+    protected readonly moduleDeclaration?: InternalModuleDeclaration
   ) {
     this.manager_ = manager
     this.inventoryItemService_ = inventoryItemService
@@ -60,7 +61,10 @@ export default class InventoryService implements IInventoryService {
    * @param config - the find configuration to use
    * @return A tuple of inventory items and their total count
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async listInventoryItems(
     selector: FilterableInventoryItemProps,
     config: FindConfig<InventoryItemDTO> = { relations: [], skip: 0, take: 10 },
@@ -79,7 +83,10 @@ export default class InventoryService implements IInventoryService {
    * @param config - the find configuration to use
    * @return A tuple of inventory levels and their total count
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async listInventoryLevels(
     selector: FilterableInventoryLevelProps,
     config: FindConfig<InventoryLevelDTO> = {
@@ -102,7 +109,10 @@ export default class InventoryService implements IInventoryService {
    * @param config - the find configuration to use
    * @return A tuple of reservation items and their total count
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async listReservationItems(
     selector: FilterableReservationItemProps,
     config: FindConfig<ReservationItemDTO> = {
@@ -125,7 +135,10 @@ export default class InventoryService implements IInventoryService {
    * @param config - the find configuration to use
    * @return The retrieved inventory item
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async retrieveInventoryItem(
     inventoryItemId: string,
     config?: FindConfig<InventoryItemDTO>,
@@ -145,7 +158,10 @@ export default class InventoryService implements IInventoryService {
    * @param locationId - the id of the location
    * @return the retrieved inventory level
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async retrieveInventoryLevel(
     inventoryItemId: string,
     locationId: string,
@@ -170,7 +186,10 @@ export default class InventoryService implements IInventoryService {
    * @param inventoryItemId - the id of the reservation item
    * @return the retrieved reservation level
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async retrieveReservationItem(
     reservationId: string,
     @MedusaContext() context: SharedContext = {}
@@ -187,7 +206,10 @@ export default class InventoryService implements IInventoryService {
    * @param input - the input object
    * @return The created reservation item
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async createReservationItem(
     input: CreateReservationItemInput,
     @MedusaContext() context: SharedContext = {}
@@ -222,7 +244,10 @@ export default class InventoryService implements IInventoryService {
    * @param input - the input object
    * @return The created inventory item
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async createInventoryItem(
     input: CreateInventoryItemInput,
     @MedusaContext() context: SharedContext = {}
@@ -239,7 +264,10 @@ export default class InventoryService implements IInventoryService {
    * @param input - the input object
    * @return The created inventory level
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async createInventoryLevel(
     input: CreateInventoryLevelInput,
     @MedusaContext() context: SharedContext = {}
@@ -253,7 +281,10 @@ export default class InventoryService implements IInventoryService {
    * @param input - the input object
    * @return The updated inventory item
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async updateInventoryItem(
     inventoryItemId: string,
     input: Partial<CreateInventoryItemInput>,
@@ -271,7 +302,10 @@ export default class InventoryService implements IInventoryService {
    * Deletes an inventory item
    * @param inventoryItemId - the id of the inventory item to delete
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async deleteInventoryItem(
     inventoryItemId: string,
     @MedusaContext() context: SharedContext = {}
@@ -284,7 +318,10 @@ export default class InventoryService implements IInventoryService {
     return await this.inventoryItemService_.delete(inventoryItemId, context)
   }
 
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async deleteInventoryItemLevelByLocationId(
     locationId: string,
     @MedusaContext() context: SharedContext = {}
@@ -295,7 +332,10 @@ export default class InventoryService implements IInventoryService {
     )
   }
 
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async deleteReservationItemByLocationId(
     locationId: string,
     @MedusaContext() context: SharedContext = {}
@@ -311,7 +351,10 @@ export default class InventoryService implements IInventoryService {
    * @param inventoryItemId - the id of the inventory item associated with the level
    * @param locationId - the id of the location associated with the level
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async deleteInventoryLevel(
     inventoryItemId: string,
     locationId: string,
@@ -337,7 +380,10 @@ export default class InventoryService implements IInventoryService {
    * @param input - the input object
    * @return The updated inventory level
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async updateInventoryLevel(
     inventoryItemId: string,
     locationId: string,
@@ -370,7 +416,10 @@ export default class InventoryService implements IInventoryService {
    * @param input - the input object
    * @return The updated inventory level
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async updateReservationItem(
     reservationItemId: string,
     input: UpdateReservationItemInput,
@@ -387,7 +436,10 @@ export default class InventoryService implements IInventoryService {
    * Deletes reservation items by line item
    * @param lineItemId - the id of the line item associated with the reservation item
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async deleteReservationItemsByLineItem(
     lineItemId: string,
     @MedusaContext() context: SharedContext = {}
@@ -402,7 +454,10 @@ export default class InventoryService implements IInventoryService {
    * Deletes a reservation item
    * @param reservationItemId - the id of the reservation item to delete
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async deleteReservationItem(
     reservationItemId: string | string[],
     @MedusaContext() context: SharedContext = {}
@@ -418,7 +473,10 @@ export default class InventoryService implements IInventoryService {
    * @return The updated inventory level
    * @throws when the inventory level is not found
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async adjustInventory(
     inventoryItemId: string,
     locationId: string,
@@ -455,7 +513,10 @@ export default class InventoryService implements IInventoryService {
    * @return The available quantity
    * @throws when the inventory item is not found
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async retrieveAvailableQuantity(
     inventoryItemId: string,
     locationIds: string[],
@@ -491,7 +552,10 @@ export default class InventoryService implements IInventoryService {
    * @return The stocked quantity
    * @throws when the inventory item is not found
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async retrieveStockedQuantity(
     inventoryItemId: string,
     locationIds: string[],
@@ -527,7 +591,10 @@ export default class InventoryService implements IInventoryService {
    * @return The reserved quantity
    * @throws when the inventory item is not found
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async retrieveReservedQuantity(
     inventoryItemId: string,
     locationIds: string[],
@@ -563,7 +630,10 @@ export default class InventoryService implements IInventoryService {
    * @param quantity - the quantity to check
    * @return Whether there is sufficient inventory
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async confirmInventory(
     inventoryItemId: string,
     locationIds: string[],

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

@@ -243,7 +243,7 @@ export default class ReservationItemService {
     )
 
     const ops: Promise<unknown>[] = [
-      itemRepository.softDelete({ line_item_id: lineItemId })
+      itemRepository.softDelete({ line_item_id: lineItemId }),
     ]
     for (const item of items) {
       ops.push(

packages/medusa/src/api/routes/admin/products/__tests__/create-product.js

@@ -55,23 +55,25 @@ describe("POST /admin/products", () => {
       expect(ProductVariantServiceMock.create).toHaveBeenCalledTimes(1)
       expect(ProductVariantServiceMock.create).toHaveBeenCalledWith(
         IdMap.getId("productWithOptions"),
-        {
+        [
-          title: "Test",
+          {
-          variant_rank: 0,
+            title: "Test",
-          prices: [
+            variant_rank: 0,
-            {
+            prices: [
-              currency_code: "USD",
+              {
-              amount: 100,
+                currency_code: "USD",
-            },
+                amount: 100,
-          ],
+              },
-          options: [
+            ],
-            {
+            options: [
-              option_id: IdMap.getId("option1"),
+              {
-              value: "100",
+                option_id: IdMap.getId("option1"),
-            },
+                value: "100",
-          ],
+              },
-          inventory_quantity: 0,
+            ],
-        }
+            inventory_quantity: 0,
+          },
+        ]
       )
     })
   })

packages/medusa/src/api/routes/admin/products/__tests__/create-variant.js

@@ -37,18 +37,20 @@ describe("POST /admin/products/:id/variants", () => {
       expect(ProductVariantServiceMock.create).toHaveBeenCalledTimes(1)
       expect(ProductVariantServiceMock.create).toHaveBeenCalledWith(
         IdMap.getId("productWithOptions"),
-        {
+        [
-          inventory_quantity: 0,
+          {
-          manage_inventory: true,
+            inventory_quantity: 0,
-          title: "Test Product Variant",
+            manage_inventory: true,
-          options: [],
+            title: "Test Product Variant",
-          prices: [
+            options: [],
-            {
+            prices: [
-              currency_code: "DKK",
+              {
-              amount: 1234,
+                currency_code: "DKK",
-            },
+                amount: 1234,
-          ],
+              },
-        }
+            ],
+          },
+        ]
       )
     })
 

packages/medusa/src/api/routes/admin/products/__tests__/update-product.js

@@ -49,7 +49,7 @@ describe("POST /admin/products/:id", () => {
 
     it("successfully updates variants and create new ones", async () => {
       expect(ProductVariantServiceMock.update).toHaveBeenCalledTimes(1)
-      expect(ProductVariantServiceMock.create).toHaveBeenCalledTimes(2)
+      expect(ProductVariantServiceMock.create).toHaveBeenCalledTimes(1)
     })
   })
 

packages/medusa/src/api/routes/admin/products/create-product.ts

@@ -39,7 +39,7 @@ import { FlagRouter } from "../../../../utils/flag-router"
 import { DistributedTransaction } from "../../../../utils/transaction"
 import { validator } from "../../../../utils/validator"
 import {
-  createVariantTransaction,
+  createVariantsTransaction,
   revertVariantTransaction,
 } from "./transaction/create-product-variant"
 
@@ -130,7 +130,7 @@ export default async (req, res) => {
 
   const entityManager: EntityManager = req.scope.resolve("manager")
 
-  const newProduct = await entityManager.transaction(async (manager) => {
+  const product = await entityManager.transaction(async (manager) => {
     const { variants } = validated
     delete validated.variants
 
@@ -185,27 +185,25 @@ export default async (req, res) => {
       }
 
       try {
-        await Promise.all(
+        const variantsInputData = variants.map((variant) => {
-          variants.map(async (variant) => {
+          const options =
-            const options =
+            variant?.options?.map((option, index) => ({
-              variant?.options?.map((option, index) => ({
+              ...option,
-                ...option,
+              option_id: optionIds[index],
-                option_id: optionIds[index],
+            })) || []
-              })) || []
 
-            const input = {
+          return {
-              ...variant,
+            ...variant,
-              options,
+            options,
-            }
+          } as CreateProductVariantInput
+        })
 
-            const varTransation = await createVariantTransaction(
+        const varTransaction = await createVariantsTransaction(
-              transactionDependencies,
+          transactionDependencies,
-              newProduct.id,
+          newProduct.id,
-              input as CreateProductVariantInput
+          variantsInputData
-            )
-            allVariantTransactions.push(varTransation)
-          })
         )
+        allVariantTransactions.push(varTransaction)
       } catch (e) {
         await Promise.all(
           allVariantTransactions.map(async (transaction) => {
@@ -220,15 +218,19 @@ export default async (req, res) => {
       }
     }
 
-    return newProduct
+    const rawProduct = await productService
-  })
+      .withTransaction(manager)
+      .retrieve(newProduct.id, {
+        select: defaultAdminProductFields,
+        relations: defaultAdminProductRelations,
+      })
 
-  const rawProduct = await productService.retrieve(newProduct.id, {
+    const [product] = await pricingService
-    select: defaultAdminProductFields,
+      .withTransaction(manager)
-    relations: defaultAdminProductRelations,
+      .setProductPrices([rawProduct])
-  })
 
-  const [product] = await pricingService.setProductPrices([rawProduct])
+    return product
+  })
 
   res.json({ product })
 }

packages/medusa/src/api/routes/admin/products/create-variant.ts

@@ -22,7 +22,7 @@ import {
   ProductVariantPricesCreateReq,
 } from "../../../../types/product-variant"
 import { validator } from "../../../../utils/validator"
-import { createVariantTransaction } from "./transaction/create-product-variant"
+import { createVariantsTransaction } from "./transaction/create-product-variant"
 
 /**
  * @oas [post] /admin/products/{id}/variants
@@ -131,7 +131,7 @@ export default async (req, res) => {
   const manager: EntityManager = req.scope.resolve("manager")
 
   await manager.transaction(async (transactionManager) => {
-    await createVariantTransaction(
+    await createVariantsTransaction(
       {
         manager: transactionManager,
         inventoryService,
@@ -139,7 +139,7 @@ export default async (req, res) => {
         productVariantService,
       },
       id,
-      validated as CreateProductVariantInput
+      [validated as CreateProductVariantInput]
     )
   })
 

packages/medusa/src/api/routes/admin/products/transaction/create-product-variant.ts

@@ -18,26 +18,26 @@ import {
 } from "../../../../../utils/transaction"
 
 enum actions {
-  createVariant = "createVariant",
+  createVariants = "createVariants",
-  createInventoryItem = "createInventoryItem",
+  createInventoryItems = "createInventoryItems",
-  attachInventoryItem = "attachInventoryItem",
+  attachInventoryItems = "attachInventoryItems",
 }
 
 const simpleFlow: TransactionStepsDefinition = {
   next: {
-    action: actions.createVariant,
+    action: actions.createVariants,
   },
 }
 
 const flowWithInventory: TransactionStepsDefinition = {
   next: {
-    action: actions.createVariant,
+    action: actions.createVariants,
     saveResponse: true,
     next: {
-      action: actions.createInventoryItem,
+      action: actions.createInventoryItems,
       saveResponse: true,
       next: {
-        action: actions.attachInventoryItem,
+        action: actions.attachInventoryItems,
         noCompensation: true,
       },
     },
@@ -61,10 +61,10 @@ type InjectedDependencies = {
   inventoryService?: IInventoryService
 }
 
-export const createVariantTransaction = async (
+export const createVariantsTransaction = async (
   dependencies: InjectedDependencies,
   productId: string,
-  input: CreateProductVariantInput
+  input: CreateProductVariantInput[]
 ): Promise<DistributedTransaction> => {
   const {
     manager,
@@ -78,51 +78,78 @@ export const createVariantTransaction = async (
 
   const productVariantServiceTx = productVariantService.withTransaction(manager)
 
-  async function createVariant(variantInput: CreateProductVariantInput) {
+  async function createVariants(variantInput: CreateProductVariantInput[]) {
     return await productVariantServiceTx.create(productId, variantInput)
   }
 
-  async function removeVariant(variant: ProductVariant) {
+  async function removeVariants(variants: ProductVariant[]) {
-    if (variant) {
+    if (variants.length) {
-      await productVariantServiceTx.delete(variant.id)
+      await productVariantServiceTx.delete(variants.map((v) => v.id))
     }
   }
 
-  async function createInventoryItem(variant: ProductVariant) {
+  async function createInventoryItems(variants: ProductVariant[] = []) {
-    if (!variant.manage_inventory) {
+    const context = { transactionManager: manager }
-      return
-    }
 
-    return await inventoryService!.createInventoryItem({
+    return await Promise.all(
-      sku: variant.sku,
+      variants.map(async (variant) => {
-      origin_country: variant.origin_country,
+        if (!variant.manage_inventory) {
-      hs_code: variant.hs_code,
+          return
-      mid_code: variant.mid_code,
+        }
-      material: variant.material,
+
-      weight: variant.weight,
+        const inventoryItem = await inventoryService!.createInventoryItem(
-      length: variant.length,
+          {
-      height: variant.height,
+            sku: variant.sku,
-      width: variant.width,
+            origin_country: variant.origin_country,
-    })
+            hs_code: variant.hs_code,
+            mid_code: variant.mid_code,
+            material: variant.material,
+            weight: variant.weight,
+            length: variant.length,
+            height: variant.height,
+            width: variant.width,
+          },
+          context
+        )
+
+        return { variant, inventoryItem }
+      })
+    )
   }
 
-  async function removeInventoryItem(inventoryItem: InventoryItemDTO) {
+  async function removeInventoryItems(
-    if (inventoryItem) {
+    data: {
-      await inventoryService!.deleteInventoryItem(inventoryItem.id)
+      variant: ProductVariant
-    }
+      inventoryItem: InventoryItemDTO
-  }
+    }[] = []
-
-  async function attachInventoryItem(
-    variant: ProductVariant,
-    inventoryItem: InventoryItemDTO
   ) {
-    if (!variant.manage_inventory) {
+    const context = { transactionManager: manager }
-      return
-    }
 
-    await productVariantInventoryServiceTx.attachInventoryItem(
+    return await Promise.all(
-      variant.id,
+      data.map(async ({ inventoryItem }) => {
-      inventoryItem.id
+        return await inventoryService!.deleteInventoryItem(
+          inventoryItem.id,
+          context
+        )
+      })
+    )
+  }
+
+  async function attachInventoryItems(
+    data: {
+      variant: ProductVariant
+      inventoryItem: InventoryItemDTO
+    }[]
+  ) {
+    return await Promise.all(
+      data
+        .filter((d) => d)
+        .map(async ({ variant, inventoryItem }) => {
+          return productVariantInventoryServiceTx.attachInventoryItem(
+            variant.id,
+            inventoryItem.id
+          )
+        })
     )
   }
 
@@ -132,46 +159,49 @@ export const createVariantTransaction = async (
     payload: TransactionPayload
   ) {
     const command = {
-      [actions.createVariant]: {
+      [actions.createVariants]: {
         [TransactionHandlerType.INVOKE]: async (
-          data: CreateProductVariantInput
+          data: CreateProductVariantInput[]
         ) => {
-          return await createVariant(data)
+          return await createVariants(data)
         },
         [TransactionHandlerType.COMPENSATE]: async (
-          data: CreateProductVariantInput,
+          data: CreateProductVariantInput[],
           { invoke }
         ) => {
-          await removeVariant(invoke[actions.createVariant])
+          await removeVariants(invoke[actions.createVariants])
         },
       },
-      [actions.createInventoryItem]: {
+      [actions.createInventoryItems]: {
         [TransactionHandlerType.INVOKE]: async (
-          data: CreateProductVariantInput,
+          data: CreateProductVariantInput[],
           { invoke }
         ) => {
-          const { [actions.createVariant]: variant } = invoke
+          const { [actions.createVariants]: variants } = invoke
 
-          return await createInventoryItem(variant)
+          return await createInventoryItems(variants)
         },
         [TransactionHandlerType.COMPENSATE]: async (
-          data: CreateProductVariantInput,
+          data: {
+            variant: ProductVariant
+            inventoryItem: InventoryItemDTO
+          }[],
           { invoke }
         ) => {
-          await removeInventoryItem(invoke[actions.createInventoryItem])
+          await removeInventoryItems(invoke[actions.createInventoryItems])
         },
       },
-      [actions.attachInventoryItem]: {
+      [actions.attachInventoryItems]: {
         [TransactionHandlerType.INVOKE]: async (
-          data: CreateProductVariantInput,
+          data: CreateProductVariantInput[],
           { invoke }
         ) => {
           const {
-            [actions.createVariant]: variant,
+            [actions.createVariants]: variants,
-            [actions.createInventoryItem]: inventoryItem,
+            [actions.createInventoryItems]: inventoryItemsResult,
           } = invoke
 
-          return await attachInventoryItem(variant, inventoryItem)
+          return await attachInventoryItems(inventoryItemsResult)
         },
       },
     }

packages/medusa/src/api/routes/admin/products/update-product.ts

@@ -41,7 +41,7 @@ import { FeatureFlagDecorators } from "../../../../utils/feature-flag-decorators
 import { DistributedTransaction } from "../../../../utils/transaction"
 import { validator } from "../../../../utils/validator"
 import {
-  createVariantTransaction,
+  createVariantsTransaction,
   revertVariantTransaction,
 } from "./transaction/create-product-variant"
 
@@ -224,29 +224,25 @@ export default async (req, res) => {
     }
 
     if (variantsToCreate.length) {
-      await Promise.all(
+      try {
-        variantsToCreate.map(async (data) => {
+        const varTransaction = await createVariantsTransaction(
-          try {
+          transactionDependencies,
-            const varTransaction = await createVariantTransaction(
+          product.id,
+          variantsToCreate as CreateProductVariantInput[]
+        )
+        allVariantTransactions.push(varTransaction)
+      } catch (e) {
+        await Promise.all(
+          allVariantTransactions.map(async (transaction) => {
+            await revertVariantTransaction(
               transactionDependencies,
-              product.id,
+              transaction
-              data as CreateProductVariantInput
+            ).catch(() => logger.warn("Transaction couldn't be reverted."))
-            )
+          })
-            allVariantTransactions.push(varTransaction)
+        )
-          } catch (e) {
-            await Promise.all(
-              allVariantTransactions.map(async (transaction) => {
-                await revertVariantTransaction(
-                  transactionDependencies,
-                  transaction
-                ).catch(() => logger.warn("Transaction couldn't be reverted."))
-              })
-            )
 
-            throw e
+        throw e
-          }
+      }
-        })
-      )
     }
   })
 

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

@@ -24,7 +24,7 @@ export default class EventBusService
   protected readonly config_: ConfigModule
   protected readonly stagedJobService_: StagedJobService
   // eslint-disable-next-line max-len
-  protected readonly eventBusModuleService_: EventBusUtils.AbstractEventBusModuleService
+  protected readonly eventBusModuleService_: EventBusTypes.IEventBusModuleService
 
   protected shouldEnqueuerRun: boolean
   protected enqueue_: Promise<void>
@@ -137,7 +137,7 @@ export default class EventBusService
   ): Promise<TResult | void> {
     const manager = this.activeManager_
     const isBulkEmit = !isString(eventNameOrData)
-    const events = isBulkEmit
+    let events: EventBusTypes.EmitData[] = isBulkEmit
       ? eventNameOrData.map((event) => ({
           eventName: event.eventName,
           data: event.data,
@@ -151,22 +151,32 @@ export default class EventBusService
           },
         ]
 
-    /**
+    events = events.filter(
-     * We store events in the database when in an ongoing transaction.
+      (event) =>
-     *
+        this.eventBusModuleService_?.retrieveSubscribers(event.eventName)
-     * If we are in a long-running transaction, the ACID properties of a
+          ?.length ?? false
-     * transaction ensure, that events are kept invisible to the enqueuer
+    )
-     * until the transaction has committed.
+
-     *
+    let stagedJobs: StagedJob[] = []
-     * This patterns also gives us at-least-once delivery of events, as events
+    if (events.length) {
-     * are only removed from the database, if they are successfully delivered.
+      /**
-     *
+       * We store events in the database when in an ongoing transaction.
-     * In case of a failing transaction, jobs stored in the database are removed
+       *
-     * as part of the rollback.
+       * If we are in a long-running transaction, the ACID properties of a
-     */
+       * transaction ensure, that events are kept invisible to the enqueuer
-    const stagedJobs = await this.stagedJobService_
+       * until the transaction has committed.
-      .withTransaction(manager)
+       *
-      .create(events)
+       * This patterns also gives us at-least-once delivery of events, as events
+       * are only removed from the database, if they are successfully delivered.
+       *
+       * In case of a failing transaction, jobs stored in the database are removed
+       * as part of the rollback.
+       */
+
+      stagedJobs = await this.stagedJobService_
+        .withTransaction(manager)
+        .create(events)
+    }
 
     return (!isBulkEmit ? stagedJobs[0] : stagedJobs) as unknown as TResult
   }

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

@@ -162,20 +162,26 @@ class ProductVariantService extends TransactionBaseService {
    * Creates an unpublished product variant. Will validate against parent product
    * to ensure that the variant can in fact be created.
    * @param productOrProductId - the product the variant will be added to
-   * @param variant - the variant to create
+   * @param variants
    * @return resolves to the creation result.
    */
-  async create(
+  async create<
+    TVariants extends CreateProductVariantInput | CreateProductVariantInput[],
+    TOutput = TVariants extends CreateProductVariantInput[]
+      ? CreateProductVariantInput[]
+      : CreateProductVariantInput
+  >(
     productOrProductId: string | Product,
-    variant: CreateProductVariantInput
+    variants: CreateProductVariantInput | CreateProductVariantInput[]
-  ): Promise<ProductVariant> {
+  ): Promise<TOutput> {
+    const isVariantsArray = Array.isArray(variants)
+    const variants_ = isVariantsArray ? variants : [variants]
+
     return await this.atomicPhase_(async (manager: EntityManager) => {
       const productRepo = manager.withRepository(this.productRepository_)
       const variantRepo = manager.withRepository(this.productVariantRepository_)
 
-      const { prices, ...rest } = variant
+      let product = productOrProductId as Product
-
-      let product = productOrProductId
 
       if (isString(product)) {
         product = (await productRepo.findOne({
@@ -195,69 +201,60 @@ class ProductVariantService extends TransactionBaseService {
         )
       }
 
-      if (product.options.length !== variant.options.length) {
+      this.validateVariantsToCreate_(product, variants_)
-        throw new MedusaError(
-          MedusaError.Types.INVALID_DATA,
-          `Product options length does not match variant options length. Product has ${product.options.length} and variant has ${variant.options.length}.`
-        )
-      }
 
-      product.options.forEach((option) => {
+      let computedRank = product.variants.length
-        if (!variant.options.find((vo) => option.id === vo.option_id)) {
+      const variantPricesToUpdate: {
-          throw new MedusaError(
+        id: string
-            MedusaError.Types.INVALID_DATA,
+        prices: ProductVariantPrice[]
-            `Variant options do not contain value for ${option.title}`
+      }[] = []
-          )
-        }
-      })
 
-      const variantExists = product.variants.find((v) => {
+      const results = await Promise.all(
-        return v.options.every((option) => {
+        variants_.map(async (variant) => {
-          const variantOption = variant.options.find(
+          const { prices, ...rest } = variant
-            (o) => option.option_id === o.option_id
-          )
 
-          return option.value === variantOption?.value
+          if (!rest.variant_rank) {
+            rest.variant_rank = computedRank
+          }
+          ++computedRank
+
+          const toCreate = {
+            ...rest,
+            product_id: product.id,
+          }
+
+          const productVariant = variantRepo.create(toCreate)
+
+          const result = await variantRepo.save(productVariant)
+
+          if (prices?.length) {
+            variantPricesToUpdate.push({ id: result.id, prices })
+          }
+
+          return result
         })
-      })
+      )
 
-      if (variantExists) {
+      if (variantPricesToUpdate.length) {
-        throw new MedusaError(
+        await this.updateVariantPrices(
-          MedusaError.Types.DUPLICATE_ERROR,
+          variantPricesToUpdate.map((v) => ({
-          `Variant with title ${variantExists.title} with provided options already exists`
+            variantId: v.id,
+            prices: v.prices,
+          }))
         )
       }
 
-      if (!rest.variant_rank) {
+      const eventsToEmit = results.map((result) => ({
-        rest.variant_rank = product.variants.length
+        eventName: ProductVariantService.Events.CREATED,
-      }
+        data: {
-
-      const toCreate = {
-        ...rest,
-        product_id: product.id,
-      }
-
-      const productVariant = variantRepo.create(toCreate)
-
-      const result = await variantRepo.save(productVariant)
-
-      if (prices) {
-        await this.updateVariantPrices([
-          {
-            variantId: result.id,
-            prices,
-          },
-        ])
-      }
-
-      await this.eventBus_
-        .withTransaction(manager)
-        .emit(ProductVariantService.Events.CREATED, {
           id: result.id,
           product_id: result.product_id,
-        })
+        },
+      }))
 
-      return result
+      await this.eventBus_.withTransaction(manager).emit(eventsToEmit)
+
+      return (isVariantsArray ? results : results[0]) as unknown as TOutput
     })
   }
 
@@ -1104,6 +1101,46 @@ class ProductVariantService extends TransactionBaseService {
 
     return qb
   }
+
+  protected validateVariantsToCreate_(
+    product: Product,
+    variants: CreateProductVariantInput[]
+  ): void {
+    for (const variant of variants) {
+      if (product.options.length !== variant.options.length) {
+        throw new MedusaError(
+          MedusaError.Types.INVALID_DATA,
+          `Product options length does not match variant options length. Product has ${product.options.length} and variant has ${variant.options.length}.`
+        )
+      }
+
+      product.options.forEach((option) => {
+        if (!variant.options.find((vo) => option.id === vo.option_id)) {
+          throw new MedusaError(
+            MedusaError.Types.INVALID_DATA,
+            `Variant options do not contain value for ${option.title}`
+          )
+        }
+      })
+
+      const variantExists = product.variants.find((v) => {
+        return v.options.every((option) => {
+          const variantOption = variant.options.find(
+            (o) => option.option_id === o.option_id
+          )
+
+          return option.value === variantOption?.value
+        })
+      })
+
+      if (variantExists) {
+        throw new MedusaError(
+          MedusaError.Types.DUPLICATE_ERROR,
+          `Variant with title ${variantExists.title} with provided options already exists`
+        )
+      }
+    }
+  }
 }
 
 export default ProductVariantService

packages/stock-location/src/services/stock-location.ts

@@ -4,6 +4,7 @@ import {
   FilterableStockLocationProps,
   FindConfig,
   IEventBusService,
+  MODULE_RESOURCE_TYPE,
   SharedContext,
   StockLocationAddressInput,
   UpdateStockLocationInput,
@@ -41,7 +42,7 @@ export default class StockLocationService {
   constructor(
     { eventBusService, manager }: InjectedDependencies,
     options?: unknown,
-    moduleDeclaration?: InternalModuleDeclaration
+    protected readonly moduleDeclaration?: InternalModuleDeclaration
   ) {
     this.manager_ = manager
     this.eventBusService_ = eventBusService
@@ -53,7 +54,10 @@ export default class StockLocationService {
    * @param config - Additional configuration for the query.
    * @return A list of stock locations.
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async list(
     selector: FilterableStockLocationProps = {},
     config: FindConfig<StockLocation> = { relations: [], skip: 0, take: 10 },
@@ -72,7 +76,10 @@ export default class StockLocationService {
    * @param config - Additional configuration for the query.
    * @return A list of stock locations and the count of matching stock locations.
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async listAndCount(
     selector: FilterableStockLocationProps = {},
     config: FindConfig<StockLocation> = { relations: [], skip: 0, take: 10 },
@@ -92,7 +99,10 @@ export default class StockLocationService {
    * @return The stock location.
    * @throws If the stock location ID is not definedor the stock location with the given ID was not found.
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async retrieve(
     stockLocationId: string,
     config: FindConfig<StockLocation> = {},
@@ -126,7 +136,10 @@ export default class StockLocationService {
    * @param data - The input data for creating a Stock Location.
    * @returns The created stock location.
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async create(
     data: CreateStockLocationInput,
     @MedusaContext() context: SharedContext = {}
@@ -170,7 +183,10 @@ export default class StockLocationService {
    * @param updateData - The update data for the stock location.
    * @returns The updated stock location.
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async update(
     stockLocationId: string,
     updateData: UpdateStockLocationInput,
@@ -216,7 +232,10 @@ export default class StockLocationService {
    * @param address - The update data for the address.
    * @returns The updated stock location address.
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   protected async updateAddress(
     addressId: string,
     address: StockLocationAddressInput,
@@ -257,7 +276,10 @@ export default class StockLocationService {
    * @param id - The ID of the stock location to delete.
    * @returns An empty promise.
    */
-  @InjectEntityManager()
+  @InjectEntityManager(
+    (target) =>
+      target.moduleDeclaration?.resources === MODULE_RESOURCE_TYPE.ISOLATED
+  )
   async delete(
     id: string,
     @MedusaContext() context: SharedContext = {}

packages/types/src/event-bus/event-bus-module.ts

@@ -1,6 +1,15 @@
-import { EmitData, Subscriber, SubscriberContext } from "./common"
+import {
+  EmitData,
+  Subscriber,
+  SubscriberContext,
+  SubscriberDescriptor,
+} from "./common"
 
 export interface IEventBusModuleService {
+  retrieveSubscribers(
+    event: string | symbol
+  ): SubscriberDescriptor[] | undefined
+
   emit<T>(
     eventName: string,
     data: T,

packages/utils/src/decorators/inject-entity-manager.ts

@@ -1,7 +1,8 @@
 import { SharedContext } from "@medusajs/types"
 
 export function InjectEntityManager(
-  managerProperty = "manager_"
+  shouldForceTransaction: (target: any) => boolean = () => false,
+  managerProperty: string = "manager_"
 ): MethodDecorator {
   return function (
     target: any,
@@ -18,9 +19,10 @@ export function InjectEntityManager(
 
     const argIndex = target.MedusaContextIndex_[propertyKey]
     descriptor.value = async function (...args: any[]) {
+      const shouldForceTransactionRes = shouldForceTransaction(target)
       const context: SharedContext = args[argIndex] ?? {}
 
-      if (context?.transactionManager) {
+      if (!shouldForceTransactionRes && context?.transactionManager) {
         return await originalMethod.apply(this, args)
       }
 

packages/utils/src/event-bus/index.ts

@@ -23,24 +23,15 @@ export abstract class AbstractEventBusModuleService
   ): Promise<void>
   abstract emit<T>(data: EventBusTypes.EmitData<T>[]): Promise<void>
 
-  public subscribe(
+  protected storeSubscribers({
-    eventName: string | symbol,
+    event,
-    subscriber: EventBusTypes.Subscriber,
+    subscriberId,
-    context?: EventBusTypes.SubscriberContext
+    subscriber,
-  ): this {
+  }: {
-    if (typeof subscriber !== `function`) {
+    event: string | symbol
-      throw new Error("Subscriber must be a function")
+    subscriberId: string
-    }
+    subscriber: EventBusTypes.Subscriber
-    /**
+  }) {
-     * If context is provided, we use the subscriberId from it
-     * otherwise we generate a random using a ulid
-     */
-
-    const randId = ulid()
-    const event = eventName.toString()
-
-    const subscriberId = context?.subscriberId ?? `${event}-${randId}`
-
     const newSubscriberDescriptor = { subscriber, id: subscriberId }
 
     const existingSubscribers = this.eventToSubscribersMap_.get(event) ?? []
@@ -57,6 +48,33 @@ export abstract class AbstractEventBusModuleService
       ...existingSubscribers,
       newSubscriberDescriptor,
     ])
+  }
+
+  public retrieveSubscribers(event: string | symbol) {
+    return this.eventToSubscribersMap_.get(event)
+  }
+
+  public subscribe(
+    eventName: string | symbol,
+    subscriber: EventBusTypes.Subscriber,
+    context?: EventBusTypes.SubscriberContext
+  ): this {
+    if (typeof subscriber !== `function`) {
+      throw new Error("Subscriber must be a function")
+    }
+    /**
+     * If context is provided, we use the subscriberId from it
+     * otherwise we generate a random using a ulid
+     */
+
+    const randId = ulid()
+    const event = eventName.toString()
+
+    this.storeSubscribers({
+      event,
+      subscriberId: context?.subscriberId ?? `${event}-${randId}`,
+      subscriber,
+    })
 
     return this
   }
@@ -70,7 +88,7 @@ export abstract class AbstractEventBusModuleService
       throw new Error("Subscriber must be a function")
     }
 
-    const existingSubscribers = this.eventToSubscribersMap_.get(eventName)
+    const existingSubscribers = this.retrieveSubscribers(eventName)
 
     if (existingSubscribers?.length) {
       const subIndex = existingSubscribers?.findIndex(