Support file path in job body

.changeset/swift-dingos-know.md

@@ -0,0 +1,6 @@
+---
+'@openfn/cli': minor
+'@openfn/deploy': minor
+---
+
+Support file paths for job body

packages/cli/src/pull/handler.ts

@@ -6,6 +6,7 @@ import {
   getProject,
   getSpec,
   getStateFromProjectPayload,
+  updatePulledSpecJobBodyPath,
 } from '@openfn/deploy';
 import type { Logger } from '../util/logger';
 import { PullOptions } from '../pull/command';
@@ -47,11 +48,7 @@ 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)
-    );
+    // defer writing to disk until we have the spec
 
     logger.always('Downloading the project spec (as YAML) from the server.');
     // Get the project.yaml from Lightning
@@ -85,9 +82,21 @@ 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 updatePulledSpecJobBodyPath(
+      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

@@ -1,8 +1,8 @@
 import { confirm } from '@inquirer/prompts';
 import { inspect } from 'node:util';
-import { DeployConfig, ProjectState } from './types';
+import { DeployConfig, ProjectState, SpecJob } from './types';
 import { readFile, writeFile } from 'fs/promises';
-import { parseAndValidate } from './validator';
+import { parseAndValidate, addSpecJobBodyPath } from './validator';
 import jsondiff from 'json-diff';
 import {
   mergeProjectPayloadIntoState,
@@ -91,7 +91,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 await parseAndValidate(body, path);
   } catch (error: any) {
     if (error.code === 'ENOENT') {
       throw new DeployError(`File not found: ${path}`, 'SPEC_ERROR');
@@ -103,6 +103,55 @@ export async function getSpec(path: string) {
 
 // =============================================
 
+async function getAllSpecJobs(
+  config: DeployConfig,
+  logger: Logger
+): Promise<SpecJob[]> {
+  const jobs: SpecJob[] = [];
+
+  try {
+    const [state, spec] = await Promise.all([
+      getState(config.statePath),
+      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;
+}
+
+export async function updatePulledSpecJobBodyPath(
+  newSpecBody: string,
+  newState: ProjectState,
+  config: DeployConfig,
+  logger: Logger
+): Promise<string> {
+  try {
+    const oldSpecJobs = await getAllSpecJobs(config, logger);
+
+    return await addSpecJobBodyPath(newSpecBody, newState, oldSpecJobs, config);
+  } catch (error: any) {
+    logger.warn(`Could not update spec job body paths: ${error.message}`);
+    return newSpecBody;
+  }
+}
+
 export async function deploy(config: DeployConfig, logger: Logger) {
   const [state, spec] = await Promise.all([
     getState(config.statePath),

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 { ProjectSpec } from './types';
+import YAML, { YAMLMap, isMap, isPair, isScalar } from 'yaml';
+import { DeployConfig, ProjectSpec, ProjectState, SpecJob } from './types';
+import { readFile, writeFile } 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);
@@ -148,3 +185,61 @@ export function parseAndValidate(input: string): {
 
   return { errors, doc: doc.toJSON() as ProjectSpec };
 }
+
+export async function addSpecJobBodyPath(
+  specBody: string,
+  state: ProjectState,
+  oldJobs: SpecJob[],
+  config: DeployConfig
+): Promise<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
+      ) {
+        // the job
+        const job = pairPath[pairPath.length - 2];
+        const jobKey = isPair(job) && isScalar(job.key) && job.key.value;
+        // the workflow
+        const workflow = pairPath[pairPath.length - 6];
+        const workflowKey =
+          isPair(workflow) && isScalar(workflow.key) && workflow.key.value;
+
+        // find the job in the state
+        const stateJob =
+          typeof jobKey === 'string' &&
+          typeof workflowKey === 'string' &&
+          state.workflows[workflowKey]?.jobs[jobKey];
+
+        // check if the state job is in the old spec jobs
+        const oldSpecJob =
+          stateJob && oldJobs.find((job) => job.id === stateJob.id);
+
+        // get the file path from the old spec job
+        const oldSpecJobPath =
+          oldSpecJob &&
+          typeof oldSpecJob?.body === 'object' &&
+          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 undefined;
+    },
+  });
+
+  return doc.toString();
+}