Declarative Workflows & Fluent DSL

Fluent DSL Syntax

Workflows are built using fluent chainable methods that produce a validated DAG specification:

import { workflow, map, when, choose, fanOut, branch, mockApi, rest } from "@vaakloom/engine";

export const checkoutFlow = workflow("checkout")
  .route("POST", "/checkout")
  .search("orderId", "tenantId")
  .tags("ecom", "orders")
  .steps(
    // Step 1: Initialize payload
    map("init", {
      mergeInput: true,
      set: {
        timestamp: "{{now}}",
        tenant: "{{attrs.tenantId}}",
      },
    }),

    // Step 2: Conditional execution (if/then/else)
    when("{{input.tier}} === 'enterprise'", [
      map("applyDiscount", { mergeInput: true, set: { discount: 0.20 } }),
    ], [
      map("standardRate", { mergeInput: true, set: { discount: 0.05 } }),
    ]),

    // Step 3: Multi-way switch/case
    choose("{{input.paymentMethod}}", {
      card: [mockApi("chargeCard", { success: true, fee: 1.5 })],
      crypto: [mockApi("chargeCrypto", { txHash: "0xabc123", fee: 0.5 })],
    }, [
      map("unsupportedPayment", { set: { error: "Unknown payment method" } })
    ]),

    // Step 4: Parallel Fan-out and Join
    fanOut("enrichment", [
      branch("profile", rest("getProfile", {
        url: "https://api.crm.internal/users/{{input.userId}}",
        method: "GET",
      })),
      branch("notifications", mockApi("slackNotify", { sent: true })),
    ], {
      id: "consolidate",
      kind: "merge",
      config: { target: "enrichment" },
    })
  )
  .build();

Built-in Step Types

Step Kind	Helper	Description
map	map(id, config)	Transforms JSON structures, sets keys, evaluates templates, and merges input.
rest	rest(id, config)	Makes HTTP/REST requests via native fetch with header and timeout support.
mock-api	mockApi(id, response)	Simulates external APIs with configurable payload and synthetic latency.
delay	delay(id, ms)	Non-blocking reactive pause (e.g., rate limit pacing or exponential backoff).
merge	merge(id, config)	Deep merges multiple step outputs into a single target dictionary.
passthrough	passthrough(id)	Passes input to output without modification (useful for trace checkpoints).

Control Flow & Resiliency Patterns

1. Parallel Fan-out & Join

Run concurrent branches with Promise.all execution semantics and aggregate results:

fanOut("parallelBranch", [
  branch("authService", mockApi("auth", { valid: true })),
  branch("fraudService", mockApi("fraud", { riskScore: 12 })),
], {
  id: "joinStep",
  kind: "map",
  config: { set: { approved: true, results: "{{input}}" } }
})

2. Automatic Retries & Backoff

Configure granular retry policies on individual steps to handle transient network errors:

rest("paymentGateway", {
  url: "https://api.stripe.com/v1/charges",
  method: "POST",
}, {
  retry: {
    max: 3,           // Maximum attempts
    delayMs: 100,     // Initial delay
    backoff: 2.0,     // Multiplier per retry (100ms -> 200ms -> 400ms)
  },
  timeoutMs: 3000,    // Hard step timeout
})

3. Soft Failures (Error Tolerance)

To prevent a non-critical step from failing the entire workflow, specify onError: "continue":

rest("analyticsPush", {
  url: "https://analytics.internal/track",
  method: "POST",
}, {
  onError: "continue", // Workflow proceeds even if analytics is down
})

Execution Modes: Batch Jobs vs. Real-Time APIs

1. Real-Time Synchronous APIs (Zero Chunking Overhead)

For synchronous request/response APIs, workflows execute immediately with direct in-memory passing through steps. Sequential and parallel branches run without chunking overhead, ensuring sub-millisecond execution dispatch.

2. Batch Workflows with Adaptive Chunking

When orchestrating high-volume collections, use the batch() node. The engine dynamically computes adaptive chunk sizes and concurrency based on individual record payload weights and system memory headroom:

import { defineWorkflow, batch, step, mapAdapter } from "@vaakloom/engine";

export const batchEnrichment = defineWorkflow({
  workflowId: "bulk-process",
  nodes: [
    batch("processRecords", "input.records", {
      id: "transformRecord",
      handler: async ({ input }) => {
        // Individual record execution with error isolation
        return { ...input, enriched: true };
      },
    }, 10), // Optional maximum concurrency
  ],
});

3. Failure Isolation & Dead-Letter Queue (DLQ)

In batch jobs and async/subscriber workflows, a failure in a single record does not crash the batch. Failed records are isolated into the DeadLetterLedger with full diagnostic context (payload, error type, message, stack, timestamp) for inspection and one-click replay, while valid records continue seamlessly.

4. Independent Modular Architecture

All core resilience and telemetry features are completely decoupled and available as independent, standalone components:

• ResourceGovernor: Standalone system & process resource monitor (75% cap, 25% GC headroom, proactive GC at 70%).
• DeadLetterLedger: Standalone dead-letter queue and replay ledger with bounded memory retention.
• EventHub: Standalone 100% capture circular ring buffer with real-time subscription & SSE broadcasting.

Dynamic Template Interpolation

VaakLoom provides mustache-style template resolution across step contexts:

Expression	Description
{{input.field}}	Current step’s input payload attribute.
{{runInput.field}}	Initial workflow invocation payload.
{{attrs.field}}	Injected request context attributes (e.g. tenantId, userId).
{{steps.stepId.output}}	Output of any previously executed step by ID.
{{requestId}}	Unique execution request identifier for tracing.
{{now}}	Current ISO timestamp (UTC).