Workflow Design

Workflows orchestrate business processes by combining events, actions, conditions, and integrations into cohesive automations.

What are Workflows?

Workflows are executable business processes defined as code. They:

React to events (triggers)
Execute actions (operations)
Make decisions (conditions)
Maintain state (durable execution)
Handle errors (retries, compensation)
Integrate systems (external services)

Workflow Patterns

1. Sequential Workflows

Execute steps in order:

// Customer onboarding workflow
const onboardingWorkflow = $.workflow('customer-onboarding', async (customer: Person) => {
  // Step 1: Send welcome email
  await $.step('welcome-email', async () => {
    send.Email({
      to: customer.email,
      template: 'welcome',
    })
  })

  // Step 2: Create account
  const account = await $.step('create-account', async () => {
    return await db.Accounts.create({
      customer: customer.$id,
      status: 'active',
      createdAt: new Date(),
    })
  })

  // Step 3: Schedule onboarding call
  await $.step('schedule-call', async () => {
    send.Meeting.schedule({
      attendees: [customer.email, '[email protected]'],
      duration: 30,
      subject: 'Welcome to Acme!',
    })
  })

  return { customer, account }
})

// Trigger workflow
on.Customer.registered(async (customer) => {
  await onboardingWorkflow.execute(customer)
})

2. Parallel Workflows

Execute steps concurrently:

// Lead enrichment workflow
const enrichLeadWorkflow = $.workflow('enrich-lead', async (lead) => {
  // Execute all enrichments in parallel
  const [email, company, social] = await Promise.all([
    $.step('verify-email', () => verifyEmail(lead.email)),
    $.step('lookup-company', () => lookupCompany(lead.domain)),
    $.step('find-social', () => findSocialProfiles(lead.name)),
  ])

  // Combine results
  return await $.step('merge-data', () => ({
    ...lead,
    emailValid: email.valid,
    company: company,
    socialProfiles: social,
  }))
})

3. Conditional Workflows

Branch based on conditions:

// Order fulfillment workflow
const fulfillOrderWorkflow = $.workflow('fulfill-order', async (order: Order) => {
  // Check inventory
  const inventory = await $.step('check-inventory', async () => {
    return await db.Inventory.findOne({
      where: { productId: { $in: order.items.map((i) => i.productId) } },
    })
  })

  if (inventory.available) {
    // In stock: process immediately
    await $.step('reserve-inventory', () => send.Inventory.reserve(order.items))

    await $.step('create-shipment', () => send.Shipment.create(order))
  } else {
    // Out of stock: backorder
    await $.step('create-backorder', () => send.Backorder.create(order))

    await $.step('notify-customer', () =>
      send.Email({
        to: order.customer.email,
        template: 'backorder-notification',
      })
    )
  }
})

4. Loop Workflows

Iterate over collections:

// Batch email campaign workflow
const emailCampaignWorkflow = $.workflow('email-campaign', async (campaign) => {
  // Get recipients
  const recipients = await $.step('get-recipients', async () => {
    return await db.Person.list({
      where: { segment: campaign.segment },
    })
  })

  // Send to each recipient
  for (const recipient of recipients) {
    await $.step(`send-${recipient.$id}`, async () => {
      send.Email({
        to: recipient.email,
        template: campaign.template,
        data: await personalize(recipient),
      })
    })

    // Rate limiting
    await $.sleep(100) // 100ms between emails
  }

  return { sent: recipients.length }
})

5. Compensating Workflows

Handle failures with rollback:

