Workflow Orchestration

Simple Operation (No Orchestration)
┌─────────────┐
│   Action    │ ← Single operation, no coordination
└─────────────┘

Orchestrated Workflow
┌─────────────────────────────────────────────────┐
│  Workflow (Orchestrator)                        │
│  ┌──────────┐   ┌──────────┐   ┌──────────┐   │
│  │  Step 1  │→→→│  Step 2  │→→→│  Step 3  │   │
│  └────┬─────┘   └────┬─────┘   └────┬─────┘   │
│       │              │              │          │
│       ▼              ▼              ▼          │
│  ┌────────┐     ┌────────┐     ┌────────┐     │
│  │Rollback│     │Rollback│     │Rollback│     │
│  │Step 1  │◀◀◀◀◀│Step 2  │◀◀◀◀◀│Step 3  │     │
│  └────────┘     └────────┘     └────────┘     │
└─────────────────────────────────────────────────┘

// Without workflows - Manual coordination and rollback
async function createBrandWithLogo(brandData, logoFile) {
  let brand
  let logoUrl

  try {
    // Step 1: Create brand
    const brandService = container.resolve("brand")
    brand = await brandService.createBrands([brandData])

    // Step 2: Upload logo
    const s3Service = container.resolve("s3Service")
    logoUrl = await s3Service.upload(logoFile)

    // Step 3: Update brand with logo URL
    await brandService.updateBrands([{
      id: brand.id,
      logo_url: logoUrl,
    }])

    return { brand, logoUrl }
  } catch (error) {
    // Manual rollback - Error-prone!
    if (brand) {
      try {
        await brandService.deleteBrands([brand.id])
      } catch (rollbackError) {
        // What if rollback fails? Data is now inconsistent!
        console.error("Rollback failed:", rollbackError)
      }
    }

    if (logoUrl) {
      try {
        await s3Service.delete(logoUrl)
      } catch (rollbackError) {
        // Orphaned file in S3!
        console.error("S3 cleanup failed:", rollbackError)
      }
    }

    throw error
  }
}

// With workflows - Automatic coordination and rollback
const createBrandStep = createStep(
  "create-brand",
  async (input, { container }) => {
    const brandService = container.resolve("brand")
    const [brand] = await brandService.createBrands([input])
    return new StepResponse(brand, brand.id)
  },
  async (brandId, { container }) => {
    if (!brandId) return
    const brandService = container.resolve("brand")
    await brandService.deleteBrands([brandId])
  }
)

const uploadLogoStep = createStep(
  "upload-logo",
  async (input, { container }) => {
    const s3Service = container.resolve("s3Service")
    const logoUrl = await s3Service.upload(input.logo)
    return new StepResponse(logoUrl, logoUrl)
  },
  async (logoUrl, { container }) => {
    if (!logoUrl) return
    const s3Service = container.resolve("s3Service")
    await s3Service.delete(logoUrl)
  }
)

const updateBrandLogoStep = createStep(
  "update-brand-logo",
  async (input, { container }) => {
    const brandService = container.resolve("brand")
    await brandService.updateBrands([{
      id: input.brandId,
      logo_url: input.logoUrl,
    }])
    return new StepResponse("updated", { brandId: input.brandId, previousLogoUrl: null })
  },
  async (compensationData, { container }) => {
    const brandService = container.resolve("brand")
    await brandService.updateBrands([{
      id: compensationData.brandId,
      logo_url: compensationData.previousLogoUrl,
    }])
  }
)

export const createBrandWithLogoWorkflow = createWorkflow(
  "create-brand-with-logo",
  function (input) {
    const brand = createBrandStep(input)
    const logoUrl = uploadLogoStep({ logo: input.logo })
    updateBrandLogoStep({
      brandId: brand.id,
      logoUrl: logoUrl,
    })

    return new WorkflowResponse({ brand, logoUrl })
  }
)

// Use it
const { result } = await createBrandWithLogoWorkflow(container)
  .run({ input: { name: "Nike", logo: file } })

// ❌ WRONG - Imperative (trying to execute)
createWorkflow("wrong", async function (input) {
  const result = await someStep(input)  // ❌ Using await!
  return result
})

