Feat: @medusajs/workflows (#4553)

feat: medusa workflows

packages/orchestration/src/workflow/global-workflow.ts

@@ -0,0 +1,94 @@
+import { Context, LoadedModule, MedusaContainer } from "@medusajs/types"
+import { WorkflowDefinition, WorkflowManager } from "./workflow-manager"
+
+import { DistributedTransaction } from "../transaction"
+import { asValue } from "awilix"
+import { createMedusaContainer } from "@medusajs/utils"
+
+export class GlobalWorkflow extends WorkflowManager {
+  protected static workflows: Map<string, WorkflowDefinition> = new Map()
+  protected container: MedusaContainer
+  protected context: Context
+
+  constructor(
+    modulesLoaded?: LoadedModule[] | MedusaContainer,
+    context?: Context
+  ) {
+    super()
+
+    const container = createMedusaContainer()
+
+    // Medusa container
+    if (!Array.isArray(modulesLoaded) && modulesLoaded) {
+      const cradle = modulesLoaded.cradle
+      for (const key in cradle) {
+        container.register(key, asValue(cradle[key]))
+      }
+    }
+    // Array of modules
+    else if (modulesLoaded?.length) {
+      for (const mod of modulesLoaded) {
+        const registrationName = mod.__definition.registrationName
+        container.register(registrationName, asValue(mod))
+      }
+    }
+
+    this.container = container
+    this.context = context ?? {}
+  }
+
+  async run(workflowId: string, uniqueTransactionId: string, input?: unknown) {
+    if (!WorkflowManager.workflows.has(workflowId)) {
+      throw new Error(`Workflow with id "${workflowId}" not found.`)
+    }
+
+    const workflow = WorkflowManager.workflows.get(workflowId)!
+
+    const orchestrator = workflow.orchestrator
+
+    const transaction = await orchestrator.beginTransaction(
+      uniqueTransactionId,
+      workflow.handler(this.container, this.context),
+      input
+    )
+
+    await orchestrator.resume(transaction)
+
+    return transaction
+  }
+
+  async registerStepSuccess(
+    workflowId: string,
+    idempotencyKey: string,
+    response?: unknown
+  ): Promise<DistributedTransaction> {
+    if (!WorkflowManager.workflows.has(workflowId)) {
+      throw new Error(`Workflow with id "${workflowId}" not found.`)
+    }
+
+    const workflow = WorkflowManager.workflows.get(workflowId)!
+    return await workflow.orchestrator.registerStepSuccess(
+      idempotencyKey,
+      workflow.handler(this.container, this.context),
+      undefined,
+      response
+    )
+  }
+
+  async registerStepFailure(
+    workflowId: string,
+    idempotencyKey: string,
+    error?: Error | any
+  ): Promise<DistributedTransaction> {
+    if (!WorkflowManager.workflows.has(workflowId)) {
+      throw new Error(`Workflow with id "${workflowId}" not found.`)
+    }
+
+    const workflow = WorkflowManager.workflows.get(workflowId)!
+    return await workflow.orchestrator.registerStepFailure(
+      idempotencyKey,
+      error,
+      workflow.handler(this.container, this.context)
+    )
+  }
+}

packages/orchestration/src/workflow/index.ts

@@ -0,0 +1,3 @@
+export * from "./workflow-manager"
+export * from "./local-workflow"
+export * from "./global-workflow"

packages/orchestration/src/workflow/local-workflow.ts

@@ -0,0 +1,209 @@
+import { Context, LoadedModule, MedusaContainer } from "@medusajs/types"
+import {
+  DistributedTransaction,
+  TransactionOrchestrator,
+  TransactionStepsDefinition,
+} from "../transaction"
+import {
+  WorkflowDefinition,
+  WorkflowManager,
+  WorkflowStepHandler,
+} from "./workflow-manager"
+
+import { OrchestratorBuilder } from "../transaction/orchestrator-builder"
+import { asValue } from "awilix"
+import { createMedusaContainer } from "@medusajs/utils"
+
+type StepHandler = {
+  invoke: WorkflowStepHandler
+  compensate?: WorkflowStepHandler
+}
+
+export class LocalWorkflow {
+  protected container: MedusaContainer
+  protected workflowId: string
+  protected flow: OrchestratorBuilder
+  protected workflow: WorkflowDefinition
+  protected handlers: Map<string, StepHandler>
+
+  constructor(
+    workflowId: string,
+    modulesLoaded?: LoadedModule[] | MedusaContainer
+  ) {
+    const globalWorkflow = WorkflowManager.getWorkflow(workflowId)
+    if (!globalWorkflow) {
+      throw new Error(`Workflow with id "${workflowId}" not found.`)
+    }
+
+    this.flow = new OrchestratorBuilder(globalWorkflow.flow_)
+    this.workflowId = workflowId
+    this.workflow = globalWorkflow
+    this.handlers = new Map(globalWorkflow.handlers_)
+
+    const container = createMedusaContainer()
+
+    // Medusa container
+    if (!Array.isArray(modulesLoaded) && modulesLoaded) {
+      const cradle = modulesLoaded.cradle
+      for (const key in cradle) {
+        container.register(key, asValue(cradle[key]))
+      }
+    }
+    // Array of modules
+    else if (modulesLoaded?.length) {
+      for (const mod of modulesLoaded) {
+        const registrationName = mod.__definition.registrationName
+        container.register(registrationName, asValue(mod))
+      }
+    }
+
+    this.container = container
+  }
+
+  protected commit() {
+    const finalFlow = this.flow.build()
+
+    this.workflow = {
+      id: this.workflowId,
+      flow_: finalFlow,
+      orchestrator: new TransactionOrchestrator(this.workflowId, finalFlow),
+      handler: WorkflowManager.buildHandlers(this.handlers),
+      handlers_: this.handlers,
+    }
+  }
+
+  async run(uniqueTransactionId: string, input?: unknown, context?: Context) {
+    if (this.flow.hasChanges) {
+      this.commit()
+    }
+
+    const { handler, orchestrator } = this.workflow
+
+    const transaction = await orchestrator.beginTransaction(
+      uniqueTransactionId,
+      handler(this.container, context),
+      input
+    )
+
+    await orchestrator.resume(transaction)
+
+    return transaction
+  }
+
+  async registerStepSuccess(
+    idempotencyKey: string,
+    response?: unknown,
+    context?: Context
+  ): Promise<DistributedTransaction> {
+    const { handler, orchestrator } = this.workflow
+    return await orchestrator.registerStepSuccess(
+      idempotencyKey,
+      handler(this.container, context),
+      undefined,
+      response
+    )
+  }
+
+  async registerStepFailure(
+    idempotencyKey: string,
+    error?: Error | any,
+    context?: Context
+  ): Promise<DistributedTransaction> {
+    const { handler, orchestrator } = this.workflow
+    return await orchestrator.registerStepFailure(
+      idempotencyKey,
+      error,
+      handler(this.container, context)
+    )
+  }
+
+  addAction(
+    action: string,
+    handler: StepHandler,
+    options: Partial<TransactionStepsDefinition> = {}
+  ) {
+    this.assertHandler(handler, action)
+    this.handlers.set(action, handler)
+
+    return this.flow.addAction(action, options)
+  }
+
+  replaceAction(
+    existingAction: string,
+    action: string,
+    handler: StepHandler,
+    options: Partial<TransactionStepsDefinition> = {}
+  ) {
+    this.assertHandler(handler, action)
+    this.handlers.set(action, handler)
+
+    return this.flow.replaceAction(existingAction, action, options)
+  }
+
+  insertActionBefore(
+    existingAction: string,
+    action: string,
+    handler: StepHandler,
+    options: Partial<TransactionStepsDefinition> = {}
+  ) {
+    this.assertHandler(handler, action)
+    this.handlers.set(action, handler)
+
+    return this.flow.insertActionBefore(existingAction, action, options)
+  }
+
+  insertActionAfter(
+    existingAction: string,
+    action: string,
+    handler: StepHandler,
+    options: Partial<TransactionStepsDefinition> = {}
+  ) {
+    this.assertHandler(handler, action)
+    this.handlers.set(action, handler)
+
+    return this.flow.insertActionAfter(existingAction, action, options)
+  }
+
+  appendAction(
+    action: string,
+    to: string,
+    handler: StepHandler,
+    options: Partial<TransactionStepsDefinition> = {}
+  ) {
+    this.assertHandler(handler, action)
+    this.handlers.set(action, handler)
+
+    return this.flow.appendAction(action, to, options)
+  }
+
+  moveAction(actionToMove: string, targetAction: string): OrchestratorBuilder {
+    return this.flow.moveAction(actionToMove, targetAction)
+  }
+
+  moveAndMergeNextAction(
+    actionToMove: string,
+    targetAction: string
+  ): OrchestratorBuilder {
+    return this.flow.moveAndMergeNextAction(actionToMove, targetAction)
+  }
+
+  mergeActions(where: string, ...actions: string[]) {
+    return this.flow.mergeActions(where, ...actions)
+  }
+
+  deleteAction(action: string, parentSteps?) {
+    return this.flow.deleteAction(action, parentSteps)
+  }
+
+  pruneAction(action: string) {
+    return this.flow.pruneAction(action)
+  }
+
+  protected assertHandler(handler: StepHandler, action: string): void | never {
+    if (!handler?.invoke) {
+      throw new Error(
+        `Handler for action "${action}" is missing invoke function.`
+      )
+    }
+  }
+}

packages/orchestration/src/workflow/workflow-manager.ts

@@ -0,0 +1,168 @@
+import { Context, MedusaContainer } from "@medusajs/types"
+import {
+  OrchestratorBuilder,
+  TransactionHandlerType,
+  TransactionMetadata,
+  TransactionOrchestrator,
+  TransactionStepHandler,
+  TransactionStepsDefinition,
+} from "../transaction"
+
+export interface WorkflowDefinition {
+  id: string
+  handler: (
+    container: MedusaContainer,
+    context?: Context
+  ) => TransactionStepHandler
+  orchestrator: TransactionOrchestrator
+  flow_: TransactionStepsDefinition
+  handlers_: Map<
+    string,
+    { invoke: WorkflowStepHandler; compensate?: WorkflowStepHandler }
+  >
+  requiredModules?: Set<string>
+  optionalModules?: Set<string>
+}
+
+export type WorkflowHandler = Map<
+  string,
+  { invoke: WorkflowStepHandler; compensate?: WorkflowStepHandler }
+>
+
+export type WorkflowStepHandler = (args: {
+  container: MedusaContainer
+  payload: unknown
+  invoke: { [actions: string]: unknown }
+  compensate: { [actions: string]: unknown }
+  metadata: TransactionMetadata
+  context?: Context
+}) => unknown
+
+export class WorkflowManager {
+  protected static workflows: Map<string, WorkflowDefinition> = new Map()
+
+  static unregister(workflowId: string) {
+    WorkflowManager.workflows.delete(workflowId)
+  }
+
+  static unregisterAll() {
+    WorkflowManager.workflows.clear()
+  }
+
+  static getWorkflows() {
+    return WorkflowManager.workflows
+  }
+
+  static getWorkflow(workflowId: string) {
+    return WorkflowManager.workflows.get(workflowId)
+  }
+
+  static getTransactionDefinition(workflowId): OrchestratorBuilder {
+    if (!WorkflowManager.workflows.has(workflowId)) {
+      throw new Error(`Workflow with id "${workflowId}" not found.`)
+    }
+
+    const workflow = WorkflowManager.workflows.get(workflowId)!
+    return new OrchestratorBuilder(workflow.flow_)
+  }
+
+  static register(
+    workflowId: string,
+    flow: TransactionStepsDefinition | OrchestratorBuilder,
+    handlers: WorkflowHandler,
+    requiredModules?: Set<string>,
+    optionalModules?: Set<string>
+  ) {
+    if (WorkflowManager.workflows.has(workflowId)) {
+      throw new Error(`Workflow with id "${workflowId}" is already defined.`)
+    }
+
+    const finalFlow = flow instanceof OrchestratorBuilder ? flow.build() : flow
+
+    WorkflowManager.workflows.set(workflowId, {
+      id: workflowId,
+      flow_: finalFlow,
+      orchestrator: new TransactionOrchestrator(workflowId, finalFlow),
+      handler: WorkflowManager.buildHandlers(handlers),
+      handlers_: handlers,
+      requiredModules,
+      optionalModules,
+    })
+  }
+
+  static update(
+    workflowId: string,
+    flow: TransactionStepsDefinition | OrchestratorBuilder,
+    handlers: Map<
+      string,
+      { invoke: WorkflowStepHandler; compensate?: WorkflowStepHandler }
+    >,
+    requiredModules?: Set<string>,
+    optionalModules?: Set<string>
+  ) {
+    if (!WorkflowManager.workflows.has(workflowId)) {
+      throw new Error(`Workflow with id "${workflowId}" not found.`)
+    }
+
+    const workflow = WorkflowManager.workflows.get(workflowId)!
+
+    for (const [key, value] of handlers.entries()) {
+      workflow.handlers_.set(key, value)
+    }
+
+    const finalFlow = flow instanceof OrchestratorBuilder ? flow.build() : flow
+
+    WorkflowManager.workflows.set(workflowId, {
+      id: workflowId,
+      flow_: finalFlow,
+      orchestrator: new TransactionOrchestrator(workflowId, finalFlow),
+      handler: WorkflowManager.buildHandlers(workflow.handlers_),
+      handlers_: workflow.handlers_,
+      requiredModules,
+      optionalModules,
+    })
+  }
+
+  public static buildHandlers(
+    handlers: Map<
+      string,
+      { invoke: WorkflowStepHandler; compensate?: WorkflowStepHandler }
+    >
+  ): (container: MedusaContainer, context?: Context) => TransactionStepHandler {
+    return (
+      container: MedusaContainer,
+      context?: Context
+    ): TransactionStepHandler => {
+      return async (
+        actionId: string,
+        handlerType: TransactionHandlerType,
+        payload?: any
+      ) => {
+        const command = handlers.get(actionId)
+
+        if (!command) {
+          throw new Error(`Handler for action "${actionId}" not found.`)
+        } else if (!command[handlerType]) {
+          throw new Error(
+            `"${handlerType}" handler for action "${actionId}" not found.`
+          )
+        }
+
+        const { invoke, compensate, payload: input } = payload.context
+        const { metadata } = payload
+
+        return await command[handlerType]!({
+          container,
+          payload: input,
+          invoke,
+          compensate,
+          metadata,
+          context,
+        })
+      }
+    }
+  }
+}
+
+global.WorkflowManager ??= WorkflowManager
+exports.WorkflowManager = global.WorkflowManager