feat: support for re-usable global functions in workflows

packages/cli/src/types.ts

@@ -18,6 +18,7 @@ export type CLIExecutionPlan = {
     id?: UUID;
     name?: string;
     steps: Array<CLIJobNode | Trigger>;
+    globals?: string;
   };
 };
 

packages/cli/src/util/load-plan.ts

@@ -156,11 +156,15 @@ const loadOldWorkflow = async (
 };
 
 const fetchFile = async (
-  jobId: string,
-  rootDir: string = '',
-  filePath: string,
+  fileInfo: {
+    id: string;
+    fileType: string;
+    rootDir?: string;
+    filePath: string;
+  },
   log: Logger
 ) => {
+  const { id, rootDir = '', filePath, fileType = 'resource' } = fileInfo;
   try {
     // Special handling for ~ feels like a necessary evil
     const fullPath = filePath.startsWith('~')
@@ -172,7 +176,7 @@ const fetchFile = async (
   } catch (e) {
     abort(
       log,
-      `File not found for job ${jobId}: ${filePath}`,
+      `File not found for ${fileType} ${id}: ${filePath}`,
       undefined,
       `This workflow references a file which cannot be found at ${filePath}\n\nPaths inside the workflow are relative to the workflow.json`
     );
@@ -182,6 +186,21 @@ const fetchFile = async (
   }
 };
 
+const importGlobals = async (
+  plan: CLIExecutionPlan,
+  rootDir: string,
+  log: Logger
+) => {
+  const fnStr = plan.workflow?.globals;
+  if (fnStr && isPath(fnStr)) {
+    // FIXME: fetchFile function isn't generic enough
+    plan.workflow.globals = await fetchFile(
+      { id: '', fileType: 'globals', rootDir, filePath: fnStr },
+      log
+    );
+  }
+};
+
 // TODO this is currently untested in load-plan
 // (but covered a bit in execute tests)
 const importExpressions = async (
@@ -204,26 +223,35 @@ const importExpressions = async (
 
     if (expressionStr && isPath(expressionStr)) {
       job.expression = await fetchFile(
-        job.id || `${idx}`,
-        rootDir,
-        expressionStr,
+        {
+          id: job.id || `${idx}`,
+          rootDir,
+          filePath: expressionStr,
+          fileType: 'job',
+        },
         log
       );
     }
     if (configurationStr && isPath(configurationStr)) {
       const configString = await fetchFile(
-        job.id || `${idx}`,
-        rootDir,
-        configurationStr,
+        {
+          id: job.id || `${idx}`,
+          rootDir,
+          filePath: configurationStr,
+          fileType: 'job configuration',
+        },
         log
       );
       job.configuration = JSON.parse(configString!);
     }
     if (stateStr && isPath(stateStr)) {
       const stateString = await fetchFile(
-        job.id || `${idx}`,
-        rootDir,
-        stateStr,
+        {
+          id: job.id || `${idx}`,
+          rootDir,
+          filePath: stateStr,
+          fileType: 'job state',
+        },
         log
       );
       job.state = JSON.parse(stateString!);
@@ -260,6 +288,9 @@ const loadXPlan = async (
   }
   ensureAdaptors(plan);
 
+  // import global functions
+  await importGlobals(plan, options.baseDir!, logger);
+
   // Note that baseDir should be set up in the default function
   await importExpressions(plan, options.baseDir!, logger);
   // expand shorthand adaptors in the workflow jobs

packages/cli/test/execute/execute.test.ts

@@ -450,3 +450,56 @@ test.serial('run a job which does not return state', async (t) => {
   // Check that no error messages have been logged
   t.is(logger._history.length, 0);
 });
+
+test.serial('globals: use a global function in an operation', async (t) => {
+  const workflow = {
+    workflow: {
+      globals: "export const prefixer = (w) => 'welcome '+w",
+      steps: [
+        {
+          id: 'a',
+          state: { data: { name: 'John' } },
+          expression: `${fn}fn(state=> { state.data.new = prefixer(state.data.name); return state; })`,
+        },
+      ],
+    },
+  };
+
+  mockFs({
+    '/workflow.json': JSON.stringify(workflow),
+  });
+
+  const options = {
+    ...defaultOptions,
+    workflowPath: '/workflow.json',
+  };
+  const result = await handler(options, logger);
+  t.deepEqual(result.data, { name: 'John', new: 'welcome John' });
+});
+
+test.serial('globals: get global functions from a filePath', async (t) => {
+  const workflow = {
+    workflow: {
+      globals: '/my-globals.js',
+      steps: [
+        {
+          id: 'a',
+          state: { data: { name: 'John' } },
+          expression: `${fn}fn(state=> { state.data.new = suffixer(state.data.name); return state; })`,
+        },
+      ],
+    },
+  };
+
+  mockFs({
+    '/workflow.json': JSON.stringify(workflow),
+    '/my-globals.js': `export const suffixer = (w) => w + " goodbye!"`,
+  });
+
+  const options = {
+    ...defaultOptions,
+    workflowPath: '/workflow.json',
+  };
+  const result = await handler(options, logger);
+  t.deepEqual(result.data, { name: 'John', new: 'John goodbye!' });
+});

packages/lexicon/core.d.ts

@@ -34,6 +34,9 @@ export type Workflow = {
   // global credentials
   // (gets applied to every configuration object)
   credentials?: Record<string, any>;
+
+  // a path to a file where functions are defined!
+  globals?: string;
 };
 
 /**

packages/runtime/src/execute/compile-plan.ts

@@ -100,6 +100,7 @@ export default (plan: ExecutionPlan) => {
   const newPlan: CompiledExecutionPlan = {
     workflow: {
       steps: {},
+      globals: workflow.globals,
     },
     options: {
       ...options,

packages/runtime/src/execute/expression.ts

@@ -21,13 +21,14 @@ import {
   assertSecurityKill,
   AdaptorError,
 } from '../errors';
-import type { JobModule, ExecutionContext } from '../types';
+import type { JobModule, ExecutionContext, GlobalsModule } from '../types';
 import { ModuleInfoMap } from '../modules/linker';
 import {
   clearNullState,
   isNullState,
   createNullState,
 } from '../util/null-state';
+import vm from '../modules/experimental-vm';
 
 export type ExecutionErrorWrapper = {
   state: any;
@@ -50,15 +51,24 @@ export default (
     try {
       const timeout = plan.options?.timeout ?? ctx.opts.defaultRunTimeoutMs;
 
+      // prepare global functions to be injected into execution context
+      let funcs = {};
+      if (plan.workflow?.globals)
+        funcs = await prepareGlobals(plan.workflow.globals);
+      const globals = { ...opts.globals, ...funcs };
+
       // Setup an execution context
-      const context = buildContext(input, opts);
+      const context = buildContext(input, { ...opts, globals });
 
+      // FIXME: when expression isn't a string, additional stuff isn't loaded
+      // eg. global functions might not be loaded!
       const { operations, execute } = await prepareJob(
         expression,
         context,
         opts,
         moduleOverrides
       );
+
       // Create the main reducer function
       const reducer = (execute || defaultExecute)(
         ...operations.map((op, idx) =>
@@ -238,3 +248,22 @@ const prepareJob = async (
     return { operations: expression as Operation[] };
   }
 };
+
+const prepareGlobals = async (
+  source: string,
+  opts: Options = {}
+): Promise<GlobalsModule> => {
+  if (typeof source === 'string' && !!source.trim()) {
+    const context = vm.createContext({ console: opts.logger });
+    const funcs = await loadModule(source || '', {
+      context,
+    }).catch((e) => {
+      // mostly syntax errors
+      // repackage errors and throw
+      e.message = `(inside globals) ${e.message}`;
+      throw e;
+    });
+    return funcs;
+  }
+  return {};
+};

packages/runtime/src/types.ts

@@ -41,6 +41,7 @@ export type Lazy<T> = string | T;
 
 export type CompiledExecutionPlan = {
   workflow: {
+    globals?: string;
     steps: Record<StepId, CompiledStep>;
     credentials?: Record<string, any>;
   };
@@ -55,6 +56,8 @@ export type JobModule = {
   // TODO lifecycle hooks
 };
 
+export type GlobalsModule = Record<string, any>;
+
 type NotifyHandler = (
   event: NotifyEvents,
   payload: NotifyEventsLookup[typeof event]

packages/runtime/test/execute/compile-plan.test.ts

@@ -433,6 +433,31 @@ test('throw for a syntax error on a job edge', (t) => {
   }
 });
 
+test('global functions. when undefined', (t) => {
+  const plan: ExecutionPlan = {
+    workflow: {
+      steps: [{ expression: 'x' }, { expression: 'y' }],
+    },
+    options: {},
+  };
+  const compiled = compilePlan(plan);
+  t.is(compiled.workflow.globals, undefined);
+});
+
+test('global functions. perform no transformation', (t) => {
+  const GLOBAL_FUNCTIONS =
+    'export const cleaner = (name) => name.replace("a", "b");';
+  const plan: ExecutionPlan = {
+    workflow: {
+      globals: GLOBAL_FUNCTIONS,
+      steps: [{ expression: 'x' }, { expression: 'y' }],
+    },
+    options: {},
+  };
+  const compiled = compilePlan(plan);
+  t.is(compiled.workflow.globals, GLOBAL_FUNCTIONS);
+});
+
 test('throw for multiple errors', (t) => {
   const plan = {
     workflow: {

packages/runtime/test/execute/expression.test.ts

@@ -51,6 +51,23 @@ test.serial('run a live no-op job with one operation', async (t) => {
   t.deepEqual(state, result);
 });
 
+test.serial('use global function defined in functions', async (t) => {
+  const job = `export default [state=> {state.data.pr = prefixer(state.data.name); return state;}]`;
+
+  const state = createState();
+  // @ts-ignore
+  state.data.name = 'john';
+  const context = createContext();
+  context.plan.workflow = {
+    steps: [] as any,
+    globals: "export const prefixer = (w) => 'welcome '+ w",
+  };
+
+  const result = await execute(context, job, state);
+  // @ts-ignore
+  t.is(result.data.pr, 'welcome ' + state.data.name);
+});
+
 test.serial('run a stringified no-op job with one operation', async (t) => {
   const job = 'export default [(s) => s]';
   const state = createState();

packages/runtime/test/execute/plan.test.ts

@@ -10,9 +10,11 @@ let mockLogger = createMockLogger(undefined, { level: 'debug' });
 
 const createPlan = (
   steps: Job[],
-  options: Partial<CompiledExecutionPlan['options']> = {}
+  options: Partial<CompiledExecutionPlan['options']> = {},
+  globals?: string
 ): ExecutionPlan => ({
   workflow: {
+    globals,
     steps,
   },
   options,
@@ -1204,3 +1206,39 @@ test('Plans log step names for each job start and end', async (t) => {
   const end = logger._find('success', /do-the-thing completed in/i);
   t.regex(end!.message as string, /do-the-thing completed in \d+ms/);
 });
+
+test.serial(
+  'global functions should be scoped per step or job code',
+  async (t) => {
+    const functions = `
+    export const addToBase = ((a) => (b) => { a = a + b; return a })(0); 
+    export const INC = 5;
+    `;
+    const plan = createPlan(
+      [
+        {
+          id: 'a',
+          name: 'do-a',
+          expression:
+            'export default [s => {addToBase(INC); return s;}, s => {state.data.a = addToBase(INC); return state;}]',
+          next: {
+            b: true,
+          },
+        },
+        {
+          id: 'b',
+          name: 'do-b',
+          expression:
+            'export default [s => {addToBase(INC); return s;}, s => {state.data.b = addToBase(INC); return state;}]',
+        },
+      ],
+      {},
+      functions
+    );
+
+    const result: any = await executePlan(plan, {}, {}, mockLogger);
+
+    t.notThrows(() => JSON.stringify(result));
+    t.deepEqual(result.data, { a: 10, b: 10 });
+  }
+);