// ✅ CORRECT - Declarative (defining flow)
createWorkflow("correct", function (input) {
  const result = someStep(input)  // ✅ No await! Just defining flow
  return new WorkflowResponse(result)
})

Workflow Definition (What)      Workflow Execution (How)
┌────────────────────┐           ┌──────────────────────┐
│ function (input) { │           │  Engine executes:    │
│   step1(input)     │──────▶    │  1. Calls step1      │
│   step2(step1)     │           │  2. Waits for result │
│   step3(step2)     │           │  3. Calls step2      │
│   return response  │           │  4. Waits for result │
│ }                  │           │  5. Calls step3      │
└────────────────────┘           │  6. Returns response │
                                 └──────────────────────┘

createWorkflow("sequential", function (input) {
  const brand = createBrandStep(input.brand)
  const product = createProductStep({
    title: input.productTitle,
    brand_id: brand.id,  // Uses output from previous step
  })
  const inventory = allocateInventoryStep({
    product_id: product.id,  // Uses output from previous step
    quantity: input.quantity,
  })

  return new WorkflowResponse({ brand, product, inventory })
})

import { createWorkflow, when } from "@medusajs/framework/workflows-sdk"

createWorkflow("conditional", function (input) {
  const brand = createBrandStep(input.brand)

  // Only send notification if brand is premium
  when({ brand }, ({ brand }) => {
    return brand.is_premium
  }).then(() => {
    sendPremiumNotificationStep(brand)
  })

  return new WorkflowResponse(brand)
})

import { createWorkflow, transform } from "@medusajs/framework/workflows-sdk"

createWorkflow("transform-example", function (input) {
  const brands = createMultipleBrandsStep(input.brands)

  // Transform array of brands to just their IDs
  const brandIds = transform({ brands }, ({ brands }) => {
    return brands.map(b => b.id)
  })

  const products = createProductsStep({
    products: input.products,
    brand_ids: brandIds,  // Use transformed data
  })

  return new WorkflowResponse({ brands, products })
})

const createBrandStep = createStep(
  "create-brand",
  async (input, { container }) => {
    const brandService = container.resolve("brand")
    const [brand] = await brandService.createBrands([input])
    return new StepResponse(brand, brand.id)
  },
  async (brandId, { container }) => {
    if (!brandId) return
    const brandService = container.resolve("brand")
    await brandService.deleteBrands([brandId])
  }
)

const updateBrandStep = createStep(
  "update-brand",
  async (input, { container }) => {
    const brandService = container.resolve("brand")

    // Get current brand to save its state
    const [currentBrand] = await brandService.retrieveBrands([input.id])

    // Update brand
    const [updatedBrand] = await brandService.updateBrands([{
      id: input.id,
      name: input.name,
    }])

    // Return updated brand as result, current brand for compensation
    return new StepResponse(updatedBrand, {
      id: currentBrand.id,
      previousName: currentBrand.name,
    })
  },
  async (compensationData, { container }) => {
    if (!compensationData) return

    const brandService = container.resolve("brand")

    // Restore previous name
    await brandService.updateBrands([{
      id: compensationData.id,
      name: compensationData.previousName,
    }])
  }
)

const getBrandStep = createStep(
  "get-brand",
  async (input, { container }) => {
    const brandService = container.resolve("brand")
    const [brand] = await brandService.retrieveBrands([input.id])
    return new StepResponse(brand)
  }
  // No compensation function - read-only operation
)

const uploadToS3Step = createStep(
  "upload-to-s3",
  async (input, { container }) => {
    const s3Service = container.resolve("s3Service")
    const result = await s3Service.upload(input.file)

    return new StepResponse(result.url, {
      url: result.url,
      bucket: result.bucket,
      key: result.key,
    })
  },
  async (compensationData, { container }) => {
    if (!compensationData) return

    const s3Service = container.resolve("s3Service")

    // Delete file from S3
    await s3Service.deleteObject({
      bucket: compensationData.bucket,
      key: compensationData.key,
    })
  }
)