// Payment workflow with compensation
const processPaymentWorkflow = $.workflow('process-payment', async (order: Order) => {
  // Reserve inventory
  const reservation = await $.step('reserve-inventory', async () => {
    return await send.Inventory.reserve(order.items)
  })

  try {
    // Charge payment
    const payment = await $.step('charge-payment', async () => {
      return await send.Payment.charge({
        amount: order.total,
        customer: order.customer,
      })
    })

    // Confirm order
    await $.step('confirm-order', async () => {
      send.Order.confirmed({ order, payment })
    })

    return { success: true, payment }
  } catch (error) {
    // Compensate: release inventory
    await $.step('release-inventory', async () => {
      send.Inventory.release(reservation)
    })

    // Notify failure
    await $.step('notify-failure', async () => {
      send.Order.failed({ order, error })
    })

    throw error
  }
})

Workflow Triggers

Event Triggers

React to system events:

// Trigger on order creation
on.Order.created(async (order) => {
  await fulfillOrderWorkflow.execute(order)
})

// Trigger on multiple events
on([$.User.registered, $.User.verified], async (user) => {
  await onboardingWorkflow.execute(user)
})

Schedule Triggers

Run on a schedule:

// Daily report workflow
const dailyReportWorkflow = $.workflow('daily-report', async () => {
  const stats = await calculateDailyStats()
  send.Report.generated(stats)
})

// Run every day at 9am
every.Daily.at('09:00', async () => {
  await dailyReportWorkflow.execute()
})

Manual Triggers

Invoke manually via API:

// API endpoint to trigger workflow
app.post('/api/workflows/process-refund', async (c) => {
  const { orderId } = await c.req.json()
  const order = await db.Orders.get(orderId)

  const result = await refundWorkflow.execute(order)
  return c.json(result)
})

Webhook Triggers

React to external webhooks:

// Webhook handler
app.post('/webhooks/stripe', async (c) => {
  const event = await c.req.json()

  if (event.type === 'payment_intent.succeeded') {
    await paymentSucceededWorkflow.execute(event.data)
  }

  return c.json({ received: true })
})

Workflow State

Durable Execution

Workflows maintain state across failures:

const longRunningWorkflow = $.workflow('long-running', async (data) => {
  // Step 1: Process data (takes 5 minutes)
  const processed = await $.step('process', async () => {
    return await processLargeDataset(data)
  })

  // If workflow crashes here, it resumes from Step 2
  // Step 1 won't re-execute

  // Step 2: Upload results (takes 10 minutes)
  const uploaded = await $.step('upload', async () => {
    return await uploadToStorage(processed)
  })

  // Step 3: Notify completion
  await $.step('notify', async () => {
    send.Notification({ result: uploaded })
  })
})

Workflow Variables

Store state across steps:

const orderWorkflow = $.workflow('order-processing', async (order) => {
  // Workflow-scoped variables
  let inventory: any
  let payment: any
  let shipment: any

  inventory = await $.step('check-inventory', async () => {
    return await checkInventory(order.items)
  })

  if (inventory.available) {
    payment = await $.step('process-payment', async () => {
      return await processPayment(order)
    })

    shipment = await $.step('create-shipment', async () => {
      return await createShipment(order, payment)
    })
  }

  return { inventory, payment, shipment }
})

Workflow Error Handling

Retries

Automatically retry failed steps:

const apiWorkflow = $.workflow('api-call', async (data) => {
  const result = await $.step(
    'call-api',
    async () => {
      return await externalAPI.call(data)
    },
    {
      retry: {
        maxAttempts: 3,
        backoff: 'exponential',
        initialDelay: 1000, // Start with 1 second
      },
    }
  )

  return result
})

Error Handling

Handle errors gracefully:

const resilientWorkflow = $.workflow('resilient', async (data) => {
  try {
    await $.step('risky-operation', async () => {
      return await riskyOperation(data)
    })
  } catch (error) {
    // Log error
    await $.step('log-error', async () => {
      send.Error.log({ error, data })
    })

    // Fallback operation
    await $.step('fallback', async () => {
      return await fallbackOperation(data)
    })
  }
})

Timeouts

Set step timeouts:

const timedWorkflow = $.workflow('timed', async (data) => {
  const result = await $.step(
    'slow-operation',
    async () => {
      return await slowOperation(data)
    },
    {
      timeout: 30000, // 30 seconds
    }
  )

  return result
})

Workflow Monitoring

Progress Tracking

Track workflow progress:

const workflow = $.workflow('tracked', async (data) => {
  // Emit progress events
  send.Workflow.progress({ step: 1, total: 5 })

  await $.step('step-1', async () => {
    await operation1(data)
    send.Workflow.progress({ step: 2, total: 5 })
  })

  await $.step('step-2', async () => {
    await operation2(data)
    send.Workflow.progress({ step: 3, total: 5 })
  })

  // ...more steps
})

Workflow Metrics

Collect workflow metrics:

const workflow = $.workflow('measured', async (data) => {
  const startTime = Date.now()

  const result = await $.step('operation', async () => {
    return await operation(data)
  })

  // Record metrics
  send.Metric.record({
    workflow: 'measured',
    duration: Date.now() - startTime,
    status: 'success',
  })

  return result
})

Workflow Composition

Sub-workflows

Compose workflows from other workflows:

// Reusable sub-workflows
const verifyEmailWorkflow = $.workflow('verify-email', async (email) => {
  // Email verification logic
})

const enrichDataWorkflow = $.workflow('enrich-data', async (data) => {
  // Data enrichment logic
})

// Parent workflow
const mainWorkflow = $.workflow('main', async (lead) => {
  // Call sub-workflows
  await $.step('verify', () => verifyEmailWorkflow.execute(lead.email))
  await $.step('enrich', () => enrichDataWorkflow.execute(lead))
})

Workflow Templates

Create reusable workflow templates:

// Template for approval workflows
const createApprovalWorkflow = (entity: string) => {
  return $.workflow(`${entity}-approval`, async (item) => {
    // Request approval
    const approval = await $.step('request-approval', async () => {
      return await send.Approval.request({
        entity,
        item,
        approvers: await getApprovers(entity),
      })
    })

    // Wait for approval
    const decision = await $.waitFor($.Approval.decided, {
      timeout: 86400000, // 24 hours
    })

    if (decision.approved) {
      await $.step('approve', () => send[$[entity].approved](item))
    } else {
      await $.step('reject', () => send[$[entity].rejected](item))
    }
  })
}

// Create specific workflows from template
const invoiceApproval = createApprovalWorkflow('Invoice')
const expenseApproval = createApprovalWorkflow('Expense')

Workflow Integration

External APIs

Integrate with external services:

const integrationWorkflow = $.workflow('integration', async (data) => {
  // Call external API
  const result = await $.step('call-salesforce', async () => {
    return await fetch('https://api.salesforce.com/leads', {
      method: 'POST',
      headers: { Authorization: `Bearer ${env.SALESFORCE_TOKEN}` },
      body: JSON.stringify(data),
    })
  })

  return result
})

Database Operations

Perform database operations:

const dbWorkflow = $.workflow('database', async (user) => {
  // Create user
  const created = await $.step('create-user', async () => {
    return await db.Person.create(user)
  })

  // Create related records
  await $.step('create-profile', async () => {
    return await db.Profile.create({
      user: created.$id,
      bio: '',
      avatar: null,
    })
  })

  return created
})

AI Operations

Use AI in workflows:

const aiWorkflow = $.workflow('ai-content', async (topic) => {
  // Generate content
  const content = await $.step('generate', async () => {
    return await ai.generate({
      model: 'gpt-5',
      prompt: `Write a blog post about ${topic}`,
      schema: $.BlogPost,
    })
  })

  // Review content
  const reviewed = await $.step('review', async () => {
    return await ai.generate({
      model: 'claude-sonnet-4.5',
      prompt: 'Review and improve this content',
      context: content,
      schema: $.BlogPost,
    })
  })

  return reviewed
})

Best Practices

Do's

