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,

integration-tests/cli/test/execute-workflow.test.ts

@@ -221,3 +221,55 @@ test.serial(
     });
   }
 );
+
+test.serial(`openfn ${jobsPath}/globals-exp.json`, async (t) => {
+  const res = await run(t.title);
+  t.falsy(res.err);
+  const out = getJSON();
+  t.deepEqual(out, {
+    alter: 'heartsfx',
+    data: {},
+    final: 'some-big-valueheartsfx',
+    val: 'some-big-value',
+  });
+});
+
+test.serial(`openfn ${jobsPath}/globals-path.json`, async (t) => {
+  const res = await run(t.title);
+  t.falsy(res.err);
+  const out = getJSON();
+  t.deepEqual(out, {
+    alter: 'heart.path.value',
+    data: {},
+    final: 'path-valueheart.path.value',
+    val: 'path-value',
+  });
+});
+
+test.serial(
+  `openfn ${jobsPath}/globals-job.js --globals="export const suffixer = w => w + '-some-suffix'" -a common`,
+  async (t) => {
+    const res = await run(t.title);
+    t.falsy(res.err);
+    const out = getJSON();
+    t.deepEqual(out, {
+      data: {
+        result: 'love-some-suffix',
+      },
+    });
+  }
+);
+
+test.serial(
+  `openfn ${jobsPath}/globals-job.js --globals ${jobsPath}/globals-path-file.js -a common`,
+  async (t) => {
+    const res = await run(t.title);
+    t.falsy(res.err);
+    const out = getJSON();
+    t.deepEqual(out, {
+      data: {
+        result: 'love-humble-suffix',
+      },
+    });
+  }
+);

integration-tests/cli/test/fixtures/globals-exp.json

@@ -0,0 +1,19 @@
+{
+  "workflow": {
+    "globals": "export const BIG_VAL = 'some-big-value'; export function suffix(name){ return name + 'sfx'}",
+    "steps": [
+      {
+        "adaptor": "common",
+        "expression": "fn(state=> ({val: BIG_VAL, alter: suffix('heart')}))",
+        "next": {
+          "b": true
+        }
+      },
+      {
+        "id": "b",
+        "adaptor": "common",
+        "expression": "fn((state) => { state.final = state.val + state.alter; return state; });"
+      }
+    ]
+  }
+}

integration-tests/cli/test/fixtures/globals-job.js

@@ -0,0 +1,3 @@
+fn(() => {
+  return { data: { result: suffixer('love') } }
+})

integration-tests/cli/test/fixtures/globals-path-file.js

@@ -0,0 +1,7 @@
+export const BIG_VAL = 'path-value';
+
+export function suffix(name) {
+    return name + '.path.value'
+}
+
+export const suffixer = w => w + '-humble-suffix'

integration-tests/cli/test/fixtures/globals-path.json

@@ -0,0 +1,19 @@
+{
+  "workflow": {
+    "globals": "./globals-path-file.js",
+    "steps": [
+      {
+        "adaptor": "common",
+        "expression": "fn(state=> ({val: BIG_VAL, alter: suffix('heart')}))",
+        "next": {
+          "b": true
+        }
+      },
+      {
+        "id": "b",
+        "adaptor": "common",
+        "expression": "fn((state) => { state.final = state.val + state.alter; return state; });"
+      }
+    ]
+  }
+}

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/compile/command.ts

@@ -17,6 +17,7 @@ export type CompileOptions = Pick<
   | 'repoDir'
   | 'path'
   | 'useAdaptorsMonorepo'
+  | 'globals'
 > & {
   workflow?: Opts['workflow'];
   repoDir?: string;

packages/cli/src/compile/compile.ts

@@ -1,4 +1,8 @@
-import compile, { preloadAdaptorExports, Options } from '@openfn/compiler';
+import compile, {
+  preloadAdaptorExports,
+  Options,
+  getExports,
+} from '@openfn/compiler';
 import { getModulePath } from '@openfn/runtime';
 import type {
   ExecutionPlan,
@@ -72,11 +76,16 @@ const compileWorkflow = async (
   opts: CompileOptions,
   log: Logger
 ) => {
+  let globalsIgnoreList: string[] = [];
+  if (plan.workflow.globals)
+    globalsIgnoreList = getExports(plan.workflow.globals);
+
   for (const step of plan.workflow.steps) {
     const job = step as Job;
     const jobOpts = {
       ...opts,
       adaptors: job.adaptors ?? opts.adaptors,
+      ignoreImports: globalsIgnoreList,
     };
     if (job.expression) {
       const { code, map } = await compileJob(

packages/cli/src/execute/command.ts

@@ -33,6 +33,7 @@ export type ExecuteOptions = Required<
     | 'timeout'
     | 'useAdaptorsMonorepo'
     | 'workflowPath'
+    | 'globals'
   >
 > &
   Pick<Opts, 'monorepoPath' | 'repoDir'>;

packages/cli/src/options.ts

@@ -18,6 +18,7 @@ export type Opts = {
   command?: CommandList;
   baseDir?: string;
 
+  globals?: string;
   adaptor?: boolean | string;
   adaptors?: string[];
   apolloUrl?: string;

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

@@ -21,6 +21,7 @@ const loadPlan = async (
     | 'baseDir'
     | 'expandAdaptors'
     | 'path'
+    | 'globals'
   > & {
     workflow?: Opts['workflow'];
   },
@@ -141,7 +142,10 @@ const maybeAssign = (a: any, b: any, keys: Array<keyof WorkflowOptions>) => {
 };
 
 const loadExpression = async (
-  options: Pick<Opts, 'expressionPath' | 'adaptors' | 'monorepoPath'>,
+  options: Pick<
+    Opts,
+    'expressionPath' | 'adaptors' | 'monorepoPath' | 'globals'
+  >,
   logger: Logger
 ): Promise<ExecutionPlan> => {
   const expressionPath = options.expressionPath!;
@@ -165,6 +169,7 @@ const loadExpression = async (
       workflow: {
         name,
         steps: [step],
+        globals: options.globals,
       },
       options: wfOptions,
     };
@@ -212,11 +217,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('~')
@@ -228,7 +236,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`
     );
@@ -238,6 +246,21 @@ const fetchFile = async (
   }
 };
 
+const importGlobals = async (
+  plan: CLIExecutionPlan,
+  rootDir: string,
+  log: Logger
+) => {
+  const fnStr = plan.workflow?.globals;
+  if (!fnStr) return;
+  if (isPath(fnStr))
+    plan.workflow.globals = await fetchFile(
+      { name: 'globals', rootDir, filePath: fnStr },
+      log
+    );
+  else plan.workflow.globals = fnStr;
+};
+
 // TODO this is currently untested in load-plan
 // (but covered a bit in execute tests)
 const importExpressions = async (
@@ -260,26 +283,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!);
@@ -303,7 +332,10 @@ const ensureAdaptors = (plan: CLIExecutionPlan) => {
 
 const loadXPlan = async (
   plan: CLIExecutionPlan,
-  options: Pick<Opts, 'monorepoPath' | 'baseDir' | 'expandAdaptors'>,
+  options: Pick<
+    Opts,
+    'monorepoPath' | 'baseDir' | 'expandAdaptors' | 'globals'
+  >,
   logger: Logger,
   defaultName: string = ''
 ) => {
@@ -316,6 +348,11 @@ const loadXPlan = async (
   }
   ensureAdaptors(plan);
 
+  // import global functions
+  // if globals is provided via cli argument. it takes precedence
+  if (options.globals) plan.workflow.globals = options.globals;
+  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!' });
+});