chore: Refactor batch product workflows and add tests (#7540)

integration-tests/modules/__tests__/product/workflows/batch-products.spec.ts

@@ -0,0 +1,156 @@
+import {
+  batchProductsWorkflow,
+  batchProductsWorkflowId,
+  batchProductVariantsWorkflow,
+  batchProductVariantsWorkflowId,
+} from "@medusajs/core-flows"
+import { ModuleRegistrationName } from "@medusajs/modules-sdk"
+import { IProductModuleService } from "@medusajs/types"
+import { medusaIntegrationTestRunner } from "medusa-test-utils/dist"
+
+jest.setTimeout(50000)
+
+medusaIntegrationTestRunner({
+  env: {},
+  testSuite: ({ getContainer }) => {
+    describe("Workflows: Batch Product", () => {
+      let appContainer
+      let service: IProductModuleService
+
+      beforeAll(async () => {
+        appContainer = getContainer()
+        service = appContainer.resolve(ModuleRegistrationName.PRODUCT)
+      })
+
+      describe("batchProductWorkflow", () => {
+        describe("compensation", () => {
+          it("should cancel created and deleted products if step throws error", async () => {
+            const workflow = batchProductsWorkflow(appContainer)
+
+            workflow.appendAction("throw", batchProductsWorkflowId, {
+              invoke: async function failStep() {
+                throw new Error(
+                  `Failed the update product workflow after creating product`
+                )
+              },
+            })
+
+            const product1 = await service.create({ title: "test1" })
+            const product2 = await service.create({ title: "test2" })
+
+            const { errors } = await workflow.run({
+              input: {
+                create: [{ title: "test3" }],
+                update: [{ id: product1.id, title: "test1-updated" }],
+                delete: [product2.id],
+              },
+              throwOnError: false,
+            })
+
+            expect(errors).toEqual([
+              {
+                action: "throw",
+                handlerType: "invoke",
+                error: expect.objectContaining({
+                  message: `Failed the update product workflow after creating product`,
+                }),
+              },
+            ])
+
+            const products = await service.list()
+
+            expect(products).toHaveLength(2)
+            expect(products).toEqual([
+              expect.objectContaining({
+                title: "test1",
+              }),
+              expect.objectContaining({
+                title: "test2",
+              }),
+            ])
+          })
+        })
+      })
+
+      describe("batchVariantWorkflow", () => {
+        describe("compensation", () => {
+          it("should cancel created and deleted variant if step throws error", async () => {
+            const workflow = batchProductVariantsWorkflow(appContainer)
+
+            workflow.appendAction("throw", batchProductVariantsWorkflowId, {
+              invoke: async function failStep() {
+                throw new Error(`Failed the update product variant workflow`)
+              },
+            })
+
+            const res = await batchProductsWorkflow(appContainer).run({
+              input: {
+                create: [
+                  {
+                    title: "test1",
+                    variants: [
+                      {
+                        title: "variant1",
+                        prices: [{ amount: 100, currency_code: "EUR" }],
+                      },
+                      {
+                        title: "variant2",
+                        prices: [{ amount: 100, currency_code: "EUR" }],
+                      },
+                    ],
+                  },
+                ],
+              },
+            })
+            const product1 = res.result.created[0]
+
+            const { errors } = await workflow.run({
+              input: {
+                create: [
+                  {
+                    title: "variant3",
+                    product_id: product1.id,
+                    prices: [{ amount: 100, currency_code: "EUR" }],
+                  },
+                ],
+                update: [
+                  {
+                    id: product1.variants[0].id,
+                    product_id: product1.id,
+                    title: "variant1-updated",
+                  },
+                ],
+                delete: [product1.variants[1].id],
+              },
+              throwOnError: false,
+            })
+
+            expect(errors).toEqual([
+              {
+                action: "throw",
+                handlerType: "invoke",
+                error: expect.objectContaining({
+                  message: `Failed the update product variant workflow`,
+                }),
+              },
+            ])
+
+            const product = await service.retrieve(product1.id, {
+              relations: ["variants"],
+            })
+
+            expect(product.variants).toHaveLength(2)
+            expect(product.variants).toEqual([
+              expect.objectContaining({
+                title: "variant1",
+              }),
+              expect.objectContaining({
+                title: "variant2",
+              }),
+            ])
+          })
+        })
+      })
+    })
+  },
+})

packages/core/core-flows/src/product/steps/batch-product-variants.ts

@@ -1,94 +0,0 @@
-import { StepResponse, createStep } from "@medusajs/workflows-sdk"
-import { deleteProductVariantsWorkflow } from "../workflows/delete-product-variants"
-import { createProductVariantsWorkflow } from "../workflows/create-product-variants"
-import { updateProductVariantsWorkflow } from "../workflows/update-product-variants"
-import {
-  BatchWorkflowInput,
-  CreateProductVariantWorkflowInputDTO,
-  UpdateProductVariantWorkflowInputDTO,
-} from "@medusajs/types"
-
-export const batchProductVariantsStepId = "batch-product-variants"
-export const batchProductVariantsStep = createStep(
-  batchProductVariantsStepId,
-  async (
-    data: BatchWorkflowInput<
-      CreateProductVariantWorkflowInputDTO,
-      UpdateProductVariantWorkflowInputDTO
-    >,
-    { container }
-  ) => {
-    const {
-      transaction: createTransaction,
-      result: created,
-      errors: createErrors,
-    } = await createProductVariantsWorkflow(container).run({
-      input: { product_variants: data.create ?? [] },
-      throwOnError: false,
-    })
-
-    if (createErrors?.length) {
-      throw createErrors[0].error
-    }
-
-    const {
-      transaction: updateTransaction,
-      result: updated,
-      errors: updateErrors,
-    } = await updateProductVariantsWorkflow(container).run({
-      input: { product_variants: data.update ?? [] },
-      throwOnError: false,
-    })
-
-    if (updateErrors?.length) {
-      throw updateErrors[0].error
-    }
-
-    const { transaction: deleteTransaction, errors: deleteErrors } =
-      await deleteProductVariantsWorkflow(container).run({
-        input: { ids: data.delete ?? [] },
-        throwOnError: false,
-      })
-
-    if (deleteErrors?.length) {
-      throw deleteErrors[0].error
-    }
-
-    return new StepResponse(
-      {
-        created,
-        updated,
-        deleted: {
-          ids: data.delete ?? [],
-          object: "product_variant",
-          deleted: true,
-        },
-      },
-      { createTransaction, updateTransaction, deleteTransaction }
-    )
-  },
-
-  async (flow, { container }) => {
-    if (!flow) {
-      return
-    }
-
-    if (flow.createTransaction) {
-      await createProductVariantsWorkflow(container).cancel({
-        transaction: flow.createTransaction,
-      })
-    }
-
-    if (flow.updateTransaction) {
-      await updateProductVariantsWorkflow(container).cancel({
-        transaction: flow.updateTransaction,
-      })
-    }
-
-    if (flow.deleteTransaction) {
-      await deleteProductVariantsWorkflow(container).cancel({
-        transaction: flow.deleteTransaction,
-      })
-    }
-  }
-)

packages/core/core-flows/src/product/steps/batch-products.ts

@@ -1,94 +0,0 @@
-import { StepResponse, createStep } from "@medusajs/workflows-sdk"
-import { createProductsWorkflow } from "../workflows/create-products"
-import { updateProductsWorkflow } from "../workflows/update-products"
-import { deleteProductsWorkflow } from "../workflows/delete-products"
-import {
-  BatchWorkflowInput,
-  CreateProductWorkflowInputDTO,
-  UpdateProductWorkflowInputDTO,
-} from "@medusajs/types"
-
-export const batchProductsStepId = "batch-products"
-export const batchProductsStep = createStep(
-  batchProductsStepId,
-  async (
-    data: BatchWorkflowInput<
-      CreateProductWorkflowInputDTO,
-      UpdateProductWorkflowInputDTO
-    >,
-    { container }
-  ) => {
-    const {
-      transaction: createTransaction,
-      result: created,
-      errors: createErrors,
-    } = await createProductsWorkflow(container).run({
-      input: { products: data.create ?? [] },
-      throwOnError: false,
-    })
-
-    if (createErrors?.length) {
-      throw createErrors[0].error
-    }
-
-    const {
-      transaction: updateTransaction,
-      result: updated,
-      errors: updateErrors,
-    } = await updateProductsWorkflow(container).run({
-      input: { products: data.update ?? [] },
-      throwOnError: false,
-    })
-
-    if (updateErrors?.length) {
-      throw updateErrors[0].error
-    }
-
-    const { transaction: deleteTransaction, errors: deleteErrors } =
-      await deleteProductsWorkflow(container).run({
-        input: { ids: data.delete ?? [] },
-        throwOnError: false,
-      })
-
-    if (deleteErrors?.length) {
-      throw deleteErrors[0].error
-    }
-
-    return new StepResponse(
-      {
-        created,
-        updated,
-        deleted: {
-          ids: data.delete ?? [],
-          object: "product",
-          deleted: true,
-        },
-      },
-      { createTransaction, updateTransaction, deleteTransaction }
-    )
-  },
-
-  async (flow, { container }) => {
-    if (!flow) {
-      return
-    }
-
-    if (flow.createTransaction) {
-      await createProductsWorkflow(container).cancel({
-        transaction: flow.createTransaction,
-      })
-    }
-
-    if (flow.updateTransaction) {
-      await updateProductsWorkflow(container).cancel({
-        transaction: flow.updateTransaction,
-      })
-    }
-
-    if (flow.deleteTransaction) {
-      await deleteProductsWorkflow(container).cancel({
-        transaction: flow.deleteTransaction,
-      })
-    }
-  }
-)

packages/core/core-flows/src/product/steps/index.ts

@@ -2,7 +2,6 @@ export * from "./create-products"
 export * from "./update-products"
 export * from "./delete-products"
 export * from "./get-products"
-export * from "./batch-products"
 export * from "./create-variant-pricing-link"
 export * from "./create-product-options"
 export * from "./update-product-options"
@@ -10,7 +9,6 @@ export * from "./delete-product-options"
 export * from "./create-product-variants"
 export * from "./update-product-variants"
 export * from "./delete-product-variants"
-export * from "./batch-product-variants"
 export * from "./create-collections"
 export * from "./update-collections"
 export * from "./delete-collections"

packages/core/core-flows/src/product/workflows/batch-product-variants.ts

@@ -1,5 +1,9 @@
-import { WorkflowData, createWorkflow } from "@medusajs/workflows-sdk"
+import {
-import { batchProductVariantsStep } from "../steps/batch-product-variants"
+  WorkflowData,
+  createWorkflow,
+  parallelize,
+  transform,
+} from "@medusajs/workflows-sdk"
 import {
   BatchWorkflowInput,
   BatchWorkflowOutput,
@@ -7,6 +11,9 @@ import {
   UpdateProductVariantWorkflowInputDTO,
   CreateProductVariantWorkflowInputDTO,
 } from "@medusajs/types"
+import { createProductVariantsWorkflow } from "./create-product-variants"
+import { updateProductVariantsWorkflow } from "./update-product-variants"
+import { deleteProductVariantsWorkflow } from "./delete-product-variants"
 
 export const batchProductVariantsWorkflowId = "batch-product-variants"
 export const batchProductVariantsWorkflow = createWorkflow(
@@ -19,6 +26,28 @@ export const batchProductVariantsWorkflow = createWorkflow(
       >
     >
   ): WorkflowData<BatchWorkflowOutput<ProductTypes.ProductVariantDTO>> => {
-    return batchProductVariantsStep(input)
+    const res = parallelize(
+      createProductVariantsWorkflow.runAsStep({
+        input: { product_variants: input.create ?? [] },
+      }),
+      updateProductVariantsWorkflow.runAsStep({
+        input: { product_variants: input.update ?? [] },
+      }),
+      deleteProductVariantsWorkflow.runAsStep({
+        input: { ids: input.delete ?? [] },
+      })
+    )
+
+    return transform({ res, input }, (data) => {
+      return {
+        created: data.res[0],
+        updated: data.res[1],
+        deleted: {
+          ids: data.input.delete ?? [],
+          object: "product_variant",
+          deleted: true,
+        },
+      }
+    })
   }
 )

packages/core/core-flows/src/product/workflows/batch-products.ts

@@ -1,5 +1,9 @@
-import { WorkflowData, createWorkflow } from "@medusajs/workflows-sdk"
+import {
-import { batchProductsStep } from "../steps/batch-products"
+  WorkflowData,
+  createWorkflow,
+  parallelize,
+  transform,
+} from "@medusajs/workflows-sdk"
 import {
   ProductTypes,
   BatchWorkflowInput,
@@ -7,6 +11,9 @@ import {
   UpdateProductWorkflowInputDTO,
 } from "@medusajs/types"
 import { BatchWorkflowOutput } from "@medusajs/types/src"
+import { createProductsWorkflow } from "./create-products"
+import { updateProductsWorkflow } from "./update-products"
+import { deleteProductsWorkflow } from "./delete-products"
 
 export const batchProductsWorkflowId = "batch-products"
 export const batchProductsWorkflow = createWorkflow(
@@ -19,6 +26,28 @@ export const batchProductsWorkflow = createWorkflow(
       >
     >
   ): WorkflowData<BatchWorkflowOutput<ProductTypes.ProductDTO>> => {
-    return batchProductsStep(input)
+    const res = parallelize(
+      createProductsWorkflow.runAsStep({
+        input: { products: input.create ?? [] },
+      }),
+      updateProductsWorkflow.runAsStep({
+        input: { products: input.update ?? [] },
+      }),
+      deleteProductsWorkflow.runAsStep({
+        input: { ids: input.delete ?? [] },
+      })
+    )
+
+    return transform({ res, input }, (data) => {
+      return {
+        created: data.res[0],
+        updated: data.res[1],
+        deleted: {
+          ids: data.input.delete ?? [],
+          object: "product",
+          deleted: true,
+        },
+      }
+    })
   }
 )

packages/core/core-flows/src/product/workflows/update-products.ts

@@ -72,7 +72,7 @@ function updateProductIds({
     return arrayDifference(productIds, discardedProductIds)
   }
 
-  return !input.update.sales_channels ? [] : productIds
+  return !input.update?.sales_channels ? [] : productIds
 }
 
 function prepareSalesChannelLinks({
@@ -97,7 +97,7 @@ function prepareSalesChannelLinks({
       )
   }
 
-  if (input.selector && input.update.sales_channels?.length) {
+  if (input.selector && input.update?.sales_channels?.length) {
     return updatedProducts.flatMap((p) =>
       input.update.sales_channels!.map((channel) => ({
         [Modules.PRODUCT]: {