Undici Workflow Engine

A dependency-aware, declarative service orchestrator for orchestrating HTTP services with variable interpolation and fallback support. Built for Node.js with Undici.

Features

• Dependency-based orchestration - Define service dependencies and execute them in the correct order
• Conditional execution - Skip services based on context using condition functions
• Error strategies - Control how critical vs. optional service failures affect the workflow
• Variable interpolation - Use {$<contextKey>.path} syntax to reference values from context
• Fallback responses - Define fallback data for services that fail
• OIDC authentication - Built-in OIDC client credentials flow support
• Cookie handling - Automatic Set-Cookie extraction and cookie management
• Response header extraction - Full header and status code capture
• Undici-powered - High-performance HTTP client built on modern Node.js APIs
• Type-safe - Full TypeScript support with comprehensive type definitions

Installation

npm install undici-workflow-engine

Quick Start

import { runOrchestration } from 'undici-workflow-engine';

const result = await runOrchestration(
  [
    {
      id: 'fetchUser',
      service: {
        url: 'https://api.example.com/users/123',
        method: 'GET',
      },
    },
    {
      id: 'fetchPosts',
      dependsOn: ['fetchUser'],
      service: {
        url: 'https://api.example.com/posts',
        method: 'GET',
        query: {
          userId: '{$fetchUser.body.id}',
        },
      },
    },
  ],
  {
    request: {
      headers: { 'authorization': 'Bearer token' },
    },
  }
);

console.log(result.success); // true/false
console.log(result.results); // { fetchUser: {...}, fetchPosts: {...} }

API Reference

runOrchestration(services, context)

Executes a workflow of orchestrated services.

Parameters:

• services: ServiceBlock[] - Array of service definitions with dependencies
• context: OrchestrationContext - Initial execution context

Returns: Promise<OrchestrationResult>

ServiceBlock

interface ServiceBlock {
  id: string;                              // Unique service identifier
  dependsOn?: string[];                    // Service IDs this depends on
  service: ServiceConfig;                  // Service configuration
  condition?: (context: any) => boolean;   // Skip service if condition is false
  errorStrategy?: 'silent' | 'throw';      // How to handle service errors
}

Properties:

• condition - Optional function to determine if service should execute. Receives the workflow context and should return boolean. If false, service is skipped.
• errorStrategy - How to handle errors:
  • 'silent' (default) - Service errors don't prevent dependent services from executing
  • 'throw' - Mark service as critical (for logging/monitoring purposes)

ServiceConfig

interface ServiceConfig {
  url: string;
  method: string;
  headers?: Record<string, string>;
  cookies?: Record<string, string>;
  query?: Record<string, string>;
  body?: any;
  timeout?: number;        // milliseconds (default: 30000)
  fallback?: { data: any };
  oidc?: {                 // Node.js only
    clientId: string;
    clientSecret: string;
    scope?: string;
    tokenUrl?: string;
  };
}

OrchestrationContext

interface OrchestrationContext {
  request: {
    body?: any;
    headers?: Record<string, string>;
    cookies?: Record<string, string>;
    query?: Record<string, string>;
  };
  env?: Record<string, string>;
}

ServiceResult

interface ServiceResult {
  status: number | null;
  body: any;
  headers?: Record<string, string>;
  cookies?: Record<string, string>;
  error?: any;
  fallbackUsed?: boolean;
}

OrchestrationResult

interface OrchestrationResult {
  success: boolean;
  results: Record<string, ServiceResult>;
}

Variable Interpolation

Reference any value from the orchestration context using {$<contextKey>.path} syntax with curly braces.

The context key can be any root-level property in the context object. Supports text before, after, and between tokens.

Syntax

{$<contextKey>.path.to.value}

Examples

{
  service: {
    body: {
      userId: '{$service01.body.id}',
      email: '{$request.body.email}',
      apiKey: '{$env.API_KEY}',
      customField: '{$customData.token}',
    },
  },
}

Bracket Notation

For keys with special characters (dots, colons, dashes):

{
  service: {
    body: {
      customField: "{$serviceId.body['custom:field']}",
      anotherField: '{$serviceId.body["field-with-dash"]}',
    },
  },
}

Text Before, After, and Between Tokens

You can combine tokens with text in the same string:

{
  service: {
    url: '{$env.HOST}/api/{$request.id}',
    body: {
      fullUrl: 'https://{$env.HOST}:{$env.PORT}/api/{$request.id}',
      prefixedId: 'user_{$request.body.userId}',
    },
  },
}

Examples

Authentication Flow

const result = await runOrchestration(
  [
    {
      id: 'authenticate',
      service: {
        url: 'https://auth.example.com/login',
        method: 'POST',
        body: {
          email: '$request.body.email',
          password: '$request.body.password',
        },
      },
    },
    {
      id: 'fetchProfile',
      dependsOn: ['authenticate'],
      service: {
        url: 'https://api.example.com/profile',
        method: 'GET',
        headers: {
          authorization: '{$authenticate.body.token}',
        },
      },
    },
  ],
  {
    request: {
      body: {
        email: '[email protected]',
        password: 'secret123',
      },
    },
  }
);

Multi-branch Orchestration

