[Bug]: Handling dynamically sized input data in workflows

[noubase]

Package.json file

{
  "name": "medusa-starter-default",
  "version": "0.0.1",
  "description": "A starter for Medusa projects.",
  "author": "Medusa (https://medusajs.com)",
  "license": "MIT",
  "keywords": [
    "sqlite",
    "postgres",
    "typescript",
    "ecommerce",
    "headless",
    "medusa"
  ],
  "scripts": {
    "build": "medusa build",
    "seed": "medusa exec ./src/scripts/seed.ts",
    "start": "medusa start",
    "dev": "medusa develop",
    "test:integration:http": "TEST_TYPE=integration:http NODE_OPTIONS=--experimental-vm-modules jest --silent=false --runInBand --forceExit",
    "test:integration:modules": "TEST_TYPE=integration:modules NODE_OPTIONS=--experimental-vm-modules jest --silent --runInBand --forceExit",
    "test:unit": "TEST_TYPE=unit NODE_OPTIONS=--experimental-vm-modules jest --silent --runInBand --forceExit"
  },
  "dependencies": {
    "@dnd-kit/modifiers": "^7.0.0",
    "@medusajs/admin-sdk": "2.0.4",
    "@medusajs/cli": "2.0.4",
    "@medusajs/framework": "2.0.4",
    "@medusajs/medusa": "2.0.4",
    "@mikro-orm/core": "5.9.7",
    "@mikro-orm/knex": "5.9.7",
    "@mikro-orm/migrations": "5.9.7",
    "@mikro-orm/postgresql": "5.9.7",
    "awilix": "^8.0.1",
    "axios-retry": "^4.5.0",
    "contentful-management": "11.35.1",
    "contentful-rich-text-html-parser": "^1.5.13",
    "lodash": "^4.17.21",
    "pg": "^8.13.1",
    "react-icons": "^5.3.0",
    "xmlrpc": "^1.3.2",
    "@directus/sdk": "^18.0.0"
  },
  "devDependencies": {
    "@mikro-orm/cli": "5.9.7",
    "@swc/core": "1.5.7",
    "@swc/jest": "^0.2.36",
    "@types/jest": "^29.5.13",
    "@types/lodash": "^4.17.12",
    "@types/node": "^20.0.0",
    "@types/react": "^18.3.2",
    "@types/react-dom": "^18.2.25",
    "jest": "^29.7.0",
    "medusa-test-utils": "rc",
    "prop-types": "^15.8.1",
    "react": "^18.2.0",
    "react-dom": "^18.2.0",
    "ts-node": "^10.9.2",
    "typescript": "^5.6.2",
    "vite": "^5.2.11",
    "@medusajs/test-utils": "latest"
  },
  "engines": {
    "node": ">=20"
  },
  "packageManager": "[email protected]+sha512.a6b2f7906b721bba3d67d4aff083df04dad64c399707841b7acf00f6b133b7ac24255f2652fa22ae3534329dc6180534e98d17432037ff6fd140556e2bb3137e"
}

Node.js version

v20

Database and its version

16.2

Operating system name and version

Mac OS X

Browser name

No response

What happended?

As I understand it, there’s no way to process dynamically sized input data in workflows because we need a for-each mechanism for steps.
Let me clarify with an example:

import {createStep, createWorkflow, parallelize, StepResponse, WorkflowResponse} from "@medusajs/workflows-sdk";


const oneStep = createStep<string, string, never>(
    {name: 'dynamic-step', noCompensation: true},
    async (input: string, context) => {
      return new StepResponse<string, never>(`dynamic-${input}`)
    })


const dynamicInputWorkflow = createWorkflow<string[], string[], any[]>(
    {name: "dynamic-workflow"},

    function (input: string[]) {
      const steps = input.map(s => oneStep(s))
      const results = parallelize(...steps)
      return new WorkflowResponse(results)
    }
)

export default dynamicInputWorkflow

This approach is not feasible.

const dynamicInputWorkflow = createWorkflow<{ input: string[] }, string[], any[]>(
    {name: "dynamic-workflow"},

    function ({input}: { input: string[] }) {
      const results: string[] = []
      for (const s of input) {
        results.push(oneStep(s))
      }
      return new WorkflowResponse(results)
    }
)

I have a workflow that takes a single product as input. It creates an entity in my CMS and updates a MedusaJS product’s metadata using the updateProductsStep from the core workflows. I want to use this workflow as a step within the product creation workflow hook. However, it’s currently not possible to process an array of products in this way.

Even if my workflow receives an array of products to create in the CMS, I can’t invoke updateProductsStep multiple times within the workflow — there’s simply no mechanism for that

createProductsWorkflow.hooks.productsCreated(
    async ({products}, {container}) => {
      for (const product of products) {
        createProductFromMedusaProduct.runAsStep({input: product})
      }
    }
)

Such an approach won’t work either because there would be no compensation if it fails on the third product.

Expected behavior

We need a mechanism to invoke steps either in parallel or sequentially within a workflow, based on dynamically sized input.

Actual behavior

There is no such mechanism

Link to reproduction repo

no need

[noubase]

A bit off-topic:
From the documentation, we know that an exception in a workflow will trigger compensation in a hook. However, the reverse is not true—any exceptions in a hook will not trigger compensation in a workflow. That’s a SIGNIFICANT architectural issue.

[noubase]

A bit off-topic: From the documentation, we know that an exception in a workflow will trigger compensation in a hook. However, the reverse is not true—any exceptions in a hook will not trigger compensation in a workflow. That’s a SIGNIFICANT architectural issue.

Sorry, this behavior is caused by another bug in a workflows engine
#10201

[carlos-r-l-rodrigues]

Your hook function should be like a createStep definition, the example above using createProductFromMedusaProduct.runAsStep is not correct, this is used when composing the workflow.
Here you want to use it as you would use in an endpoint, wf(container).run(..)

Regarding multiple executions, currently we don't yet support a "loop" in the workflow composition.
The alternative to it will be available in the next release, that is handling the errors in your loop, and responding the step as a failure, with data to be used to compensate that step.

basically you wrap a try/catch in you createProduct workflow, storing the necessary data to revert it, and if there is failure you can revert it on the compensation step.

You can see how it is used in this test:
https://github.com/medusajs/medusa/blob/develop/packages/core/workflows-sdk/src/utils/composer/__tests__/compose.ts#L2526

[noubase]

basically you wrap a try/catch in you createProduct workflow, storing the necessary data to revert it, and if there is failure you can revert it on the compensation step.

You can see how it is used in this test: https://github.com/medusajs/medusa/blob/develop/packages/core/workflows-sdk/src/utils/composer/__tests__/compose.ts#L2526

Nah, that’s not a very elegant solution because I essentially have to re-implement the updateProductsStep within my step.

Something like this

const steps = input.map(s => oneStep(s))
const results = parallelize(...steps)
return new WorkflowResponse(results)

would be much better, IMHO

[carlos-r-l-rodrigues]

A proper loop functionality that you can run multiple batches inside the same step will come in the future.

I'm not sure why you would need to reimplement what you mentioned, in theory you either call another workflow to delete what you created or you could call the workflow method to cancel the transaction, which essentially is the same as reverting all steps after the workflow is done. However this isn't yet documented.