feat(workflows): Data aggregation (#4732)

* apply the aggregator automatically

* add comment

* apply aggregate

* improve pipe aggregation

* improve test cases

* improvements

* clean tests

* renameing to merge

* fix merge apply

* move merge apply

* cleanup cart workflow and end point

* fixes and naming

packages/medusa/src/api/routes/store/carts/create-cart.ts

@@ -1,7 +1,7 @@
 import { MedusaContainer } from "@medusajs/modules-sdk"
 import {
-  Workflows,
   createCart as createCartWorkflow,
+  Workflows,
 } from "@medusajs/workflows"
 import { Type } from "class-transformer"
 import {
@@ -12,15 +12,13 @@ import {
   IsString,
   ValidateNested,
 } from "class-validator"
-import { MedusaError, isDefined } from "medusa-core-utils"
+import { isDefined, MedusaError } from "medusa-core-utils"
 import reqIp from "request-ip"
 import { EntityManager } from "typeorm"
-
-import { Logger } from "@medusajs/types"
 import { FlagRouter } from "@medusajs/utils"
 import { defaultStoreCartFields, defaultStoreCartRelations } from "."
 import SalesChannelFeatureFlag from "../../../../loaders/feature-flags/sales-channels"
-import { Cart, LineItem } from "../../../../models"
+import { LineItem } from "../../../../models"
 import {
   CartService,
   LineItemService,
@@ -83,8 +81,9 @@ import { FeatureFlagDecorators } from "../../../../utils/feature-flag-decorators
 export default async (req, res) => {
   const entityManager: EntityManager = req.scope.resolve("manager")
   const featureFlagRouter: FlagRouter = req.scope.resolve("featureFlagRouter")
+  const cartService: CartService = req.scope.resolve("cartService")
+
   const validated = req.validatedBody as StorePostCartReq
-  const logger: Logger = req.scope.resolve("logger")
 
   const reqContext = {
     ip: reqIp.getClientIp(req),
@@ -95,6 +94,8 @@ export default async (req, res) => {
     workflows: Workflows.CreateCart,
   })
 
+  let cart
+
   if (isWorkflowEnabled) {
     const cartWorkflow = createCartWorkflow(req.scope as MedusaContainer)
     const input = {
@@ -104,12 +105,6 @@ export default async (req, res) => {
         ...reqContext,
         ...validated.context,
       },
-      config: {
-        retrieveConfig: {
-          select: defaultStoreCartFields,
-          relations: defaultStoreCartRelations,
-        },
-      },
     }
     const { result, errors } = await cartWorkflow.run({
       input,
@@ -125,11 +120,10 @@ export default async (req, res) => {
       }
     }
 
-    return res.status(200).json({ cart: cleanResponseData(result, []) })
-  }
-
-  const lineItemService: LineItemService = req.scope.resolve("lineItemService")
-  const cartService: CartService = req.scope.resolve("cartService")
+    cart = result
+  } else {
+    const lineItemService: LineItemService =
+      req.scope.resolve("lineItemService")
     const regionService: RegionService = req.scope.resolve("regionService")
 
     let regionId!: string
@@ -181,15 +175,15 @@ export default async (req, res) => {
         )
       }
 
-    toCreate.sales_channel_id = req.publishableApiKeyScopes.sales_channel_ids[0]
+      toCreate.sales_channel_id =
+        req.publishableApiKeyScopes.sales_channel_ids[0]
     }
 
-  let cart: Cart
-  await entityManager.transaction(async (manager) => {
+    cart = await entityManager.transaction(async (manager) => {
       const cartServiceTx = cartService.withTransaction(manager)
       const lineItemServiceTx = lineItemService.withTransaction(manager)
 
-    cart = await cartServiceTx.create(toCreate)
+      const createdCart = await cartServiceTx.create(toCreate)
 
       if (validated.items?.length) {
         const generateInputData = validated.items.map((item) => {
@@ -206,12 +200,19 @@ export default async (req, res) => {
           }
         )
 
-      await cartServiceTx.addOrUpdateLineItems(cart.id, generatedLineItems, {
+        await cartServiceTx.addOrUpdateLineItems(
+          createdCart.id,
+          generatedLineItems,
+          {
             validateSalesChannels:
               featureFlagRouter.isFeatureEnabled("sales_channels"),
+          }
+        )
+      }
+
+      return createdCart
     })
   }
-  })
 
   cart = await cartService.retrieveWithTotals(cart!.id, {
     select: defaultStoreCartFields,

packages/types/src/workflow/cart/create-cart.ts

@@ -1,5 +1,4 @@
 import { AddressDTO } from "../../address"
-import { WorkflowInputConfig } from "../common"
 
 export interface CreateLineItemInputDTO {
   variant_id: string
@@ -7,7 +6,6 @@ export interface CreateLineItemInputDTO {
 }
 
 export interface CreateCartWorkflowInputDTO {
-  config?: WorkflowInputConfig
   region_id?: string
   country_code?: string
   items?: CreateLineItemInputDTO[]

packages/workflows/src/definition/cart/create-cart.ts

@@ -13,10 +13,9 @@ import {
   RegionHandlers,
   SalesChannelHandlers,
 } from "../../handlers"
-import { aggregateData, exportWorkflow, pipe } from "../../helper"
+import { exportWorkflow, pipe } from "../../helper"
 
 enum CreateCartActions {
-  setConfig = "setConfig",
   setContext = "setContext",
   attachLineItems = "attachLineItems",
   findRegion = "findRegion",
@@ -26,7 +25,6 @@ enum CreateCartActions {
   findOrCreateCustomer = "findOrCreateCustomer",
   removeCart = "removeCart",
   removeAddresses = "removeAddresses",
-  retrieveCart = "retrieveCart",
 }
 
 const workflowAlias = "cart"
@@ -40,10 +38,6 @@ const getWorkflowInput = (alias = workflowAlias) => ({
 
 const workflowSteps: TransactionStepsDefinition = {
   next: [
-    {
-      action: CreateCartActions.setConfig,
-      noCompensation: true,
-    },
     {
       action: CreateCartActions.findOrCreateCustomer,
       noCompensation: true,
@@ -67,10 +61,6 @@ const workflowSteps: TransactionStepsDefinition = {
           next: {
             action: CreateCartActions.attachLineItems,
             noCompensation: true,
-            next: {
-              action: CreateCartActions.retrieveCart,
-              noCompensation: true,
-            },
           },
         },
       },
@@ -79,16 +69,6 @@ const workflowSteps: TransactionStepsDefinition = {
 }
 
 const handlers = new Map([
-  [
-    CreateCartActions.setConfig,
-    {
-      invoke: pipe(
-        getWorkflowInput(CommonHandlers.setConfig.aliases.Config),
-        aggregateData(),
-        CommonHandlers.setConfig
-      ),
-    },
-  ],
   [
     CreateCartActions.findOrCreateCustomer,
     {
@@ -206,26 +186,6 @@ const handlers = new Map([
       ),
     },
   ],
-  [
-    CreateCartActions.retrieveCart,
-    {
-      invoke: pipe(
-        {
-          invoke: [
-            {
-              from: CreateCartActions.setConfig,
-              alias: CommonHandlers.setConfig.aliases.Config,
-            },
-            {
-              from: CreateCartActions.createCart,
-              alias: CartHandlers.retrieveCart.aliases.Cart,
-            },
-          ],
-        },
-        CartHandlers.retrieveCart
-      ),
-    },
-  ],
 ])
 
 WorkflowManager.register(Workflows.CreateCart, workflowSteps, handlers)
@@ -235,4 +195,4 @@ type CreateCartWorkflowOutput = Record<any, any>
 export const createCart = exportWorkflow<
   CartWorkflow.CreateCartWorkflowInputDTO,
   CreateCartWorkflowOutput
->(Workflows.CreateCart, CreateCartActions.retrieveCart)
+>(Workflows.CreateCart, CreateCartActions.createCart)

packages/workflows/src/definition/product/create-products.ts

@@ -3,7 +3,7 @@ import {
   TransactionStepsDefinition,
   WorkflowManager,
 } from "@medusajs/orchestration"
-import { aggregateData, exportWorkflow, pipe } from "../../helper"
+import { exportWorkflow, pipe } from "../../helper"
 
 import { ProductTypes, WorkflowTypes } from "@medusajs/types"
 import {
@@ -57,12 +57,12 @@ const handlers = new Map([
     {
       invoke: pipe(
         {
+          merge: true,
           inputAlias: InputAlias.ProductsInputData,
           invoke: {
             from: InputAlias.ProductsInputData,
           },
         },
-        aggregateData(),
         ProductHandlers.createProductsPrepareData
       ),
     },
@@ -72,21 +72,21 @@ const handlers = new Map([
     {
       invoke: pipe(
         {
+          merge: true,
           invoke: {
             from: CreateProductsActions.prepare,
           },
         },
-        aggregateData(),
         ProductHandlers.createProducts
       ),
       compensate: pipe(
         {
+          merge: true,
           invoke: {
             from: CreateProductsActions.createProducts,
             alias: ProductHandlers.removeProducts.aliases.products,
           },
         },
-        aggregateData(),
         ProductHandlers.removeProducts
       ),
     },
@@ -96,6 +96,7 @@ const handlers = new Map([
     {
       invoke: pipe(
         {
+          merge: true,
           invoke: [
             {
               from: CreateProductsActions.prepare,
@@ -108,11 +109,11 @@ const handlers = new Map([
             },
           ],
         },
-        aggregateData(),
         ProductHandlers.attachShippingProfileToProducts
       ),
       compensate: pipe(
         {
+          merge: true,
           invoke: [
             {
               from: CreateProductsActions.prepare,
@@ -125,7 +126,6 @@ const handlers = new Map([
             },
           ],
         },
-        aggregateData(),
         ProductHandlers.detachShippingProfileFromProducts
       ),
     },
@@ -135,6 +135,7 @@ const handlers = new Map([
     {
       invoke: pipe(
         {
+          merge: true,
           invoke: [
             {
               from: CreateProductsActions.prepare,
@@ -146,11 +147,11 @@ const handlers = new Map([
             },
           ],
         },
-        aggregateData(),
         ProductHandlers.attachSalesChannelToProducts
       ),
       compensate: pipe(
         {
+          merge: true,
           invoke: [
             {
               from: CreateProductsActions.prepare,
@@ -162,7 +163,6 @@ const handlers = new Map([
             },
           ],
         },
-        aggregateData(),
         ProductHandlers.detachSalesChannelFromProducts
       ),
     },
@@ -172,23 +172,23 @@ const handlers = new Map([
     {
       invoke: pipe(
         {
+          merge: true,
           invoke: {
             from: CreateProductsActions.createProducts,
             alias: InventoryHandlers.createInventoryItems.aliases.products,
           },
         },
-        aggregateData(),
         InventoryHandlers.createInventoryItems
       ),
       compensate: pipe(
         {
+          merge: true,
           invoke: {
             from: CreateProductsActions.createInventoryItems,
             alias:
               InventoryHandlers.removeInventoryItems.aliases.inventoryItems,
           },
         },
-        aggregateData(),
         InventoryHandlers.removeInventoryItems
       ),
     },
@@ -198,24 +198,24 @@ const handlers = new Map([
     {
       invoke: pipe(
         {
+          merge: true,
           invoke: {
             from: CreateProductsActions.createInventoryItems,
             alias:
               InventoryHandlers.attachInventoryItems.aliases.inventoryItems,
           },
         },
-        aggregateData(),
         InventoryHandlers.attachInventoryItems
       ),
       compensate: pipe(
         {
+          merge: true,
           invoke: {
             from: CreateProductsActions.createInventoryItems,
             alias:
               InventoryHandlers.detachInventoryItems.aliases.inventoryItems,
           },
         },
-        aggregateData(),
         InventoryHandlers.detachInventoryItems
       ),
     },
@@ -225,6 +225,7 @@ const handlers = new Map([
     {
       invoke: pipe(
         {
+          merge: true,
           invoke: [
             {
               from: CreateProductsActions.prepare,
@@ -236,14 +237,18 @@ const handlers = new Map([
             },
           ],
         },
-        aggregateData(),
         ProductHandlers.updateProductsVariantsPrices
       ),
       compensate: pipe(
         {
+          merge: true,
           invoke: [
             {
               from: CreateProductsActions.prepare,
+              alias:
+                MiddlewaresHandlers
+                  .createProductsPrepareCreatePricesCompensation.aliases
+                  .preparedData,
             },
             {
               from: CreateProductsActions.createProducts,
@@ -252,7 +257,6 @@ const handlers = new Map([
             },
           ],
         },
-        aggregateData(),
         MiddlewaresHandlers.createProductsPrepareCreatePricesCompensation,
         ProductHandlers.updateProductsVariantsPrices
       ),

packages/workflows/src/handlers/common/index.ts

@@ -1,2 +1 @@
-export * from "./set-config"
 export * from "./set-context"

packages/workflows/src/handlers/common/set-config.ts

@@ -1,31 +0,0 @@
-import { WorkflowArguments } from "../../helper"
-
-type ConfigDTO = {
-  retrieveConfig?: {
-    select?: string[]
-    relations?: string[]
-  }
-}
-
-enum Aliases {
-  Config = "config",
-}
-
-type HandlerInputData = {
-  config: {
-    retrieveConfig?: {
-      select?: string[]
-      relations?: string[]
-    }
-  }
-}
-
-export async function setConfig({
-  data,
-}: WorkflowArguments<HandlerInputData>): Promise<ConfigDTO> {
-  return {
-    retrieveConfig: data[Aliases.Config].retrieveConfig,
-  }
-}
-
-setConfig.aliases = Aliases

packages/workflows/src/handlers/middlewares/create-products-prepare-create-prices-compensation.ts

@@ -10,33 +10,37 @@ type VariantIndexAndPrices = {
 export async function createProductsPrepareCreatePricesCompensation({
   data,
 }: WorkflowArguments<{
+  preparedData: {
     productsHandleVariantsIndexPricesMap: Map<
       ProductHandle,
       VariantIndexAndPrices[]
     >
+  }
   products: ProductTypes.ProductDTO[]
 }>) {
   const productsHandleVariantsIndexPricesMap =
-    data.productsHandleVariantsIndexPricesMap
+    data.preparedData.productsHandleVariantsIndexPricesMap
   const products = data.products
 
   const updatedProductsHandleVariantsIndexPricesMap = new Map()
-  productsHandleVariantsIndexPricesMap.forEach((items, productHandle) => {
-    const existingItems =
+  productsHandleVariantsIndexPricesMap.forEach(
+    (existingItems, productHandle) => {
+      const items =
         updatedProductsHandleVariantsIndexPricesMap.get(productHandle) ?? []
 
-    items.forEach(({ index }) => {
-      existingItems.push({
+      existingItems.forEach(({ index }) => {
+        items.push({
           index,
           prices: [],
         })
       })
 
       updatedProductsHandleVariantsIndexPricesMap.set(productHandle, items)
-  })
+    }
+  )
 
   return {
-    alias: "",
+    alias: createProductsPrepareCreatePricesCompensation.aliases.output,
     value: {
       productsHandleVariantsIndexPricesMap:
         updatedProductsHandleVariantsIndexPricesMap,
@@ -46,5 +50,6 @@ export async function createProductsPrepareCreatePricesCompensation({
 }
 
 createProductsPrepareCreatePricesCompensation.aliases = {
-  payload: "payload",
+  preparedData: "preparedData",
+  output: "createProductsPrepareCreatePricesCompensationOutput",
 }

packages/workflows/src/helper/__tests__/merge-data.spec.ts

@@ -1,8 +1,8 @@
-import { aggregateData } from "../aggregate"
+import { mergeData } from "../merge-data"
 import { WorkflowStepMiddlewareReturn } from "../pipe"
 
-describe("aggregate", function () {
-  it("should aggregate a new object from the source into a specify target", async function () {
+describe("merge", function () {
+  it("should merge a new object from the source into a specify target", async function () {
     const source = {
       stringProp: "stringProp",
       anArray: ["anArray"],
@@ -14,13 +14,14 @@ describe("aggregate", function () {
       },
     }
 
-    const { value: result } = (await aggregateData(
+    const result = (await mergeData(
       ["input", "another", "stringProp", "anArray"],
       "payload"
     )({ data: source } as any)) as unknown as WorkflowStepMiddlewareReturn
 
     expect(result).toEqual({
-      payload: {
+      alias: "payload",
+      value: {
         ...source.input,
         ...source.another,
         anArray: source.anArray,
@@ -29,7 +30,7 @@ describe("aggregate", function () {
     })
   })
 
-  it("should aggregate a new object from the entire source into the resul object", async function () {
+  it("should merge a new object from the entire source into the resul object", async function () {
     const source = {
       stringProp: "stringProp",
       anArray: ["anArray"],
@@ -41,7 +42,7 @@ describe("aggregate", function () {
       },
     }
 
-    const { value: result } = (await aggregateData()({
+    const { value: result } = (await mergeData()({
       data: source,
     } as any)) as unknown as WorkflowStepMiddlewareReturn
 

packages/workflows/src/helper/__tests__/pipe.spec.ts

@@ -46,6 +46,152 @@ describe("Pipe", function () {
     expect(result).toEqual(output)
   })
 
+  it("should evaluate the input object and append the values to the data object using the merge and return the result from the handler", async function () {
+    const payload = { input: "input" }
+    const output = { test: "test" }
+    const invoke = {
+      input: payload,
+      step1: { step1Data: { test: "test" } },
+      step2: [{ test: "test" }],
+    }
+
+    const handler = jest.fn().mockImplementation(async () => output)
+    const input = {
+      inputAlias: "payload",
+      merge: true,
+      invoke: [
+        {
+          from: "payload",
+        },
+        {
+          from: "step1",
+        },
+        {
+          from: "step2",
+          alias: "step2Data",
+        },
+      ],
+    }
+
+    const result = await pipe(input, handler)({ invoke, payload } as any)
+
+    expect(handler).toHaveBeenCalled()
+    expect(handler).toHaveBeenCalledWith(
+      expect.objectContaining({
+        data: {
+          ...payload,
+          ...invoke.step1,
+          step2Data: invoke.step2,
+        },
+      })
+    )
+
+    expect(result).toBeDefined()
+    expect(result).toEqual(output)
+  })
+
+  it("should evaluate the input object and append the values to the data object using the merge to store on the merge alias and return the result from the handler", async function () {
+    const payload = { input: "input" }
+    const output = { test: "test" }
+    const invoke = {
+      input: payload,
+      step1: { step1Data: { test: "test" } },
+      step2: { step2Data: { test: "test" } },
+    }
+
+    const handler = jest.fn().mockImplementation(async () => output)
+    const input = {
+      inputAlias: "payload",
+      mergeAlias: "mergedData",
+      invoke: [
+        {
+          from: "payload",
+          alias: "input",
+        },
+        {
+          from: "step1",
+          alias: "step1Data",
+        },
+        {
+          from: "step2",
+          alias: "step2Data",
+        },
+      ],
+    }
+
+    const result = await pipe(input, handler)({ invoke, payload } as any)
+
+    expect(handler).toHaveBeenCalled()
+    expect(handler).toHaveBeenCalledWith(
+      expect.objectContaining({
+        data: {
+          input: payload,
+          step1Data: invoke.step1,
+          step2Data: invoke.step2,
+          mergedData: {
+            ...payload,
+            ...invoke.step1,
+            ...invoke.step2,
+          },
+        },
+      })
+    )
+
+    expect(result).toBeDefined()
+    expect(result).toEqual(output)
+  })
+
+  it("should evaluate the input object and append the values to the data object using the merge to store on the merge alias from the merge from values and return the result from the handler", async function () {
+    const payload = { input: "input" }
+    const output = { test: "test" }
+    const invoke = {
+      input: payload,
+      step1: { step1Data: { test: "test" } },
+      step2: { step2Data: { test: "test" } },
+    }
+
+    const handler = jest.fn().mockImplementation(async () => output)
+    const input = {
+      inputAlias: "payload",
+      mergeAlias: "mergedData",
+      mergeFrom: ["input", "step1Data"],
+      invoke: [
+        {
+          from: "payload",
+          alias: "input",
+        },
+        {
+          from: "step1",
+          alias: "step1Data",
+        },
+        {
+          from: "step2",
+          alias: "step2Data",
+        },
+      ],
+    }
+
+    const result = await pipe(input, handler)({ invoke, payload } as any)
+
+    expect(handler).toHaveBeenCalled()
+    expect(handler).toHaveBeenCalledWith(
+      expect.objectContaining({
+        data: {
+          input: payload,
+          step1Data: invoke.step1,
+          step2Data: invoke.step2,
+          mergedData: {
+            ...payload,
+            ...invoke.step1,
+          },
+        },
+      })
+    )
+
+    expect(result).toBeDefined()
+    expect(result).toEqual(output)
+  })
+
   it("should execute onComplete function if available but the output result shouldn't change", async function () {
     const payload = { input: "input" }
     const output = { test: "test" }

packages/workflows/src/helper/index.ts

@@ -1,4 +1,4 @@
-export * from "./aggregate"
+export * from "./merge-data"
 export * from "./empty-handler"
 export * from "./pipe"
 export * from "./workflow-export"

packages/workflows/src/helper/merge-data.ts

@@ -2,12 +2,12 @@ import { PipelineHandler, WorkflowArguments } from "./pipe"
 import { isObject } from "@medusajs/utils"
 
 /**
- * Pipe utils that aggregates data from an object into a new object.
- * The new object will have a target key with the aggregated data from the keys.
+ * Pipe utils that merges data from an object into a new object.
+ * The new object will have a target key with the merged data from the keys if specified.
  * @param keys
  * @param target
  */
-export function aggregateData<
+export function mergeData<
   T extends Record<string, unknown> = Record<string, unknown>,
   TKeys extends keyof T = keyof T,
   Target extends "payload" | string = string
@@ -43,7 +43,7 @@ export function aggregateData<
 
     return {
       alias: target,
-      value,
+      value: target ? value[target as string] : value,
     }
   }
 }

packages/workflows/src/helper/pipe.ts

@@ -1,11 +1,11 @@
 import {
+  DistributedTransaction,
   TransactionMetadata,
   WorkflowStepHandler,
 } from "@medusajs/orchestration"
 import { Context, MedusaContainer, SharedContext } from "@medusajs/types"
-
-import { DistributedTransaction } from "@medusajs/orchestration"
 import { InputAlias } from "../definitions"
+import { mergeData } from "./merge-data"
 
 export type WorkflowStepMiddlewareReturn = {
   alias?: string
@@ -18,10 +18,28 @@ export type WorkflowStepMiddlewareInput = {
 }
 
 interface PipelineInput {
+  /**
+   * The alias of the input data to store in
+   */
   inputAlias?: InputAlias | string
+  /**
+   * Descriptors to get the data from
+   */
   invoke?: WorkflowStepMiddlewareInput | WorkflowStepMiddlewareInput[]
   compensate?: WorkflowStepMiddlewareInput | WorkflowStepMiddlewareInput[]
-  onComplete?: (args: WorkflowOnCompleteArguments) => {}
+  onComplete?: (args: WorkflowOnCompleteArguments) => Promise<void>
+  /**
+   * Apply the data merging
+   */
+  merge?: boolean
+  /**
+   * Store the merged data in a new key, if this is present no need to set merge: true
+   */
+  mergeAlias?: string
+  /**
+   * Store the merged data from the chosen aliases, if this is present no need to set merge: true
+   */
+  mergeFrom?: string[]
 }
 
 export type WorkflowArguments<T = any> = {
@@ -53,6 +71,15 @@ export function pipe<T>(
   input: PipelineInput,
   ...functions: [...PipelineHandler[], PipelineHandler<T>]
 ): WorkflowStepHandler {
+  // Apply the aggregator just before the last handler
+  if (
+    (input.merge || input.mergeAlias || input.mergeFrom) &&
+    functions.length
+  ) {
+    const handler = functions.pop()!
+    functions.push(mergeData(input.mergeFrom, input.mergeAlias), handler)
+  }
+
   return async ({
     container,
     payload,