@hemia/workflow-engine

Motor de flujos de trabajo flexible y extensible desarrollado por Hemia. Orquesta pasos definidos en JSON, integra servicios HTTP, pausa por webhooks, evalúa reglas y maneja errores con rutas dedicadas.

Instalación

npm install @hemia/workflow-engine

Características clave

• Definición de workflows en JSON.
• Nodos soportados: http, wait-for-webhook, validator (+ nodos personalizados).
• Trigger por webhook con autenticación Bearer (startFromWebhook).
• Configuración global y plantillas con ${config.*} y {{ ... }}.
• Pausa y reanudación (WAITING) con wait-for-webhook.
• Manejo de errores: error.next por paso y workflow.errorHandlers.
• Reintentos automáticos en HTTP con params.retry y política global (retryPolicy).
• Templates con JEXL y transforms globales (pipes).
• Persistencia de contexto (save/load) inyectable.

Uso básico

import { WorkflowEngine, NodeRegistry } from "@hemia/workflow-engine";
import { httpNode } from "@hemia/workflow-node-http";
import { waitForWebhookNode } from "@hemia/workflow-node-wait-for-webhook";
import { validatorNode } from "@hemia/workflow-node-validator";
import workflow from "./workflows/mi-flujo.json";

// Registro de nodos
const registry = new NodeRegistry();
registry.register("http", httpNode);
registry.register("wait-for-webhook", waitForWebhookNode);
registry.register("validator", validatorNode);

// Persistencia (implementa según tu app)
const saveContext = async (ctx: any, execId: string, workflowId: string, status?: "RUNNING"|"WAITING"|"FAILED"|"COMPLETED") => { /* ... */ };
const loadContext = async (execId: string) => ({ /* ctx persistido */ });

// Crear y ejecutar
const engine = new WorkflowEngine(workflow, "exec-123", registry, saveContext, loadContext);
await engine.run();

Trigger por Webhook + Bearer

Define el trigger en tu workflow y usa startFromWebhook en tu servidor HTTP.

{
  "trigger": { "type": "webhook", "path": "/webhooks/vacation-request-flow/start", "authentication": { "type": "bearer" } }
}

Ejemplo con Express:

import express from "express";
import { WorkflowEngine, NodeRegistry } from "@hemia/workflow-engine";
// ...registra nodos y prepara save/load...

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

app.post("/webhooks/vacation-request-flow/start", async (req, res) => {
  const engine = new WorkflowEngine(workflow, "exec-abc", registry, saveContext, loadContext);
  const { accepted, reason } = await engine.startFromWebhook({ headers: req.headers as any, path: req.path, body: req.body });
  if (!accepted) {
    const code = reason === "MISSING_BEARER" || reason === "INVALID_TOKEN" ? 401 : 404;
    return res.status(code).json({ reason });
  }
  return res.sendStatus(202);
});

Configuración global e interpolación

• Incluye un bloque config en el workflow; el engine lo expone en variables.config.
• En strings puedes usar:
  • ${config.apiUrl}/path
  • "{{config.apiUrl}}/path" (JEXL)

{
  "config": {
    "apiUrl": "https://api.example.com",
    "retryPolicy": { "maxAttempts": 3, "backoff": { "type": "exponential", "initialDelay": "1000" } }
  }
}

Templates y funciones con JEXL (transforms)

Registra transforms globales en @hemia/workflow-core y úsalos con la sintaxis de pipe:

import { registerJexlTransforms } from "@hemia/workflow-core";

registerJexlTransforms({
  getManagerEmail: (employeeId: string) => `manager+${employeeId}@example.com`,
  generateApprovalUrl: (requestId: string) => `${process.env.APP_URL}/approve/${requestId}`
});

En el workflow:

{ "to": "{{ input.employeeId | getManagerEmail }}" }

Validaciones con el nodo validator

Valida datos de entrada u otras variables con expresiones JEXL.

{
  "id": "validateInput",
  "type": "validator",
  "params": { "rules": [
    "input.startDate < input.endDate",
    "input.employeeId != null",
    "input.email != null"
  ]},
  "next": [{ "id": "checkVacationBalance" }]
}

Reintentos automáticos en HTTP

• Actívalos por paso con "params.retry": true.
• Configúralos globalmente en config.retryPolicy (maxAttempts y backoff: fixed|exponential, initialDelay en ms).
• Ejemplo:

{
  "id": "saveRequest",
  "type": "http",
  "params": {
    "method": "POST",
    "url": "${config.apiUrl}/vacations",
    "body": "{{input}}",
    "retry": true,
    "timeout": 30000
  },
  "error": { "next": "handleError" }
}

Manejo de errores (routing)

• Por paso: usa "error.next" para redirigir a un handler cuando falle.
• Global: define "errorHandlers" en el workflow.

{
  "errorHandlers": {
    "handleError": {
      "id": "handleError",
      "type": "http",
      "params": {
        "method": "POST",
        "url": "${config.errorHandlerUrl}",
        "body": { "workflow": "{{context.workflowId}}", "error": "{{error}}", "context": "{{context}}" }
      }
    }
  }
}

El engine intentará:

1. Redirigir a step.error.next si existe.
2. Usar un handler global definido en errorHandlers.
3. Si no hay ruta de error, marca FAILED y detiene.

Pausa y reanudación (wait-for-webhook)

• Un paso "wait-for-webhook" pausará la ejecución (status WAITING).
• Reanuda luego con engine.resume(fromStepId) o invocando el webhook que tú manejes.

Ejemplo de pausa:

{
  "id": "waitExtraData",
  "type": "wait-for-webhook",
  "params": { "stepId": "waitExtraData", "reason": "extra-data" }
}

Estados del workflow

• RUNNING: ejecutando.
• WAITING: pausado esperando webhook.
• FAILED: error sin handler o tras agotar reintentos.
• COMPLETED: flujo finalizado.

Notas y buenas prácticas

• Establece variables.input en el contexto inicial antes de run() o via startFromWebhook(body).
• En nodos HTTP usa saveAs para referenciar datos en condiciones: p.ej., "balance.data.availableDays".
• Usa timeouts numéricos