Use descriptive workflow names
Break complex workflows into steps
Handle errors gracefully
Implement retries for flaky operations
Use timeouts to prevent hanging
Track workflow progress
Compose workflows from sub-workflows
Test workflows thoroughly

Don'ts

Don't create monolithic workflows
Don't ignore error handling
Don't skip retries for external calls
Don't forget timeouts
Don't lose workflow state
Don't make workflows too granular
Don't skip monitoring

Human-in-the-Loop Workflows

Approval Workflows

Implement approval processes:

const approvalWorkflow = $.workflow('expense-approval', async (expense) => {
  // Submit for approval
  await $.step('submit', async () => {
    await send($.Expense.submitted, {
      expenseId: expense.$id,
      amount: expense.amount,
      submittedBy: expense.employee,
    })
  })

  // Wait for manager approval
  const managerApproval = await $.waitFor(
    $.Approval.decided,
    {
      filter: (event) => event.expenseId === expense.$id && event.approver === expense.manager,
      timeout: 172800000, // 48 hours
    }
  )

  if (!managerApproval || managerApproval.decision === 'reject') {
    await $.step('reject', () => send($.Expense.rejected, { expenseId: expense.$id }))
    return { status: 'rejected' }
  }

  // For amounts over $1000, require VP approval
  if (expense.amount > 1000) {
    await $.step('escalate', () =>
      send($.Approval.requested, {
        expenseId: expense.$id,
        approver: expense.vp,
      })
    )

    const vpApproval = await $.waitFor(
      $.Approval.decided,
      {
        filter: (event) => event.expenseId === expense.$id && event.approver === expense.vp,
        timeout: 172800000,
      }
    )

    if (!vpApproval || vpApproval.decision === 'reject') {
      await $.step('reject-final', () => send($.Expense.rejected, { expenseId: expense.$id }))
      return { status: 'rejected' }
    }
  }

  // Process payment
  await $.step('approve', async () => {
    await send($.Expense.approved, { expenseId: expense.$id })
    await send($.Payment.process, {
      recipient: expense.employee,
      amount: expense.amount,
    })
  })

  return { status: 'approved' }
})

Review Workflows

Implement document review processes:

const documentReviewWorkflow = $.workflow('document-review', async (document) => {
  // Assign reviewers
  const reviewers = await $.step('assign-reviewers', async () => {
    return await assignReviewers(document.type, document.department)
  })

  // Request reviews in parallel
  const reviewRequests = reviewers.map((reviewer) =>
    $.step(`request-review-${reviewer.id}`, () =>
      send($.Review.requested, {
        documentId: document.$id,
        reviewerId: reviewer.id,
      })
    )
  )

  await Promise.all(reviewRequests)

  // Wait for all reviews (with timeout)
  const reviews = []
  for (const reviewer of reviewers) {
    try {
      const review = await $.waitFor(
        $.Review.submitted,
        {
          filter: (event) => event.documentId === document.$id && event.reviewerId === reviewer.id,
          timeout: 604800000, // 7 days
        }
      )
      reviews.push(review)
    } catch (timeoutError) {
      // Send reminder
      await $.step(`remind-${reviewer.id}`, () =>
        send($.Email.send, {
          to: reviewer.email,
          template: 'review-reminder',
          data: { document },
        })
      )

      // Wait again with shorter timeout
      const review = await $.waitFor(
        $.Review.submitted,
        {
          filter: (event) => event.documentId === document.$id && event.reviewerId === reviewer.id,
          timeout: 172800000, // 2 more days
        }
      )
      reviews.push(review)
    }
  }

  // Aggregate feedback
  await $.step('aggregate-feedback', async () => {
    const approved = reviews.filter((r) => r.status === 'approved').length
    const changesRequested = reviews.filter((r) => r.status === 'changes-requested').length
    const rejected = reviews.filter((r) => r.status === 'rejected').length

    if (rejected > 0) {
      await send($.Document.rejected, { documentId: document.$id, reviews })
    } else if (changesRequested > 0) {
      await send($.Document.needs_changes, { documentId: document.$id, reviews })
    } else {
      await send($.Document.approved, { documentId: document.$id, reviews })
    }
  })
})