export const createOrderWorkflow = createWorkflow(
  "create-order",
  function (input) {
    // Step 1: Validate inventory (read-only, no compensation)
    const inventoryCheck = validateInventoryStep(input.items)

    // Step 2: Create order
    const order = createOrderStep({
      customer_id: input.customer_id,
      items: input.items,
    })

    // Step 3: Reserve inventory (parallel with payment)
    const reservation = reserveInventoryStep({
      order_id: order.id,
      items: input.items,
    })

    // Step 4: Process payment (parallel with inventory)
    const payment = processPaymentStep({
      order_id: order.id,
      amount: input.amount,
      payment_method: input.payment_method,
    })

    // Step 5: Send confirmation (only after payment succeeds)
    when({ payment }, ({ payment }) => payment.status === "succeeded")
      .then(() => {
        sendOrderConfirmationStep({
          order_id: order.id,
          customer_email: input.customer_email,
        })
      })

    // Step 6: Allocate to warehouse
    const allocation = allocateToWarehouseStep({
      order_id: order.id,
      items: input.items,
      warehouse_id: input.warehouse_id,
    })

    return new WorkflowResponse({ order, payment, reservation, allocation })
  }
)

// Core Medusa workflow
export const createProductsWorkflow = createWorkflow(
  "create-products",
  function (input) {
    const products = createProductsStep(input)

    // Hook point: productsCreated
    // Your custom code runs here

    return new WorkflowResponse(products)
  }
)

// Your application - Subscribe to hook
createProductsWorkflow.hooks.productsCreated(
  async ({ products, additional_data }, { container }) => {
    // Your custom logic
    const link = container.resolve("link")

    if (additional_data?.brand_id) {
      await link.create({
        [Modules.BRAND]: { brand_id: additional_data.brand_id },
        [Modules.PRODUCT]: { product_id: products[0].id },
      })
    }

    return new StepResponse("done")
  },
  async (compensationData, { container }) => {
    // Your custom compensation
    if (compensationData?.linkId) {
      const link = container.resolve("link")
      await link.dismiss([compensationData.linkId])
    }
  }
)

// ❌ WRONG
createWorkflow("wrong", async function (input) {
  const result = await someStep(input)  // ❌ Async/await not allowed!
  return result
})

// ✅ CORRECT
createWorkflow("correct", function (input) {
  const result = someStep(input)  // ✅ Synchronous definition
  return new WorkflowResponse(result)
})

// ❌ WRONG - No compensation for state change
const createBrandStep = createStep(
  "create-brand",
  async (input, { container }) => {
    const brandService = container.resolve("brand")
    const [brand] = await brandService.createBrands([input])
    return new StepResponse(brand)
  }
  // Missing compensation! Brand remains if workflow fails
)

// ✅ CORRECT
const createBrandStep = createStep(
  "create-brand",
  async (input, { container }) => {
    const brandService = container.resolve("brand")
    const [brand] = await brandService.createBrands([input])
    return new StepResponse(brand, brand.id)
  },
  async (brandId, { container }) => {
    if (!brandId) return
    const brandService = container.resolve("brand")
    await brandService.deleteBrands([brandId])
  }
)

// ❌ WRONG - Logic in workflow function
createWorkflow("wrong", function (input) {
  const brand = createBrandStep(input)

  // ❌ Business logic in workflow function
  if (brand.name.startsWith("Nike")) {
    const premiumBrand = { ...brand, is_premium: true }
    return new WorkflowResponse(premiumBrand)
  }

  return new WorkflowResponse(brand)
})

// ✅ CORRECT - Logic in steps
createWorkflow("correct", function (input) {
  const brand = createBrandStep(input)

  // Conditional step based on brand data
  when({ brand }, ({ brand }) => brand.name.startsWith("Nike"))
    .then(() => {
      markAsPremiumStep(brand.id)
    })

  return new WorkflowResponse(brand)
})

// ❌ WRONG - Direct database access
createWorkflow("wrong", function (input) {
  const brand = someStepThatDirectlyQueriesDB(input)  // ❌ DB access outside module
  return new WorkflowResponse(brand)
})

// ✅ CORRECT - Database access in steps, steps use modules
const createBrandStep = createStep(
  "create-brand",
  async (input, { container }) => {
    // ✅ Use module service
    const brandService = container.resolve("brand")
    const [brand] = await brandService.createBrands([input])
    return new StepResponse(brand, brand.id)
  },
  async (brandId, { container }) => {
    const brandService = container.resolve("brand")
    await brandService.deleteBrands([brandId])
  }
)