Feature: Workflow Engine
Relevant source files
The following files were used as context for generating this wiki page:The Workflow Engine is a core component of the Kodus Runtime, designed to orchestrate complex, multi-step event processing. It allows developers to define a sequence of operations—such as transformations, filtering, and routing—that are executed on an event as it passes through the system. This provides a flexible and powerful way to manage business logic and data flow for events.
The engine is implemented in Rust for high performance and is exposed to the Node.js environment through a native addon. This allows for both low-level, high-performance workflow definitions and high-level integration with JavaScript/TypeScript applications, including a REST API for remote management and execution.
Sources: apps/runtime/crates/core/src/workflow.rs, README.md, apps/runtime/crates/ts-addon/src/lib.rs
Architecture
The Workflow Engine is built around a few key Rust structs that define its structure and behavior. The WorkflowEngine is the central component that manages and executes workflows. Each workflow is described by a WorkflowDefinition, which contains a series of WorkflowSteps.
Sources: apps/runtime/crates/core/src/workflow.rs:9-62
Core Components
The following diagram illustrates the main structural components of the Workflow Engine.
This diagram shows that a WorkflowEngine holds multiple WorkflowDefinitions. Each definition is composed of WorkflowSteps and has its own ErrorHandling policy.
Sources: apps/runtime/crates/core/src/workflow.rs:9-62
Workflow Definition
A WorkflowDefinition is the blueprint for a workflow. It is a serializable struct that contains all the information needed to execute a series of steps on an event.
| Field | Type | Description |
|---|---|---|
id | Uuid | A unique identifier for the workflow. |
name | String | A human-readable name for the workflow. |
description | Option<String> | An optional description of the workflow's purpose. |
steps | Vec<WorkflowStep> | An ordered vector of steps that make up the workflow. |
error_handling | ErrorHandling | Configuration for how to handle errors during execution. |
retry_policy | crate::queue::RetryPolicy | Configuration for retrying a failed workflow execution. |
Sources: apps/runtime/crates/core/src/workflow.rs:9-16
Workflow Steps
Each WorkflowStep represents a single unit of work within a workflow. Steps are linked together via the next_step field, forming a chain of operations.
| Field | Type | Description |
|---|---|---|
id | String | A unique identifier for the step within the workflow. |
name | String | A human-readable name for the step. |
step_type | StepType | The type of operation this step performs. See Step Types. |
config | HashMap<String, serde_json::Value> | A key-value map for step-specific configuration. |
next_step | Option<String> | The id of the next step to execute. None indicates the end of the workflow. |
error_step | Option<String> | The id of a step to jump to if this step fails. |
Sources: apps/runtime/crates/core/src/workflow.rs:18-26
Step Types
The StepType enum defines the different kinds of operations a step can perform.
| Type | Description |
|---|---|
Transform | Modifies the event. The current implementation adds a processed_by tag to the event's metadata. |
Filter | Conditionally stops the workflow. The current implementation checks for a specific tag in the event's metadata, defined by config["filter_tag"]. |
Route | Adds routing information to the event. The current implementation adds a routed_to tag, with the value taken from config["route_to"]. |
ExternalCall | Simulates a call to an external system. The current implementation adds an external_call: completed tag. |
Delay | Pauses the workflow execution for a specified duration. The delay in milliseconds is read from config["delay_ms"]. |
Condition | Checks for a specific condition on the event. It can check for the presence of user_id or correlation_id. If the condition fails, the step returns an error. |
Sources: apps/runtime/crates/core/src/workflow.rs:28-36, apps/runtime/crates/core/src/workflow.rs:142-207
Execution Flow
The WorkflowEngine executes a workflow by processing its steps sequentially.
Basic Execution
The execute_workflow method iterates through the steps defined in the WorkflowDefinition. It starts with the first step, executes it, and then uses the next_step ID to find and execute the subsequent step until no next_step is defined.
The following flowchart shows the basic execution loop.
Execution with Retries and DLQ
The execute_workflow_with_retry method provides a more robust execution model. It wraps execute_workflow in a loop that incorporates the workflow's RetryPolicy. If an execution fails, it calculates a delay, waits, and retries. If the number of retries exceeds the configured maximum, and if the Dead Letter Queue (DLQ) is enabled, it sends a DeadLetterEvent to the DLQ.
The sequence diagram below illustrates this process.
Creating Workflows
WorkflowBuilder
To simplify the creation of WorkflowDefinition objects, the WorkflowBuilder provides a fluent API. It automatically chains steps together by managing the next_step pointers.
// Example from apps/runtime/crates/core/examples/workflow.rs
let workflow = WorkflowBuilder::new("payment-processing".to_string())
.with_description("Process payment events with validation and transformation".to_string())
.add_filter_step("validate-user".to_string(), "user_id".to_string())
.add_transform_step("enrich-payment".to_string(), {
let mut config = HashMap::new();
config.insert("enrich_type".to_string(), serde_json::Value::String("payment".to_string()));
config
})
.add_delay_step("processing-delay".to_string(), 1000) // 1 segundo
.with_error_handling(ErrorHandling {
max_retries: 3,
dead_letter_queue: true,
error_step: None,
})
.build();
Sources: apps/runtime/crates/core/src/workflow.rs:210-344, apps/runtime/crates/core/examples/workflow.rs:27-46
The following diagram shows a sample workflow created with the builder, demonstrating how different step types can be chained.
This diagram represents the workflow created in the code example above.
Sources: apps/runtime/crates/core/examples/workflow.rs:27-38
Node.js Integration & API
The Rust-based Workflow Engine is exposed to Node.js applications via a native addon, which is then used by the API layer.
Node.js Addon
The kodus-runtime N-API addon exposes the KodusRuntime class, which includes methods to interact with the workflow engine.
createWorkflow(name: string, description?: string): Promise<string>: Creates a new workflow and returns its ID.executeWorkflow(workflowId: string, eventData: EventData): Promise<EventData>: Executes a registered workflow with the given event data.
// Example from apps/api/examples/nodejs-usage.js
console.log('\n🔄 Criando workflow...');
const workflowId = await runtime.createWorkflow(
'payment-processing',
'Process payment events with validation'
);
console.log(`✅ Workflow criado com ID: ${workflowId}`);
console.log('🔄 Executando workflow...');
const inputEvent = { /* ... event data ... */ };
const processedEvent = await runtime.executeWorkflow(workflowId, inputEvent);
console.log('✅ Workflow executado com sucesso!');
Sources: apps/runtime/crates/ts-addon/src/lib.rs:125-194, apps/api/examples/nodejs-usage.js:65-87
REST API Endpoints
The API server provides REST endpoints for creating and executing workflows. All workflow routes require authentication.
| Endpoint | Method | Description |
|---|---|---|
POST /api/workflows | POST | Creates a new workflow. The request body should contain name and an optional description. Returns the workflowId on success. |
POST /api/workflows/:id/execute | POST | Executes the workflow specified by :id. The request body should be a JSON object representing the EventData to be processed. Returns the processed event data. |
Sources: apps/api/src/routes/workflows.ts:6-52
Summary
The Kodus Workflow Engine is a versatile system for defining and executing event processing logic. Its Rust core ensures high performance, while the Node.js addon and REST API provide easy integration and management. With support for various step types, configurable error handling, and retry policies, it is a powerful tool for building robust, event-driven applications.