const result = await runOrchestration(
  [
    {
      id: 'getUser',
      service: {
        url: 'https://api.example.com/users/123',
        method: 'GET',
      },
    },
    {
      id: 'getPosts',
      dependsOn: ['getUser'],
      service: {
        url: 'https://api.example.com/posts',
        method: 'GET',
        query: { userId: '{$getUser.body.id}' },
      },
    },
    {
      id: 'getComments',
      dependsOn: ['getUser'],
      service: {
        url: 'https://api.example.com/comments',
        method: 'GET',
        query: { userId: '{$getUser.body.id}' },
      },
    },
    {
      id: 'aggregate',
      dependsOn: ['getPosts', 'getComments'],
      service: {
        url: 'https://aggregation.example.com/combine',
        method: 'POST',
        body: {
          posts: '{$getPosts.body}',
          comments: '{$getComments.body}',
        },
      },
    },
  ],
  { request: {} }
);

With Fallbacks

const result = await runOrchestration(
  [
    {
      id: 'criticalService',
      service: {
        url: 'https://api.example.com/data',
        method: 'GET',
        fallback: {
          data: { status: 'unavailable', items: [] },
        },
      },
    },
  ],
  { request: {} }
);

With Conditional Execution

Skip services based on conditions:

const result = await runOrchestration(
  [
    {
      id: 'authenticate',
      service: {
        url: 'https://auth.example.com/login',
        method: 'POST',
        body: {
          email: '{$request.body.email}',
          password: '{$request.body.password}',
        },
      },
    },
    {
      id: 'fetchAdminPanel',
      dependsOn: ['authenticate'],
      condition: (context) => {
        const auth = context.getAll().authenticate;
        return auth?.body?.role === 'admin';
      },
      service: {
        url: 'https://api.example.com/admin',
        method: 'GET',
        headers: {
          authorization: '{$authenticate.body.token}',
        },
      },
    },
    {
      id: 'fetchUserProfile',
      dependsOn: ['authenticate'],
      service: {
        url: 'https://api.example.com/profile',
        method: 'GET',
        headers: {
          authorization: '{$authenticate.body.token}',
        },
      },
    },
  ],
  {
    request: {
      body: {
        email: '[email protected]',
        password: 'secret123',
      },
    },
  }
);

With Error Strategies

Handle critical vs. optional failures differently:

const result = await runOrchestration(
  [
    {
      id: 'validatePayment',
      errorStrategy: 'throw', // Mark as critical
      service: {
        url: 'https://payment.example.com/validate',
        method: 'POST',
        body: { amount: '{$request.body.amount}' },
      },
    },
    {
      id: 'processPayment',
      dependsOn: ['validatePayment'],
      service: {
        url: 'https://payment.example.com/process',
        method: 'POST',
        body: { token: '{$validatePayment.body.token}' },
      },
    },
    {
      id: 'sendNotification',
      errorStrategy: 'silent', // Optional - don't block workflow
      service: {
        url: 'https://notifications.example.com/send',
        method: 'POST',
        body: { message: 'Payment processed' },
        fallback: { data: { status: 'skipped' } },
      },
    },
  ],
  { request: { body: { amount: 100 } } }
);

Requirements

• Node.js 18+ - Required for Undici and modern async/await support
• Undici - High-performance HTTP client (included as dependency)
• async-flow-orchestrator - Dependency orchestration engine
• undici-oidc-interceptor - Optional, for OIDC authentication support

Diagnostics and Monitoring

The engine emits Node.js diagnostic channel events for service lifecycle events, enabling monitoring and debugging without modifying your code.

Diagnostic Channels

Three diagnostic channels are available for subscription:

• workflow:service:start - Emitted when a service execution begins
• workflow:service:complete - Emitted when a service execution completes successfully
• workflow:service:error - Emitted when a service execution fails

Example: Monitoring Service Execution

import { diagnosticsChannel } from 'node:diagnostics_channel';

// Subscribe to service start events
const startChannel = diagnosticsChannel.channel('workflow:service:start');
startChannel.subscribe((message) => {
  console.log(`Service ${message.serviceId} starting...`);
  console.log(`URL: ${message.request.url}`);
  console.log(`Method: ${message.request.method}`);
});

// Subscribe to service completion events
const completeChannel = diagnosticsChannel.channel('workflow:service:complete');
completeChannel.subscribe((message) => {
  console.log(`Service ${message.serviceId} completed`);
  console.log(`Status: ${message.status}`);
  console.log(`Processing time: ${message.processingTime}ms`);
  if (message.fallbackUsed) {
    console.log('(Fallback was used)');
  }
});

// Subscribe to service error events
const errorChannel = diagnosticsChannel.channel('workflow:service:error');
errorChannel.subscribe((message) => {
  console.log(`Service ${message.serviceId} failed`);
  console.log(`Error: ${message.error.message}`);
  console.log(`Processing time: ${message.processingTime}ms`);
});

Diagnostic Message Format

All diagnostic messages include:

• serviceId - The unique service identifier
• request - Request details (url, method, headers, body, query)
• timestamp - When the event occurred (milliseconds since epoch)
• processingTime - How long the service took to execute (for complete/error events)
• status - HTTP status code (complete events only, null if fallback used)
• fallbackUsed - Whether a fallback response was used (complete events only)
• error - Error details including message and code (error events only)

Running Tests

npm test

Tests use Node.js native test runner (node --test).

License

MIT