deploy: support webhook_reply and cron_cursor_job (#1344)

midigofrank · josephjclark · web-flow · commit 267cc158322e · 2026-04-01T14:44:13.000+01:00
* deploy: support webhook_reply and cron_cursor_job

* oops, did not handle scenario where both state and spec trigger exist

* versions

---------

Co-authored-by: Joe Clark &lt;jclark@openfn.org&gt;
diff --git a/packages/cli/CHANGELOG.md b/packages/cli/CHANGELOG.md
@@ -1,5 +1,20 @@
 # @openfn/cli
 
+## 1.31.0
+
+### Minor Changes
+
+- In deploy v1, add support for support for new `webhook_reply` and `cron_cursor_job` keys on triggers.
+
+  When syncing projects with these keys, versions of lightning prior to `2.16.0` will return errors. In other words, projects pulled from the latest cloud may fail when deployed to older instances.
+
+  To workaround, update the target lightning version or use an older CLI version.
+
+### Patch Changes
+
+- Updated dependencies
+  - @openfn/deploy@0.12.0
+
 ## 1.30.6
 
 ### Patch Changes
diff --git a/packages/cli/package.json b/packages/cli/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@openfn/cli",
-  "version": "1.30.6",
+  "version": "1.31.0",
   "description": "CLI devtools for the OpenFn toolchain",
   "engines": {
     "node": ">=18",
diff --git a/packages/deploy/CHANGELOG.md b/packages/deploy/CHANGELOG.md
@@ -1,5 +1,11 @@
 # @openfn/deploy
 
+## 0.12.0
+
+### Minor Changes
+
+- Support webhook_reply and cron_cursor_job.
+
 ## 0.11.7
 
 ### Patch Changes
diff --git a/packages/deploy/package.json b/packages/deploy/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@openfn/deploy",
-  "version": "0.11.7",
+  "version": "0.12.0",
   "description": "Deploy projects to Lightning instances",
   "type": "module",
   "exports": {
diff --git a/packages/deploy/src/stateTransform.ts b/packages/deploy/src/stateTransform.ts
@@ -63,6 +63,20 @@ function getStateJobCredential(
   return stateCredentials[specJobCredential].id;
 }
 
+function getStateJob(
+  specJob: string,
+  stateJobs: WorkflowState['jobs']
+): string {
+  if (!stateJobs[specJob]) {
+    throw new DeployError(
+      `Could not find a job with key: ${specJob}`,
+      'VALIDATION_ERROR'
+    );
+  }
+
+  return stateJobs[specJob].id;
+}
+
 function mergeJobs(
   credentials: ProjectState['project_credentials'],
   stateJobs: WorkflowState['jobs'],
@@ -134,6 +148,7 @@ function pickValue(
 }
 
 function mergeTriggers(
+  jobs: WorkflowState['jobs'],
   stateTriggers: WorkflowState['triggers'],
   specTriggers: WorkflowSpec['triggers']
 ): WorkflowState['triggers'] {
@@ -146,8 +161,19 @@ function mergeTriggers(
             ...pickKeys(specTrigger, ['type', 'enabled']),
           };
 
+          if (specTrigger.type === 'webhook' && specTrigger.webhook_reply) {
+            trigger.webhook_reply = specTrigger.webhook_reply;
+          }
+
           if (specTrigger.type === 'cron') {
             trigger.cron_expression = specTrigger.cron_expression;
+
+            if (specTrigger.cron_cursor_job) {
+              trigger.cron_cursor_job_id = getStateJob(
+                specTrigger.cron_cursor_job,
+                jobs
+              );
+            }
           }
 
           if (specTrigger.type === 'kafka') {
@@ -173,8 +199,19 @@ function mergeTriggers(
           enabled: pickValue(specTrigger!, stateTrigger!, 'enabled', true),
         };
 
+        if (specTrigger!.type === 'webhook' && specTrigger!.webhook_reply) {
+          trigger.webhook_reply = specTrigger!.webhook_reply;
+        }
+
         if (specTrigger!.type === 'cron') {
           trigger.cron_expression = specTrigger!.cron_expression;
+
+          if (specTrigger!.cron_cursor_job) {
+            trigger.cron_cursor_job_id = getStateJob(
+              specTrigger!.cron_cursor_job,
+              jobs
+            );
+          }
         }
 
         if (specTrigger!.type === 'kafka') {
@@ -334,6 +371,7 @@ export function mergeSpecIntoState(
         );
 
         const nextTriggers = mergeTriggers(
+          nextJobs,
           stateWorkflow?.triggers || {},
           specWorkflow?.triggers || {}
         );
diff --git a/packages/deploy/src/types.ts b/packages/deploy/src/types.ts
@@ -38,9 +38,13 @@ export type SpecKafkaConfiguration = {
   connect_timeout: number;
 };
 
+export type WebhookReply = 'before_start' | 'after_completion';
+
 export type SpecTrigger = {
   type: string;
   cron_expression?: string;
+  cron_cursor_job?: string;
+  webhook_reply?: WebhookReply;
   enabled?: boolean;
   kafka_configuration?: SpecKafkaConfiguration;
 };
@@ -49,6 +53,8 @@ export type StateTrigger = {
   id: string;
   type: string;
   cron_expression?: string;
+  cron_cursor_job_id?: string | null;
+  webhook_reply?: WebhookReply;
   delete?: boolean;
   enabled?: boolean;
   kafka_configuration?: StateKafkaConfiguration;
diff --git a/packages/deploy/test/stateTransform.test.ts b/packages/deploy/test/stateTransform.test.ts
@@ -475,6 +475,247 @@ test('toNextState with for kafka trigger', (t) => {
   });
 });
 
+test('toNextState sets webhook_reply when specified', (t) => {
+  const state = { workflows: {} };
+  const spec = {
+    name: 'my project',
+    workflows: {
+      w: {
+        name: 'workflow',
+        jobs: {},
+        triggers: {
+          t: { type: 'webhook', webhook_reply: 'before_start' },
+        },
+        edges: {},
+      },
+    },
+  };
+
+  const result = mergeSpecIntoState(state, spec);
+  t.is(result.workflows.w.triggers.t.webhook_reply, 'before_start');
+});
+
+test('toNextState omits webhook_reply when not specified', (t) => {
+  const state = { workflows: {} };
+  const spec = {
+    name: 'my project',
+    workflows: {
+      w: {
+        name: 'workflow',
+        jobs: {},
+        triggers: {
+          t: { type: 'webhook' },
+        },
+        edges: {},
+      },
+    },
+  };
+
+  const result = mergeSpecIntoState(state, spec);
+  t.false('webhook_reply' in result.workflows.w.triggers.t);
+});
+
+test('toNextState sets cron_cursor_job_id when specified', (t) => {
+  const state = { workflows: {} };
+  const spec = {
+    name: 'my project',
+    workflows: {
+      w: {
+        name: 'workflow',
+        jobs: {
+          'job-a': {
+            name: 'job a',
+            adaptor: '@openfn/language-http',
+            body: 'fn()',
+          },
+        },
+        triggers: {
+          t: {
+            type: 'cron',
+            cron_expression: '0 * * * *',
+            cron_cursor_job: 'job-a',
+          },
+        },
+        edges: {},
+      },
+    },
+  };
+
+  const result = mergeSpecIntoState(state, spec);
+  const jobId = result.workflows.w.jobs['job-a'].id;
+  t.is(result.workflows.w.triggers.t.cron_cursor_job_id, jobId);
+});
+
+test('toNextState omits cron_cursor_job_id when not specified', (t) => {
+  const state = { workflows: {} };
+  const spec = {
+    name: 'my project',
+    workflows: {
+      w: {
+        name: 'workflow',
+        jobs: {
+          'job-a': {
+            name: 'job a',
+            adaptor: '@openfn/language-http',
+            body: 'fn()',
+          },
+        },
+        triggers: {
+          t: { type: 'cron', cron_expression: '0 * * * *' },
+        },
+        edges: {},
+      },
+    },
+  };
+
+  const result = mergeSpecIntoState(state, spec);
+  t.false('cron_cursor_job_id' in result.workflows.w.triggers.t);
+});
+
+test('toNextState sets webhook_reply on existing trigger', (t) => {
+  const triggerId = 'aaa-bbb-ccc';
+  const state = {
+    workflows: {
+      w: {
+        id: 'wf-1',
+        name: 'workflow',
+        jobs: {},
+        triggers: {
+          t: { id: triggerId, type: 'webhook', enabled: true },
+        },
+        edges: {},
+      },
+    },
+  };
+  const spec = {
+    name: 'my project',
+    workflows: {
+      w: {
+        name: 'workflow',
+        jobs: {},
+        triggers: {
+          t: { type: 'webhook', webhook_reply: 'after_completion' },
+        },
+        edges: {},
+      },
+    },
+  };
+
+  const result = mergeSpecIntoState(state, spec);
+  t.is(result.workflows.w.triggers.t.id, triggerId);
+  t.is(result.workflows.w.triggers.t.webhook_reply, 'after_completion');
+});
+
+test('toNextState omits webhook_reply on existing trigger when not specified', (t) => {
+  const triggerId = 'aaa-bbb-ccc';
+  const state = {
+    workflows: {
+      w: {
+        id: 'wf-1',
+        name: 'workflow',
+        jobs: {},
+        triggers: {
+          t: { id: triggerId, type: 'webhook', enabled: true },
+        },
+        edges: {},
+      },
+    },
+  };
+  const spec = {
+    name: 'my project',
+    workflows: {
+      w: {
+        name: 'workflow',
+        jobs: {},
+        triggers: {
+          t: { type: 'webhook' },
+        },
+        edges: {},
+      },
+    },
+  };
+
+  const result = mergeSpecIntoState(state, spec);
+  t.false('webhook_reply' in result.workflows.w.triggers.t);
+});
+
+test('toNextState sets cron_cursor_job_id on existing trigger', (t) => {
+  const triggerId = 'aaa-bbb-ccc';
+  const jobId = 'job-uuid-111';
+  const state = {
+    workflows: {
+      w: {
+        id: 'wf-1',
+        name: 'workflow',
+        jobs: {
+          'job-a': { id: jobId, name: 'job a' },
+        },
+        triggers: {
+          t: { id: triggerId, type: 'cron', enabled: true },
+        },
+        edges: {},
+      },
+    },
+  };
+  const spec = {
+    name: 'my project',
+    workflows: {
+      w: {
+        name: 'workflow',
+        jobs: {
+          'job-a': { name: 'job a', adaptor: '@openfn/language-http', body: 'fn()' },
+        },
+        triggers: {
+          t: { type: 'cron', cron_expression: '0 * * * *', cron_cursor_job: 'job-a' },
+        },
+        edges: {},
+      },
+    },
+  };
+
+  const result = mergeSpecIntoState(state, spec);
+  t.is(result.workflows.w.triggers.t.id, triggerId);
+  t.is(result.workflows.w.triggers.t.cron_cursor_job_id, jobId);
+});
+
+test('toNextState omits cron_cursor_job_id on existing trigger when not specified', (t) => {
+  const triggerId = 'aaa-bbb-ccc';
+  const jobId = 'job-uuid-111';
+  const state = {
+    workflows: {
+      w: {
+        id: 'wf-1',
+        name: 'workflow',
+        jobs: {
+          'job-a': { id: jobId, name: 'job a' },
+        },
+        triggers: {
+          t: { id: triggerId, type: 'cron', enabled: true },
+        },
+        edges: {},
+      },
+    },
+  };
+  const spec = {
+    name: 'my project',
+    workflows: {
+      w: {
+        name: 'workflow',
+        jobs: {
+          'job-a': { name: 'job a', adaptor: '@openfn/language-http', body: 'fn()' },
+        },
+        triggers: {
+          t: { type: 'cron', cron_expression: '0 * * * *' },
+        },
+        edges: {},
+      },
+    },
+  };
+
+  const result = mergeSpecIntoState(state, spec);
+  t.false('cron_cursor_job_id' in result.workflows.w.triggers.t);
+});
+
 test('mergeProjectIntoState with no changes', (t) => {
   let existingState = fullExampleState();
   const workflowOne = existingState.workflows['workflow-one'];

Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,6 @@`
`1`	`1`	`{`
`2`	`2`	`"name": "@openfn/cli",`
`3`		`- "version": "1.30.6",`
	`3`	`+ "version": "1.31.0",`
`4`	`4`	`"description": "CLI devtools for the OpenFn toolchain",`
`5`	`5`	`"engines": {`
`6`	`6`	`"node": ">=18",`
Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,6 @@`
`1`	`1`	`{`
`2`	`2`	`"name": "@openfn/deploy",`
`3`		`- "version": "0.11.7",`
	`3`	`+ "version": "0.12.0",`
`4`	`4`	`"description": "Deploy projects to Lightning instances",`
`5`	`5`	`"type": "module",`
`6`	`6`	`"exports": {`