From 610a3dfd732fbf37c68bc3edacaed6c99610a2ab Mon Sep 17 00:00:00 2001
From: Luca <102960244+Luca-Timo@users.noreply.github.com>
Date: Tue, 23 Jun 2026 02:09:57 +0200
Subject: [PATCH] feat(workflows): scheduler resumes elapsed wait nodes
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Adds engine.runDueWaits() — polls waiting runs whose wake_at has passed and
resumes the ones parked on a wait node (gate timeouts handled later by the
approvals layer). Flag-gated (fails closed when workflows is off). Wired into
the existing hourly invoiceScheduler tick in its own try/catch so a workflow
failure never suppresses the invoice/reminder jobs. Test covers not-due vs
elapsed resume.
---
 .../integration/workflowEngine.test.js        | 24 +++++++++++
 .../src/services/invoiceSchedulerService.js   |  9 ++++
 backend/src/services/workflows/engine.js      | 41 +++++++++++++++++++
 3 files changed, 74 insertions(+)

diff --git a/backend/__tests__/integration/workflowEngine.test.js b/backend/__tests__/integration/workflowEngine.test.js
index 65119b6f..433b169c 100644
--- a/backend/__tests__/integration/workflowEngine.test.js
+++ b/backend/__tests__/integration/workflowEngine.test.js
@@ -134,4 +134,28 @@ describe('workflow engine', () => {
     run = await db('workflow_runs').where({ id: run0.id }).first();
     expect(run.status).toBe('done');
   });
+
+  test('runDueWaits resumes only elapsed wait nodes', async () => {
+    await makeWorkflow({
+      trigger: 'wait.event',
+      nodes: [
+        { key: 'w1', type: 'trigger' },
+        { key: 'w2', type: 'wait', config: { delayMinutes: 60 } },
+        { key: 'w3', type: 'action', config: { action: 'noop' } },
+      ],
+      edges: [{ from: 'w1', to: 'w2' }, { from: 'w2', to: 'w3' }],
+    });
+    const runIds = await engine.emitWorkflowEvent('wait.event', { entityType: 'e', entityId: 7 });
+    const runId = runIds[0];
+    let run = await db('workflow_runs').where({ id: runId }).first();
+    expect(run.status).toBe('waiting');
+
+    expect(await engine.runDueWaits()).toBe(0); // wake_at ~60min out → not due
+
+    await db('workflow_runs').where({ id: runId }).update({ wake_at: new Date(Date.now() - 1000).toISOString() });
+    const resumed = await engine.runDueWaits();
+    expect(resumed).toBeGreaterThanOrEqual(1);
+    run = await db('workflow_runs').where({ id: runId }).first();
+    expect(run.status).toBe('done');
+  });
 });
diff --git a/backend/src/services/invoiceSchedulerService.js b/backend/src/services/invoiceSchedulerService.js
index 181ab99e..614349b1 100644
--- a/backend/src/services/invoiceSchedulerService.js
+++ b/backend/src/services/invoiceSchedulerService.js
@@ -42,6 +42,15 @@ async function runTick() {
   } catch (err) {
     logger.error('Event reminder pass failed', { err: err.message });
   }
+  try {
+    // Resume workflow runs whose wait has elapsed. No-op (fails closed) when
+    // the `workflows` feature flag is off. Independent try/catch so a workflow
+    // failure never suppresses the invoice/reminder jobs above.
+    const resumed = await require('./workflows').runDueWaits();
+    if (resumed) logger.info('Workflow scheduler: resumed waiting runs', { resumed });
+  } catch (err) {
+    logger.error('Workflow resume pass failed', { err: err.message });
+  }
 }
 
 function startInvoiceScheduler() {
diff --git a/backend/src/services/workflows/engine.js b/backend/src/services/workflows/engine.js
index 97caa737..b4bb401f 100644
--- a/backend/src/services/workflows/engine.js
+++ b/backend/src/services/workflows/engine.js
@@ -270,8 +270,49 @@ async function emitWorkflowEvent(triggerType, { entityType = null, entityId = nu
   }
 }
 
+/**
+ * Resume runs whose wait has elapsed. Called from the cron scheduler tick.
+ * Only advances `wait` nodes — gate timeouts are handled by the approvals
+ * layer. Fails CLOSED if the workflows flag is off (master kill-switch).
+ */
+async function runDueWaits(limit = 100) {
+  try {
+    const { isFeatureEnabled } = require('../../middleware/requireFeatureFlag');
+    let enabled = false;
+    try { enabled = await isFeatureEnabled('workflows'); } catch (e) { return 0; }
+    if (!enabled) return 0;
+
+    const nowIso = new Date().toISOString();
+    const due = await db('workflow_runs')
+      .where({ status: 'waiting' })
+      .whereNotNull('wake_at')
+      .where('wake_at', '<=', nowIso)
+      .limit(limit);
+
+    let resumed = 0;
+    for (const run of due) {
+      try {
+        const node = await db('workflow_nodes')
+          .where({ workflow_id: run.workflow_id, version: run.version, node_key: run.current_node })
+          .first();
+        if (node && node.type === 'wait') {
+          await resumeRun(run.id);
+          resumed += 1;
+        }
+      } catch (err) {
+        logger.error('[workflow] runDueWaits item failed', { runId: run.id, error: err.message });
+      }
+    }
+    return resumed;
+  } catch (e) {
+    logger.error('[workflow] runDueWaits failed', { error: e.message });
+    return 0;
+  }
+}
+
 module.exports = {
   emitWorkflowEvent,
+  runDueWaits,
   startRun,
   advanceRun,
   resumeRun,