fix(workflows-sdk): name for when/then step (#10459)

.changeset/calm-needles-buy.md

@@ -0,0 +1,8 @@
+---
+"@medusajs/workflow-engine-redis": patch
+"@medusajs/workflows-sdk": patch
+"@medusajs/core-flows": patch
+"@medusajs/utils": patch
+---
+
+fix: when/then step name

packages/core/core-flows/src/cart/workflows/complete-cart.ts

@@ -65,7 +65,7 @@ export const completeCartWorkflow = createWorkflow(
     })
 
     // If order ID does not exist, we are completing the cart for the first time
-    const order = when({ orderId }, ({ orderId }) => {
+    const order = when("create-order", { orderId }, ({ orderId }) => {
       return !orderId
     }).then(() => {
       const cart = useRemoteQueryStep({

packages/core/core-flows/src/cart/workflows/list-shipping-options-for-cart.ts

@@ -70,9 +70,13 @@ export const listShippingOptionsForCartWorkflow = createWorkflow(
       }
     )
 
-    const customerGroupIds = when({ cart }, ({ cart }) => {
+    const customerGroupIds = when(
-      return !!cart.id
+      "get-customer-group",
-    }).then(() => {
+      { cart },
+      ({ cart }) => {
+        return !!cart.id
+      }
+    ).then(() => {
       const customerQuery = useQueryGraphStep({
         entity: "customer",
         filters: { id: cart.customer_id },

packages/core/utils/src/common/promise-all.ts

@@ -20,6 +20,10 @@ export async function promiseAll<T extends readonly unknown[] | []>(
   promises: T,
   { aggregateErrors } = { aggregateErrors: false }
 ): Promise<{ -readonly [P in keyof T]: Awaited<T[P]> }> {
+  if (!promises.length) {
+    return [] as unknown as Promise<{ -readonly [P in keyof T]: Awaited<T[P]> }>
+  }
+
   const states = await Promise.allSettled(promises)
 
   const rejected = (states as PromiseSettledResult<unknown>[]).filter(

packages/core/workflows-sdk/src/helper/workflow-export.ts

@@ -529,8 +529,6 @@ function attachOnFinishReleaseEvents(
         )
     }
 
-    await onFinish?.(args)
-
     const eventBusService = (
       flow.container as MedusaContainer
     ).resolve<IEventBusModuleService>(Modules.EVENT_BUS, {
@@ -538,6 +536,7 @@ function attachOnFinishReleaseEvents(
     })
 
     if (!eventBusService || !flowEventGroupId) {
+      await onFinish?.(args)
       return
     }
 
@@ -553,14 +552,19 @@ function attachOnFinishReleaseEvents(
         })
     }
 
-    await eventBusService.releaseGroupedEvents(flowEventGroupId).catch((e) => {
+    await eventBusService
-      logger.error(
+      .releaseGroupedEvents(flowEventGroupId)
-        `Failed to release grouped events for eventGroupId: ${flowEventGroupId}`,
+      .then(async () => {
-        e
+        await onFinish?.(args)
-      )
+      })
+      .catch((e) => {
+        logger.error(
+          `Failed to release grouped events for eventGroupId: ${flowEventGroupId}`,
+          e
+        )
 
-      return flow.cancel(transaction)
+        return flow.cancel(transaction)
-    })
+      })
   }
 
   events.onFinish = wrappedOnFinish

packages/core/workflows-sdk/src/utils/composer/__tests__/index.spec.ts

@@ -243,8 +243,9 @@ describe("Workflow composer", () => {
         return new StepResponse({ result: input })
       })
 
+      const wfId = getNewWorkflowId()
       const subWorkflow = createWorkflow(
-        getNewWorkflowId(),
+        wfId,
         function (input: WorkflowData<string>) {
           childWorkflowStep1()
           return new WorkflowResponse(childWorkflowStep2(input))
@@ -269,7 +270,9 @@ describe("Workflow composer", () => {
       expect(result).toEqual({ result: "hi from outside" })
 
       expect(parentContext.transactionId).toEqual(expect.any(String))
-      expect(parentContext.transactionId).toEqual(childContext.transactionId)
+      expect(childContext.transactionId).toEqual(
+        wfId + "-as-step-" + parentContext.transactionId
+      )
 
       expect(parentContext.eventGroupId).toEqual("eventGroupId")
       expect(parentContext.eventGroupId).toEqual(childContext.eventGroupId)
@@ -293,8 +296,9 @@ describe("Workflow composer", () => {
         return new StepResponse({ result: input })
       })
 
+      const wfId = getNewWorkflowId()
       const subWorkflow = createWorkflow(
-        getNewWorkflowId(),
+        wfId,
         function (input: WorkflowData<string>) {
           childWorkflowStep1()
           return new WorkflowResponse(childWorkflowStep2(input))
@@ -315,7 +319,9 @@ describe("Workflow composer", () => {
       expect(result).toEqual({ result: "hi from outside" })
 
       expect(parentContext.transactionId).toBeTruthy()
-      expect(parentContext.transactionId).toEqual(childContext.transactionId)
+      expect(childContext.transactionId).toEqual(
+        wfId + "-as-step-" + parentContext.transactionId
+      )
 
       expect(parentContext.eventGroupId).toBeTruthy()
       expect(parentContext.eventGroupId).toEqual(childContext.eventGroupId)

packages/core/workflows-sdk/src/utils/composer/create-workflow.ts

@@ -194,8 +194,9 @@ export function createWorkflow<TData, TResult, THooks extends any[]>(
           input: stepInput as any,
           container,
           context: {
-            transactionId: ulid(),
             ...sharedContext,
+            transactionId:
+              step.__step__ + "-" + (stepContext.transactionId ?? ulid()),
             parentStepIdempotencyKey: stepContext.idempotencyKey,
           },
         })

packages/core/workflows-sdk/src/utils/composer/when.ts

@@ -1,4 +1,4 @@
-import { OrchestrationUtils } from "@medusajs/utils"
+import { isDefined, OrchestrationUtils } from "@medusajs/utils"
 import { ulid } from "ulid"
 import { createStep } from "./create-step"
 import { StepResponse } from "./helpers/step-response"
@@ -26,7 +26,26 @@ export function when<T extends object | WorkflowData, Then extends Function>(
   then: ThenFunc
 }
 
-export function when(input, condition) {
+export function when<T extends object | WorkflowData, Then extends Function>(
+  name: string,
+  values: T,
+  condition: ConditionFunction<T>
+): {
+  then: ThenFunc
+}
+
+export function when(...args) {
+  let [name, input, condition] = args
+  if (args.length === 2) {
+    condition = input
+    input = name
+    name = undefined
+  }
+
+  if (typeof condition !== "function") {
+    throw new Error(`"when condition" must be a function`)
+  }
+
   global[OrchestrationUtils.SymbolMedusaWorkflowComposerCondition] = {
     input,
     condition,
@@ -49,9 +68,23 @@ export function when(input, condition) {
       const applyCondition =
         global[OrchestrationUtils.SymbolMedusaWorkflowComposerCondition].steps
 
-      if (ret?.__type !== OrchestrationUtils.SymbolWorkflowStep) {
+      if (
+        isDefined(ret) &&
+        ret?.__type !== OrchestrationUtils.SymbolWorkflowStep
+      ) {
+        if (!isDefined(name)) {
+          name = "when-then-" + ulid()
+          const context =
+            global[OrchestrationUtils.SymbolMedusaWorkflowComposerContext]
+
+          console.warn(
+            `${context.workflowId}: "when" name should be defined. A random one will be assigned to it, which is not recommended for production.\n`,
+            condition.toString()
+          )
+        }
+
         const retStep = createStep(
-          "when-then-" + ulid(),
+          name,
           ({ input }: { input: any }) => new StepResponse(input)
         )
         returnStep = retStep({ input: ret })

packages/modules/workflow-engine-redis/integration-tests/__fixtures__/index.ts

@@ -3,3 +3,4 @@ export * from "./workflow_2"
 export * from "./workflow_async"
 export * from "./workflow_step_timeout"
 export * from "./workflow_transaction_timeout"
+export * from "./workflow_when"

packages/modules/workflow-engine-redis/integration-tests/__fixtures__/workflow_async.ts

@@ -5,7 +5,6 @@ import {
   StepResponse,
   WorkflowResponse,
 } from "@medusajs/framework/workflows-sdk"
-import { setTimeout } from "timers/promises"
 
 const step_1_background = createStep(
   {
@@ -13,8 +12,6 @@ const step_1_background = createStep(
     async: true,
   },
   jest.fn(async (input) => {
-    await setTimeout(Math.random() * 300)
-
     return new StepResponse(input)
   })
 )

packages/modules/workflow-engine-redis/integration-tests/__fixtures__/workflow_when.ts

@@ -0,0 +1,52 @@
+import {
+  createStep,
+  createWorkflow,
+  StepResponse,
+  when,
+  WorkflowData,
+  WorkflowResponse,
+} from "@medusajs/framework/workflows-sdk"
+
+const step1 = createStep(
+  {
+    name: "step1",
+    async: true,
+  },
+  async (_, context) => {
+    await new Promise((resolve) => setTimeout(resolve, 2000))
+    return new StepResponse({ result: "step1" })
+  }
+)
+const step2 = createStep("step2", async (input: string, context) => {
+  return new StepResponse({ result: input })
+})
+const step3 = createStep(
+  "step3",
+  async (input: string | undefined, context) => {
+    return new StepResponse({ result: input ?? "default response" })
+  }
+)
+
+const subWorkflow = createWorkflow(
+  "wf-when-sub",
+  function (input: WorkflowData<string>) {
+    return new WorkflowResponse(step2(input))
+  }
+)
+
+createWorkflow("wf-when", function (input: { callSubFlow: boolean }) {
+  step1()
+  const subWorkflowRes = when("sub-flow", { input }, ({ input }) => {
+    return input.callSubFlow
+  }).then(() => {
+    const res = subWorkflow.runAsStep({
+      input: "hi from outside",
+    })
+
+    return {
+      result: res,
+    }
+  }) as any
+
+  return new WorkflowResponse(step3(subWorkflowRes.result))
+})

packages/modules/workflow-engine-redis/integration-tests/__tests__/index.spec.ts

@@ -18,6 +18,7 @@ import {
 import { moduleIntegrationTestRunner } from "@medusajs/test-utils"
 import { asValue } from "awilix"
 import { setTimeout } from "timers/promises"
+import { setTimeout as setTimeoutSync } from "timers"
 import { WorkflowsModuleService } from "../../src/services"
 import "../__fixtures__"
 import { createScheduled } from "../__fixtures__/workflow_scheduled"
@@ -25,6 +26,18 @@ import { TestDatabase } from "../utils"
 
 jest.setTimeout(999900000)
 
+const failTrap = (done) => {
+  setTimeoutSync(() => {
+    // REF:https://stackoverflow.com/questions/78028715/jest-async-test-with-event-emitter-isnt-ending
+    console.warn(
+      "Jest is breaking the event emit with its debouncer. This allows to continue the test by managing the timeout of the test manually."
+    )
+    done()
+  }, 5000)
+}
+
+// REF:https://stackoverflow.com/questions/78028715/jest-async-test-with-event-emitter-isnt-ending
+
 moduleIntegrationTestRunner<IWorkflowEngineService>({
   moduleName: Modules.WORKFLOW_ENGINE,
   resolve: __dirname + "/../..",
@@ -359,9 +372,9 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
           ).toBe(true)
         })
 
-        it.skip("should complete an async workflow that returns a StepResponse", (done) => {
+        it("should complete an async workflow that returns a StepResponse", (done) => {
           const transactionId = "transaction_1"
-          void workflowOrcModule
+          workflowOrcModule
             .run("workflow_async_background", {
               input: {
                 myInput: "123",
@@ -369,7 +382,7 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
               transactionId,
               throwOnError: true,
             })
-            .then(({ transaction, result }) => {
+            .then(({ transaction, result }: any) => {
               expect(transaction.flow.state).toEqual(
                 TransactionStepState.INVOKING
               )
@@ -385,14 +398,14 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
               }
             },
           })
+
+          failTrap(done)
         })
 
-        it.skip("should subscribe to a async workflow and receive the response when it finishes", (done) => {
+        it("should subscribe to a async workflow and receive the response when it finishes", (done) => {
           const transactionId = "trx_123"
 
-          const onFinish = jest.fn(() => {
+          const onFinish = jest.fn()
-            done()
-          })
 
           void workflowOrcModule.run("workflow_async_background", {
             input: {
@@ -408,11 +421,36 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
             subscriber: (event) => {
               if (event.eventType === "onFinish") {
                 onFinish()
+                done()
               }
             },
           })
 
           expect(onFinish).toHaveBeenCalledTimes(0)
+
+          failTrap(done)
+        })
+
+        it("should not skip step if condition is true", function (done) {
+          void workflowOrcModule.run("wf-when", {
+            input: {
+              callSubFlow: true,
+            },
+            transactionId: "trx_123_when",
+            throwOnError: true,
+            logOnError: true,
+          })
+
+          void workflowOrcModule.subscribe({
+            workflowId: "wf-when",
+            subscriber: (event) => {
+              if (event.eventType === "onFinish") {
+                done()
+              }
+            },
+          })
+
+          failTrap(done)
         })
       })
 

packages/modules/workflow-engine-redis/src/services/workflow-orchestrator.ts

@@ -9,6 +9,7 @@ import {
 import {
   ContainerLike,
   Context,
+  Logger,
   MedusaContainer,
 } from "@medusajs/framework/types"
 import {
@@ -24,6 +25,7 @@ import {
   ReturnWorkflow,
 } from "@medusajs/framework/workflows-sdk"
 import Redis from "ioredis"
+import { setTimeout } from "timers"
 import { ulid } from "ulid"
 import type { RedisDistributedTransactionStorage } from "../utils"
 
@@ -92,6 +94,8 @@ export class WorkflowOrchestratorService {
   private subscribers: Subscribers = new Map()
   private activeStepsCount: number = 0
 
+  readonly #logger: Logger
+
   protected redisDistributedTransactionStorage_: RedisDistributedTransactionStorage
 
   constructor({
@@ -112,6 +116,9 @@ export class WorkflowOrchestratorService {
     this.redisPublisher = redisPublisher
     this.redisSubscriber = redisSubscriber
 
+    this.#logger =
+      this.container_.resolve("logger", { allowUnregistered: true }) ?? console
+
     redisDistributedTransactionStorage.setWorkflowOrchestratorService(this)
 
     if (!dataLoaderOnly) {
@@ -149,6 +156,7 @@ export class WorkflowOrchestratorService {
   private async triggerParentStep(transaction, result) {
     const metadata = transaction.flow.metadata
     const { parentStepIdempotencyKey } = metadata ?? {}
+
     if (parentStepIdempotencyKey) {
       const hasFailed = [
         TransactionState.REVERTED,
@@ -159,11 +167,17 @@ export class WorkflowOrchestratorService {
         await this.setStepFailure({
           idempotencyKey: parentStepIdempotencyKey,
           stepResponse: result,
+          options: {
+            logOnError: true,
+          },
         })
       } else {
         await this.setStepSuccess({
           idempotencyKey: parentStepIdempotencyKey,
           stepResponse: result,
+          options: {
+            logOnError: true,
+          },
         })
       }
     }
@@ -209,6 +223,9 @@ export class WorkflowOrchestratorService {
       throw new Error(`Workflow with id "${workflowId}" not found.`)
     }
 
+    const originalOnFinishHandler = events.onFinish!
+    delete events.onFinish
+
     const ret = await exportedWorkflow.run({
       input,
       throwOnError: false,
@@ -235,13 +252,11 @@ export class WorkflowOrchestratorService {
       hasFailed,
     }
 
-    if (ret.transaction.hasFinished()) {
+    if (hasFinished) {
       const { result, errors } = ret
 
-      await this.notify({
+      await originalOnFinishHandler({
-        eventType: "onFinish",
+        transaction: ret.transaction,
-        workflowId,
-        transactionId: context.transactionId,
         result,
         errors,
       })
@@ -327,6 +342,9 @@ export class WorkflowOrchestratorService {
       workflowId,
     })
 
+    const originalOnFinishHandler = events.onFinish!
+    delete events.onFinish
+
     const ret = await exportedWorkflow.registerStepSuccess({
       idempotencyKey: idempotencyKey_,
       context,
@@ -341,10 +359,8 @@ export class WorkflowOrchestratorService {
     if (ret.transaction.hasFinished()) {
       const { result, errors } = ret
 
-      await this.notify({
+      await originalOnFinishHandler({
-        eventType: "onFinish",
+        transaction: ret.transaction,
-        workflowId,
-        transactionId,
         result,
         errors,
       })
@@ -397,6 +413,9 @@ export class WorkflowOrchestratorService {
       workflowId,
     })
 
+    const originalOnFinishHandler = events.onFinish!
+    delete events.onFinish
+
     const ret = await exportedWorkflow.registerStepFailure({
       idempotencyKey: idempotencyKey_,
       context,
@@ -411,10 +430,8 @@ export class WorkflowOrchestratorService {
     if (ret.transaction.hasFinished()) {
       const { result, errors } = ret
 
-      await this.notify({
+      await originalOnFinishHandler({
-        eventType: "onFinish",
+        transaction: ret.transaction,
-        workflowId,
-        transactionId,
         result,
         errors,
       })
@@ -517,7 +534,6 @@ export class WorkflowOrchestratorService {
 
     if (publish) {
       const channel = this.getChannelName(options.workflowId)
-
       const message = JSON.stringify({
         instanceId: this.instanceId,
         data: options,
@@ -540,7 +556,7 @@ export class WorkflowOrchestratorService {
 
     const notifySubscribers = (handlers: SubscriberHandler[]) => {
       handlers.forEach((handler) => {
-        handler({
+        const args = {
           eventType,
           workflowId,
           transactionId,
@@ -548,13 +564,30 @@ export class WorkflowOrchestratorService {
           response,
           result,
           errors,
-        })
+        }
+        const isPromise = "then" in handler
+        if (isPromise) {
+          ;(handler(args) as unknown as Promise<any>).catch((e) => {
+            this.#logger.error(e)
+          })
+        } else {
+          try {
+            handler(args)
+          } catch (e) {
+            this.#logger.error(e)
+          }
+        }
       })
     }
 
     if (transactionId) {
       const transactionSubscribers = subscribers.get(transactionId) ?? []
       notifySubscribers(transactionSubscribers)
+
+      // removes transaction id subscribers on finish
+      if (eventType === "onFinish") {
+        subscribers.delete(transactionId)
+      }
     }
 
     const workflowSubscribers = subscribers.get(AnySubscriber) ?? []
@@ -613,8 +646,8 @@ export class WorkflowOrchestratorService {
         await notify({ eventType: "onCompensateBegin" })
       },
       onFinish: async ({ transaction, result, errors }) => {
-        // TODO: unsubscribe transaction handlers on finish
         customEventHandlers?.onFinish?.({ transaction, result, errors })
+        await notify({ eventType: "onFinish" })
       },
 
       onStepBegin: async ({ step, transaction }) => {

packages/modules/workflow-engine-redis/src/utils/workflow-orchestrator-storage.ts

@@ -94,7 +94,10 @@ export class RedisDistributedTransactionStorage
           )
         }
       },
-      { connection: this.redisWorkerConnection }
+      {
+        connection:
+          this.redisWorkerConnection /*, runRetryDelay: 100000 for tests */,
+      }
     )
   }