'use strict' /** * flow-runtime.js — Execution engine for Flow Templates. * * Responsibilities * ──────────────── * 1. startFlow(templateName, ctx) → creates a Flow Run + kicks off the first wave of steps * 2. advanceFlow(runName) → evaluates which steps are ready and runs them * 3. completeStep(runName, stepId, result) → mark a step done and advance the flow * 4. Kind dispatcher: KIND_HANDLERS[step.kind](step, ctx) → { status, result } * 5. Scheduling: any step with trigger.type === 'after_delay' | 'on_date' creates * a Flow Step Pending row instead of executing inline * * Context shape * ───────────── * ctx = { * run: { name, variables, step_state }, // the Flow Run doc * template: { flow_definition, … }, // the Flow Template doc * customer: { name, customer_name, … }, // resolved customer doc * doc: { … }, // the trigger doc (contract, quotation …) * doctype, docname, // trigger doc reference * } * * Step state (persisted in Flow Run.step_state) * ───────────────────────────────────────────── * { [stepId]: { status: 'pending|running|done|failed|scheduled|skipped', * started_at, completed_at, result, error, retry_count } } * * See docs/FLOW_EDITOR_ARCHITECTURE.md for full data model + trigger wiring. */ const { log, erpFetch } = require('./helpers') const cfg = require('./config') // ─────────────────────────────────────────────────────────────────────────── // Constants // ─────────────────────────────────────────────────────────────────────────── const FT_DOCTYPE = 'Flow Template' const FR_DOCTYPE = 'Flow Run' const FS_DOCTYPE = 'Flow Step Pending' const ENC_FT = encodeURIComponent(FT_DOCTYPE) const ENC_FR = encodeURIComponent(FR_DOCTYPE) const ENC_FS = encodeURIComponent(FS_DOCTYPE) /** All step statuses (for type-safety when reading step_state). */ const STATUS = { PENDING: 'pending', RUNNING: 'running', DONE: 'done', FAILED: 'failed', SCHEDULED: 'scheduled', SKIPPED: 'skipped', } // ─────────────────────────────────────────────────────────────────────────── // Template rendering (simple {{var.path}} interpolation, no Mustache lib) // ─────────────────────────────────────────────────────────────────────────── const TEMPLATE_RE = /\{\{\s*([\w.]+)\s*\}\}/g /** Follow dotted paths through nested objects. Safe on missing keys. */ function getPath (obj, path) { if (!obj || !path) return undefined return path.split('.').reduce((o, k) => (o == null ? o : o[k]), obj) } /** Replace {{a.b.c}} tokens with ctx values; leaves non-strings alone. */ function render (tpl, ctx) { if (typeof tpl !== 'string') return tpl return tpl.replace(TEMPLATE_RE, (match, path) => { const val = getPath(ctx, path) return val == null ? '' : String(val) }) } /** Deep-render every string inside an object. Recursive; arrays preserved. */ function renderDeep (input, ctx) { if (input == null) return input if (typeof input === 'string') return render(input, ctx) if (Array.isArray(input)) return input.map(v => renderDeep(v, ctx)) if (typeof input === 'object') { const out = {} for (const [k, v] of Object.entries(input)) out[k] = renderDeep(v, ctx) return out } return input } // ─────────────────────────────────────────────────────────────────────────── // Condition evaluator — simple predicate language (no eval) // ─────────────────────────────────────────────────────────────────────────── /** * Evaluate a condition payload against the context. * Shape: { field, op, value } * Ops: == != < > <= >= in not_in empty not_empty contains starts_with ends_with * Returns boolean. */ function evalCondition (p, ctx) { const left = getPath(ctx, p.field) const right = typeof p.value === 'string' ? render(p.value, ctx) : p.value switch (p.op) { case '==': return String(left) == String(right) // eslint-disable-line eqeqeq case '!=': return String(left) != String(right) // eslint-disable-line eqeqeq case '<': return Number(left) < Number(right) case '>': return Number(left) > Number(right) case '<=': return Number(left) <= Number(right) case '>=': return Number(left) >= Number(right) case 'in': return String(right).split(',').map(s => s.trim()).includes(String(left)) case 'not_in':return !String(right).split(',').map(s => s.trim()).includes(String(left)) case 'empty': return left == null || left === '' case 'not_empty': return left != null && left !== '' case 'contains': return String(left || '').includes(String(right)) case 'starts_with':return String(left || '').startsWith(String(right)) case 'ends_with': return String(left || '').endsWith(String(right)) default: return false } } // ─────────────────────────────────────────────────────────────────────────── // Kind handlers — one function per step kind. All return: // { status: 'done'|'failed'|'scheduled', result?, error? } // ─────────────────────────────────────────────────────────────────────────── /** Create a Dispatch Job. Returns the new job name in result. */ async function handleDispatchJob (step, ctx) { const p = renderDeep(step.payload || {}, ctx) const payload = { doctype: 'Dispatch Job', customer: ctx.customer?.name || p.customer || '', customer_name: ctx.customer?.customer_name || '', subject: p.subject || step.label || 'Tâche', job_type: p.job_type || 'Autre', priority: p.priority || 'medium', duration_h: p.duration_h || 1, assigned_group: p.assigned_group || 'Tech Targo', status: 'open', flow_run: ctx.run?.name, flow_step_id: step.id, } if (p.service_location) payload.service_location = p.service_location else if (ctx.doc?.service_location) payload.service_location = ctx.doc.service_location if (p.merge_key) payload.merge_key = p.merge_key if (p.on_open_webhook) payload.on_open_webhook = p.on_open_webhook if (p.on_close_webhook) payload.on_close_webhook = p.on_close_webhook if (p.notes) payload.description = p.notes const r = await erpFetch('/api/resource/Dispatch Job', { method: 'POST', body: JSON.stringify(payload), }) if (r.status !== 200 && r.status !== 201) { return { status: STATUS.FAILED, error: 'Dispatch Job creation failed: ' + JSON.stringify(r.data).slice(0, 200) } } return { status: STATUS.DONE, result: { job: r.data.data?.name } } } /** Create an ERPNext Issue (ticket). */ async function handleIssue (step, ctx) { const p = renderDeep(step.payload || {}, ctx) const payload = { doctype: 'Issue', subject: p.subject || step.label || 'Suivi', description: p.description || '', priority: p.priority || 'Medium', issue_type: p.issue_type || 'Suivi', status: 'Open', customer: ctx.customer?.name || null, flow_run: ctx.run?.name, flow_step_id: step.id, } const r = await erpFetch('/api/resource/Issue', { method: 'POST', body: JSON.stringify(payload), }) if (r.status !== 200 && r.status !== 201) { return { status: STATUS.FAILED, error: 'Issue creation failed: ' + JSON.stringify(r.data).slice(0, 200) } } return { status: STATUS.DONE, result: { issue: r.data.data?.name } } } /** Send a notification (SMS or email) via existing Hub modules. */ async function handleNotify (step, ctx) { const p = renderDeep(step.payload || {}, ctx) const channel = (p.channel || 'sms').toLowerCase() const to = p.to || ctx.customer?.cell_phone || ctx.customer?.primary_phone || ctx.customer?.email_id if (!to) return { status: STATUS.FAILED, error: 'notify: no recipient resolved' } try { if (channel === 'sms') { const { sendSms } = require('./twilio') const body = p.body || (p.template_id ? _lookupTemplateText(p.template_id, ctx) : '') if (!body) return { status: STATUS.FAILED, error: 'notify: empty SMS body' } const r = await sendSms({ to, body }) return { status: STATUS.DONE, result: { channel, sid: r?.sid, simulated: r?.simulated } } } if (channel === 'email') { const { sendEmail } = require('./email') const subject = p.subject || 'Notification' const html = p.body || _lookupTemplateText(p.template_id, ctx) || '' const r = await sendEmail({ to, subject, html }) return { status: STATUS.DONE, result: { channel, id: r?.id } } } return { status: STATUS.FAILED, error: `notify: unknown channel "${channel}"` } } catch (e) { return { status: STATUS.FAILED, error: 'notify failed: ' + e.message } } } /** Look up a template body from email-templates.js (stubbed with empty fallback). */ function _lookupTemplateText (templateId, ctx) { if (!templateId) return '' try { const tpls = require('./email-templates') const tpl = tpls[templateId] || tpls.TEMPLATES?.[templateId] if (!tpl) return '' const body = typeof tpl === 'string' ? tpl : (tpl.body || tpl.text || '') return render(body, ctx) } catch { return '' } } /** POST/GET/PUT/DELETE an external webhook. Body JSON-rendered from context. */ async function handleWebhook (step, ctx) { const p = renderDeep(step.payload || {}, ctx) if (!p.url) return { status: STATUS.FAILED, error: 'webhook: url required' } const method = (p.method || 'POST').toUpperCase() let body if (p.body_template) { try { body = JSON.parse(p.body_template) } catch (e) { return { status: STATUS.FAILED, error: 'webhook: body_template is not valid JSON: ' + e.message } } } try { const r = await fetch(p.url, { method, headers: { 'Content-Type': 'application/json' }, body: body ? JSON.stringify(body) : undefined, }) return { status: r.ok ? STATUS.DONE : STATUS.FAILED, result: { http_status: r.status }, error: r.ok ? undefined : `HTTP ${r.status}`, } } catch (e) { return { status: STATUS.FAILED, error: 'webhook request failed: ' + e.message } } } /** Update a field (or set of fields) on an existing ERPNext doc. */ async function handleErpUpdate (step, ctx) { const p = renderDeep(step.payload || {}, ctx) if (!p.doctype || !p.docname_ref) { return { status: STATUS.FAILED, error: 'erp_update: doctype and docname_ref required' } } let fields try { fields = typeof p.fields_json === 'string' ? JSON.parse(p.fields_json) : (p.fields_json || {}) } catch (e) { return { status: STATUS.FAILED, error: 'erp_update: fields_json invalid: ' + e.message } } const path = `/api/resource/${encodeURIComponent(p.doctype)}/${encodeURIComponent(p.docname_ref)}` const r = await erpFetch(path, { method: 'PUT', body: JSON.stringify(fields) }) if (r.status !== 200) { return { status: STATUS.FAILED, error: `erp_update failed: HTTP ${r.status}` } } return { status: STATUS.DONE, result: { doctype: p.doctype, docname: p.docname_ref } } } /** * Activate a Subscription (ERPNext Subscription doctype). * Expects payload.subscription_ref → name of the Subscription doc. */ async function handleSubscriptionActivate (step, ctx) { const p = renderDeep(step.payload || {}, ctx) const subName = p.subscription_ref || ctx.doc?.subscription if (!subName) return { status: STATUS.FAILED, error: 'subscription_activate: subscription_ref required' } const path = `/api/resource/Subscription/${encodeURIComponent(subName)}` const r = await erpFetch(path, { method: 'PUT', body: JSON.stringify({ status: 'Active' }) }) if (r.status !== 200) { return { status: STATUS.FAILED, error: 'subscription_activate failed: HTTP ' + r.status } } return { status: STATUS.DONE, result: { subscription: subName } } } /** Wait step — purely declarative; always resolves "done" inline. */ function handleWait () { return { status: STATUS.DONE, result: { waited: true } } } /** Condition step — evaluates the predicate; returns DONE with a branch hint. */ function handleCondition (step, ctx) { const ok = evalCondition(step.payload || {}, ctx) return { status: STATUS.DONE, result: { branch: ok ? 'yes' : 'no' } } } const KIND_HANDLERS = { dispatch_job: handleDispatchJob, issue: handleIssue, notify: handleNotify, webhook: handleWebhook, erp_update: handleErpUpdate, subscription_activate: handleSubscriptionActivate, wait: handleWait, condition: handleCondition, } // ─────────────────────────────────────────────────────────────────────────── // Run-level persistence helpers // ─────────────────────────────────────────────────────────────────────────── async function _fetchTemplate (name) { const r = await erpFetch(`/api/resource/${ENC_FT}/${encodeURIComponent(name)}`) if (r.status !== 200) throw new Error(`Template ${name} not found`) const tpl = r.data.data try { tpl.flow_definition = JSON.parse(tpl.flow_definition || '{}') } catch { tpl.flow_definition = { steps: [] } } if (!tpl.flow_definition.steps) tpl.flow_definition.steps = [] return tpl } async function _fetchRun (name) { const r = await erpFetch(`/api/resource/${ENC_FR}/${encodeURIComponent(name)}`) if (r.status !== 200) throw new Error(`Run ${name} not found`) const run = r.data.data run.variables = _parseJson(run.variables, {}) run.step_state = _parseJson(run.step_state, {}) return run } function _parseJson (s, fallback) { if (!s) return fallback if (typeof s === 'object') return s try { return JSON.parse(s) } catch { return fallback } } async function _persistRun (run, patch) { const body = { ...patch } if (body.variables && typeof body.variables !== 'string') body.variables = JSON.stringify(body.variables) if (body.step_state && typeof body.step_state !== 'string') body.step_state = JSON.stringify(body.step_state) const r = await erpFetch(`/api/resource/${ENC_FR}/${encodeURIComponent(run.name)}`, { method: 'PUT', body: JSON.stringify(body), }) if (r.status !== 200) throw new Error('Failed to persist Flow Run: HTTP ' + r.status) } async function _schedulePending (run, step, triggerAt) { const payload = { doctype: FS_DOCTYPE, flow_run: run.name, step_id: step.id, status: 'pending', trigger_at: triggerAt, context_snapshot: JSON.stringify({ step }), retry_count: 0, } const r = await erpFetch(`/api/resource/${ENC_FS}`, { method: 'POST', body: JSON.stringify(payload), }) if (r.status !== 200 && r.status !== 201) { throw new Error('Failed to schedule step: HTTP ' + r.status) } log(`[flow] scheduled ${step.id} @ ${triggerAt} → ${r.data.data?.name}`) return r.data.data?.name } // ─────────────────────────────────────────────────────────────────────────── // Trigger resolution (decides when a step is ready to run) // ─────────────────────────────────────────────────────────────────────────── /** * Compute a "trigger fire time" in ISO format for delayed steps. * Returns null for immediate triggers. */ function resolveTriggerAt (step, now = new Date()) { const t = step.trigger || {} if (t.type === 'after_delay') { const hours = Number(t.delay_hours || 0) + Number(t.delay_days || 0) * 24 if (!hours) return null return new Date(now.getTime() + hours * 3600 * 1000).toISOString() } if (t.type === 'on_date' && t.at) { return new Date(t.at).toISOString() } return null } /** A step is "ready" when all its depends_on are done + parent branch matches. */ function isStepReady (step, state, def) { // Already handled? const s = state[step.id] if (s && (s.status === STATUS.DONE || s.status === STATUS.SCHEDULED || s.status === STATUS.FAILED)) return false // All depends_on must be done for (const dep of step.depends_on || []) { if (state[dep]?.status !== STATUS.DONE) return false } // Parent step must be done (if any), AND branch must match if (step.parent_id) { const parentState = state[step.parent_id] if (parentState?.status !== STATUS.DONE) return false const parentStep = def.steps.find(s2 => s2.id === step.parent_id) if (parentStep?.kind === 'condition' || parentStep?.kind === 'switch') { if (step.branch && parentState.result?.branch !== step.branch) return false } } return true } // ─────────────────────────────────────────────────────────────────────────── // Public API // ─────────────────────────────────────────────────────────────────────────── /** * Start a new flow run. Creates the Flow Run doc + immediately advances it. * * @param {string} templateName e.g. "FT-00005" * @param {Object} opts * - doctype, docname Trigger doc (contract, quotation, …) * - customer Customer docname (optional; derived from doc if absent) * - variables Runtime bag merged into context * - triggerEvent For audit (e.g. "on_contract_signed") * @returns {Object} { run, executed: [...stepIds], scheduled: [...stepIds] } */ async function startFlow (templateName, opts = {}) { const template = await _fetchTemplate(templateName) if (!template.is_active) { throw new Error(`Template ${templateName} is inactive`) } // Create Flow Run const runPayload = { doctype: FR_DOCTYPE, flow_template: template.name, template_version: template.version || 1, status: 'running', trigger_event: opts.triggerEvent || template.trigger_event || 'manual', context_doctype: opts.doctype || template.applies_to || '', context_docname: opts.docname || '', customer: opts.customer || '', variables: JSON.stringify(opts.variables || {}), step_state: JSON.stringify({}), started_at: new Date().toISOString(), } const r = await erpFetch(`/api/resource/${ENC_FR}`, { method: 'POST', body: JSON.stringify(runPayload), }) if (r.status !== 200 && r.status !== 201) { throw new Error('Failed to create Flow Run: HTTP ' + r.status + ' ' + JSON.stringify(r.data).slice(0, 200)) } const run = r.data.data log(`[flow] started ${run.name} from ${templateName}`) // Advance immediately (fires the first wave of ready steps) const result = await advanceFlow(run.name) return { run: result.run, ...result } } /** * Evaluate all steps and run those that became ready. Idempotent: can be * called again after a pending step resolves (via completeStep), after a * webhook trigger, or from the scheduler. * * @param {string} runName * @returns {Object} { run, executed, scheduled, errors } */ async function advanceFlow (runName) { const run = await _fetchRun(runName) if (run.status === 'completed' || run.status === 'failed' || run.status === 'cancelled') { return { run, executed: [], scheduled: [], errors: [], done: true } } const template = await _fetchTemplate(run.flow_template) const def = template.flow_definition // Resolve trigger context once (lookup doc + customer) const ctx = await _buildContext(run, template) const state = run.step_state const executed = [] const scheduled = [] const errors = [] let mutations = 0 // Linear pass — tree is small, we can re-check after mutations // This loop continues until no more steps become ready (max 3 waves for safety) for (let wave = 0; wave < 50; wave++) { let progressed = false for (const step of def.steps) { if (!isStepReady(step, state, def)) continue const trigType = step.trigger?.type // Delayed triggers → persist as pending if (trigType === 'after_delay' || trigType === 'on_date') { const at = resolveTriggerAt(step) if (at) { state[step.id] = { status: STATUS.SCHEDULED, scheduled_for: at } try { await _schedulePending(run, step, at) } catch (e) { errors.push({ step: step.id, error: e.message }) } scheduled.push(step.id) mutations++ progressed = true continue } } // Webhook / manual triggers → leave in pending state, caller advances via event if (trigType === 'on_webhook' || trigType === 'manual') { state[step.id] = { status: STATUS.PENDING, reason: trigType } mutations++ continue } // Inline execution state[step.id] = { status: STATUS.RUNNING, started_at: new Date().toISOString() } const handler = KIND_HANDLERS[step.kind] if (!handler) { state[step.id] = { status: STATUS.FAILED, error: `no handler for kind "${step.kind}"`, completed_at: new Date().toISOString() } errors.push({ step: step.id, error: 'no handler' }) } else { try { const r2 = await handler(step, { ...ctx, run }) state[step.id] = { status: r2.status, result: r2.result, error: r2.error, started_at: state[step.id].started_at, completed_at: new Date().toISOString(), } if (r2.status === STATUS.FAILED) errors.push({ step: step.id, error: r2.error }) executed.push(step.id) } catch (e) { state[step.id] = { status: STATUS.FAILED, error: e.message, completed_at: new Date().toISOString() } errors.push({ step: step.id, error: e.message }) } } mutations++ progressed = true } if (!progressed) break } // Persist new state + compute run-level status const allDone = def.steps.every(s => { const st = state[s.id]?.status return st === STATUS.DONE || st === STATUS.FAILED || st === STATUS.SKIPPED }) const anyFailed = def.steps.some(s => state[s.id]?.status === STATUS.FAILED) const patch = { step_state: state } if (allDone) { patch.status = anyFailed ? 'failed' : 'completed' patch.completed_at = new Date().toISOString() } if (mutations > 0) { await _persistRun(run, patch) Object.assign(run, patch) } return { run, executed, scheduled, errors, done: allDone } } /** * Mark a scheduled/manual/webhook step as complete externally. Useful for * the scheduler, manual "Je confirme" buttons, or webhook callbacks. * * @param {string} runName * @param {string} stepId * @param {Object} [result] */ async function completeStep (runName, stepId, result = {}) { const run = await _fetchRun(runName) const state = run.step_state state[stepId] = { ...(state[stepId] || {}), status: STATUS.DONE, result, completed_at: new Date().toISOString(), } await _persistRun(run, { step_state: state }) return advanceFlow(runName) } /** * Build the execution context: pulls in the trigger doc + customer. * Kept narrow to limit payload size on downstream template renders. */ async function _buildContext (run, template) { const ctx = { run: { name: run.name, variables: run.variables, step_state: run.step_state }, template: { name: template.name, version: template.version }, variables: run.variables || {}, now: new Date().toISOString(), } if (run.context_doctype && run.context_docname) { try { const r = await erpFetch(`/api/resource/${encodeURIComponent(run.context_doctype)}/${encodeURIComponent(run.context_docname)}`) if (r.status === 200) { ctx.doc = r.data.data ctx.doctype = run.context_doctype ctx.docname = run.context_docname } } catch (e) { log('[flow] context fetch error:', e.message) } } if (run.customer) { try { const r = await erpFetch(`/api/resource/Customer/${encodeURIComponent(run.customer)}`) if (r.status === 200) ctx.customer = r.data.data } catch (e) { log('[flow] customer fetch error:', e.message) } } else if (ctx.doc?.customer) { try { const r = await erpFetch(`/api/resource/Customer/${encodeURIComponent(ctx.doc.customer)}`) if (r.status === 200) ctx.customer = r.data.data } catch { /* noop */ } } return ctx } // ─────────────────────────────────────────────────────────────────────────── // Trigger helpers — called from event hooks (contracts, payments, …) // ─────────────────────────────────────────────────────────────────────────── /** * Find all active templates listening for `eventName` and start them. * Templates with a trigger_condition must match the provided context * (simple ==/!= JSON check, expanded later). * * @returns {Array} list of { template, run, executed, scheduled } results */ async function dispatchEvent (eventName, opts = {}) { const filters = JSON.stringify([['trigger_event', '=', eventName], ['is_active', '=', 1]]) const r = await erpFetch(`/api/resource/${ENC_FT}?filters=${encodeURIComponent(filters)}&fields=${encodeURIComponent(JSON.stringify(['name', 'template_name', 'trigger_condition']))}&limit_page_length=50`) if (r.status !== 200) { log('[flow] dispatchEvent list failed:', r.status) return [] } const templates = r.data.data || [] const results = [] for (const t of templates) { if (t.trigger_condition && !_matchCondition(t.trigger_condition, opts)) continue try { const res = await startFlow(t.name, { ...opts, triggerEvent: eventName }) results.push({ template: t.name, ...res }) } catch (e) { log(`[flow] dispatchEvent "${eventName}" failed for ${t.name}:`, e.message) results.push({ template: t.name, error: e.message }) } } return results } /** Minimal condition match (JSON == check). Extend later for full JSONLogic. */ function _matchCondition (condJson, opts) { if (!condJson) return true try { const c = typeof condJson === 'string' ? JSON.parse(condJson) : condJson if (c['==']) { const [a, b] = c['=='] const left = a?.var ? getPath(opts, a.var) : a return String(left) === String(b) } return true } catch { return true } } // ─────────────────────────────────────────────────────────────────────────── module.exports = { startFlow, advanceFlow, completeStep, dispatchEvent, // for testing KIND_HANDLERS, evalCondition, render, renderDeep, isStepReady, resolveTriggerAt, STATUS, }