Support file path in job body (#742)

* support file paths

* add changeset file

* add test

* refactor for easy readability

* update fixtures to use body content

* move relevant functions to pull.ts

.changeset/swift-dingos-know.md

@@ -0,0 +1,6 @@
+---
+'@openfn/cli': minor
+'@openfn/deploy': minor
+---
+
+Deploy: allow job body to be loaded from a file path in workflow.yaml

packages/cli/src/pull/handler.ts

@@ -6,6 +6,7 @@ import {
   getProject,
   getSpec,
   getStateFromProjectPayload,
+  syncRemoteSpec,
 } from '@openfn/deploy';
 import type { Logger } from '../util/logger';
 import { PullOptions } from '../pull/command';
@@ -47,12 +48,6 @@ async function pullHandler(options: PullOptions, logger: Logger) {
     // Build the state.json
     const state = getStateFromProjectPayload(project!);
 
-    // Write the final project state to disk
-    await fs.writeFile(
-      path.resolve(config.statePath),
-      JSON.stringify(state, null, 2)
-    );
-
     logger.always('Downloading the project spec (as YAML) from the server.');
     // Get the project.yaml from Lightning
     const queryParams = new URLSearchParams();
@@ -85,9 +80,20 @@ async function pullHandler(options: PullOptions, logger: Logger) {
     const resolvedPath = path.resolve(config.specPath);
     logger.debug('reading spec from', resolvedPath);
 
-    // Write the yaml to disk
-    // @ts-ignore
-    await fs.writeFile(resolvedPath, res.body);
+    const updatedSpec = await syncRemoteSpec(
+      await res.text(),
+      state,
+      config,
+      logger
+    );
+
+    // Write the final project state and yaml to disk
+    await fs.writeFile(
+      path.resolve(config.statePath),
+      JSON.stringify(state, null, 2)
+    );
+
+    await fs.writeFile(resolvedPath, updatedSpec);
 
     // Read the spec back in a parsed yaml
     const spec = await getSpec(resolvedPath);

packages/deploy/src/index.ts

@@ -10,6 +10,7 @@ import {
   toProjectPayload,
   getStateFromProjectPayload,
 } from './stateTransform';
+import { syncRemoteSpec } from './pull';
 import { deployProject, getProject } from './client';
 import { DeployError } from './deployError';
 import { Logger } from '@openfn/logger';
@@ -33,6 +34,7 @@ export {
   mergeSpecIntoState,
   mergeProjectPayloadIntoState,
   getStateFromProjectPayload,
+  syncRemoteSpec,
 };
 
 export async function getConfig(path?: string): Promise<DeployConfig> {
@@ -91,7 +93,7 @@ function writeState(config: DeployConfig, nextState: {}): Promise<void> {
 export async function getSpec(path: string) {
   try {
     const body = await readFile(path, 'utf8');
-    return parseAndValidate(body);
+    return parseAndValidate(body, path);
   } catch (error: any) {
     if (error.code === 'ENOENT') {
       throw new DeployError(`File not found: ${path}`, 'SPEC_ERROR');

packages/deploy/src/pull.ts

@@ -0,0 +1,122 @@
+import YAML, { Pair, Scalar, isPair, isScalar } from 'yaml';
+import { DeployConfig, ProjectState, SpecJob } from './types';
+import { Logger } from '@openfn/logger';
+import { writeFile } from 'fs/promises';
+import path from 'path';
+import { getState, getSpec } from './index';
+
+async function getAllSpecJobs(
+  config: DeployConfig,
+  logger: Logger
+): Promise<SpecJob[]> {
+  const jobs: SpecJob[] = [];
+
+  try {
+    const state = await getState(config.statePath);
+    const spec = await getSpec(config.specPath);
+
+    for (const [workflowKey, workflow] of Object.entries(spec.doc.workflows)) {
+      if (workflow.jobs) {
+        for (const [jobKey, specJob] of Object.entries(workflow.jobs)) {
+          const stateJob = state.workflows[workflowKey]?.jobs[jobKey];
+          stateJob &&
+            jobs.push({
+              id: stateJob.id,
+              name: specJob.name,
+              adaptor: specJob.adaptor,
+              body: specJob.body,
+            });
+        }
+      }
+    }
+  } catch (error: any) {
+    logger.debug(`Could not read the spec and state: ${error.message}`);
+  }
+
+  return jobs;
+}
+
+async function extractJobsToDisk(
+  specBody: string,
+  state: ProjectState,
+  oldJobs: SpecJob[],
+  config: DeployConfig
+): Promise<string> {
+  function isPairWithScalarKey(
+    node: any
+  ): node is Pair & { key: Scalar & { value: string } } {
+    return (
+      isPair(node) && isScalar(node.key) && typeof node.key.value === 'string'
+    );
+  }
+
+  const doc = YAML.parseDocument(specBody);
+
+  await YAML.visitAsync(doc, {
+    async Pair(_, pair: any, pairPath) {
+      if (
+        !pair.key ||
+        pair.key.value !== 'body' ||
+        !isScalar(pair.value) ||
+        pairPath.length <= 6
+      ) {
+        return;
+      }
+
+      const jobPair = pairPath[pairPath.length - 2];
+      const workflowPair = pairPath[pairPath.length - 6];
+
+      if (!isPairWithScalarKey(jobPair) || !isPairWithScalarKey(workflowPair)) {
+        return;
+      }
+
+      const jobKey = jobPair.key.value;
+      const workflowKey = workflowPair.key.value;
+
+      // find the job in the state
+      const stateJob = state.workflows[workflowKey]?.jobs[jobKey];
+
+      if (!stateJob) {
+        return;
+      }
+
+      // check if the state job is in the old spec jobs
+      const oldSpecJob = oldJobs.find((job) => job.id === stateJob.id);
+
+      if (!oldSpecJob || typeof oldSpecJob?.body !== 'object') {
+        return;
+      }
+
+      const oldSpecJobPath = oldSpecJob.body.path;
+
+      if (oldSpecJobPath) {
+        const basePath = path.dirname(config.specPath);
+        const resolvedPath = path.resolve(basePath, oldSpecJobPath);
+        await writeFile(resolvedPath, pair.value.value);
+
+        // set the body path in the spec
+        const map = doc.createNode({ path: oldSpecJobPath });
+
+        pair.value = map;
+      }
+    },
+  });
+
+  return doc.toString();
+}
+
+export async function syncRemoteSpec(
+  remoteSpecBody: string,
+  newState: ProjectState,
+  config: DeployConfig,
+  logger: Logger
+): Promise<string> {
+  try {
+    const oldSpecJobs = await getAllSpecJobs(config, logger);
+
+    return extractJobsToDisk(remoteSpecBody, newState, oldSpecJobs, config);
+  } catch (error: any) {
+    logger.warn(`Could not update spec job body paths: ${error.message}`);
+    return remoteSpecBody;
+  }
+}

packages/deploy/src/stateTransform.ts

@@ -5,6 +5,7 @@ import {
   ProjectSpec,
   ProjectState,
   SpecEdge,
+  SpecJobBody,
   StateEdge,
   WorkflowSpec,
   WorkflowState,
@@ -20,6 +21,14 @@ import {
 import { DeployError } from './deployError';
 import { Logger } from '@openfn/logger/dist';
 
+function stringifyJobBody(body: SpecJobBody): string {
+  if (typeof body === 'object') {
+    return body.content;
+  } else {
+    return body;
+  }
+}
+
 function mergeJobs(
   stateJobs: WorkflowState['jobs'],
   specJobs: WorkflowSpec['jobs']
@@ -33,7 +42,7 @@ function mergeJobs(
             id: crypto.randomUUID(),
             name: specJob.name,
             adaptor: specJob.adaptor,
-            body: specJob.body,
+            body: stringifyJobBody(specJob.body),
           },
         ];
       }
@@ -49,7 +58,7 @@ function mergeJobs(
             id: stateJob.id,
             name: specJob.name,
             adaptor: specJob.adaptor,
-            body: specJob.body,
+            body: stringifyJobBody(specJob.body),
           },
         ];
       }

packages/deploy/src/types.ts

@@ -1,11 +1,25 @@
-export type Job = {
+export type StateJob = {
   id?: string;
   name: string;
   adaptor: string;
   body: string;
   delete?: boolean;
 };
 
+export type SpecJobBody =
+  | string
+  | {
+      path?: string;
+      content: string;
+    };
+
+export type SpecJob = {
+  id?: string;
+  name: string;
+  adaptor: string;
+  body: SpecJobBody;
+};
+
 export type Trigger = {
   id?: string;
   type?: string;
@@ -38,7 +52,7 @@ export type SpecEdge = {
 export type WorkflowSpec = {
   id?: string;
   name: string;
-  jobs?: Record<string | symbol, Job>;
+  jobs?: Record<string | symbol, SpecJob>;
   triggers?: Record<string | symbol, Trigger>;
   edges?: Record<string | symbol, SpecEdge>;
 };
@@ -52,7 +66,7 @@ export interface ProjectSpec {
 export interface WorkflowState {
   id: string;
   name: string;
-  jobs: Record<string | symbol, Concrete<Job>>;
+  jobs: Record<string | symbol, Concrete<StateJob>>;
   triggers: Record<string | symbol, Concrete<Trigger>>;
   edges: Record<string | symbol, Concrete<StateEdge>>;
   delete?: boolean;
@@ -78,7 +92,7 @@ export interface ProjectPayload {
     id: string;
     name: string;
     project_id?: string;
-    jobs: Concrete<Job>[];
+    jobs: Concrete<StateJob>[];
     triggers: Concrete<Trigger>[];
     edges: Concrete<StateEdge>[];
   }[];

packages/deploy/src/validator.ts

@@ -1,5 +1,7 @@
-import YAML, { YAMLMap, isMap, isPair } from 'yaml';
+import YAML, { YAMLMap, isMap, isPair, isScalar } from 'yaml';
 import { ProjectSpec } from './types';
+import { readFile } from 'fs/promises';
+import path from 'path';
 
 export interface Error {
   context: any;
@@ -8,12 +10,16 @@ export interface Error {
   range?: [number, number, number];
 }
 
-export function parseAndValidate(input: string): {
+export async function parseAndValidate(
+  input: string,
+  specPath: string = '.'
+): Promise<{
   errors: Error[];
   doc: ProjectSpec;
-} {
+}> {
   let errors: Error[] = [];
   const doc = YAML.parseDocument(input);
+  const basePath = path.dirname(specPath);
 
   function ensureUniqueId(key: string, arr: string[]) {
     if (arr.includes(key)) {
@@ -97,8 +103,8 @@ export function parseAndValidate(input: string): {
     }
   }
 
-  YAML.visit(doc, {
-    Pair(_, pair: any) {
+  await YAML.visitAsync(doc, {
+    async Pair(_, pair: any, pairPath) {
       if (pair.key && pair.key.value === 'workflows') {
         if (pair.value.value === null) {
           errors.push({
@@ -124,6 +130,37 @@ export function parseAndValidate(input: string): {
         }
       }
 
+      if (
+        pair.key &&
+        pair.key.value === 'body' &&
+        pairPath.length > 4 &&
+        isMap(pair.value)
+      ) {
+        const pathValue = pair.value.get('path');
+        const grandPair = pairPath[pairPath.length - 4];
+
+        if (
+          isPair(grandPair) &&
+          isScalar(grandPair.key) &&
+          grandPair.key.value === 'jobs' &&
+          typeof pathValue === 'string'
+        ) {
+          const filePath = path.resolve(basePath, pathValue);
+          try {
+            const content = await readFile(filePath, 'utf8');
+            pair.value.set('content', content);
+          } catch (error: any) {
+            errors.push({
+              path: `job/body/path`,
+              context: pair,
+              message: `Failed to read file ${pathValue}: ${error.message}`,
+              range: pair.value.range,
+            });
+          }
+          return undefined;
+        }
+      }
+
       if (pair.key && pair.key.value === 'condition_expression') {
         if (typeof pair.value.value !== 'string') {
           pair.value.value = String(pair.value.value);

packages/deploy/test/fixtures.ts

@@ -11,7 +11,10 @@ export function fullExampleSpec() {
           'job-a': {
             name: 'job a',
             adaptor: '@openfn/language-common@latest',
-            body: '',
+            body: {
+              path: 'somefile.js',
+              content: '',
+            },
           },
           'job-b': {
             name: 'job b',

packages/deploy/test/util.ts

@@ -0,0 +1,14 @@
+import mock from 'mock-fs';
+import path from 'node:path';
+
+export const mockFs = (files: Record<string, string>) => {
+  const pnpm = path.resolve('../../node_modules/.pnpm');
+  mock({
+    [pnpm]: mock.load(pnpm, {}),
+    ...files,
+  });
+};
+
+export const resetMockFs = () => {
+  mock.restore();
+};