chore(product): revamp upsertWithReplace and Remove its usage from product creation (#11585)

**What**
- Move create product to use native create by structuring the data appropriately, it means no more `upsertWithReplace` being very poorly performant and got 20x better performances on staging
- Improvements in `upsertWithReplace` to still get performance boost for places that still relies on it. Mostly bulking the operations when possible

Co-authored-by: Carlos R. L. Rodrigues <37986729+carlos-r-l-rodrigues@users.noreply.github.com>

.changeset/clean-cars-build.md

@@ -0,0 +1,6 @@
+---
+"@medusajs/product": patch
+"@medusajs/utils": patch
+---
+
+chore(product): Remove upsertWithReplace where needed

integration-tests/http/__tests__/product/admin/product-import.spec.ts

@@ -533,7 +533,7 @@ medusaIntegrationTestRunner({
         )
       })
 
-      it("should fail on non-existent product fields being present in the CSV", async () => {
+      it("should successfully skip non-existent product fields being present in the CSV", async () => {
         const subscriberExecution = TestEventUtils.waitSubscribersExecution(
           `${Modules.NOTIFICATION}.notification.${CommonEvents.CREATED}`,
           eventBus
@@ -582,10 +582,21 @@ medusaIntegrationTestRunner({
           expect.objectContaining({
             data: expect.objectContaining({
               title: "Product import",
-              description: `Failed to import products from file test.csv`,
+              description:
+                "Product import of file test.csv completed successfully!",
             }),
           })
         )
+
+        const [importedProduct] = (
+          await api.get("/admin/products?limit=1&order=-id", adminHeaders)
+        ).data.products
+
+        expect(importedProduct).not.toEqual(
+          expect.objectContaining({
+            field: "Test product",
+          })
+        )
       })
 
       it("supports importing the v1 template", async () => {

packages/core/utils/src/dal/mikro-orm/integration-tests/__tests__/mikro-orm-repository.spec.ts

@@ -693,6 +693,96 @@ describe("mikroOrmRepository", () => {
       )
     })
 
+    it("should successfully update, create, and delete subentities an entity with a one-to-many relation within a transaction", async () => {
+      const entity1 = {
+        id: "1",
+        title: "en1",
+        entity2: [
+          { id: "2", title: "en2-1", handle: "some-handle" },
+          { id: "3", title: "en2-2", handle: "some-other-handle" },
+        ] as any[],
+      }
+
+      const { entities: entities1, performedActions: performedActions1 } =
+        await manager1().transaction(async (txManager) => {
+          return await manager1().upsertWithReplace(
+            [entity1],
+            {
+              relations: ["entity2"],
+            },
+            {
+              transactionManager: txManager,
+            }
+          )
+        })
+
+      expect(performedActions1).toEqual({
+        created: {
+          [Entity1.name]: [expect.objectContaining({ id: entity1.id })],
+          [Entity2.name]: entities1[0].entity2.map((entity2) =>
+            expect.objectContaining({ id: entity2.id })
+          ),
+        },
+        updated: {},
+        deleted: {},
+      })
+
+      entity1.entity2 = [
+        { id: "2", title: "newen2-1" },
+        { title: "en2-3", handle: "some-new-handle" },
+      ]
+
+      const { entities: entities2, performedActions: performedActions2 } =
+        await manager1().transaction(async (txManager) => {
+          return await manager1().upsertWithReplace(
+            [entity1],
+            {
+              relations: ["entity2"],
+            },
+            { transactionManager: txManager }
+          )
+        })
+
+      const entity2En23 = entities2[0].entity2.find((e) => e.title === "en2-3")!
+
+      expect(performedActions2).toEqual({
+        created: {
+          [Entity2.name]: [expect.objectContaining({ id: entity2En23.id })],
+        },
+        updated: {
+          [Entity1.name]: [expect.objectContaining({ id: entity1.id })],
+          [Entity2.name]: [expect.objectContaining({ id: "2" })],
+        },
+        deleted: {
+          [Entity2.name]: [expect.objectContaining({ id: "3" })],
+        },
+      })
+
+      const listedEntities = await manager1().find({
+        where: { id: "1" },
+        options: { populate: ["entity2"] },
+      })
+
+      expect(listedEntities).toHaveLength(1)
+      expect(listedEntities[0]).toEqual(
+        expect.objectContaining({
+          id: "1",
+          title: "en1",
+        })
+      )
+      expect(listedEntities[0].entity2.getItems()).toHaveLength(2)
+      expect(listedEntities[0].entity2.getItems()).toEqual(
+        expect.arrayContaining([
+          expect.objectContaining({
+            title: "newen2-1",
+          }),
+          expect.objectContaining({
+            title: "en2-3",
+          }),
+        ])
+      )
+    })
+
     it("should update an entity with a one-to-many relation that has the same unique constraint key", async () => {
       const entity1 = {
         id: "1",
@@ -1106,7 +1196,9 @@ describe("mikroOrmRepository", () => {
   describe("error mapping", () => {
     it("should map UniqueConstraintViolationException to MedusaError on upsertWithReplace", async () => {
       const entity3 = { title: "en3" }
+
       await manager3().upsertWithReplace([entity3])
+
       const err = await manager3()
         .upsertWithReplace([entity3])
         .catch((e) => e.message)

packages/core/utils/src/dal/mikro-orm/mikro-orm-repository.ts

@@ -803,15 +803,20 @@ export function mikroOrmBaseRepositoryFactory<const T extends object>(
           }
         })
 
-        const qb = manager.qb(relation.pivotEntity)
+        await promiseAll([
-        await qb.insert(pivotData).onConflict().ignore().execute()
+          manager
-
+            .qb(relation.pivotEntity)
-        await manager.nativeDelete(relation.pivotEntity, {
+            .insert(pivotData)
-          [parentPivotColumn]: (data as any).id,
+            .onConflict()
-          [currentPivotColumn]: {
+            .ignore()
-            $nin: pivotData.map((d) => d[currentPivotColumn]),
+            .execute(),
-          },
+          manager.nativeDelete(relation.pivotEntity, {
-        })
+            [parentPivotColumn]: (data as any).id,
+            [currentPivotColumn]: {
+              $nin: pivotData.map((d) => d[currentPivotColumn]),
+            },
+          }),
+        ])
 
         return { entities: normalizedData, performedActions }
       }
@@ -826,27 +831,23 @@ export function mikroOrmBaseRepositoryFactory<const T extends object>(
           joinColumnsConstraints[joinColumn] = data[referencedColumnName]
         })
 
-        const toDeleteEntities = await manager.find<any, any, "id">(
+        const deletedRelations = await (
-          relation.type,
+          manager.getTransactionContext() ?? manager.getKnex()
-          {
-            ...joinColumnsConstraints,
-            id: { $nin: normalizedData.map((d: any) => d.id) },
-          },
-          {
-            fields: ["id"],
-          }
         )
-        const toDeleteIds = toDeleteEntities.map((d: any) => d.id)
+          .queryBuilder()
+          .from(relation.targetMeta!.collection)
+          .delete()
+          .where(joinColumnsConstraints)
+          .whereNotIn(
+            "id",
+            normalizedData.map((d: any) => d.id)
+          )
+          .returning("id")
 
-        await manager.nativeDelete(relation.type, {
+        if (deletedRelations.length) {
-          ...joinColumnsConstraints,
-          id: { $in: toDeleteIds },
-        })
-
-        if (toDeleteEntities.length) {
           performedActions.deleted[relation.type] ??= []
           performedActions.deleted[relation.type].push(
-            ...toDeleteEntities.map((d) => ({ id: d.id }))
+            ...deletedRelations.map((row) => ({ id: row.id }))
           )
         }
 
@@ -970,38 +971,46 @@ export function mikroOrmBaseRepositoryFactory<const T extends object>(
         deleted: {},
       }
 
-      await promiseAll(
+      const promises: Promise<any>[] = []
-        entries.map(async (data) => {
+      const toInsert: unknown[] = []
-          const existingEntity = existingEntitiesMap.get(data.id)
+      let shouldInsert = false
-          orderedEntities.push(data)
+
-          if (existingEntity) {
+      entries.map(async (data) => {
-            if (skipUpdate) {
+        const existingEntity = existingEntitiesMap.get(data.id)
-              return
+        orderedEntities.push(data)
-            }
+        if (existingEntity) {
-            await manager.nativeUpdate(entityName, { id: data.id }, data)
+          if (skipUpdate) {
-            performedActions.updated[entityName] ??= []
+            return
-            performedActions.updated[entityName].push({ id: data.id })
-          } else {
-            const qb = manager.qb(entityName)
-            if (skipUpdate) {
-              const res = await qb
-                .insert(data)
-                .onConflict()
-                .ignore()
-                .execute("all", true)
-              if (res) {
-                performedActions.created[entityName] ??= []
-                performedActions.created[entityName].push({ id: data.id })
-              }
-            } else {
-              await manager.insert(entityName, data)
-              performedActions.created[entityName] ??= []
-              performedActions.created[entityName].push({ id: data.id })
-              // await manager.insert(entityName, data)
-            }
           }
-        })
+          const update = manager.nativeUpdate(entityName, { id: data.id }, data)
-      )
+          promises.push(update)
+
+          performedActions.updated[entityName] ??= []
+          performedActions.updated[entityName].push({ id: data.id })
+        } else {
+          shouldInsert = true
+          toInsert.push(data)
+        }
+      })
+
+      if (shouldInsert) {
+        let insertQb = manager.qb(entityName).insert(toInsert).returning("id")
+
+        if (skipUpdate) {
+          insertQb = insertQb.onConflict().ignore()
+        }
+
+        promises.push(
+          insertQb.execute("all", true).then((res: { id: string }[]) => {
+            performedActions.created[entityName] ??= []
+            performedActions.created[entityName].push(
+              ...res.map((data) => ({ id: data.id }))
+            )
+          })
+        )
+      }
+
+      await promiseAll(promises)
 
       return { orderedEntities, performedActions }
     }

packages/core/utils/src/dml/helpers/entity-builder/define-relationship.ts

@@ -283,14 +283,17 @@ export function defineHasManyRelationship(
 ) {
   const shouldRemoveRelated = !!cascades.delete?.includes(relationship.name)
 
-  OneToMany({
+  const options: Parameters<typeof OneToMany>[0] = {
     entity: relatedModelName,
     orphanRemoval: true,
     mappedBy: relationship.mappedBy || camelToSnakeCase(MikroORMEntity.name),
-    cascade: shouldRemoveRelated
+  }
-      ? (["persist", "soft-remove"] as any)
+
-      : undefined,
+  if (shouldRemoveRelated) {
-  })(MikroORMEntity.prototype, relationship.name)
+    options.cascade = ["persist", "soft-remove"] as any
+  }
+
+  OneToMany(options)(MikroORMEntity.prototype, relationship.name)
 }
 
 /**

packages/modules/product/integration-tests/__fixtures__/product/data/create-product.ts

@@ -48,11 +48,11 @@ export const buildProductAndRelationsData = ({
   images,
   status,
   type_id,
-  tags,
+  tag_ids,
   options,
   variants,
   collection_id,
-}: Partial<ProductTypes.CreateProductDTO> & { tags: { value: string }[] }) => {
+}: Partial<ProductTypes.CreateProductDTO>) => {
   const defaultOptionTitle = "test-option"
   const defaultOptionValue = "test-value"
 
@@ -66,7 +66,7 @@ export const buildProductAndRelationsData = ({
     status: status ?? ProductStatus.PUBLISHED,
     images: (images ?? []) as ProductImage[],
     type_id,
-    tags: tags ?? [{ value: "tag-1" }],
+    tag_ids,
     collection_id,
     options: options ?? [
       {

packages/modules/product/integration-tests/__tests__/product-module-service/products.spec.ts

@@ -30,7 +30,7 @@ import {
   createTypes,
 } from "../../__fixtures__/product"
 
-jest.setTimeout(300000)
+jest.setTimeout(3000000)
 
 moduleIntegrationTestRunner<IProductModuleService>({
   moduleName: Modules.PRODUCT,
@@ -181,6 +181,7 @@ moduleIntegrationTestRunner<IProductModuleService>({
         })
 
         it("should update a product and upsert relations that are not created yet", async () => {
+          const tags = await service.createProductTags([{ value: "tag-1" }])
           const data = buildProductAndRelationsData({
             images,
             thumbnail: images[0].url,
@@ -190,6 +191,7 @@ moduleIntegrationTestRunner<IProductModuleService>({
                 values: ["val-1", "val-2"],
               },
             ],
+            tag_ids: [tags[0].id],
           })
 
           const variantTitle = data.variants[0].title
@@ -217,7 +219,7 @@ moduleIntegrationTestRunner<IProductModuleService>({
           productBefore.options = data.options
           productBefore.images = data.images
           productBefore.thumbnail = data.thumbnail
-          productBefore.tags = data.tags
+          productBefore.tag_ids = data.tag_ids
           const updatedProducts = await service.upsertProducts([productBefore])
           expect(updatedProducts).toHaveLength(1)
 
@@ -273,7 +275,7 @@ moduleIntegrationTestRunner<IProductModuleService>({
               tags: expect.arrayContaining([
                 expect.objectContaining({
                   id: expect.any(String),
-                  value: productBefore.tags?.[0].value,
+                  value: tags[0].value,
                 }),
               ]),
               variants: expect.arrayContaining([
@@ -856,9 +858,11 @@ moduleIntegrationTestRunner<IProductModuleService>({
       describe("create", function () {
         let images = [{ url: "image-1" }]
         it("should create a product", async () => {
+          const tags = await service.createProductTags([{ value: "tag-1" }])
           const data = buildProductAndRelationsData({
             images,
             thumbnail: images[0].url,
+            tag_ids: [tags[0].id],
           })
 
           const productsCreated = await service.createProducts([data])
@@ -917,7 +921,7 @@ moduleIntegrationTestRunner<IProductModuleService>({
               tags: expect.arrayContaining([
                 expect.objectContaining({
                   id: expect.any(String),
-                  value: data.tags[0].value,
+                  value: tags[0].value,
                 }),
               ]),
               variants: expect.arrayContaining([
@@ -1164,15 +1168,17 @@ moduleIntegrationTestRunner<IProductModuleService>({
           productCollectionOne = collections[0]
           productCollectionTwo = collections[1]
 
+          const tags = await service.createProductTags([{ value: "tag-1" }])
+
           const resp = await service.createProducts([
             buildProductAndRelationsData({
               collection_id: productCollectionOne.id,
               options: [{ title: "size", values: ["large", "small"] }],
               variants: [{ title: "variant 1", options: { size: "small" } }],
+              tag_ids: [tags[0].id],
             }),
             buildProductAndRelationsData({
               collection_id: productCollectionTwo.id,
-              tags: [],
             }),
           ])
 

packages/modules/product/package.json

@@ -29,7 +29,7 @@
     "resolve:aliases": "tsc --showConfig -p tsconfig.json > tsconfig.resolved.json && tsc-alias -p tsconfig.resolved.json && rimraf tsconfig.resolved.json",
     "build": "rimraf dist && tsc --build && npm run resolve:aliases",
     "test": "jest --runInBand --bail --forceExit -- src/**/__tests__/**/*.ts",
-    "test:integration": "jest --forceExit -- integration-tests/__tests__/**/*.ts",
+    "test:integration": "jest --runInBand --bail --forceExit -- integration-tests/__tests__/**/*.ts",
     "migration:initial": " MIKRO_ORM_CLI_CONFIG=./mikro-orm.config.dev.ts medusa-mikro-orm migration:create --initial",
     "migration:create": " MIKRO_ORM_CLI_CONFIG=./mikro-orm.config.dev.ts medusa-mikro-orm migration:create",
     "migration:up": " MIKRO_ORM_CLI_CONFIG=./mikro-orm.config.dev.ts medusa-mikro-orm migration:up",

packages/modules/product/src/services/product-module-service.ts

@@ -26,6 +26,7 @@ import { ProductCategoryService } from "@services"
 import {
   arrayDifference,
   EmitEvents,
+  generateEntityId,
   InjectManager,
   InjectTransactionManager,
   isDefined,
@@ -1568,75 +1569,82 @@ export default class ProductModuleService
       })
     )
 
-    const { entities: productData } =
+    const tagIds = normalizedInput
-      await this.productService_.upsertWithReplace(
+      .flatMap((d) => (d as any).tags ?? [])
-        normalizedInput,
+      .map((t) => t.id)
+    let existingTags: InferEntityType<typeof ProductTag>[] = []
+
+    if (tagIds.length) {
+      existingTags = await this.productTagService_.list(
         {
-          relations: ["tags", "categories"],
+          id: tagIds,
         },
+        {},
         sharedContext
       )
+    }
 
-    await promiseAll(
+    const existingTagsMap = new Map(existingTags.map((tag) => [tag.id, tag]))
-      // Note: It's safe to rely on the order here as `upsertWithReplace` preserves the order of the input
-      normalizedInput.map(async (product, i) => {
-        const upsertedProduct: any = productData[i]
-        upsertedProduct.options = []
-        upsertedProduct.variants = []
 
-        if (product.options?.length) {
+    const productsToCreate = normalizedInput.map((product) => {
-          const { entities: productOptions } =
+      const productId = generateEntityId(product.id, "prod")
-            await this.productOptionService_.upsertWithReplace(
+      product.id = productId
-              product.options?.map((option) => ({
-                ...option,
-                product_id: upsertedProduct.id,
-              })) ?? [],
-              { relations: ["values"] },
-              sharedContext
-            )
-          upsertedProduct.options = productOptions
-        }
 
-        if (product.variants?.length) {
+      if ((product as any).categories?.length) {
-          const { entities: productVariants } =
+        ;(product as any).categories = (product as any).categories.map(
-            await this.productVariantService_.upsertWithReplace(
+          (category: { id: string }) => category.id
-              ProductModuleService.assignOptionsToVariants(
+        )
-                product.variants?.map((v) => ({
+      }
-                  ...v,
-                  product_id: upsertedProduct.id,
-                })) ?? [],
-                upsertedProduct.options
-              ),
-              { relations: ["options"] },
-              sharedContext
-            )
-          upsertedProduct.variants = productVariants
-        }
 
-        if (Array.isArray(product.images)) {
+      if (product.variants?.length) {
-          if (product.images.length) {
+        const normalizedVariants = product.variants.map((variant) => {
-            const { entities: productImages } =
+          const variantId = generateEntityId((variant as any).id, "variant")
-              await this.productImageService_.upsertWithReplace(
+          ;(variant as any).id = variantId
-                product.images.map((image, rank) => ({
+
-                  ...image,
+          Object.entries(variant.options ?? {}).forEach(([key, value]) => {
-                  product_id: upsertedProduct.id,
+            const productOption = product.options?.find(
-                  rank,
+              (option) => option.title === key
-                })),
+            )!
-                {},
+            const productOptionValue = productOption.values?.find(
-                sharedContext
+              (optionValue) => (optionValue as any).value === value
-              )
+            )!
-            upsertedProduct.images = productImages
+            ;(productOptionValue as any).variants ??= []
-          } else {
+            ;(productOptionValue as any).variants.push(variant)
-            await this.productImageService_.delete(
+          })
-              { product_id: upsertedProduct.id },
+
-              sharedContext
+          delete variant.options
+
+          return variant
+        })
+
+        product.variants = normalizedVariants
+      }
+
+      if ((product as any).tags?.length) {
+        ;(product as any).tags = (product as any).tags.map(
+          (tag: { id: string }) => {
+            const existingTag = existingTagsMap.get(tag.id)
+            if (existingTag) {
+              return existingTag
+            }
+
+            throw new MedusaError(
+              MedusaError.Types.INVALID_DATA,
+              `Tag with id ${tag.id} not found. Please create the tag before associating it with the product.`
             )
           }
-        }
+        )
-      })
+      }
+
+      return product
+    })
+
+    const createdProducts = await this.productService_.create(
+      productsToCreate,
+      sharedContext
     )
 
-    return productData
+    return createdProducts
   }
 
   @InjectTransactionManager()
@@ -1916,6 +1924,17 @@ export default class ProductModuleService
       productData.thumbnail = productData.images[0].url
     }
 
+    if (productData.images?.length) {
+      productData.images = productData.images.map((image, index) =>
+        (image as { rank?: number }).rank != null
+          ? image
+          : {
+              ...image,
+              rank: index,
+            }
+      )
+    }
+
     return productData
   }
 
@@ -1929,6 +1948,7 @@ export default class ProductModuleService
     }
 
     if (productData.options?.length) {
+      // TODO: Instead of fetching per product, this should fetch for all product allowing for only one query instead of X
       const dbOptions = await this.productOptionService_.list(
         { product_id: productData.id },
         { relations: ["values"] },