Skip to content

Engine Architecture

The workflow engine is a NestJS application built around an event-driven core. Database changes trigger events that are dispatched to handlers through a registry pattern. This architecture enables clean separation of lifecycle phases, testability, and recovery from restarts.

Module Structure

flowchart LR
    src["src/"]

    src --> engine["engine/<br/>Core engine"]
    engine --> core["core-engine/<br/>Event handler registry,<br/>lifecycle executors"]
    core --> eh["event-handler/<br/>Base handler, registry,<br/>event types"]
    core --> lc["lifecycle/<br/>Workflow and step executors"]
    core --> cms["cms-client/<br/>CMS GraphQL client<br/>(entity acquisition, locking)"]
    core --> ea["event-aggregator/<br/>RxJS-based event pub/sub,<br/>DB subscribers, startup replay"]

    src --> resources["resources/<br/>REST API modules"]
    resources --> bb["blackboard/<br/>Job polling and result pushing"]
    resources --> fb["function-blocks/<br/>FB registration and resolution"]
    resources --> wk["workers/<br/>Worker registration and health"]
    resources --> wd["workflow-definition/<br/>Publish, read, soft-delete, restore"]
    resources --> we["workflow-execution/<br/>Execution lifecycle"]
    resources --> hl["health/<br/>Health checks"]
    resources --> sc["schema/<br/>Schema endpoints"]
    resources --> vr["version/<br/>Version and compatibility"]
    resources --> lic["license/<br/>License reporting endpoint"]

    src --> model["model/<br/>Domain models"]
    model --> ent["entities/<br/>MikroORM entity classes"]
    model --> exe["execution/<br/>Execution types, status enums, schemas"]

    src --> pers["persistence/<br/>Repositories and DB-specific code"]
    src --> val["validator/<br/>JSON Schema validation, JSON Forms,<br/>compatibility checks, publish version floor"]
    src --> licsrc["license/<br/>License model, quotas,<br/>workflow-definition admission"]
    src --> log["logging/<br/>Logger module, HTTP interceptor"]
    src --> common["common/<br/>Cross-cutting HTTP concerns (@BodyLimit)"]
    src --> utils["utils/<br/>JMESPath, semver, Mermaid,<br/>typing utilities"]

Event-Driven Core

The engine processes workflow executions through an event-driven pipeline:

graph LR
    DB["DB Flush<br/>(MikroORM)"] --> Sub["DB Subscribers"]
    Sub --> EA["Event<br/>Aggregator"]
    EA --> CE["Core Engine"]
    CE --> Reg["Handler<br/>Registry"]
    Reg --> H["Handler"]
    H --> DB

DB Subscribers

MikroORM entity subscribers detect changes to workflow executions and job results on flush. They create events:

  • WorkflowExecutionUpdatedEvent – When a workflow execution status changes
  • JobResultUpdatedEvent – When a job result is created or updated

Event Aggregator

The EventAggregatorService is an RxJS Subject-based pub/sub system. It receives events from DB subscribers and forwards them to the core engine. It also handles event buffering and deduplication.

Core Engine

CoreEngineService subscribes to the event aggregator and dispatches events to the handler registry. The handleOrDispatchEvent() method:

  1. Looks up the appropriate handler via the registry
  2. Runs the handler within a database transaction
  3. Flushes the entity manager (which may trigger new events, continuing the cycle)

Handler Registry

The EventHandlerRegistry maps event types to handler instances. It selects handlers based on:

  1. Event class – WorkflowExecutionUpdatedEvent or JobResultUpdatedEvent
  2. Event conditions – The handler’s eventConditions() (workflow status, job status, etc.)

All handlers are registered during initEngine() in CoreEngineService.

Workflow Execution Handlers

Each state transition has a dedicated handler:

State Handler Responsibility
NEW NewWorkflowExecutionHandler Resolve FBs, validate definition → VALID
VALID ValidWorkflowExecutionHandler Check preconditions → READY
READY ReadyWorkflowExecutionHandler Create ACQUIRE jobs → RESOURCE_DISCOVERY
RESOURCE_DISCOVERY ResourceDiscoveryWorkflowExecutionHandler Wait for acquire results
(acquire job results) JobResultAcquireResultHandler Process acquire results → RESOURCES_DISCOVERED
RESOURCES_DISCOVERED ResourceDiscoveredWorkflowExecutionHandler Prepare scheduling → SCHEDULED
SCHEDULED ScheduledWorkflowExecutionHandler Send lock request → LOCKING
LOCKING LockingWorkflowExecutionHandler Process lock response → LOCKED
LOCKED LockedWorkflowExecutionHandler Start execution → RUNNING
RUNNING RunningWorkflowExecutionHandler (most work arrives via job results) Process step results, advance workflow
ERROR ErrorWorkflowExecutionHandler Classify the failure via the RollbackCoordinator → FAILED_SAFE (nothing non-pure executed), ROLLBACK (all executed non-pure FBs reversible, first wave created) or FAILED_UNSAFE (irreversible FB executed)
ROLLBACK RollbackWorkflowExecutionHandler Keep the rollback phase progressing and resume-safe: recreate the current wave after a restart, finish the phase → FAILED_SAFE, or halt on a failed rollback job → FAILED_UNSAFE. Waves normally advance via JobResultRollbackResultHandler (see Retry & Rollback)
COMPLETED CompletedWorkflowExecutionHandler Apply DB updates, release locks → COMPLETED_ACK
FAILED_SAFE FailedSafeWorkflowExecutionHandler Release locks → FAILED_SAFE_ACK
FAILED_UNSAFE FailedUnsafeWorkflowExecutionHandler Keep locks (deliberately no unlock), log at error level → FAILED_UNSAFE_ACK
COMPLETED_ACK CompletedAckWorkflowExecutionHandler Terminal. No-op — registered so the terminal state has an owner
FAILED_SAFE_ACK FailedSafeAckWorkflowExecutionHandler Terminal. No-op
FAILED_UNSAFE_ACK FailedUnsafeAckWorkflowExecutionHandler Terminal. No-op — note the execution still holds its CMS locks (why)

Job Result Handlers

Scope Handler Responsibility
Acquire JobResultAcquireResultHandler Process acquire results → RESOURCES_DISCOVERED
Running JobResultRunningResultHandler Process execution results, advance steps
Rollback JobResultRollbackResultHandler Process rollback results
Success ACK JobResultSuccessAckHandler Post-success cleanup
Failed ACK JobResultFailedAckHandler Post-failure cleanup

Startup Recovery

The InitEventService runs on application startup. It loads all non-terminal workflow executions and pending job results from the database and re-publishes their events to the aggregator. This ensures that executions interrupted by an engine restart are resumed correctly.

Key NestJS Modules

Module Purpose Key dependencies
AppModule Root module wiring all engine capabilities. Imports all modules below.
CoreEngineModule Orchestrates lifecycle and event handling. EventAggregatorModule, CmsClientModule
EventAggregatorModule Event pub/sub and buffering. Used by engine, subscribers, and init.
DbEventSubscriberModule MikroORM subscribers that emit engine events. EventAggregatorModule
InitEventModule Startup recovery and event replay. EventAggregatorModule
BlackboardModule Job polling and result endpoints. REST resources layer.
FunctionBlocksModule Function block registration and resolution. REST resources layer.
WorkflowDefinitionModule Workflow CRUD and validation. REST resources layer.
WorkflowExecutionModule Execution lifecycle endpoints. REST resources layer.
WorkersModule Worker registration and health tracking. REST resources layer.
JobCleanupModule Cleanup of stale jobs and timeouts. Scheduler/maintenance.
CmsClientModule CMS GraphQL client for entities and locks. Consumed by engine.
SchemaModule Serves the generated RootWorkflow JSON Schema. REST resources layer.
VersionModule Engine version and SDK compatibility checks. REST resources layer.
LicenseModule Active license, quota guard, and admission. Provides License by DI to every consumer.
LoggerModule Logger provider and HTTP logging interceptor. Global.