feat(workflows): data-touching action + condition handlers

Adds send_email (INTERNAL/admin = immediate, EXTERNAL/customer = business-
hours floor via queueEmail's respectBusinessHours) and the invoice_paid
condition (paid_at / status / cumulative paid_amount). Registers the
prepare_quote/contract/event/gallery/invoice + send_document + reserve_date
document actions as recognized-but-not-yet-wired (record an observable
skipped step rather than crashing a flow). index.js side-effect-imports the
handlers. Tests cover the customer-mail routing + the invoice_paid logic.

backend/__tests__/integration/workflowEngine.test.js

@@ -158,4 +158,33 @@ describe('workflow engine', () => {
     run = await db('workflow_runs').where({ id: runId }).first();
     expect(run.status).toBe('done');
   });
+
+  test('send_email queues a customer mail with business-hours routing', async () => {
+    await makeWorkflow({
+      trigger: 'mail.event',
+      nodes: [
+        { key: 'm1', type: 'trigger' },
+        { key: 'm2', type: 'action', config: { action: 'send_email', recipientClass: 'customer', emailType: 'workflow_test' } },
+      ],
+      edges: [{ from: 'm1', to: 'm2' }],
+    });
+    const runIds = await engine.emitWorkflowEvent('mail.event', {
+      entityType: 'invoice', entityId: 3, payload: { customerEmail: '[email protected]' },
+    });
+    const run = await db('workflow_runs').where({ id: runIds[0] }).first();
+    expect(run.status).toBe('done');
+    const queued = await db('email_queue').where({ recipient_email: '[email protected]' }).first();
+    expect(queued).toBeTruthy();
+    const step = await db('workflow_run_steps').where({ run_id: runIds[0], node_key: 'm2' }).first();
+    expect(JSON.parse(step.result).respectBusinessHours).toBe(true);
+  });
+
+  test('invoice_paid condition reads the entity', async () => {
+    const registry = require('../../src/services/workflows/registry');
+    const cond = registry.getCondition('invoice_paid');
+    const makeCtx = (row) => ({ run: { entity_id: 1 }, db: () => ({ where: () => ({ first: async () => row }) }) });
+    expect(await cond(makeCtx({ paid_at: '2026-01-01', status: 'sent' }))).toBe(true);
+    expect(await cond(makeCtx({ paid_at: null, status: 'paid' }))).toBe(true);
+    expect(await cond(makeCtx({ paid_at: null, status: 'sent', paid_amount_minor: 0, total_amount_minor: 1000 }))).toBe(false);
+  });
 });

backend/src/services/workflows/actions.js

@@ -0,0 +1,78 @@
+/**
+ * Workflow action + condition handlers that touch real picpeak data.
+ *
+ * Registered at load time (index.js requires this module). Kept separate from
+ * registry.js (which holds only primitives) so the I/O-coupled handlers don't
+ * bloat the pure core.
+ *
+ * Email routing rule (locked requirement): INTERNAL/admin mail sends
+ * immediately; EXTERNAL/customer mail respects the business-hours floor. The
+ * action sets queueEmail's `respectBusinessHours` from the recipient class.
+ *
+ * The create/prepare-document actions (quote/contract/event/gallery/invoice)
+ * are registered so flows validate, but are intentionally NOT wired to the
+ * services yet — they record a `skipped` step with a clear reason so the gap
+ * is observable rather than silent. Wiring is a follow-up commit.
+ */
+const registry = require('./registry');
+
+const DOCUMENT_ACTIONS = [
+  'prepare_quote',
+  'prepare_contract',
+  'prepare_event',
+  'prepare_gallery',
+  'prepare_invoice',
+  'send_document',
+  'reserve_date',
+];
+
+// --- Conditions ---
+
+// True once the run's invoice entity is settled (paid_at set, status paid, or
+// the cumulative paid amount covers the total).
+registry.registerCondition('invoice_paid', async (ctx) => {
+  const id = ctx.run.entity_id;
+  if (!id) return false;
+  const inv = await ctx.db('invoices').where({ id }).first();
+  if (!inv) return false;
+  if (inv.paid_at) return true;
+  if (inv.status === 'paid') return true;
+  const paid = Number(inv.paid_amount_minor) || 0;
+  const total = Number(inv.total_amount_minor);
+  return Number.isFinite(total) && total > 0 && paid >= total;
+});
+
+// --- Actions ---
+
+// Queue an email. recipientClass 'admin' (internal) sends immediately;
+// anything else (customer/external) respects the business-hours floor.
+registry.registerAction('send_email', async (ctx) => {
+  const cfg = ctx.node.config || {};
+  const recipientClass = cfg.recipientClass || cfg.recipient || 'customer';
+  const isInternal = recipientClass === 'admin' || recipientClass === 'internal';
+  const to = cfg.to
+    || ctx.vars[isInternal ? 'adminEmail' : 'customerEmail']
+    || ctx.vars.recipientEmail;
+  if (!to) return { skipped: true, reason: 'no recipient resolved' };
+
+  const emailProcessor = require('../emailProcessor');
+  const eventId = ctx.vars.eventId || null;
+  const emailType = cfg.emailType || cfg.template || 'workflow_notification';
+  const emailData = { ...(cfg.emailData || {}), ...(ctx.vars.emailData || {}) };
+
+  // INTERNAL/admin = immediate; EXTERNAL/customer = business-hours floor.
+  const respectBusinessHours = !isInternal;
+  await emailProcessor.queueEmail(eventId, to, emailType, emailData, { respectBusinessHours });
+  return { sent_to: to, recipientClass, respectBusinessHours };
+});
+
+// Create/prepare-document actions — registered so flows referencing them are
+// valid; service wiring is a follow-up. Records a skipped step (observable).
+for (const key of DOCUMENT_ACTIONS) {
+  registry.registerAction(key, async (ctx) => {
+    ctx.logger?.warn?.('[workflow] document action not yet wired', { action: key, runId: ctx.run.id });
+    return { skipped: true, reason: `action ${key} not yet implemented` };
+  });
+}
+
+module.exports = { DOCUMENT_ACTIONS };

backend/src/services/workflows/index.js

@@ -9,6 +9,9 @@
  */
 const engine = require('./engine');
 const registry = require('./registry');
+// Side-effect import: registers the data-touching action/condition handlers
+// (send_email, invoice_paid, prepare_* document actions) onto the registry.
+require('./actions');
 
 module.exports = {
   ...engine,