feat: support for re-usable global functions in workflows

integration-tests/cli/test/errors.test.ts

@@ -88,7 +88,11 @@ test.serial("can't find config referenced in a workflow", async (t) => {
 
   const stdlogs = extractLogs(stdout);
 
-  assertLog(t, stdlogs, /File not found for job 1: does-not-exist.js/i);
+  assertLog(
+    t,
+    stdlogs,
+    /File not found for job configuration 1: does-not-exist.js/i
+  );
   assertLog(
     t,
     stdlogs,

packages/cli/README.md

@@ -317,6 +317,7 @@ A workflow has a structure like this:
 {
   "workflow": {
     "name": "my-workflow", // human readable name used in logging
+    "globals": "./common-funcs.js", // code or path to functions that can be accessed in any step. globally scoped
     "steps": [
       {
         "name": "a", // human readable name used in logging

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,14 @@ const loadOldWorkflow = async (
 };
 
 const fetchFile = async (
-  jobId: string,
-  rootDir: string = '',
-  filePath: string,
+  fileInfo: {
+    name: string;
+    rootDir?: string;
+    filePath: string;
+  },
   log: Logger
 ) => {
+  const { rootDir = '', filePath, name } = fileInfo;
   try {
     // Special handling for ~ feels like a necessary evil
     const fullPath = filePath.startsWith('~')
@@ -172,7 +175,7 @@ const fetchFile = async (
   } catch (e) {
     abort(
       log,
-      `File not found for job ${jobId}: ${filePath}`,
+      `File not found for ${name}: ${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 +185,20 @@ const fetchFile = async (
   }
 };
 
+const importGlobals = async (
+  plan: CLIExecutionPlan,
+  rootDir: string,
+  log: Logger
+) => {
+  const fnStr = plan.workflow?.globals;
+  if (fnStr && isPath(fnStr)) {
+    plan.workflow.globals = await fetchFile(
+      { name: '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 +221,32 @@ const importExpressions = async (
 
     if (expressionStr && isPath(expressionStr)) {
       job.expression = await fetchFile(
-        job.id || `${idx}`,
-        rootDir,
-        expressionStr,
+        {
+          name: `job ${job.id || idx}`,
+          rootDir,
+          filePath: expressionStr,
+        },
         log
       );
     }
     if (configurationStr && isPath(configurationStr)) {
       const configString = await fetchFile(
-        job.id || `${idx}`,
-        rootDir,
-        configurationStr,
+        {
+          name: `job configuration ${job.id || idx}`,
+          rootDir,
+          filePath: configurationStr,
+        },
         log
       );
       job.configuration = JSON.parse(configString!);
     }
     if (stateStr && isPath(stateStr)) {
       const stateString = await fetchFile(
-        job.id || `${idx}`,
-        rootDir,
-        stateStr,
+        {
+          name: `job state ${job.id || idx}`,
+          rootDir,
+          filePath: stateStr,
+        },
         log
       );
       job.state = JSON.parse(stateString!);
@@ -260,6 +283,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/lexicon/lightning.d.ts

@@ -34,6 +34,8 @@ export type LightningPlan = {
   edges: LightningEdge[];
 
   options?: LightningPlanOptions;
+
+  globals?: string;
 };
 
 /**

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

@@ -106,6 +106,10 @@ export default (plan: ExecutionPlan) => {
       start: options.start ?? workflow.steps[0]?.id!,
     },
   };
+
+  if (typeof workflow.globals === 'string')
+    newPlan.workflow.globals = workflow.globals;
+
   if (workflow.credentials) {
     newPlan.workflow.credentials = workflow.credentials;
   }

packages/runtime/src/execute/expression.ts

@@ -21,7 +21,7 @@ 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,
@@ -50,15 +50,26 @@ export default (
     try {
       const timeout = plan.options?.timeout ?? ctx.opts.defaultRunTimeoutMs;
 
+      // prepare global functions to be injected into execution context
+      const globals = {
+        ...opts.globals,
+        ...(plan.workflow?.globals
+          ? await prepareGlobals(plan.workflow.globals, opts)
+          : {}),
+      };
+
       // 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 +249,21 @@ 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 = buildContext({}, opts);
+    return await loadModule(source || '', {
+      context,
+    }).catch((e) => {
+      // mostly syntax errors
+      // repackage errors and throw
+      e.message = `[globals] ${e.message}`;
+      throw e;
+    });
+  }
+  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,43 @@ test('throw for a syntax error on a job edge', (t) => {
   }
 });
 
+test('globals should be undefined if no globals are passed', (t) => {
+  const plan: ExecutionPlan = {
+    workflow: {
+      steps: [{ expression: 'x' }, { expression: 'y' }],
+    },
+    options: {},
+  };
+  const compiled = compilePlan(plan);
+  t.is(compiled.workflow.globals, undefined);
+});
+
+test('globals should be undefined if non-string is passed', (t) => {
+  const plan: ExecutionPlan = {
+    workflow: {
+      globals: { some: 'val' } as any,
+      steps: [{ expression: 'x' }, { expression: 'y' }],
+    },
+    options: {},
+  };
+  const compiled = compilePlan(plan);
+  t.is(compiled.workflow.globals, undefined);
+});
+
+test('global functions should perform no transformation inline code', (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();