From a268d2cb0be9a4068883a8504c0d1d419a4382c6 Mon Sep 17 00:00:00 2001 From: Adrien de Peretti Date: Thu, 10 Aug 2023 14:01:56 +0200 Subject: [PATCH] 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 --- .../src/api/routes/store/carts/create-cart.ts | 185 +++++++++--------- .../types/src/workflow/cart/create-cart.ts | 2 - .../src/definition/cart/create-cart.ts | 44 +---- .../src/definition/product/create-products.ts | 32 +-- .../workflows/src/handlers/common/index.ts | 1 - .../src/handlers/common/set-config.ts | 31 --- ...ucts-prepare-create-prices-compensation.ts | 39 ++-- .../{aggregate.spec.ts => merge-data.spec.ts} | 15 +- .../src/helper/__tests__/pipe.spec.ts | 146 ++++++++++++++ packages/workflows/src/helper/index.ts | 2 +- .../helper/{aggregate.ts => merge-data.ts} | 8 +- packages/workflows/src/helper/pipe.ts | 33 +++- 12 files changed, 324 insertions(+), 214 deletions(-) delete mode 100644 packages/workflows/src/handlers/common/set-config.ts rename packages/workflows/src/helper/__tests__/{aggregate.spec.ts => merge-data.spec.ts} (71%) rename packages/workflows/src/helper/{aggregate.ts => merge-data.ts} (83%) diff --git a/packages/medusa/src/api/routes/store/carts/create-cart.ts b/packages/medusa/src/api/routes/store/carts/create-cart.ts index c6a655be41..07d807c768 100644 --- a/packages/medusa/src/api/routes/store/carts/create-cart.ts +++ b/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,93 +120,99 @@ 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") - const regionService: RegionService = req.scope.resolve("regionService") - - let regionId!: string - if (isDefined(validated.region_id)) { - regionId = validated.region_id as string + cart = result } else { - const regions = await regionService.list({}) + const lineItemService: LineItemService = + req.scope.resolve("lineItemService") + const regionService: RegionService = req.scope.resolve("regionService") - if (!regions?.length) { - throw new MedusaError( - MedusaError.Types.INVALID_DATA, - `A region is required to create a cart` - ) + let regionId!: string + if (isDefined(validated.region_id)) { + regionId = validated.region_id as string + } else { + const regions = await regionService.list({}) + + if (!regions?.length) { + throw new MedusaError( + MedusaError.Types.INVALID_DATA, + `A region is required to create a cart` + ) + } + + regionId = regions[0].id } - regionId = regions[0].id - } - - const toCreate: Partial = { - region_id: regionId, - sales_channel_id: validated.sales_channel_id, - context: { - ...reqContext, - ...validated.context, - }, - } - - if (req.user && req.user.customer_id) { - const customerService = req.scope.resolve("customerService") - const customer = await customerService.retrieve(req.user.customer_id) - toCreate["customer_id"] = customer.id - toCreate["email"] = customer.email - } - - if (validated.country_code) { - toCreate["shipping_address"] = { - country_code: validated.country_code.toLowerCase(), - } - } - - if ( - !toCreate.sales_channel_id && - req.publishableApiKeyScopes?.sales_channel_ids.length - ) { - if (req.publishableApiKeyScopes.sales_channel_ids.length > 1) { - throw new MedusaError( - MedusaError.Types.UNEXPECTED_STATE, - "The PublishableApiKey provided in the request header has multiple associated sales channels." - ) + const toCreate: Partial = { + region_id: regionId, + sales_channel_id: validated.sales_channel_id, + context: { + ...reqContext, + ...validated.context, + }, } - toCreate.sales_channel_id = req.publishableApiKeyScopes.sales_channel_ids[0] - } - - let cart: Cart - await entityManager.transaction(async (manager) => { - const cartServiceTx = cartService.withTransaction(manager) - const lineItemServiceTx = lineItemService.withTransaction(manager) - - cart = await cartServiceTx.create(toCreate) - - if (validated.items?.length) { - const generateInputData = validated.items.map((item) => { - return { - variantId: item.variant_id, - quantity: item.quantity, - } - }) - const generatedLineItems: LineItem[] = await lineItemServiceTx.generate( - generateInputData, - { - region_id: regionId, - customer_id: req.user?.customer_id, - } - ) - - await cartServiceTx.addOrUpdateLineItems(cart.id, generatedLineItems, { - validateSalesChannels: - featureFlagRouter.isFeatureEnabled("sales_channels"), - }) + if (req.user && req.user.customer_id) { + const customerService = req.scope.resolve("customerService") + const customer = await customerService.retrieve(req.user.customer_id) + toCreate["customer_id"] = customer.id + toCreate["email"] = customer.email } - }) + + if (validated.country_code) { + toCreate["shipping_address"] = { + country_code: validated.country_code.toLowerCase(), + } + } + + if ( + !toCreate.sales_channel_id && + req.publishableApiKeyScopes?.sales_channel_ids.length + ) { + if (req.publishableApiKeyScopes.sales_channel_ids.length > 1) { + throw new MedusaError( + MedusaError.Types.UNEXPECTED_STATE, + "The PublishableApiKey provided in the request header has multiple associated sales channels." + ) + } + + toCreate.sales_channel_id = + req.publishableApiKeyScopes.sales_channel_ids[0] + } + + cart = await entityManager.transaction(async (manager) => { + const cartServiceTx = cartService.withTransaction(manager) + const lineItemServiceTx = lineItemService.withTransaction(manager) + + const createdCart = await cartServiceTx.create(toCreate) + + if (validated.items?.length) { + const generateInputData = validated.items.map((item) => { + return { + variantId: item.variant_id, + quantity: item.quantity, + } + }) + const generatedLineItems: LineItem[] = await lineItemServiceTx.generate( + generateInputData, + { + region_id: regionId, + customer_id: req.user?.customer_id, + } + ) + + await cartServiceTx.addOrUpdateLineItems( + createdCart.id, + generatedLineItems, + { + validateSalesChannels: + featureFlagRouter.isFeatureEnabled("sales_channels"), + } + ) + } + + return createdCart + }) + } cart = await cartService.retrieveWithTotals(cart!.id, { select: defaultStoreCartFields, diff --git a/packages/types/src/workflow/cart/create-cart.ts b/packages/types/src/workflow/cart/create-cart.ts index ecdd322b8f..99195caacd 100644 --- a/packages/types/src/workflow/cart/create-cart.ts +++ b/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[] diff --git a/packages/workflows/src/definition/cart/create-cart.ts b/packages/workflows/src/definition/cart/create-cart.ts index 8c09ecc67f..034c597e97 100644 --- a/packages/workflows/src/definition/cart/create-cart.ts +++ b/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 export const createCart = exportWorkflow< CartWorkflow.CreateCartWorkflowInputDTO, CreateCartWorkflowOutput ->(Workflows.CreateCart, CreateCartActions.retrieveCart) +>(Workflows.CreateCart, CreateCartActions.createCart) diff --git a/packages/workflows/src/definition/product/create-products.ts b/packages/workflows/src/definition/product/create-products.ts index 8ea4a304b3..d4fc7a2956 100644 --- a/packages/workflows/src/definition/product/create-products.ts +++ b/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 ), diff --git a/packages/workflows/src/handlers/common/index.ts b/packages/workflows/src/handlers/common/index.ts index b5c1adca54..0bef8d0ff2 100644 --- a/packages/workflows/src/handlers/common/index.ts +++ b/packages/workflows/src/handlers/common/index.ts @@ -1,2 +1 @@ -export * from "./set-config" export * from "./set-context" diff --git a/packages/workflows/src/handlers/common/set-config.ts b/packages/workflows/src/handlers/common/set-config.ts deleted file mode 100644 index 75c505c6ed..0000000000 --- a/packages/workflows/src/handlers/common/set-config.ts +++ /dev/null @@ -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): Promise { - return { - retrieveConfig: data[Aliases.Config].retrieveConfig, - } -} - -setConfig.aliases = Aliases diff --git a/packages/workflows/src/handlers/middlewares/create-products-prepare-create-prices-compensation.ts b/packages/workflows/src/handlers/middlewares/create-products-prepare-create-prices-compensation.ts index 9387b9d29f..6b4c58a6a3 100644 --- a/packages/workflows/src/handlers/middlewares/create-products-prepare-create-prices-compensation.ts +++ b/packages/workflows/src/handlers/middlewares/create-products-prepare-create-prices-compensation.ts @@ -10,33 +10,37 @@ type VariantIndexAndPrices = { export async function createProductsPrepareCreatePricesCompensation({ data, }: WorkflowArguments<{ - productsHandleVariantsIndexPricesMap: Map< - ProductHandle, - VariantIndexAndPrices[] - > + 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 = - updatedProductsHandleVariantsIndexPricesMap.get(productHandle) ?? [] + productsHandleVariantsIndexPricesMap.forEach( + (existingItems, productHandle) => { + const items = + updatedProductsHandleVariantsIndexPricesMap.get(productHandle) ?? [] - items.forEach(({ index }) => { - existingItems.push({ - index, - prices: [], + existingItems.forEach(({ index }) => { + items.push({ + index, + prices: [], + }) }) - }) - updatedProductsHandleVariantsIndexPricesMap.set(productHandle, items) - }) + 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", } diff --git a/packages/workflows/src/helper/__tests__/aggregate.spec.ts b/packages/workflows/src/helper/__tests__/merge-data.spec.ts similarity index 71% rename from packages/workflows/src/helper/__tests__/aggregate.spec.ts rename to packages/workflows/src/helper/__tests__/merge-data.spec.ts index febbf5e0f2..2abe8233b6 100644 --- a/packages/workflows/src/helper/__tests__/aggregate.spec.ts +++ b/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 diff --git a/packages/workflows/src/helper/__tests__/pipe.spec.ts b/packages/workflows/src/helper/__tests__/pipe.spec.ts index 29ca51e34b..141b28f45f 100644 --- a/packages/workflows/src/helper/__tests__/pipe.spec.ts +++ b/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" } diff --git a/packages/workflows/src/helper/index.ts b/packages/workflows/src/helper/index.ts index 6382e76182..a75943d8bb 100644 --- a/packages/workflows/src/helper/index.ts +++ b/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" diff --git a/packages/workflows/src/helper/aggregate.ts b/packages/workflows/src/helper/merge-data.ts similarity index 83% rename from packages/workflows/src/helper/aggregate.ts rename to packages/workflows/src/helper/merge-data.ts index fca711c2a3..134d41c240 100644 --- a/packages/workflows/src/helper/aggregate.ts +++ b/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 = Record, 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, } } } diff --git a/packages/workflows/src/helper/pipe.ts b/packages/workflows/src/helper/pipe.ts index a55f221c27..225a1f2c4c 100644 --- a/packages/workflows/src/helper/pipe.ts +++ b/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 + /** + * 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 = { @@ -53,6 +71,15 @@ export function pipe( input: PipelineInput, ...functions: [...PipelineHandler[], PipelineHandler] ): 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,