Workflow State Machines

Define explicit state machines:

interface StateMachine {
  states: Record<string, State>
  initial: string
  context?: any
}

interface State {
  on: Record<string, Transition>
  entry?: () => Promise<void>
  exit?: () => Promise<void>
}

interface Transition {
  target: string
  guard?: (context: any) => boolean
  action?: (context: any) => Promise<any>
}

// Order state machine
const orderStateMachine: StateMachine = {
  initial: 'draft',
  states: {
    draft: {
      on: {
        submit: {
          target: 'pending',
          guard: (order) => order.items.length > 0,
          action: async (order) => {
            await send($.Order.submitted, order)
          },
        },
      },
    },
    pending: {
      entry: async () => {
        console.log('Order pending, waiting for payment')
      },
      on: {
        pay: {
          target: 'processing',
          action: async (order) => {
            const payment = await processPayment(order)
            return { ...order, paymentId: payment.id }
          },
        },
        cancel: {
          target: 'cancelled',
        },
      },
    },
    processing: {
      entry: async () => {
        console.log('Processing order')
      },
      on: {
        fulfill: {
          target: 'completed',
          action: async (order) => {
            await createShipment(order)
          },
        },
        fail: {
          target: 'failed',
        },
      },
    },
    completed: {
      entry: async () => {
        console.log('Order completed')
      },
      on: {
        return: {
          target: 'returned',
          guard: (order) => {
            // Can only return within 30 days
            const daysSince = (Date.now() - order.completedAt.getTime()) / (1000 * 60 * 60 * 24)
            return daysSince <= 30
          },
        },
      },
    },
    cancelled: {},
    failed: {
      on: {
        retry: {
          target: 'processing',
        },
      },
    },
    returned: {},
  },
}

// Workflow using state machine
const orderWorkflow = $.workflow('order', async (order) => {
  let currentState = orderStateMachine.initial
  let context = order

  const transition = async (event: string, data?: any) => {
    const state = orderStateMachine.states[currentState]
    const transition = state.on[event]

    if (!transition) {
      throw new Error(`Invalid transition ${event} from state ${currentState}`)
    }

    // Check guard condition
    if (transition.guard && !transition.guard(context)) {
      throw new Error(`Guard condition failed for ${event}`)
    }

    // Exit current state
    if (state.exit) {
      await state.exit()
    }

    // Execute transition action
    if (transition.action) {
      context = await transition.action({ ...context, ...data })
    }

    // Move to target state
    currentState = transition.target

    // Enter new state
    const newState = orderStateMachine.states[currentState]
    if (newState.entry) {
      await newState.entry()
    }

    // Update order state in database
    await db.Orders.update(order.$id, {
      status: currentState,
      ...context,
    })

    return currentState
  }

  // Handle events
  on($.Order.submitted, async () => await transition('submit'))
  on($.Order.paid, async (payment) => await transition('pay', { paymentId: payment.id }))
  on($.Order.fulfilled, async () => await transition('fulfill'))
  on($.Order.cancelled, async () => await transition('cancel'))

  return { currentState, context }
})

Workflow Orchestration

Parent-Child Workflows

Orchestrate complex workflows:

