Flow Engine - Simple Workflow Framework for Node.js

Quick Start

Installation

npm install node-workflow-engine
# or
yarn add node-workflow-engine

Basic Usage

import { createFlow } from 'node-workflow-engine';

// Create a simple workflow
const flow = createFlow()
  .setCommonPayload({ apiKey: 'your-api-key' })
  .step('validate', async (data) => {
    if (!data.email) throw new Error('Email required');
    return { isValid: true };
  })
  .step('process', async (data) => {
    // Your processing logic here
    return { processed: true };
  });

// Execute the workflow
const result = await flow.execute({ email: 'user@example.com' });
console.log(result.success); // true
console.log(result.data); // { isValid: true, processed: true }

Framework Integration

Express.js Integration

import express from 'express';
import { createFlow, expressFlow } from 'node-workflow-engine';

const app = express();
app.use(express.json());

// Create user registration flow
const userFlow = createFlow()
  .setCommonPayload({ timestamp: new Date().toISOString() })
  .step('validate', async (data) => {
    if (!data.email || !data.password) {
      throw new Error('Email and password required');
    }
    return { isValid: true };
  })
  .step('createUser', async (data) => {
    const user = { id: 1, email: data.email };
    return { user };
  })
  .step('sendEmail', async (data) => {
    console.log(`Welcome email sent to ${data.user.email}`);
    return { emailSent: true };
  });

// Use as Express middleware
app.post('/api/users', expressFlow(userFlow), (req, res) => {
  res.json({
    success: true,
    user: req.flowResult.data.user
  });
});

app.listen(3000);

Fastify Integration

import Fastify from 'fastify';
import { createFlow } from 'node-workflow-engine';

const fastify = Fastify();

const orderFlow = createFlow()
  .setCommonPayload({ currency: 'USD' })
  .step('validateOrder', async (data) => {
    if (!data.items?.length) throw new Error('No items');
    return { valid: true };
  })
  .step('calculateTotal', async (data) => {
    const total = data.items.reduce((sum, item) => sum + item.price, 0);
    return { total };
  })
  .step('processPayment', async (data) => {
    return { paymentId: 'pay_123' };
  });

fastify.post('/api/orders', async (request, reply) => {
  const result = await orderFlow.execute(request.body);
  
  if (result.success) {
    return { order: result.data };
  } else {
    reply.code(400);
    return { error: result.error };
  }
});

Koa Integration

import Koa from 'koa';
import Router from 'koa-router';
import { createFlow } from 'node-workflow-engine';

const app = new Koa();
const router = new Router();

const authFlow = createFlow()
  .step('validateCredentials', async (data) => {
    if (!data.email || !data.password) {
      throw new Error('Invalid credentials');
    }
    return { valid: true };
  })
  .step('generateToken', async (data) => {
    const token = `jwt_${Math.random().toString(36).substr(2, 9)}`;
    return { token };
  });

router.post('/api/auth', async (ctx) => {
  const result = await authFlow.execute(ctx.request.body);
  
  if (result.success) {
    ctx.body = { token: result.data.token };
  } else {
    ctx.status = 401;
    ctx.body = { error: result.error };
  }
});

Standalone Usage

import { createFlow } from 'node-workflow-engine';

// Create a simple workflow
const workflow = createFlow()
  .setCommonPayload({
    environment: 'production',
    version: '1.0.0'
  })
  .step('step1', async (data) => {
    console.log('Executing step 1');
    return { step1Result: 'completed' };
  })
  .step('step2', async (data) => {
    console.log('Executing step 2');
    return { step2Result: 'completed' };
  });

// Execute the workflow
const result = await workflow.execute({
  input: 'test data'
});

console.log('Workflow result:', result);

Advanced Features

🔧

Middleware Support

Add authentication, logging, and other middleware that runs before each step.

flow.use(async (data, context) => {
  // Authentication middleware
  if (!data.apiKey) throw new Error('Unauthorized');
  context.metadata.userId = 'user_123';
});

🔄

Retry Logic & Timeouts

Built-in retry with exponential backoff and configurable timeouts.

flow.step('externalAPI', async (data) => {
  const response = await fetch('https://api.example.com/data');
  return { data: await response.json() };
}, { retries: 3, timeout: 10000 });

🔗

Step Dependencies

Define dependencies between steps to control execution order.

flow.step('fetchOrders', async (data) => {
  return { orders: await getOrders(data.user.id) };
}, { dependencies: ['fetchUser'] });

📊

Execution Tracking

Know exactly which steps executed, execution time, and detailed metadata.

const result = await flow.execute(inputData);
console.log(result.steps);        // ['validate', 'process', 'save']
console.log(result.executionTime); // 245
console.log(result.metadata);     // { requestId: 'exec_123' }

Real-World Examples

E-commerce Order Processing

Complete Order Flow

const orderFlow = createFlow()
  .setCommonPayload({ 
    currency: 'USD',
    taxRate: 0.08,
    shippingRate: 5.99
  })
  .step('validateOrder', async (data) => {
    if (!data.items?.length) throw new Error('No items in order');
    if (!data.customerId) throw new Error('Customer ID required');
    return { orderValid: true };
  })
  .step('checkInventory', async (data) => {
    for (const item of data.items) {
      const available = await checkStock(item.productId);
      if (available < item.quantity) {
        throw new Error(`Insufficient stock for ${item.productId}`);
      }
    }
    return { inventoryChecked: true };
  })
  .step('calculatePricing', async (data) => {
    const subtotal = data.items.reduce((sum, item) => sum + item.price, 0);
    const tax = subtotal * data.taxRate;
    const shipping = subtotal > 50 ? 0 : data.shippingRate;
    const total = subtotal + tax + shipping;
    return { subtotal, tax, shipping, total };
  })
  .step('processPayment', async (data) => {
    const payment = await stripe.charges.create({
      amount: Math.round(data.total * 100),
      currency: data.currency,
      source: data.paymentToken
    });
    return { paymentId: payment.id };
  })
  .step('createOrder', async (data) => {
    const order = await db.orders.create({
      customerId: data.customerId,
      items: data.items,
      total: data.total,
      paymentId: data.paymentId,
      status: 'confirmed'
    });
    return { order };
  })
  .step('sendConfirmation', async (data) => {
    await emailService.send({
      to: data.customer.email,
      template: 'order-confirmation',
      data: { order: data.order }
    });
    return { emailSent: true };
  });

// Use in Express route
app.post('/api/orders', expressFlow(orderFlow), (req, res) => {
  res.json({ order: req.flowResult.data.order });
});

API Reference

createFlow()

Creates a new workflow instance.

const flow = createFlow();

.setCommonPayload(payload)

Sets data available to all steps.

flow.setCommonPayload({ apiKey: 'your-key', version: '1.0' });

.use(middleware)

Adds middleware that runs before each step.

flow.use(async (data, context) => {
  console.log(`Executing step with data:`, data);
});

.step(id, handler, options?)

Adds a step to the workflow.

flow.step('process', async (data) => {
  return { processed: true };
}, { retries: 3, timeout: 5000 });

.execute(inputData)

Executes the workflow with input data.

const result = await flow.execute({ email: 'user@example.com' });