feat(medusa, types, utils, workflow): Migrate medusa workflow to the workflow package (#4682)

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

@@ -0,0 +1,55 @@
+import { aggregateData } from "../aggregate"
+import { WorkflowStepMiddlewareReturn } from "../pipe"
+
+describe("aggregate", function () {
+  it("should aggregate a new object from the source into a specify target", async function () {
+    const source = {
+      stringProp: "stringProp",
+      anArray: ["anArray"],
+      input: {
+        test: "test",
+      },
+      another: {
+        anotherTest: "anotherTest",
+      },
+    }
+
+    const { value: result } = (await aggregateData(
+      ["input", "another", "stringProp", "anArray"],
+      "payload"
+    )({ data: source } as any)) as unknown as WorkflowStepMiddlewareReturn
+
+    expect(result).toEqual({
+      payload: {
+        ...source.input,
+        ...source.another,
+        anArray: source.anArray,
+        stringProp: source.stringProp,
+      },
+    })
+  })
+
+  it("should aggregate a new object from the entire source into the resul object", async function () {
+    const source = {
+      stringProp: "stringProp",
+      anArray: ["anArray"],
+      input: {
+        test: "test",
+      },
+      another: {
+        anotherTest: "anotherTest",
+      },
+    }
+
+    const { value: result } = (await aggregateData()({
+      data: source,
+    } as any)) as unknown as WorkflowStepMiddlewareReturn
+
+    expect(result).toEqual({
+      ...source.input,
+      ...source.another,
+      anArray: source.anArray,
+      stringProp: source.stringProp,
+    })
+  })
+})

packages/workflows/src/helper/aggregate.ts

@@ -0,0 +1,49 @@
+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.
+ * @param keys
+ * @param target
+ */
+export function aggregateData<
+  T extends Record<string, unknown> = Record<string, unknown>,
+  TKeys extends keyof T = keyof T,
+  Target extends "payload" | string = string
+>(keys: TKeys[] = [], target?: Target): PipelineHandler {
+  return async function ({ data }: WorkflowArguments<T>) {
+    const workingKeys = (keys.length ? keys : Object.keys(data)) as TKeys[]
+    const value = workingKeys.reduce((acc, key) => {
+      let targetAcc = { ...(target ? acc[target as string] : acc) }
+      targetAcc ??= {}
+
+      if (Array.isArray(data[key as string])) {
+        targetAcc[key as string] = data[key as string]
+      } else if (isObject(data[key as string])) {
+        targetAcc = {
+          ...targetAcc,
+          ...(data[key as string] as object),
+        }
+      } else {
+        targetAcc[key as string] = data[key as string]
+      }
+
+      if (target) {
+        acc[target as string] = {
+          ...acc[target as string],
+          ...targetAcc,
+        }
+      } else {
+        acc = targetAcc
+      }
+
+      return acc
+    }, {})
+
+    return {
+      alias: target,
+      value,
+    }
+  }
+}

packages/workflows/src/helper/pipe.ts

@@ -6,14 +6,14 @@ import {
 
 import { InputAlias } from "../definitions"
 
-type WorkflowStepMiddlewareReturn = {
-  alias: string
+export type WorkflowStepMiddlewareReturn = {
+  alias?: string
   value: any
 }
 
-type WorkflowStepMiddlewareInput = {
+export type WorkflowStepMiddlewareInput = {
   from: string
-  alias: string
+  alias?: string
 }
 
 interface PipelineInput {
@@ -32,11 +32,13 @@ export type WorkflowArguments<T = any> = {
 
 export type PipelineHandler<T extends any = undefined> = (
   args: WorkflowArguments
-) => T extends undefined
-  ? Promise<WorkflowStepMiddlewareReturn | WorkflowStepMiddlewareReturn[]>
-  : T
+) => Promise<
+  T extends undefined
+    ? WorkflowStepMiddlewareReturn | WorkflowStepMiddlewareReturn[]
+    : T
+>
 
-export function pipe<T = undefined>(
+export function pipe<T>(
   input: PipelineInput,
   ...functions: [...PipelineHandler[], PipelineHandler<T>]
 ): WorkflowStepHandler {
@@ -48,7 +50,7 @@ export function pipe<T = undefined>(
     metadata,
     context,
   }) => {
-    const data = {}
+    let data = {}
 
     const original = {
       invoke: invoke ?? {},
@@ -60,7 +62,7 @@ export function pipe<T = undefined>(
     }
 
     for (const key in input) {
-      if (!input[key]) {
+      if (!input[key] || key === "inputAlias") {
         continue
       }
 
@@ -69,13 +71,16 @@ export function pipe<T = undefined>(
       }
 
       for (const action of input[key]) {
-        if (action?.alias) {
+        if (action.alias) {
           data[action.alias] = original[key][action.from]
+        } else {
+          data[action.from] = original[key][action.from]
         }
       }
     }
 
-    return functions.reduce(async (_, fn) => {
+    let finalResult
+    for (const fn of functions) {
       let result = await fn({
         container,
         payload,
@@ -90,13 +95,22 @@ export function pipe<T = undefined>(
             data[action.alias] = action.value
           }
         }
-      } else if ((result as WorkflowStepMiddlewareReturn)?.alias) {
-        data[(result as WorkflowStepMiddlewareReturn).alias] = (
-          result as WorkflowStepMiddlewareReturn
-        ).value
+      } else if (
+        result &&
+        "alias" in (result as WorkflowStepMiddlewareReturn)
+      ) {
+        if ((result as WorkflowStepMiddlewareReturn).alias) {
+          data[(result as WorkflowStepMiddlewareReturn).alias!] = (
+            result as WorkflowStepMiddlewareReturn
+          ).value
+        } else {
+          data = (result as WorkflowStepMiddlewareReturn).value
+        }
       }
 
-      return result
-    }, {})
+      finalResult = result
+    }
+
+    return finalResult
   }
 }