// Parent workflow
const ecommerceOrderWorkflow = $.workflow('ecommerce-order', async (order) => {
  // Start child workflows in parallel
  const [inventoryReservation, paymentProcessing, customerNotification] = await Promise.all([
    $.startChildWorkflow('reserve-inventory', order.items),
    $.startChildWorkflow('process-payment', {
      orderId: order.$id,
      amount: order.total,
    }),
    $.startChildWorkflow('send-confirmation', {
      email: order.customer.email,
      orderId: order.$id,
    }),
  ])

  // Wait for critical workflows
  const inventoryResult = await inventoryReservation.result()
  const paymentResult = await paymentProcessing.result()

  if (!paymentResult.success) {
    // Cancel inventory reservation
    await $.startChildWorkflow('release-inventory', inventoryReservation.data)
    throw new Error('Payment failed')
  }

  // Start fulfillment workflow
  const fulfillment = await $.startChildWorkflow('fulfill-order', {
    order,
    inventory: inventoryResult,
    payment: paymentResult,
  })

  return await fulfillment.result()
})

// Child workflow
const reserveInventoryWorkflow = $.workflow('reserve-inventory', async (items) => {
  const reservations = []

  for (const item of items) {
    const reservation = await $.step(`reserve-${item.productId}`, async () => {
      const product = await db.Products.get(item.productId)

      if (product.inventory < item.quantity) {
        throw new Error(`Insufficient inventory for ${product.name}`)
      }

      await db.Products.update(item.productId, {
        inventory: { $decrement: item.quantity },
        reserved: { $increment: item.quantity },
      })

      return {
        productId: item.productId,
        quantity: item.quantity,
        reservedAt: new Date(),
      }
    })

    reservations.push(reservation)
  }

  return reservations
})

Workflow Scheduling

Schedule workflow executions:

// Cron-based scheduling
const scheduledWorkflows = [
  {
    name: 'daily-report',
    schedule: '0 9 * * *', // 9 AM daily
    workflow: dailyReportWorkflow,
  },
  {
    name: 'weekly-cleanup',
    schedule: '0 2 * * 0', // 2 AM Sunday
    workflow: weeklyCleanupWorkflow,
  },
  {
    name: 'hourly-sync',
    schedule: '0 * * * *', // Every hour
    workflow: hourlySyncWorkflow,
  },
]

// Register scheduled workflows
scheduledWorkflows.forEach(({ name, schedule, workflow }) => {
  $.schedule(name, schedule, workflow)
})

// Dynamic scheduling
const scheduleWorkflow = async (workflow: any, runAt: Date, data: any) => {
  const delay = runAt.getTime() - Date.now()

  if (delay > 0) {
    await $.sleep(delay)
  }

  return await workflow.execute(data)
}

// Schedule order reminder for abandoned carts
on($.Cart.abandoned, async (cart) => {
  // Send first reminder after 1 hour
  await scheduleWorkflow(cartReminderWorkflow, addHours(new Date(), 1), cart)

  // Send second reminder after 24 hours
  await scheduleWorkflow(cartReminderWorkflow, addHours(new Date(), 24), cart)
})

Workflow Performance

Parallel Execution

Optimize workflow performance:

const parallelWorkflow = $.workflow('parallel-processing', async (data) => {
  // Sequential (slow)
  const slow = await $.step('sequential', async () => {
    const result1 = await operation1(data)
    const result2 = await operation2(data)
    const result3 = await operation3(data)
    return [result1, result2, result3]
  })

  // Parallel (fast)
  const fast = await $.step('parallel', async () => {
    return await Promise.all([operation1(data), operation2(data), operation3(data)])
  })

  // Parallel with concurrency limit
  const limited = await $.step('parallel-limited', async () => {
    const operations = Array.from({ length: 100 }, (_, i) => () => operation(data, i))

    // Process 10 at a time
    return await batchProcess(operations, 10)
  })

  return { slow, fast, limited }
})

// Batch processing helper
const batchProcess = async (operations: Function[], concurrency: number) => {
  const results = []
  const queue = [...operations]

  const worker = async () => {
    while (queue.length > 0) {
      const operation = queue.shift()
      if (operation) {
        const result = await operation()
        results.push(result)
      }
    }
  }

  // Create worker pool
  const workers = Array.from({ length: concurrency }, worker)
  await Promise.all(workers)

  return results
}

