feat: export webhook_reply and cron_cursor_job in provisioner (#4597)

* feat: export webhook_reply and cron_cursor_job in provisioner

Include webhook_reply and cron_cursor_job_id in the provisioning JSON
API response and YAML export. webhook_reply is only emitted when set;
cron_cursor_job resolves to the hyphenated job name in YAML.

Also fixes webhook_reply atom-to-string conversion in the YAML
serialiser to avoid a FunctionClauseError on export.

* support webhook_reply during yaml import in the editor

* ensure jobs are inserted first

* add test for importing cron trigger with cron_cursor_job_id field

* refactor tests

Author: midigofrank

CHANGELOG.md

@@ -17,6 +17,9 @@ and this project adheres to
 
 ### Added
 
+- Support `webhook_reply` and `cron_cursor_job` in the provisioner
+  [#4587](https://github.com/OpenFn/lightning/issues/4587)
+
 ### Changed
 
 ### Fixed
@@ -49,7 +52,6 @@ and this project adheres to
   and triggers can still be freely deleted from workflows since the snapshot
   system preserves their data for every run.
   [#4538](https://github.com/OpenFn/lightning/issues/4538)
-
 - Allow users to select which workflow to merge for sandbox merging
   [#4002](https://github.com/OpenFn/lightning/issues/4002)
 - Bump ws-worker to `v1.23.1`

assets/js/collaborative-editor/adapters/YAMLStateToYDoc.ts

@@ -66,9 +66,18 @@ export class YAMLStateToYDoc {
 
     // Session.Trigger always requires cron_expression
     // Default to null for non-cron triggers
-    const cronExpression =
-      trigger.type === 'cron' ? trigger.cron_expression : null;
-    triggerMap.set('cron_expression', cronExpression);
+    triggerMap.set('cron_expression', null);
+    triggerMap.set('cron_cursor_job_id', null);
+    triggerMap.set('webhook_reply', null);
+
+    if (trigger.type === 'cron') {
+      triggerMap.set('cron_expression', trigger.cron_expression ?? null);
+      triggerMap.set('cron_cursor_job_id', trigger.cron_cursor_job_id ?? null);
+    }
+
+    if (trigger.type === 'webhook') {
+      triggerMap.set('webhook_reply', trigger.webhook_reply ?? null);
+    }
 
     return triggerMap;
   }

assets/js/yaml/schema/workflow-spec.json

@@ -55,6 +55,13 @@
             "cron_expression": {
               "type": "string"
             },
+            "cron_cursor_job": {
+              "type": ["string", "null"]
+            },
+            "webhook_reply": {
+              "type": ["string", "null"],
+              "enum": ["before_start", "after_completion", "custom", null]
+            },
             "pos": {
               "type": "object",
               "properties": {
@@ -71,7 +78,9 @@
           "additionalProperties": false,
           "oneOf": [
             {
-              "properties": { "type": { "const": "webhook" } }
+              "properties": {
+                "type": { "const": "webhook" }
+              }
             },
             {
               "properties": {

assets/js/yaml/types.ts

@@ -21,12 +21,14 @@ export type StateCronTrigger = {
   type: 'cron';
   enabled: boolean;
   cron_expression: string;
+  cron_cursor_job_id: string | null;
 };
 
 export type StateWebhookTrigger = {
   id: string;
   enabled: boolean;
   type: 'webhook';
+  webhook_reply: 'before_start' | 'after_completion' | 'custom' | null;
 };
 
 export type StateKafkaTrigger = {
@@ -81,13 +83,15 @@ export type SpecCronTrigger = {
   type: 'cron';
   enabled: boolean;
   cron_expression: string;
+  cron_cursor_job: string | null;
   pos: Position | undefined;
 };
 
 export type SpecWebhookTrigger = {
   id?: string;
   type: 'webhook';
   enabled: boolean;
+  webhook_reply: string | null;
   pos: Position | undefined;
 };
 

assets/js/yaml/util.ts

@@ -61,17 +61,29 @@ export const convertWorkflowStateToSpec = (
   const triggers: { [key: string]: SpecTrigger } = {};
   workflowState.triggers.forEach(trigger => {
     const pos = workflowState.positions?.[trigger.id];
+
     const triggerDetails: SpecTrigger = {
       ...(includeIds && { id: trigger.id }),
       type: trigger.type,
       enabled: trigger.enabled,
       pos: trigger.type !== 'kafka' && pos ? roundPosition(pos) : undefined,
-      cron_expression:
-        trigger.type === 'cron' && 'cron_expression' in trigger
-          ? trigger.cron_expression
-          : undefined,
     } as SpecTrigger;
 
+    if (trigger.type === 'cron') {
+      const cursorJob = trigger.cron_cursor_job_id
+        ? workflowState.jobs.find(job => job.id === trigger.cron_cursor_job_id)
+        : null;
+
+      triggerDetails.cron_expression = trigger.cron_expression ?? null;
+      triggerDetails.cron_cursor_job = cursorJob
+        ? hyphenate(cursorJob.name)
+        : null;
+    }
+
+    if (trigger.type === 'webhook') {
+      triggerDetails.webhook_reply = trigger.webhook_reply ?? null;
+    }
+
     // TODO: handle kafka config
     triggers[trigger.type] = triggerDetails;
   });
@@ -158,17 +170,22 @@ export const convertWorkflowSpecToState = (
 
     let trigger: StateTrigger;
     if (specTrigger.type === 'cron') {
+      const cursorJob = specTrigger.cron_cursor_job
+        ? (stateJobs[specTrigger.cron_cursor_job] ?? null)
+        : null;
       trigger = {
         id: uId,
         type: 'cron',
         enabled,
         cron_expression: specTrigger.cron_expression,
+        cron_cursor_job_id: cursorJob ? cursorJob.id : null,
       };
     } else if (specTrigger.type === 'webhook') {
       trigger = {
         id: uId,
         type: 'webhook',
         enabled,
+        webhook_reply: specTrigger.webhook_reply,
       };
     } else {
       trigger = {

lib/lightning/export_utils.ex

@@ -28,7 +28,14 @@ defmodule Lightning.ExportUtils do
     credential: [:name, :owner],
     workflow: [:name, :jobs, :triggers, :edges],
     job: [:name, :adaptor, :credential, :globals, :body],
-    trigger: [:type, :cron_expression, :enabled, :kafka_configuration],
+    trigger: [
+      :type,
+      :webhook_reply,
+      :cron_expression,
+      :cron_cursor_job,
+      :enabled,
+      :kafka_configuration
+    ],
     edge: [
       :source_trigger,
       :source_job,
@@ -58,7 +65,7 @@ defmodule Lightning.ExportUtils do
 
     %{
       # The identifier here for our YAML reducer will be the hyphenated name
-      id: hyphenate(job.name),
+      id: job_key(job),
       name: job.name,
       node_type: :job,
       adaptor: job.adaptor,
@@ -68,7 +75,7 @@ defmodule Lightning.ExportUtils do
     }
   end
 
-  defp trigger_to_treenode(trigger) do
+  defp trigger_to_treenode(trigger, jobs) do
     base = %{
       id: trigger.id,
       enabled: trigger.enabled,
@@ -79,7 +86,18 @@ defmodule Lightning.ExportUtils do
 
     case trigger.type do
       :cron ->
-        Map.put(base, :cron_expression, trigger.cron_expression)
+        base
+        |> Map.put(:cron_expression, trigger.cron_expression)
+        |> then(fn cron ->
+          if trigger.cron_cursor_job_id do
+            cursor_job =
+              Enum.find(jobs, fn j -> j.id == trigger.cron_cursor_job_id end)
+
+            Map.put(cron, :cron_cursor_job, cursor_job && job_key(cursor_job))
+          else
+            cron
+          end
+        end)
 
       :kafka ->
         kafka_config =
@@ -102,8 +120,12 @@ defmodule Lightning.ExportUtils do
 
         Map.put(base, :kafka_configuration, kafka_config)
 
-      _ ->
-        base
+      :webhook ->
+        if trigger.webhook_reply do
+          Map.put(base, :webhook_reply, Atom.to_string(trigger.webhook_reply))
+        else
+          base
+        end
     end
   end
 
@@ -113,7 +135,7 @@ defmodule Lightning.ExportUtils do
 
     target_job = Enum.find(jobs, fn j -> j.id == edge.target_job_id end)
     trigger_name = to_string(source_trigger.type)
-    target_job_name = hyphenate(target_job.name)
+    target_job_name = job_key(target_job)
 
     %{
       name: "#{trigger_name}->#{target_job_name}",
@@ -125,8 +147,8 @@ defmodule Lightning.ExportUtils do
   defp edge_to_treenode(%{source_trigger_id: nil} = edge, _triggers, jobs) do
     target_job = Enum.find(jobs, fn j -> j.id == edge.target_job_id end)
     source_job = Enum.find(jobs, fn j -> j.id == edge.source_job_id end)
-    source_job_name = hyphenate(source_job.name)
-    target_job_name = hyphenate(target_job.name)
+    source_job_name = job_key(source_job)
+    target_job_name = job_key(target_job)
 
     %{
       name: "#{source_job_name}->#{target_job_name}",
@@ -138,7 +160,7 @@ defmodule Lightning.ExportUtils do
   defp merge_edge_common_fields(json, edge, target_job) do
     json
     |> Map.merge(%{
-      target_job: hyphenate(target_job.name),
+      target_job: job_key(target_job),
       condition_type: edge.condition_type |> Atom.to_string(),
       enabled: edge.enabled,
       node_type: :edge
@@ -348,6 +370,10 @@ defmodule Lightning.ExportUtils do
     }
   end
 
+  defp job_key(job) do
+    hyphenate(job.name)
+  end
+
   defp project_credential_key(project_credential) do
     hyphenate(
       "#{project_credential.credential.user.email} #{project_credential.credential.name}"
@@ -378,7 +404,7 @@ defmodule Lightning.ExportUtils do
     triggers =
       workflow.triggers
       |> Enum.sort_by(& &1.inserted_at, NaiveDateTime)
-      |> Enum.map(fn t -> trigger_to_treenode(t) end)
+      |> Enum.map(fn t -> trigger_to_treenode(t, workflow.jobs) end)
 
     edges =
       workflow.edges

lib/lightning/workflows/workflow.ex

@@ -44,9 +44,12 @@ defmodule Lightning.Workflows.Workflow do
     field :enable_job_logs, :boolean, default: true
     field :positions, :map
 
+    # the ordering of edges, triggers and jobs are intentional
+    # ecto reverses the relations when inserting. so jobs->triggers->edges
+    # triggers depend on jobs for cron_cursor_job_id
     has_many :edges, Edge, on_replace: :delete_if_exists
-    has_many :jobs, Job, on_replace: :delete
     has_many :triggers, Trigger, on_replace: :delete_if_exists
+    has_many :jobs, Job, on_replace: :delete
     has_many :versions, WorkflowVersion, foreign_key: :workflow_id
 
     has_many :work_orders, Lightning.WorkOrder

lib/lightning_web/controllers/api/provisioning_json.ex

@@ -103,7 +103,9 @@ defmodule LightningWeb.API.ProvisioningJSON do
         )
 
     trigger
-    |> Map.take(~w(id type cron_expression enabled)a)
+    |> Map.take(
+      ~w(id type cron_expression enabled webhook_reply cron_cursor_job_id)a
+    )
     |> Map.put(:kafka_configuration, kafka_configuration)
     |> drop_keys_with_nil_value()
   end

test/fixtures/webhook_reply_and_cron_cursor_project.yaml

@@ -0,0 +1,46 @@
+name: webhook-reply-and-cron-cursor-project
+description: null
+collections: null
+credentials: null
+workflows:
+  cron-cursor-workflow:
+    name: cron cursor workflow
+    jobs:
+      cursor-job:
+        name: cursor job
+        adaptor: '@openfn/language-common@latest'
+        credential: null
+        body: |
+          fn(state => state)
+    triggers:
+      cron:
+        type: cron
+        cron_expression: '0 6 * * *'
+        cron_cursor_job: cursor-job
+        enabled: true
+    edges:
+      cron->cursor-job:
+        source_trigger: cron
+        target_job: cursor-job
+        condition_type: always
+        enabled: true
+  webhook-reply-workflow:
+    name: webhook reply workflow
+    jobs:
+      reply-job:
+        name: reply job
+        adaptor: '@openfn/language-common@latest'
+        credential: null
+        body: |
+          fn(state => state)
+    triggers:
+      webhook:
+        type: webhook
+        webhook_reply: after_completion
+        enabled: true
+    edges:
+      webhook->reply-job:
+        source_trigger: webhook
+        target_job: reply-job
+        condition_type: always
+        enabled: true