Configurable webhook responses

CHANGELOG.md

@@ -17,6 +17,9 @@ and this project adheres to
 
 ### Added
 
+- Allow users to respond back with custom webhook responses via the
+  `webhookResponse` field in the job state.
+  [#3102](https://github.com/OpenFn/lightning/issues/3102)
 - New `usage_caps_input` view-extension slot on the project settings page
   (`/projects/:project_id/settings`). Same pattern as the existing
   `concurrency_input` slot: downstream apps register a component via
@@ -43,13 +46,13 @@ and this project adheres to
   The sandbox's `project_users` are now derived from the parent project: every
   parent user is copied with their role preserved, the parent owner is demoted
   to `:admin`, and the actor is set as the sandbox owner. To add a user who is
-  not already on the parent, call `Lightning.Projects.add_project_users/3`
-  after `provision/3` returns.
+  not already on the parent, call `Lightning.Projects.add_project_users/3` after
+  `provision/3` returns.
   [#4744](https://github.com/OpenFn/lightning/issues/4744)
-- `Lightning.Projects.delete_project_user!/1` now raises `ArgumentError`
-  when called with a project's `:owner` row. The settings UI already
-  prevented this; the guard closes the gap for Mix tasks, IEx, and
-  scripted callers that would otherwise have left a project ownerless.
+- `Lightning.Projects.delete_project_user!/1` now raises `ArgumentError` when
+  called with a project's `:owner` row. The settings UI already prevented this;
+  the guard closes the gap for Mix tasks, IEx, and scripted callers that would
+  otherwise have left a project ownerless.
 - `./bin/bootstrap` on aarch64 Linux now requires Rust upfront and builds the
   Rambo native binary via `mix compile.rambo` post-compile, matching the darwin
   path. x86_64 Linux is unchanged.

DEPLOYMENT.md

@@ -61,11 +61,11 @@ Note that for secure deployments, it's recommended to use a combination of
 - `MAX_CREDENTIAL_SENSITIVE_VALUES` - the maximum number of sensitive values
   that can be stored in a credential. Defaults to 50.
 - `MAX_SANDBOX_NESTING_DEPTH` - the maximum depth at which sandboxes can be
-  nested. A direct child sandbox is depth 1, a sandbox of a sandbox is depth
-  2, etc. Provision attempts beyond this depth return
+  nested. A direct child sandbox is depth 1, a sandbox of a sandbox is depth 2,
+  etc. Provision attempts beyond this depth return
   `{:error, :nesting_too_deep}`, and the **Create Sandbox** button on the
-  Sandboxes page is disabled with a tooltip when the current project is at
-  the cap. Defaults to 5. Set to `0` to disable sandbox creation entirely.
+  Sandboxes page is disabled with a tooltip when the current project is at the
+  cap. Defaults to 5. Set to `0` to disable sandbox creation entirely.
 
 ### GitHub
 

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

@@ -77,6 +77,10 @@ export class YAMLStateToYDoc {
 
     if (trigger.type === 'webhook') {
       triggerMap.set('webhook_reply', trigger.webhook_reply ?? null);
+      triggerMap.set(
+        'webhook_response_config',
+        trigger.webhook_response_config ?? null
+      );
     }
 
     return triggerMap;

assets/js/collaborative-editor/hooks/useUnsavedChanges.ts

@@ -106,6 +106,7 @@ function transformTrigger(trigger: Trigger) {
       break;
     case 'webhook':
       output.webhook_reply = trigger.webhook_reply ?? 'before_start';
+      output.webhook_response_config = trigger.webhook_response_config ?? null;
       break;
   }
   return output;

assets/js/collaborative-editor/types/session.ts

@@ -71,6 +71,10 @@ export namespace Session {
     cron_expression: string | null;
     has_auth_method: boolean;
     webhook_reply: 'before_start' | 'after_completion' | null;
+    webhook_response_config: {
+      success_code: number | null;
+      error_code: number | null;
+    } | null;
     webhook_auth_methods: Array<{
       id: string;
       name: string;

assets/js/collaborative-editor/types/trigger.ts

@@ -24,6 +24,13 @@ const webhookTriggerSchema = baseTriggerSchema.extend({
     .enum(['before_start', 'after_completion'])
     .nullable()
     .default('before_start'),
+  webhook_response_config: z
+    .object({
+      success_code: z.number().int().nullable().default(null),
+      error_code: z.number().int().nullable().default(null),
+    })
+    .nullable()
+    .default(null),
 });
 
 // Cron trigger schema with professional validation using cron-validator
@@ -49,6 +56,7 @@ const cronTriggerSchema = baseTriggerSchema.extend({
     ),
   cron_cursor_job_id: z.string().uuid().nullable().default(null),
   kafka_configuration: z.null().default(null),
+  webhook_response_config: z.null().default(null),
   webhook_reply: z.null().default(null).catch(null),
 });
 
@@ -103,6 +111,7 @@ const kafkaTriggerSchema = baseTriggerSchema.extend({
   cron_expression: z.null().default(null),
   cron_cursor_job_id: z.null().default(null),
   kafka_configuration: kafkaConfigSchema,
+  webhook_response_config: z.null().default(null),
   webhook_reply: z.null().default(null).catch(null),
 });
 
@@ -137,6 +146,7 @@ export const createDefaultTrigger = (
         cron_cursor_job_id: null,
         kafka_configuration: null,
         webhook_reply: 'before_start' as const,
+        webhook_response_config: null,
       };
 
     case 'cron':

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

@@ -60,7 +60,15 @@
             },
             "webhook_reply": {
               "type": ["string", "null"],
-              "enum": ["before_start", "after_completion", "custom", null]
+              "enum": ["before_start", "after_completion", null]
+            },
+            "webhook_response_config": {
+              "type": ["object", "null"],
+              "properties": {
+                "success_code": { "type": ["integer", "null"] },
+                "error_code": { "type": ["integer", "null"] }
+              },
+              "additionalProperties": false
             },
             "pos": {
               "type": "object",

assets/js/yaml/types.ts

@@ -28,7 +28,11 @@ export type StateWebhookTrigger = {
   id: string;
   enabled: boolean;
   type: 'webhook';
-  webhook_reply: 'before_start' | 'after_completion' | 'custom' | null;
+  webhook_reply: 'before_start' | 'after_completion' | null | undefined;
+  webhook_response_config?: {
+    success_code?: number | null;
+    error_code?: number | null;
+  } | null;
 };
 
 export type StateKafkaTrigger = {
@@ -87,11 +91,17 @@ export type SpecCronTrigger = {
   pos: Position | undefined;
 };
 
+export type WebhookResponseConfig = {
+  success_code?: number | null;
+  error_code?: number | null;
+};
+
 export type SpecWebhookTrigger = {
   id?: string;
   type: 'webhook';
   enabled: boolean;
   webhook_reply: string | null;
+  webhook_response_config?: WebhookResponseConfig | null;
   pos: Position | undefined;
 };
 

assets/js/yaml/util.ts

@@ -82,6 +82,18 @@ export const convertWorkflowStateToSpec = (
 
     if (trigger.type === 'webhook') {
       triggerDetails.webhook_reply = trigger.webhook_reply ?? null;
+      const config = trigger.webhook_response_config;
+      if (
+        config &&
+        (config.success_code != null || config.error_code != null)
+      ) {
+        triggerDetails.webhook_response_config = {
+          ...(config.success_code != null && {
+            success_code: config.success_code,
+          }),
+          ...(config.error_code != null && { error_code: config.error_code }),
+        };
+      }
     }
 
     // TODO: handle kafka config
@@ -186,6 +198,7 @@ export const convertWorkflowSpecToState = (
         type: 'webhook',
         enabled,
         webhook_reply: specTrigger.webhook_reply,
+        webhook_response_config: specTrigger.webhook_response_config ?? null,
       };
     } else {
       trigger = {

assets/test/collaborative-editor/components/inspector/TriggerForm.test.tsx

@@ -192,9 +192,11 @@ describe('TriggerForm - Response Mode Field', () => {
 
     // Check both options exist (use exact text to avoid "Async" matching "sync")
     expect(
-      screen.getByRole('option', { name: 'Async (default)' })
+      screen.getByRole('option', { name: 'Async (Before Start)' })
+    ).toBeInTheDocument();
+    expect(
+      screen.getByRole('option', { name: 'Sync (After Completion)' })
     ).toBeInTheDocument();
-    expect(screen.getByRole('option', { name: 'Sync' })).toBeInTheDocument();
   });
 
   test('shows Async as default selected value', async () => {
@@ -261,7 +263,7 @@ describe('TriggerForm - Response Mode Field', () => {
     await waitFor(() => {
       expect(
         screen.getByText(
-          /responds with the final output state after the run completes/i
+          /holds the http connection open and responds when the run completes/i
         )
       ).toBeInTheDocument();
     });

lib/lightning/collaboration/workflow_serializer.ex

@@ -228,7 +228,18 @@ defmodule Lightning.Collaboration.WorkflowSerializer do
           "id" => trigger.id,
           "type" => trigger.type |> to_string(),
           "webhook_reply" =>
-            trigger.webhook_reply && to_string(trigger.webhook_reply)
+            trigger.webhook_reply && to_string(trigger.webhook_reply),
+          "webhook_response_config" =>
+            case trigger.webhook_response_config do
+              nil ->
+                nil
+
+              config ->
+                Yex.MapPrelim.from(%{
+                  "success_code" => config.success_code,
+                  "error_code" => config.error_code
+                })
+            end
         })
 
       Yex.Array.push(triggers_array, trigger_map)
@@ -274,9 +285,11 @@ defmodule Lightning.Collaboration.WorkflowSerializer do
     |> Enum.map(fn trigger ->
       trigger
       |> Map.take(
-        ~w(id type enabled cron_expression cron_cursor_job_id webhook_reply kafka_configuration)
+        ~w(id type enabled cron_expression cron_cursor_job_id webhook_reply
+           kafka_configuration webhook_response_config)
       )
       |> normalize_kafka_configuration()
+      |> normalize_webhook_response_config()
     end)
   end
 
@@ -299,6 +312,27 @@ defmodule Lightning.Collaboration.WorkflowSerializer do
 
   defp normalize_kafka_configuration(trigger), do: trigger
 
+  # Y.Doc serialises numbers as floats; convert integer codes back.
+  defp normalize_webhook_response_config(
+         %{"webhook_response_config" => %{} = config} = trigger
+       ) do
+    normalized =
+      config
+      |> normalize_integer_field("success_code")
+      |> normalize_integer_field("error_code")
+
+    Map.put(trigger, "webhook_response_config", normalized)
+  end
+
+  defp normalize_webhook_response_config(trigger), do: trigger
+
+  defp normalize_integer_field(map, key) do
+    case Map.fetch(map, key) do
+      {:ok, value} when is_float(value) -> Map.put(map, key, trunc(value))
+      _ -> map
+    end
+  end
+
   defp extract_positions(positions_map) do
     Yex.Map.to_json(positions_map)
   end

lib/lightning/export_utils.ex

@@ -15,6 +15,8 @@ defmodule Lightning.ExportUtils do
     :connect_timeout
   ]
 
+  @webhook_response_config_fields [:success_code, :error_code]
+
   @ordering_map %{
     project: [
       :name,
@@ -31,6 +33,7 @@ defmodule Lightning.ExportUtils do
     trigger: [
       :type,
       :webhook_reply,
+      :webhook_response_config,
       :cron_expression,
       :cron_cursor_job,
       :enabled,
@@ -121,14 +124,43 @@ defmodule Lightning.ExportUtils do
         Map.put(base, :kafka_configuration, kafka_config)
 
       :webhook ->
-        if trigger.webhook_reply do
-          Map.put(base, :webhook_reply, Atom.to_string(trigger.webhook_reply))
-        else
-          base
-        end
+        base
+        |> maybe_put_webhook_reply(trigger.webhook_reply)
+        |> maybe_put_webhook_response_config(trigger.webhook_response_config)
+    end
+  end
+
+  defp maybe_put_webhook_reply(map, nil), do: map
+
+  defp maybe_put_webhook_reply(map, reply) when is_atom(reply) do
+    Map.put(map, :webhook_reply, Atom.to_string(reply))
+  end
+
+  defp maybe_put_webhook_response_config(map, %{} = config) do
+    webhook_response =
+      Map.reject(
+        %{
+          success_code: config.success_code,
+          error_code: config.error_code
+        },
+        fn {_k, v} -> is_nil(v) end
+      )
+      |> Enum.sort_by(
+        fn {key, _val} ->
+          Enum.find_index(@webhook_response_config_fields, &(&1 == key))
+        end,
+        :asc
+      )
+
+    if length(webhook_response) > 0 do
+      Map.put(map, :webhook_response_config, webhook_response)
+    else
+      map
     end
   end
 
+  defp maybe_put_webhook_response_config(map, _), do: map
+
   defp edge_to_treenode(%{source_job_id: nil} = edge, triggers, jobs) do
     source_trigger =
       Enum.find(triggers, fn t -> t.id == edge.source_trigger_id end)

lib/lightning/projects/provisioner.ex

@@ -32,6 +32,7 @@ defmodule Lightning.Projects.Provisioner do
   alias Lightning.Workflows.Snapshot
   alias Lightning.Workflows.Trigger
   alias Lightning.Workflows.Triggers.KafkaConfiguration
+  alias Lightning.Workflows.Triggers.WebhookResponseConfig
   alias Lightning.Workflows.Workflow
   alias Lightning.Workflows.WorkflowUsageLimiter
   alias Lightning.WorkflowVersions
@@ -465,6 +466,11 @@ defmodule Lightning.Projects.Provisioner do
       required: false,
       with: &kafka_config_changeset/2
     )
+    |> cast_embed(
+      :webhook_response_config,
+      required: false,
+      with: &WebhookResponseConfig.changeset/2
+    )
     |> Trigger.validate()
     |> cast(attrs, [:delete])
     |> validate_required([:id])

lib/lightning/workflow_versions.ex

@@ -21,6 +21,7 @@ defmodule Lightning.WorkflowVersions do
   alias Ecto.Multi
   alias Lightning.Repo
   alias Lightning.Validators.Hex
+  alias Lightning.Workflows.Triggers.WebhookResponseConfig
   alias Lightning.Workflows.Workflow
   alias Lightning.Workflows.WorkflowVersion
 
@@ -251,6 +252,7 @@ defmodule Lightning.WorkflowVersions do
       :cron_expression,
       :enabled,
       :webhook_reply,
+      :webhook_response_config,
       :cron_cursor_job_id
     ]
 
@@ -325,6 +327,10 @@ defmodule Lightning.WorkflowVersions do
     |> binary_part(0, 12)
   end
 
+  defp serialize_value(%WebhookResponseConfig{} = val) do
+    val |> Map.take([:success_code, :error_code]) |> serialize_value()
+  end
+
   defp serialize_value(val) when is_map(val) do
     val
     |> round_numeric_values()