Workflow Caching

Cache workflow results:

const cachedWorkflow = $.workflow('expensive-operation', async (input) => {
  // Check cache
  const cacheKey = `workflow:expensive-operation:${JSON.stringify(input)}`
  const cached = await cache.get(cacheKey)

  if (cached) {
    return cached
  }

  // Execute expensive operation
  const result = await $.step('expensive', async () => {
    // Long-running computation
    return await complexCalculation(input)
  })

  // Cache result
  await cache.set(cacheKey, result, { ttl: 3600 })

  return result
})

Workflow Debouncing

Debounce high-frequency workflows:

const debouncedWorkflow = $.workflow('search-index-update', async (data) => {
  // Debounce: only execute if no new events for 5 seconds
  const debounceKey = `debounce:search-index:${data.entityId}`

  await $.debounce(debounceKey, 5000, async () => {
    await $.step('update-index', async () => {
      await updateSearchIndex(data.entityId)
    })
  })
})

// Debounce implementation
const debounceMap = new Map()

const debounce = async (key: string, delay: number, fn: Function) => {
  // Clear existing timer
  if (debounceMap.has(key)) {
    clearTimeout(debounceMap.get(key))
  }

  // Set new timer
  return new Promise((resolve) => {
    const timer = setTimeout(async () => {
      debounceMap.delete(key)
      const result = await fn()
      resolve(result)
    }, delay)

    debounceMap.set(key, timer)
  })
}

Workflow Testing

Unit Testing Workflows

Test individual workflow steps:

import { describe, it, expect, beforeEach, vi } from 'vitest'

describe('orderWorkflow', () => {
  beforeEach(async () => {
    // Reset database and mocks
    await db.test.reset()
    vi.clearAllMocks()
  })

  it('should process order successfully', async () => {
    // Mock dependencies
    const mockProcessPayment = vi.fn().mockResolvedValue({ success: true, id: 'payment-123' })
    const mockCreateShipment = vi.fn().mockResolvedValue({ id: 'shipment-123' })

    // Execute workflow
    const order = await db.Orders.create({
      customerId: 'customer-123',
      items: [{ productId: 'product-1', quantity: 2 }],
      total: 99.99,
    })

    const result = await orderWorkflow.execute(order, {
      mocks: {
        processPayment: mockProcessPayment,
        createShipment: mockCreateShipment,
      },
    })

    // Assertions
    expect(result.status).toBe('completed')
    expect(mockProcessPayment).toHaveBeenCalledWith(order)
    expect(mockCreateShipment).toHaveBeenCalledWith(order)
  })

  it('should handle payment failure', async () => {
    const mockProcessPayment = vi.fn().mockResolvedValue({ success: false })

    const order = await db.Orders.create({
      customerId: 'customer-123',
      items: [{ productId: 'product-1', quantity: 2 }],
      total: 99.99,
    })

    await expect(
      orderWorkflow.execute(order, {
        mocks: { processPayment: mockProcessPayment },
      })
    ).rejects.toThrow('Payment failed')

    // Verify order was cancelled
    const cancelled = await db.Orders.get(order.$id)
    expect(cancelled.status).toBe('cancelled')
  })

  it('should retry on transient failures', async () => {
    const mockOperation = vi
      .fn()
      .mockRejectedValueOnce(new Error('Transient error'))
      .mockResolvedValueOnce({ success: true })

    const order = await db.Orders.create({
      customerId: 'customer-123',
      items: [],
      total: 0,
    })

    const result = await orderWorkflow.execute(order, {
      mocks: { operation: mockOperation },
    })

    expect(mockOperation).toHaveBeenCalledTimes(2)
    expect(result.status).toBe('completed')
  })
})

Integration Testing Workflows

Test workflow interactions:

