Merge pull request #797 from OpenFn/multi-adaptors-compiler-imports

Support multiple adaptors in the compiler

.changeset/neat-lies-begin.md

@@ -0,0 +1,5 @@
+---
+'@openfn/engine-multi': minor
+---
+
+Support multiple adaptors in job structures

.changeset/sharp-needles-peel.md

@@ -0,0 +1,5 @@
+---
+'@openfn/ws-worker': patch
+---
+
+Update worker to use adaptors as an array on xplans. Internal only change.

.changeset/sharp-tips-pretend.md

@@ -0,0 +1,5 @@
+---
+'@openfn/compiler': minor
+---
+
+support multiple adaptors when adding imports

.changeset/wild-donuts-check.md

@@ -0,0 +1,5 @@
+---
+'@openfn/cli': patch
+---
+
+Support multiple adaptors

integration-tests/execute/src/execute.ts

@@ -6,10 +6,12 @@ const execute = async (job: string, state: any, adaptor = 'common') => {
   // compile with common and dumb imports
   const options = {
     'add-imports': {
-      adaptor: {
-        name: `@openfn/language-${adaptor}`,
-        exportAll: true,
-      },
+      adaptors: [
+        {
+          name: `@openfn/language-${adaptor}`,
+          exportAll: true,
+        },
+      ],
     },
   };
   const compiled = compiler(job, options);

packages/cli/src/compile/compile.ts

@@ -56,10 +56,8 @@ const compileWorkflow = async (
     const job = step as Job;
     const jobOpts = {
       ...opts,
+      adaptors: job.adaptors ?? opts.adaptors,
     };
-    if (job.adaptor) {
-      jobOpts.adaptors = [job.adaptor];
-    }
     if (job.expression) {
       job.expression = await compileJob(
         job.expression as string,
@@ -115,37 +113,41 @@ export const loadTransformOptions = async (
   // If an adaptor is passed in, we need to look up its declared exports
   // and pass them along to the compiler
   if (opts.adaptors?.length && opts.ignoreImports != true) {
-    let exports;
-    const [pattern] = opts.adaptors;
-    const [specifier] = pattern.split('=');
-
-    // Preload exports from a path, optionally logging errors in case of a failure
-    log.debug(`Trying to preload types for ${specifier}`);
-    const path = await resolveSpecifierPath(pattern, opts.repoDir, log);
-    if (path) {
-      try {
-        exports = await preloadAdaptorExports(
-          path,
-          opts.useAdaptorsMonorepo,
-          log
-        );
-      } catch (e) {
-        log.error(`Failed to load adaptor typedefs from path ${path}`);
-        log.error(e);
+    const adaptorsConfig = [];
+    for (const adaptorInput of opts.adaptors) {
+      let exports;
+      const [specifier] = adaptorInput.split('=');
+
+      // Preload exports from a path, optionally logging errors in case of a failure
+      log.debug(`Trying to preload types for ${specifier}`);
+      const path = await resolveSpecifierPath(adaptorInput, opts.repoDir, log);
+      if (path) {
+        try {
+          exports = await preloadAdaptorExports(
+            path,
+            opts.useAdaptorsMonorepo,
+            log
+          );
+        } catch (e) {
+          log.error(`Failed to load adaptor typedefs from path ${path}`);
+          log.error(e);
+        }
       }
-    }
 
-    if (!exports || exports.length === 0) {
-      log.debug(`No module exports found for ${pattern}`);
-    }
+      if (!exports || exports.length === 0) {
+        log.debug(`No module exports found for ${adaptorInput}`);
+      }
 
-    options['add-imports'] = {
-      ignore: opts.ignoreImports as string[],
-      adaptor: {
+      adaptorsConfig.push({
         name: stripVersionSpecifier(specifier),
         exports,
         exportAll: true,
-      },
+      });
+    }
+
+    options['add-imports'] = {
+      ignore: opts.ignoreImports as string[],
+      adaptors: adaptorsConfig,
     };
   }
 

packages/cli/src/execute/execute.ts

@@ -67,14 +67,16 @@ export function parseAdaptors(plan: ExecutionPlan) {
 
   const adaptors: ModuleInfoMap = {};
 
-  // TODO what if there are different versions of the same adaptor?
-  // This structure can't handle it - we'd need to build it for every job
   Object.values(plan.workflow.steps).forEach((step) => {
     const job = step as Job;
-    if (job.adaptor) {
-      const { name, ...maybeVersionAndPath } = extractInfo(job.adaptor);
+    // Usually every job should have an adaptors array
+    // But there are a couple of cases mostly in test, when validation is skipped,
+    // when the array may not be set
+    // It's mostly redundant nbut harmless to optionally chain here
+    job.adaptors?.forEach((adaptor) => {
+      const { name, ...maybeVersionAndPath } = extractInfo(adaptor);
       adaptors[name] = maybeVersionAndPath;
-    }
+    });
   });
 
   return adaptors;

packages/cli/src/execute/get-autoinstall-targets.ts

@@ -5,9 +5,11 @@ const getAutoinstallTargets = (plan: ExecutionPlan) => {
   Object.values(plan.workflow.steps).forEach((step) => {
     const job = step as Job;
     // Do not autoinstall adaptors with a path
-    if (job.adaptor && !/=/.test(job.adaptor)) {
-      adaptors[job.adaptor] = true;
-    }
+    job.adaptors
+      ?.filter((adaptor) => !/=/.test(adaptor))
+      .forEach((adaptor) => {
+        adaptors[adaptor] = true;
+      });
   });
   return Object.keys(adaptors);
 };

packages/cli/src/execute/handler.ts

@@ -85,7 +85,7 @@ const executeHandler = async (options: ExecuteOptions, logger: Logger) => {
     if (options.start) {
       customStart = matchStep(
         plan,
-        options.start ?? plan.options.start,
+        options.start ?? plan.options!.start,
         'start',
         logger
       );
@@ -95,7 +95,7 @@ const executeHandler = async (options: ExecuteOptions, logger: Logger) => {
     if (options.end) {
       customEnd = matchStep(
         plan,
-        options.end ?? plan.options.end,
+        options.end ?? plan.options!.end,
         'end',
         logger
       );
@@ -113,8 +113,8 @@ const executeHandler = async (options: ExecuteOptions, logger: Logger) => {
   const finalPlan = {
     ...plan,
     options: {
-      ...plan.options,
-      start: customStart || plan.options.start,
+      ...plan.options!,
+      start: customStart || plan.options!.start,
       end: customEnd,
     },
     workflow: plan.workflow,

packages/cli/src/types.ts

@@ -1,4 +1,7 @@
 // the executionPLan for the CLI is a bit differnt to the runtime's perspective
+
+import { Trigger, UUID, WorkflowOptions } from '@openfn/lexicon';
+
 // Ie config can be a string
 export type JobNodeID = string;
 
@@ -9,19 +12,26 @@ export type OldCLIWorkflow = {
 };
 
 export type CLIExecutionPlan = {
-  id?: string; // UUID for this plan
-  start?: JobNodeID;
-  jobs: CLIJobNode[];
+  id?: string;
+  options?: WorkflowOptions;
+  workflow: {
+    id?: UUID;
+    name?: string;
+    steps: Array<CLIJobNode | Trigger>;
+  };
 };
 
 export type CLIJobNode = {
   id?: string;
-  adaptor?: string;
   expression?: string; // path or expression
   configuration?: string | object; // path or credential object
   data?: any;
-
   next?: string | Record<JobNodeID, true | CLIJobEdge>;
+
+  // We can accept a single adaptor or multiple
+  // The CLI will convert it to adaptors as an array
+  adaptor?: string;
+  adaptors?: string[];
 };
 
 export type CLIJobEdge = {

packages/cli/src/util/expand-adaptors.ts

@@ -14,8 +14,6 @@ const expand = (name: string) => {
 
 type ArrayOrPlan<T> = T extends string[] ? string[] : ExecutionPlan;
 
-// TODO typings here aren't good,I can't get this to work!
-// At least this looks nice externally
 export default <T extends Array<string> | ExecutionPlan>(
   input: T
 ): ArrayOrPlan<T> => {
@@ -26,8 +24,8 @@ export default <T extends Array<string> | ExecutionPlan>(
   const plan = input as ExecutionPlan;
   Object.values(plan.workflow.steps).forEach((step) => {
     const job = step as Job;
-    if (job.adaptor) {
-      job.adaptor = expand(job.adaptor);
+    if (job.adaptors) {
+      job.adaptors = job.adaptors.map(expand);
     }
   });
 

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

@@ -8,7 +8,7 @@ import mapAdaptorsToMonorepo from './map-adaptors-to-monorepo';
 import type { ExecutionPlan, Job, WorkflowOptions } from '@openfn/lexicon';
 import type { Opts } from '../options';
 import type { Logger } from './logger';
-import type { OldCLIWorkflow } from '../types';
+import type { CLIExecutionPlan, CLIJobNode, OldCLIWorkflow } from '../types';
 
 const loadPlan = async (
   options: Pick<
@@ -36,6 +36,7 @@ const loadPlan = async (
 
   const json = await loadJson(jsonPath!, logger);
   const defaultName = path.parse(jsonPath!).name;
+
   if (json.workflow) {
     return loadXPlan(json, options, logger, defaultName);
   } else {
@@ -94,21 +95,17 @@ const loadExpression = async (
     const expression = await fs.readFile(expressionPath, 'utf8');
     const name = path.parse(expressionPath).name;
 
-    const step: Job = { expression };
-
-    // The adaptor should have been expanded nicely already, so we don't need intervene here
-    if (options.adaptors) {
-      const [adaptor] = options.adaptors;
-      if (adaptor) {
-        step.adaptor = adaptor;
-      }
-    }
+    const step: Job = {
+      expression,
+      // The adaptor should have been expanded nicely already, so we don't need intervene here
+      adaptors: options.adaptors ?? [],
+    };
 
     const wfOptions: WorkflowOptions = {};
     // TODO support state props to remove?
     maybeAssign(options, wfOptions, ['timeout']);
 
-    const plan: ExecutionPlan = {
+    const plan: CLIExecutionPlan = {
       workflow: {
         name,
         steps: [step],
@@ -126,7 +123,7 @@ const loadExpression = async (
     );
 
     // This will never execute
-    return {} as ExecutionPlan;
+    return {} as CLIExecutionPlan;
   }
 };
 
@@ -188,7 +185,7 @@ const fetchFile = async (
 // TODO this is currently untested in load-plan
 // (but covered a bit in execute tests)
 const importExpressions = async (
-  plan: ExecutionPlan,
+  plan: CLIExecutionPlan,
   rootDir: string,
   log: Logger
 ) => {
@@ -234,8 +231,22 @@ const importExpressions = async (
   }
 };
 
+// Allow users to specify a single adaptor on a job,
+// but convert the internal representation into an array
+const ensureAdaptors = (plan: CLIExecutionPlan) => {
+  Object.values(plan.workflow.steps).forEach((step) => {
+    const job = step as CLIJobNode;
+    if (job.adaptor) {
+      job.adaptors = [job.adaptor];
+      delete job.adaptor;
+    }
+    // Also, ensure there is an empty adaptors array, which makes everything else easier
+    job.adaptors ??= [];
+  });
+};
+
 const loadXPlan = async (
-  plan: ExecutionPlan,
+  plan: CLIExecutionPlan,
   options: Pick<Opts, 'monorepoPath' | 'baseDir' | 'expandAdaptors'>,
   logger: Logger,
   defaultName: string = ''
@@ -247,6 +258,8 @@ const loadXPlan = async (
   if (!plan.workflow.name && defaultName) {
     plan.workflow.name = defaultName;
   }
+  ensureAdaptors(plan);
+
   // Note that baseDir should be set up in the default function
   await importExpressions(plan, options.baseDir!, logger);
   // expand shorthand adaptors in the workflow jobs
@@ -261,5 +274,5 @@ const loadXPlan = async (
 
   logger.info(`Loaded workflow ${plan.workflow.name ?? ''}`);
 
-  return plan;
+  return plan as ExecutionPlan;
 };

packages/cli/src/util/map-adaptors-to-monorepo.ts

@@ -58,8 +58,10 @@ const mapAdaptorsToMonorepo = (
     const plan = input as ExecutionPlan;
     Object.values(plan.workflow.steps).forEach((step) => {
       const job = step as Job;
-      if (job.adaptor) {
-        job.adaptor = updatePath(job.adaptor, monorepoPath, log);
+      if (job.adaptors) {
+        job.adaptors = job.adaptors.map((a) =>
+          updatePath(a, monorepoPath, log)
+        );
       }
     });