From 1eaef67c36a1a1eae6d574d2d44f2a3bdfb43d4c Mon Sep 17 00:00:00 2001 From: Luca <102960244+Luca-Timo@users.noreply.github.com> Date: Tue, 23 Jun 2026 02:05:34 +0200 Subject: [PATCH] feat(workflows): execution engine core + registry + tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Graph executor that walks nodes/edges per run: trigger, condition/branch (registered conditions → yes/no edge), bounded loop (counter in context + maxIterations cap), wait (status=waiting + wake_at for the scheduler), gate (status=waiting; resumed via confirm/deny edge), action/webhook (registered handlers). emitWorkflowEvent creates one idempotent run per matching enabled workflow (unique dedup_key) and fails CLOSED if the flag system is unavailable; never throws into callers (safe to call after commit). Every node records a workflow_run_steps row. Registry seeds primitive conditions (always/never/expr) + actions (noop/log/set_context). Integration test covers loop+wait resume, gate confirm, and dedup. --- .../integration/workflowEngine.test.js | 137 +++++++++ backend/src/services/workflows/engine.js | 284 ++++++++++++++++++ backend/src/services/workflows/index.js | 16 + backend/src/services/workflows/registry.js | 61 ++++ 4 files changed, 498 insertions(+) create mode 100644 backend/__tests__/integration/workflowEngine.test.js create mode 100644 backend/src/services/workflows/engine.js create mode 100644 backend/src/services/workflows/index.js create mode 100644 backend/src/services/workflows/registry.js diff --git a/backend/__tests__/integration/workflowEngine.test.js b/backend/__tests__/integration/workflowEngine.test.js new file mode 100644 index 00000000..65119b6f --- /dev/null +++ b/backend/__tests__/integration/workflowEngine.test.js @@ -0,0 +1,137 @@ +/** + * Workflow engine — graph execution integration tests. + * + * Exercises the engine against a real (temp SQLite) DB with migration 142 + * applied: branching, bounded loops, wait pauses + scheduler-style resume, + * gate pauses + confirm/deny resume, dedup idempotency, and step recording. + */ +const { bootCrmDb } = require('./helpers/crmDb'); + +let db; +let cleanup; +let engine; + +async function makeWorkflow({ nodes, edges, trigger = 'test.event', enabled = true }) { + const ins = await db('workflows').insert({ name: 'wf', trigger_type: trigger, version: 1, enabled }); + const workflowId = ins[0]; + for (const n of nodes) { + await db('workflow_nodes').insert({ + workflow_id: workflowId, version: 1, node_key: n.key, type: n.type, + config: JSON.stringify(n.config || {}), + }); + } + for (const e of edges) { + await db('workflow_edges').insert({ + workflow_id: workflowId, version: 1, from_node: e.from, from_handle: e.handle || null, to_node: e.to, + loop_back: e.loopBack || false, + }); + } + return workflowId; +} + +beforeAll(async () => { + ({ db, cleanup } = await bootCrmDb()); + // Engine requires the singleton db — require AFTER bootCrmDb wired the test path. + engine = require('../../src/services/workflows'); + // Enable the workflows flag so emitWorkflowEvent doesn't fail closed. + await db('feature_flags').insert({ key: 'workflows', value: true }); +}); + +afterAll(async () => { await cleanup(); }); + +describe('workflow engine', () => { + test('condition + bounded loop + wait pauses, resumes to completion', async () => { + // trigger → set paid=false → condition(paid?) --no--> loop(max2) + // loop --loop--> reminder(noop) → wait → (back to condition) + // loop --exit--> lateFee(noop) → end + // condition --yes--> lateFee (paid path, not taken here) + const wfId = await makeWorkflow({ + nodes: [ + { key: 'n1', type: 'trigger' }, + { key: 'n2', type: 'action', config: { action: 'set_context', set: { paid: false } } }, + { key: 'n3', type: 'condition', config: { condition: 'expr', field: 'paid', op: 'truthy' } }, + { key: 'n4', type: 'loop', config: { maxIterations: 2 } }, + { key: 'n5', type: 'action', config: { action: 'noop' } }, + { key: 'n6', type: 'wait', config: { delayMinutes: 0 } }, + { key: 'n7', type: 'action', config: { action: 'noop' } }, + ], + edges: [ + { from: 'n1', to: 'n2' }, + { from: 'n2', to: 'n3' }, + { from: 'n3', handle: 'no', to: 'n4' }, + { from: 'n3', handle: 'yes', to: 'n7' }, + { from: 'n4', handle: 'loop', to: 'n5' }, + { from: 'n4', handle: 'exit', to: 'n7' }, + { from: 'n5', to: 'n6' }, + { from: 'n6', to: 'n3', loopBack: true }, + ], + }); + + const runIds = await engine.emitWorkflowEvent('test.event', { entityType: 'invoice', entityId: 1 }); + expect(runIds.length).toBe(1); + const runId = runIds[0]; + + let run = await db('workflow_runs').where({ id: runId }).first(); + expect(run.status).toBe('waiting'); // paused at first wait (loop iter 1) + expect(run.current_node).toBe('n6'); + + await engine.resumeRun(runId); + run = await db('workflow_runs').where({ id: runId }).first(); + expect(run.status).toBe('waiting'); // paused again (loop iter 2) + + await engine.resumeRun(runId); + run = await db('workflow_runs').where({ id: runId }).first(); + expect(run.status).toBe('done'); // loop exhausted → exit → end + + const ctx = JSON.parse(run.context); + expect(ctx.vars.__loop_n4).toBe(3); // counter incremented past the cap + void wfId; + + const steps = await db('workflow_run_steps').where({ run_id: runId }); + expect(steps.length).toBeGreaterThan(0); + }); + + test('emit is idempotent on dedup_key', async () => { + await makeWorkflow({ + trigger: 'dedup.event', + nodes: [{ key: 'n1', type: 'trigger' }, { key: 'n2', type: 'action', config: { action: 'noop' } }], + edges: [{ from: 'n1', to: 'n2' }], + }); + const first = await engine.emitWorkflowEvent('dedup.event', { entityType: 'x', entityId: 9 }); + const second = await engine.emitWorkflowEvent('dedup.event', { entityType: 'x', entityId: 9 }); + expect(first.length).toBe(1); + expect(second.length).toBe(0); // same entity → no duplicate run + }); + + test('gate pauses and resumes via the confirm edge', async () => { + const wfId = await makeWorkflow({ + trigger: 'gate.event', + nodes: [ + { key: 'g1', type: 'trigger' }, + { key: 'g2', type: 'gate', config: { type: 'payment_confirm' } }, + { key: 'g3', type: 'action', config: { action: 'noop' } }, + { key: 'g4', type: 'action', config: { action: 'noop' } }, + ], + edges: [ + { from: 'g1', to: 'g2' }, + { from: 'g2', handle: 'confirm', to: 'g3' }, + { from: 'g2', handle: 'deny', to: 'g4' }, + ], + }); + // create + start a run directly + await db('workflow_runs').insert({ + workflow_id: wfId, version: 1, trigger_event: 'gate.event', status: 'pending', + context: JSON.stringify({ vars: {} }), dedup_key: 'gate-test', + }); + const run0 = await db('workflow_runs').where({ dedup_key: 'gate-test' }).first(); + await engine.startRun(run0.id); + + let run = await db('workflow_runs').where({ id: run0.id }).first(); + expect(run.status).toBe('waiting'); + expect(run.current_node).toBe('g2'); + + await engine.resumeRun(run0.id, { decisionHandle: 'confirm' }); + run = await db('workflow_runs').where({ id: run0.id }).first(); + expect(run.status).toBe('done'); + }); +}); diff --git a/backend/src/services/workflows/engine.js b/backend/src/services/workflows/engine.js new file mode 100644 index 00000000..97caa737 --- /dev/null +++ b/backend/src/services/workflows/engine.js @@ -0,0 +1,284 @@ +/** + * Workflow execution engine — walks a flow GRAPH (nodes + edges) per run. + * + * Node types: trigger | condition/branch | loop | wait | gate | action | webhook. + * - condition/branch: run a registered condition → follow the yes/no edge. + * - loop: increment a per-node counter in run.context → follow loop/exit edge + * (bounded by config.maxIterations — no infinite runs). + * - wait: set status='waiting' + wake_at; the scheduler resumes it later. + * - gate: set status='waiting'; an approval (email confirm/deny or the webview + * inbox) resumes it via the matching confirm/deny edge. + * - action/webhook: dispatch to a registered action handler. + * + * Runs are idempotent (unique dedup_key per trigger+entity) and every node + * writes a workflow_run_steps row for observability / System Health. Designed + * to be called AFTER the caller's DB commit (emit never throws into callers). + */ +const { db } = require('../../database/db'); +const logger = require('../../utils/logger'); +const registry = require('./registry'); + +const MAX_STEPS_PER_ADVANCE = 200; + +function parseJson(value, fallback) { + if (value == null) return fallback; + if (typeof value === 'object') return value; + try { return JSON.parse(value); } catch (e) { return fallback; } +} + +async function loadGraph(workflowId, version) { + const nodes = await db('workflow_nodes').where({ workflow_id: workflowId, version }); + const edges = await db('workflow_edges').where({ workflow_id: workflowId, version }); + const nodeByKey = new Map(nodes.map((n) => [n.node_key, { ...n, config: parseJson(n.config, {}) }])); + return { nodeByKey, edges }; +} + +// Pick the outgoing edge from `fromNode`. With a handle, prefer the matching +// handle; otherwise fall back to a default (null-handle) edge or the sole edge. +function outEdge(edges, fromNode, handle) { + const candidates = edges.filter((e) => e.from_node === fromNode); + if (handle != null) { + const exact = candidates.find((e) => (e.from_handle || null) === handle); + if (exact) return exact; + } + return candidates.find((e) => e.from_handle == null) || (candidates.length === 1 ? candidates[0] : null); +} + +function computeWakeAt(config = {}, vars = {}) { + const cfg = config || {}; + if (cfg.untilVar && vars[cfg.untilVar]) return new Date(vars[cfg.untilVar]).toISOString(); + const ms = (Number(cfg.delayDays || 0) * 86400000) + + (Number(cfg.delayHours || 0) * 3600000) + + (Number(cfg.delayMinutes || 0) * 60000); + return new Date(Date.now() + ms).toISOString(); +} + +function gateTimeout(config = {}) { + const days = Number((config || {}).timeoutDays || 0); + return days > 0 ? new Date(Date.now() + days * 86400000).toISOString() : null; +} + +function matchFilter(filter, payload) { + if (!filter || typeof filter !== 'object') return true; + const { field, op = 'eq', value } = filter; + const actual = payload ? payload[field] : undefined; + switch (op) { + case 'neq': return actual != value; // eslint-disable-line eqeqeq + case 'truthy': return Boolean(actual); + case 'falsy': return !actual; + case 'eq': + default: return actual == value; // eslint-disable-line eqeqeq + } +} + +async function recordStep(runId, node, status, result, error) { + await db('workflow_run_steps').insert({ + run_id: runId, + node_key: node.node_key, + node_type: node.type, + status, + result: result ? JSON.stringify(result) : null, + error: error || null, + finished_at: db.fn.now(), + }); +} + +async function failRun(runId, error) { + await db('workflow_runs').where({ id: runId }).update({ status: 'failed', error, finished_at: db.fn.now() }); + logger.error('[workflow] run failed', { runId, error }); +} + +async function finishRun(runId) { + await db('workflow_runs').where({ id: runId }).update({ status: 'done', finished_at: db.fn.now() }); +} + +/** + * Walk the graph from the run's current node until it ends, fails, or pauses + * (wait / gate). Persists context + current_node after each node. + */ +async function advanceRun(runId) { + let run = await db('workflow_runs').where({ id: runId }).first(); + if (!run || run.status !== 'running') return; + const { nodeByKey, edges } = await loadGraph(run.workflow_id, run.version); + const context = parseJson(run.context, { vars: {} }); + if (!context.vars) context.vars = {}; + + let currentKey = run.current_node; + let steps = 0; + + while (currentKey) { + if (++steps > MAX_STEPS_PER_ADVANCE) { await failRun(runId, 'max steps per advance exceeded'); return; } + const node = nodeByKey.get(currentKey); + if (!node) { await failRun(runId, `node not found: ${currentKey}`); return; } + + const ctx = { run, node, vars: context.vars, db, logger }; + let nextKey = null; + + try { + switch (node.type) { + case 'trigger': { + const e = outEdge(edges, currentKey, null); + nextKey = e ? e.to_node : null; + await recordStep(runId, node, 'done', null); + break; + } + case 'condition': + case 'branch': { + const cond = registry.getCondition(node.config?.condition || 'expr'); + const result = cond ? await cond(ctx) : false; + const handle = result ? (node.config?.trueHandle || 'yes') : (node.config?.falseHandle || 'no'); + const e = outEdge(edges, currentKey, handle) || outEdge(edges, currentKey, result ? 'true' : 'false'); + nextKey = e ? e.to_node : null; + await recordStep(runId, node, 'done', { result, handle }); + break; + } + case 'loop': { + const counterKey = `__loop_${node.node_key}`; + const count = (Number(context.vars[counterKey]) || 0) + 1; + context.vars[counterKey] = count; + const max = Number(node.config?.maxIterations ?? node.config?.max ?? 3); + const handle = count > max ? (node.config?.exitHandle || 'exit') : (node.config?.loopHandle || 'loop'); + const e = outEdge(edges, currentKey, handle); + nextKey = e ? e.to_node : null; + await recordStep(runId, node, 'done', { count, max, handle }); + break; + } + case 'wait': { + const wakeAt = computeWakeAt(node.config, context.vars); + await db('workflow_runs').where({ id: runId }) + .update({ status: 'waiting', wake_at: wakeAt, current_node: currentKey, context: JSON.stringify(context) }); + await recordStep(runId, node, 'waiting', { wake_at: wakeAt }); + return; // paused — scheduler resumes when wake_at passes + } + case 'gate': { + await db('workflow_runs').where({ id: runId }) + .update({ status: 'waiting', wake_at: gateTimeout(node.config), current_node: currentKey, context: JSON.stringify(context) }); + await recordStep(runId, node, 'waiting', { gate: true }); + // Optional setup hook (create approval + send admin email) — registered + // by the approval phase. Engine still pauses cleanly without it. + const setup = registry.getAction('gate_setup'); + if (setup) { + try { await setup(ctx); } catch (e) { logger.error('[workflow] gate setup failed', { runId, error: e.message }); } + } + return; // paused — an approval (email or inbox) resumes via resumeRun + } + case 'action': + case 'webhook': { + const actionKey = node.config?.action || (node.type === 'webhook' ? 'webhook' : 'noop'); + const action = registry.getAction(actionKey); + const result = action ? (await action(ctx)) || {} : { skipped: true, reason: `unknown action ${actionKey}` }; + if (result.set && typeof result.set === 'object') Object.assign(context.vars, result.set); + const e = outEdge(edges, currentKey, null); + nextKey = e ? e.to_node : null; + await recordStep(runId, node, result.skipped ? 'skipped' : 'done', result); + break; + } + default: { + await recordStep(runId, node, 'skipped', { reason: `unknown node type ${node.type}` }); + const e = outEdge(edges, currentKey, null); + nextKey = e ? e.to_node : null; + } + } + } catch (err) { + await recordStep(runId, node, 'failed', null, err.message); + await failRun(runId, `node ${currentKey} failed: ${err.message}`); + return; + } + + currentKey = nextKey; + await db('workflow_runs').where({ id: runId }).update({ current_node: currentKey || null, context: JSON.stringify(context) }); + } + + await finishRun(runId); +} + +/** Begin a freshly-created run at its trigger node. */ +async function startRun(runId) { + const run = await db('workflow_runs').where({ id: runId }).first(); + if (!run || ['done', 'failed', 'cancelled'].includes(run.status)) return; + const { nodeByKey } = await loadGraph(run.workflow_id, run.version); + let entry = null; + for (const n of nodeByKey.values()) { if (n.type === 'trigger') { entry = n; break; } } + if (!entry) { await failRun(runId, 'no trigger node'); return; } + await db('workflow_runs').where({ id: runId }).update({ status: 'running', current_node: entry.node_key }); + await advanceRun(runId); +} + +/** + * Resume a paused (waiting) run. For a wait node, pass no handle. For a gate, + * pass decisionHandle = 'confirm' | 'deny' so the matching edge is taken. + */ +async function resumeRun(runId, { decisionHandle = null } = {}) { + const run = await db('workflow_runs').where({ id: runId }).first(); + if (!run || run.status !== 'waiting') return; + const { edges } = await loadGraph(run.workflow_id, run.version); + const e = outEdge(edges, run.current_node, decisionHandle); + const nextKey = e ? e.to_node : null; + await db('workflow_runs').where({ id: runId }).update({ status: 'running', current_node: nextKey, wake_at: null }); + if (!nextKey) { await finishRun(runId); return; } + await advanceRun(runId); +} + +/** + * Entry point for lifecycle events. Creates one run per matching enabled + * workflow (idempotent via dedup_key) and starts it. Never throws — safe to + * call after a caller's commit. Fails CLOSED if the flag system is unavailable. + */ +async function emitWorkflowEvent(triggerType, { entityType = null, entityId = null, payload = {} } = {}) { + try { + const { isFeatureEnabled } = require('../../middleware/requireFeatureFlag'); + let enabled = false; + try { enabled = await isFeatureEnabled('workflows'); } catch (e) { + logger.warn('[workflow] flag check failed — treating workflows as disabled', { error: e.message }); + return []; + } + if (!enabled) return []; + + const workflows = await db('workflows').where({ enabled: true, trigger_type: triggerType }); + const runIds = []; + for (const wf of workflows) { + const tcfg = parseJson(wf.trigger_config, {}); + if (tcfg && tcfg.filter && !matchFilter(tcfg.filter, payload)) continue; + + const dedupKey = `${wf.id}:${wf.version}:${triggerType}:${entityType || ''}:${entityId || ''}`; + const existing = await db('workflow_runs').where({ dedup_key: dedupKey }).first(); + if (existing) continue; + + try { + await db('workflow_runs').insert({ + workflow_id: wf.id, + version: wf.version, + trigger_event: triggerType, + entity_type: entityType, + entity_id: entityId, + status: 'pending', + context: JSON.stringify({ vars: { ...payload } }), + dedup_key: dedupKey, + }); + } catch (e) { + continue; // unique race — another emitter created it + } + const row = await db('workflow_runs').where({ dedup_key: dedupKey }).first(); + if (!row) continue; + runIds.push(row.id); + await startRun(row.id).catch((err) => logger.error('[workflow] start failed', { runId: row.id, error: err.message })); + } + return runIds; + } catch (e) { + logger.error('[workflow] emit failed', { triggerType, error: e.message }); + return []; + } +} + +module.exports = { + emitWorkflowEvent, + startRun, + advanceRun, + resumeRun, + finishRun, + failRun, + // exported for tests / introspection + loadGraph, + outEdge, + computeWakeAt, +}; diff --git a/backend/src/services/workflows/index.js b/backend/src/services/workflows/index.js new file mode 100644 index 00000000..cfff36bf --- /dev/null +++ b/backend/src/services/workflows/index.js @@ -0,0 +1,16 @@ +/** + * Workflow engine public surface. + * + * const { emitWorkflowEvent } = require('./services/workflows'); + * + * `engine` holds the executor (start/advance/resume), `registry` the catalog + * of conditions/actions. Action/condition handler modules require `registry` + * and call registerAction/registerCondition at load time. + */ +const engine = require('./engine'); +const registry = require('./registry'); + +module.exports = { + ...engine, + registry, +}; diff --git a/backend/src/services/workflows/registry.js b/backend/src/services/workflows/registry.js new file mode 100644 index 00000000..ae21ca87 --- /dev/null +++ b/backend/src/services/workflows/registry.js @@ -0,0 +1,61 @@ +/** + * Workflow registry — the curated catalog of CONDITIONS and ACTIONS the engine + * can run. Node `config.condition` / `config.action` keys map to handlers here. + * + * Handlers are async `(ctx) => result`, where ctx = { run, node, vars, db, + * logger }. `vars` is the run's mutable context bag (loop counters, accumulated + * values, the trigger payload). A condition returns a boolean; an action may + * return `{ set: {...} }` to merge values back into `vars`. + * + * Keep handlers curated and typed — this is NOT arbitrary code execution. New + * triggers/actions register here; the canvas palette is derived from these. + */ +const conditions = new Map(); +const actions = new Map(); + +function registerCondition(key, fn) { conditions.set(key, fn); } +function registerAction(key, fn) { actions.set(key, fn); } +function getCondition(key) { return conditions.get(key); } +function getAction(key) { return actions.get(key); } +function listConditions() { return Array.from(conditions.keys()); } +function listActions() { return Array.from(actions.keys()); } + +// --- Primitive conditions --- +registerCondition('always', async () => true); +registerCondition('never', async () => false); +// Generic field/op/value compare against the run's `vars` bag. +registerCondition('expr', async (ctx) => { + const { field, op = 'truthy', value } = ctx.node.config || {}; + const actual = field != null ? ctx.vars[field] : undefined; + switch (op) { + case 'eq': return actual == value; // eslint-disable-line eqeqeq + case 'neq': return actual != value; // eslint-disable-line eqeqeq + case 'gt': return Number(actual) > Number(value); + case 'gte': return Number(actual) >= Number(value); + case 'lt': return Number(actual) < Number(value); + case 'lte': return Number(actual) <= Number(value); + case 'falsy': return !actual; + case 'truthy': + default: return Boolean(actual); + } +}); + +// --- Primitive actions --- +registerAction('noop', async () => ({})); +registerAction('log', async (ctx) => { + ctx.logger?.info?.('[workflow] log action', { runId: ctx.run.id, message: ctx.node.config?.message }); + return { logged: true }; +}); +// Merge a static object into the run context (handy for tests + seeding flags). +registerAction('set_context', async (ctx) => ({ set: ctx.node.config?.set || {} })); + +module.exports = { + registerCondition, + registerAction, + getCondition, + getAction, + listConditions, + listActions, + conditions, + actions, +};