feat(workflows): execution engine core + registry + tests

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.

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');
+  });
+});

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,
+};

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,
+};

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,
+};