@onlineapps/conn-orch-orchestrator

Workflow orchestration coordinator managing complete workflow execution for OA Drive microservices.

Features

• 🎯 Workflow coordination - Complete workflow execution management
• 🔀 Control flow handling - If/else, switch, foreach, fork-join patterns
• 📊 State management - Workflow state tracking and persistence
• 🚦 Step routing - Intelligent routing to next services
• 📈 Statistics tracking - Workflow execution metrics
• ⚡ Async processing - Non-blocking workflow execution

Installation

npm install @onlineapps/conn-orch-orchestrator

Quick Start

const { OrchestratorConnector } = require('@onlineapps/conn-orch-orchestrator');

const orchestrator = new OrchestratorConnector({
  amqpUrl: process.env.AMQP_URL,
  storageUrl: process.env.MINIO_URL
});

// Process workflow message
const result = await orchestrator.processWorkflowMessage({
  workflowId: 'wf-123',
  steps: [
    { service: 'validation', operation: 'validate' },
    { service: 'processing', operation: 'process' },
    { service: 'notification', operation: 'notify' }
  ],
  currentStep: 0,
  context: { /* workflow data */ }
});

// Get statistics
const stats = orchestrator.getStats();
console.log(`Processed: ${stats.processed}, Failed: ${stats.failed}`);

Cache Invalidation for Mutating Operations (operationMutatesLookup)

The orchestrator memoizes successful task-step results in the optional cache connector under keys of the form workflow:step:<step.service>:<operation>:<hash> so that idempotent reads inside the same workflow can be reused. Mutating operations (writes, deletes, RPC side effects) must NOT be served from this cache, and any cached reads against the same target service must be purged the moment a write succeeds.

To enable that behavior, pass an operationMutatesLookup hook to the constructor. The hook is typically wired by @onlineapps/service-wrapper from each service's operations.json map.

Signature

/**
 * @param {string} executingServiceName Name of the service running the cookbook.
 * @param {Object} step                 The cookbook step about to execute.
 * @returns {{ mutates?: boolean, resource_type?: string|null }|null|undefined}
 */
operationMutatesLookup(executingServiceName, step)

Return shape

• { mutates: true } — force a read-cache bypass for this step AND, after the handler returns success (status 2xx), invoke cache.deleteByPattern('workflow:step:<step.service>:*') to purge stale reads.
• { mutates: false } (or undefined / null) — treat as a read; the orchestrator may reuse cache.get and store fresh results via cache.set.
• The resource_type field is informational and is not consumed by the orchestrator (yet).

Error contract

If the hook throws, _stepMutatesFlag re-throws the same error. The orchestrator does NOT silently swallow lookup failures — per architecture-principles.mdc §4 we fail loudly so misconfigured operations.json is caught at workflow runtime, not by data drift.

Fallback behavior without the hook

If operationMutatesLookup is omitted or is not a function, the orchestrator only treats a step as mutating when the cookbook step itself sets step.mutates === true. Otherwise every task step is eligible for read-cache reuse.

Example

const WorkflowOrchestrator = require('@onlineapps/conn-orch-orchestrator');

const operations = {
  'get-invoice': { mutates: false, resource_type: 'invoice' },
  'update-invoice': { mutates: true, resource_type: 'invoice' }
};

const orchestrator = new WorkflowOrchestrator({
  mqClient,
  registryClient,
  invoker,
  cookbook,
  cache,
  logger,
  defaultTimeout: 30000,
  operationMutatesLookup: (executingServiceName, step) => {
    const spec = operations[step.operation];
    return {
      mutates: !!(spec && spec.mutates === true),
      resource_type: spec ? spec.resource_type : null
    };
  }
});

Configuration

| Variable | Description | Default | |----------|-------------|---------| | AMQP_URL | RabbitMQ connection URL | required | | STORAGE_URL | MinIO storage URL | required | | WORKFLOW_TIMEOUT | Max workflow execution time (ms) | 300000 | | MAX_RETRIES | Maximum retry attempts | 3 |

Control Flow Patterns

If/Else

{
  type: 'if',
  condition: { field: 'status', operator: 'equals', value: 'active' },
  then: { service: 'activeHandler' },
  else: { service: 'inactiveHandler' }
}

Switch/Case

{
  type: 'switch',
  field: 'priority',
  cases: {
    'high': { service: 'urgentHandler' },
    'normal': { service: 'standardHandler' },
    'low': { service: 'batchHandler' }
  },
  default: { service: 'defaultHandler' }
}

Foreach

{
  type: 'foreach',
  items: 'context.items',
  step: { service: 'itemProcessor' }
}

Fork-Join

{
  type: 'fork-join',
  parallel: [
    { service: 'validator' },
    { service: 'enricher' },
    { service: 'scorer' }
  ],
  join: { service: 'aggregator' }
}

API Reference

OrchestratorConnector

Main orchestrator class for workflow execution.

Methods

• processWorkflowMessage(message) - Process a workflow message
• processStep(message) - Process a single workflow step
• handleControlFlow(step, message) - Handle control flow steps
• routeToNextService(message, nextService, result, nextIndex) - Route to next service
• completeWorkflow(message, result) - Complete workflow execution
• handleWorkflowFailure(message, error) - Handle workflow failures
• getStats() - Get execution statistics
• resetStats() - Reset statistics

Testing

# Run all tests
npm test

# Run unit tests
npm run test:unit

# Run integration tests
npm run test:integration

Error Classification

The WorkflowOrchestrator classifies errors before retry decisions using error.type and error.statusCode:

| Error Type | Retry? | Action | |---|---|---| | TRANSIENT (500, 502, 503) | Yes | Retry with exponential backoff | | TIMEOUT (504, 408) | Yes | Retry with backoff | | BUSINESS (404, 403, 409) | No | Straight to DLQ | | VALIDATION (400, 422) | No | Straight to DLQ |

DLQ records include error_type, error_code, status_code, and details[] for monitoring. This works in tandem with the ApiMapper's error context preservation -- see Error Handling Standard.

Documentation

• Connectors Architecture
• Workflow Architecture
• Error Handling Standard

Version History

• 1.0.101: Error classification in catch block -- BUSINESS/VALIDATION skip retries
• 1.0.13: Added variable reference resolution in step inputs