FlowEngine

A no-code workflow orchestration engine for business process automation. Define workflows as JSON, and FlowEngine handles execution, dependency resolution, parallel processing, and state management.

Features

• DAG-based Workflows — Define step dependencies, FlowEngine resolves execution order
• Parallel Execution — Independent steps run concurrently across multiple workers
• Real-time Monitoring — Visual dashboard with live status updates
• Retry Policies — Configurable retry with exponential backoff
• Polyglot Architecture — NodeJS orchestrator + Python workers for flexibility
• Production Ready — Kubernetes deployment with auto-scaling

Architecture

flowchart TB
    subgraph Clients
        Dashboard["🖥️ Dashboard<br/>(Web UI)"]
        API["📡 REST Client<br/>(curl/Postman)"]
    end

    subgraph Orchestrator["⚡ Orchestrator (NodeJS)"]
        Scheduler["Scheduler<br/>• Queue tasks<br/>• Track dependencies"]
        StateMachine["State Machine<br/>• pending → queued<br/>• running → completed"]
        DAG["DAG Parser<br/>• Cycle detection<br/>• Topological sort"]
        Validation["Zod Validation<br/>• Schema check"]
    end

    subgraph Storage
        MongoDB[("🍃 MongoDB<br/>• Executions<br/>• Workflow snapshots<br/>• Step states")]
        Redis[("🔴 Redis<br/>• Job Queue (BRPOP)<br/>• Pub/Sub results")]
    end

    subgraph Workers["🐍 Workers (Python)"]
        W1["Worker 1"]
        W2["Worker 2"]
        WN["Worker N"]
        Handlers["Handlers<br/>• http<br/>• script<br/>• sleep<br/>• process"]
    end

    Dashboard <-->|REST API| Orchestrator
    API -->|POST /executions| Orchestrator
    
    Scheduler --> DAG
    Scheduler --> StateMachine
    DAG --> Validation
    
    Orchestrator -->|Save state| MongoDB
    Orchestrator -->|Push tasks| Redis
    
    Redis -->|BRPOP| W1
    Redis -->|BRPOP| W2
    Redis -->|BRPOP| WN
    
    W1 --> Handlers
    W2 --> Handlers
    WN --> Handlers
    
    Workers -->|Pub/Sub results| Redis
    Redis -->|Notify| Orchestrator

    style Orchestrator fill:#2d3748,stroke:#4a5568,color:#fff
    style Workers fill:#1a365d,stroke:#2c5282,color:#fff
    style Storage fill:#234e52,stroke:#285e61,color:#fff

Loading

Component Responsibilities

Component	Technology	Responsibility
Orchestrator	NodeJS/TypeScript	Parse workflows, manage state, resolve dependencies, schedule tasks
Workers	Python	Execute tasks (HTTP calls, scripts), report results
Redis	Redis 7	Job queue (BRPOP), result notification (Pub/Sub)
MongoDB	MongoDB 7	Persist executions, workflow snapshots, step states
Dashboard	HTML/CSS/JS	Visual workflow monitoring, real-time updates

Quick Start

Local Development

docker-compose up --build

# Dashboard: http://localhost:8081
# API: http://localhost:3001

Run a Workflow

Via Dashboard:

1. Open http://localhost:8081
2. Click "+ New" to see workflow options
3. Select a workflow (ETL Pipeline, Order Processing, etc.)
4. Watch real-time execution

Via API:

curl -X POST http://localhost:3001/api/executions \
  -H "Content-Type: application/json" \
  -d @examples/order-processing.json

Workflow Format

Workflows are JSON documents describing a directed acyclic graph (DAG):

{
  "workflow": {
    "id": "order-processing",
    "name": "Order Processing",
    "version": "1.0.0",
    "steps": [
      {
        "id": "validate",
        "type": "script",
        "config": { "handler": "echo" },
        "dependsOn": []
      },
      {
        "id": "process-a",
        "type": "script",
        "config": { "handler": "transform" },
        "dependsOn": ["validate"]
      },
      {
        "id": "process-b",
        "type": "script", 
        "config": { "handler": "echo" },
        "dependsOn": ["validate"]
      },
      {
        "id": "finalize",
        "type": "script",
        "config": { "handler": "echo" },
        "dependsOn": ["process-a", "process-b"]
      }
    ]
  }
}

This creates: validate → (process-a ‖ process-b) → finalize

Step Types

Type	Description
http	Make HTTP requests. Config: url, method, headers, body
script	Run Python handlers. Config: handler, args

Available Handlers

Handler	Description
echo	Returns input as output
transform	Transforms data (uppercase strings, double numbers)
validate	Validates required fields
sleep	Sleeps for specified duration
process	Simulates processing with random delay

API Reference

Method	Endpoint	Description
POST	/api/executions	Start workflow execution
GET	/api/executions	List all executions
GET	/api/executions/:id	Get execution details
GET	/api/executions/:id/steps	Get step-level status
GET	/api/health	Health check

Kubernetes Deployment

cd deploy/helm/flowengine

helm repo add bitnami https://charts.bitnami.com/bitnami
helm dependency update
helm install flowengine .

kubectl get pods -l app.kubernetes.io/name=flowengine

Auto-scaling

Workers scale automatically based on CPU:

autoscaling:
  enabled: true
  minReplicas: 2
  maxReplicas: 10
  targetCPUUtilizationPercentage: 70

Project Structure

flow-engine/
├── orchestrator/           # NodeJS orchestration service
│   └── src/
│       ├── api/            # REST endpoints (executions, health)
│       ├── core/           # DAG parser, state machine, scheduler
│       ├── models/         # Mongoose schemas (Execution, Workflow)
│       └── queue/          # Redis client (job queue, pub/sub)
├── worker/                 # Python worker service
│   └── src/
│       ├── handlers/       # Task handlers (http, script)
│       ├── consumer.py     # Redis BRPOP consumer loop
│       └── executor.py     # Task dispatch and execution
├── dashboard/              # Web UI for monitoring
├── deploy/helm/            # Kubernetes Helm charts
├── examples/               # Sample workflow JSON files
└── docker-compose.yml      # Local development setup

Adding Custom Handlers

Register new handlers in worker/src/handlers/script_handler.py:

@register("my_handler")
def my_handler(data: dict) -> Any:
    # Your business logic here
    result = process_data(data)
    return {"status": "success", "result": result}

Then use in workflows:

{
  "type": "script",
  "config": { "handler": "my_handler", "args": {"key": "value"} }
}