describe('Order Processing Integration', () => {
  it('should complete full order lifecycle', async () => {
    // Create test data
    await db.test.seed('products', 'customers', 'inventory')

    // Create order
    const order = await db.Orders.create({
      customerId: 'customer-1',
      items: [
        { productId: 'product-1', quantity: 2, price: 29.99 },
        { productId: 'product-2', quantity: 1, price: 49.99 },
      ],
      total: 109.97,
    })

    // Execute workflow
    const result = await orderWorkflow.execute(order)

    // Verify order completed
    expect(result.status).toBe('completed')

    // Verify inventory was updated
    const product1 = await db.Products.get('product-1')
    expect(product1.inventory).toBe(8) // 10 - 2

    // Verify payment was processed
    const payment = await db.Payments.findOne({
      where: { orderId: order.$id },
    })
    expect(payment.status).toBe('success')

    // Verify shipment was created
    const shipment = await db.Shipments.findOne({
      where: { orderId: order.$id },
    })
    expect(shipment.status).toBe('pending')

    // Verify email was sent
    const emails = await db.test.getEmails()
    expect(emails).toHaveLength(1)
    expect(emails[0].to).toBe('[email protected]')
    expect(emails[0].template).toBe('order-confirmation')
  })
})

Workflow Patterns Library

Retry Pattern

const retryPattern = {
  exponentialBackoff: (attempt: number) => Math.min(1000 * Math.pow(2, attempt), 30000),
  linearBackoff: (attempt: number) => 1000 * attempt,
  fixedBackoff: () => 5000,
}

const withRetry = async (
  fn: Function,
  maxAttempts = 3,
  backoff = retryPattern.exponentialBackoff
) => {
  for (let attempt = 0; attempt < maxAttempts; attempt++) {
    try {
      return await fn()
    } catch (error) {
      if (attempt === maxAttempts - 1) throw error

      const delay = backoff(attempt)
      await $.sleep(delay)
    }
  }
}

Circuit Breaker Pattern

class CircuitBreaker {
  private failures = 0
  private lastFailureTime?: number
  private state: 'closed' | 'open' | 'half-open' = 'closed'

  constructor(
    private threshold: number = 5,
    private timeout: number = 60000,
    private halfOpenAttempts: number = 1
  ) {}

  async execute(fn: Function) {
    if (this.state === 'open') {
      if (Date.now() - this.lastFailureTime! > this.timeout) {
        this.state = 'half-open'
      } else {
        throw new Error('Circuit breaker is open')
      }
    }

    try {
      const result = await fn()

      if (this.state === 'half-open') {
        this.reset()
      }

      return result
    } catch (error) {
      this.recordFailure()
      throw error
    }
  }

  private recordFailure() {
    this.failures++
    this.lastFailureTime = Date.now()

    if (this.failures >= this.threshold) {
      this.state = 'open'
    }
  }

  private reset() {
    this.failures = 0
    this.state = 'closed'
    this.lastFailureTime = undefined
  }
}

Bulkhead Pattern

class Bulkhead {
  private activeCount = 0
  private queue: Function[] = []

  constructor(private maxConcurrent: number) {}

  async execute(fn: Function) {
    if (this.activeCount >= this.maxConcurrent) {
      await this.enqueue(fn)
    }

    this.activeCount++

    try {
      return await fn()
    } finally {
      this.activeCount--
      this.processQueue()
    }
  }

  private async enqueue(fn: Function) {
    return new Promise((resolve) => {
      this.queue.push(async () => {
        const result = await fn()
        resolve(result)
      })
    })
  }

  private processQueue() {
    if (this.queue.length > 0 && this.activeCount < this.maxConcurrent) {
      const fn = this.queue.shift()
      if (fn) this.execute(fn)
    }
  }
}

Next Steps

Examples → - See complete workflows
Deploy → - Deploy your workflows
Grow → - Use workflows for growth

Workflow Tip: Great workflows are resilient, observable, and composable. Design for failure and make workflows easy to monitor and debug.

Workflow Design

On this page