Stop session crash on malformed or stale workflow references

Harden the collaborative save path against two crash modes that killed
the session GenServer and its channel:

- Stale cron_cursor_job_id reference: a cron trigger could point at a job
  that no longer exists in the workflow. The trigger changeset lacked a
  foreign_key_constraint, so the FK violation raised Ecto.ConstraintError
  instead of a changeset error. Declare foreign_key_constraint and clear
  the reference client-side in removeJob.

- Malformed UUIDs in workflow changesets: :binary_id fields are not
  format-checked by cast/3, so an unsubstituted import placeholder only
  raised Ecto.ChangeError on insert/update. Add a reusable validate_uuid/2
  helper to Lightning.Validators and apply it in the Job, Edge and Trigger
  changesets.

Plus changelog entries for the collaborative save crash fixes.

Author: midigofrank

CHANGELOG.md

@@ -61,6 +61,17 @@ and this project adheres to
   [#4810](https://github.com/OpenFn/lightning/issues/4810)
 - Workflow channel raises an exception when fetching trigger auth methods for an
   unpersisted trigger [#4819](https://github.com/OpenFn/lightning/issues/4819)
+- Collaborative session no longer crashes when saving a cron trigger whose
+  `cron_cursor_job_id` references a job that no longer exists. The `Trigger`
+  changeset now declares the foreign key constraint, and `removeJob` clears any
+  cron cursor pointing at a deleted job, so the violation surfaces as a
+  validation error instead of an `Ecto.ConstraintError`.
+  [#4816](https://github.com/OpenFn/lightning/issues/4816)
+- Collaborative session no longer crashes when a workflow payload contains a
+  malformed UUID (e.g. an unsubstituted template placeholder) for a job,
+  trigger, or edge id. These ids are now validated in the changesets, so the bad
+  value returns a changeset error instead of raising an `Ecto.ChangeError`
+  during insert. [#4816](https://github.com/OpenFn/lightning/issues/4816)
 
 ## [2.16.6] - 2026-05-27
 

assets/js/collaborative-editor/stores/createWorkflowStore.ts

@@ -941,12 +941,24 @@ export const createWorkflowStore = () => {
       // Find all incoming edges (where this job is the target)
       const incomingEdgeIndices = getIncomingEdgeIndices(edges, id);
 
+      // Find triggers whose cron cursor points at the job being removed.
+      const triggersArray = ydoc.getArray('triggers');
+      const triggers = triggersArray.toArray() as Y.Map<unknown>[];
+      const cursorTriggers = triggers.filter(
+        trigger => trigger.get('cron_cursor_job_id') === id
+      );
+
       ydoc.transact(() => {
         // Delete incoming edges first (highest index to lowest)
         incomingEdgeIndices.forEach(edgeIndex => {
           edgesArray.delete(edgeIndex, 1);
         });
 
+        // Clear any cron cursor reference to the job being removed
+        cursorTriggers.forEach(trigger => {
+          trigger.set('cron_cursor_job_id', null);
+        });
+
         // Then delete the job
         jobsArray.delete(jobIndex, 1);
       });

assets/test/collaborative-editor/stores/createWorkflowStore.test.ts

@@ -1139,6 +1139,54 @@ describe('WorkflowStore - removeJob with edge cleanup', () => {
     expect(snapshotAfter.jobs).toHaveLength(1);
     expect(snapshotAfter.jobs?.[0]?.id).toBe('job-a');
   });
+
+  test('clears cron_cursor_job_id on a cron trigger pointing at the removed job', () => {
+    // Setup: cron trigger whose cursor points at Job B (no edge between them,
+    // mirroring the unfiltered "Cron Input Source" dropdown).
+    const triggersArray = ydoc.getArray('triggers');
+    const triggerMap = new Y.Map();
+    triggerMap.set('id', 'trigger-1');
+    triggerMap.set('type', 'cron');
+    triggerMap.set('enabled', true);
+    triggerMap.set('cron_cursor_job_id', 'job-b');
+    triggersArray.push([triggerMap]);
+
+    store.addJob({ id: 'job-a', name: 'Job A', body: 'fn(state => state)' });
+    store.addJob({ id: 'job-b', name: 'Job B', body: 'fn(state => state)' });
+
+    // Verify setup
+    let snapshot = store.getSnapshot();
+    expect(snapshot.triggers?.[0]?.cron_cursor_job_id).toBe('job-b');
+
+    // Action: Delete the cursor job
+    store.removeJob('job-b');
+
+    // Assert: Job B gone and the trigger's cursor reference cleared
+    snapshot = store.getSnapshot();
+    expect(snapshot.jobs).toHaveLength(1);
+    expect(snapshot.jobs?.[0]?.id).toBe('job-a');
+    expect(snapshot.triggers?.[0]?.cron_cursor_job_id).toBeNull();
+  });
+
+  test('leaves cron_cursor_job_id untouched when a different job is removed', () => {
+    const triggersArray = ydoc.getArray('triggers');
+    const triggerMap = new Y.Map();
+    triggerMap.set('id', 'trigger-1');
+    triggerMap.set('type', 'cron');
+    triggerMap.set('enabled', true);
+    triggerMap.set('cron_cursor_job_id', 'job-a');
+    triggersArray.push([triggerMap]);
+
+    store.addJob({ id: 'job-a', name: 'Job A', body: 'fn(state => state)' });
+    store.addJob({ id: 'job-b', name: 'Job B', body: 'fn(state => state)' });
+
+    // Action: Delete a job that is NOT the cursor target
+    store.removeJob('job-b');
+
+    // Assert: cursor still points at the surviving job
+    const snapshot = store.getSnapshot();
+    expect(snapshot.triggers?.[0]?.cron_cursor_job_id).toBe('job-a');
+  });
 });
 
 describe('WorkflowStore - AI Workflow Apply Coordination', () => {

lib/lightning/utils/validators.ex

@@ -80,6 +80,37 @@ defmodule Lightning.Validators do
     end
   end
 
+  @doc """
+  Validates that the given field(s) contain a well-formed UUID.
+
+  `:binary_id` fields are not format-checked by `cast/3` — a malformed value
+  (e.g. an unsubstituted import placeholder) passes casting and only raises
+  `Ecto.ChangeError` when dumped on insert/update. This converts that into a
+  changeset error instead.
+
+  Only runs when a non-nil change is present for the field, so optional
+  foreign keys left unset are unaffected.
+
+  ```
+  changeset
+  |> validate_uuid([:id, :workflow_id])
+  ```
+  """
+  @spec validate_uuid(Ecto.Changeset.t(), atom() | [atom()]) ::
+          Ecto.Changeset.t()
+  def validate_uuid(changeset, fields) when is_list(fields) do
+    Enum.reduce(fields, changeset, &validate_uuid(&2, &1))
+  end
+
+  def validate_uuid(changeset, field) when is_atom(field) do
+    validate_change(changeset, field, fn _, value ->
+      case Ecto.UUID.cast(value) do
+        {:ok, _} -> []
+        :error -> [{field, "is not a valid UUID"}]
+      end
+    end)
+  end
+
   @doc """
   Validates a URL in a changeset field.
 

lib/lightning/workflows/edge.ex

@@ -91,6 +91,12 @@ defmodule Lightning.Workflows.Edge do
 
   def validate(changeset) do
     changeset
+    |> validate_uuid([
+      :id,
+      :source_job_id,
+      :source_trigger_id,
+      :target_job_id
+    ])
     |> assoc_constraint(:workflow)
     |> assoc_constraint(:source_trigger)
     |> assoc_constraint(:source_job)

lib/lightning/workflows/job.ex

@@ -97,6 +97,7 @@ defmodule Lightning.Workflows.Job do
     |> validate_required(:name, message: "job name can't be blank")
     |> validate_required(:body, message: "job body can't be blank")
     |> validate_required(:adaptor, message: "job adaptor can't be blank")
+    |> Validators.validate_uuid([:id])
     |> Validators.validate_exclusive(
       [:project_credential_id, :keychain_credential_id],
       "cannot be set when the other credential type is also set"

lib/lightning/workflows/trigger.ex

@@ -13,6 +13,7 @@ defmodule Lightning.Workflows.Trigger do
   """
   use Lightning.Schema
   import Ecto.Query
+  import Lightning.Validators
 
   alias Lightning.Workflows.Job
   alias Lightning.Workflows.Triggers.KafkaConfiguration
@@ -121,7 +122,11 @@ defmodule Lightning.Workflows.Trigger do
     |> validate_required([:type])
     |> assoc_constraint(:workflow)
     |> validate_by_type()
+    |> validate_uuid([:id, :cron_cursor_job_id])
     |> unique_constraint(:id, name: "triggers_pkey")
+    |> foreign_key_constraint(:cron_cursor_job_id,
+      message: "the referenced cursor job does not exist"
+    )
   end
 
   defp validate_cron(changeset, _options \\ []) do

test/lightning/workflows/edge_test.exs

@@ -14,6 +14,19 @@ defmodule Lightning.Workflows.EdgeTest do
       assert changeset.valid?
     end
 
+    test "a malformed uuid field is a changeset error, not an Ecto.ChangeError on save" do
+      changeset =
+        Edge.changeset(%Edge{}, %{
+          workflow_id: Ecto.UUID.generate(),
+          source_job_id: "__ID_JOB_Envoyer-dans-DHIS2__",
+          target_job_id: Ecto.UUID.generate(),
+          condition_type: :on_job_success
+        })
+
+      refute changeset.valid?
+      assert changeset.errors[:source_job_id] == {"is not a valid UUID", []}
+    end
+
     test "edges must have a condition" do
       changeset =
         Edge.changeset(%Edge{}, %{

test/lightning/workflows/job_test.exs

@@ -16,6 +16,22 @@ defmodule Lightning.Workflows.JobTest do
   end
 
   describe "changeset/2" do
+    test "a malformed id is a changeset error, not an Ecto.ChangeError on save" do
+      # An unsubstituted import placeholder reaching :id (a :binary_id field)
+      # passes cast/3 and would only raise when dumped on insert. validate_uuid
+      # surfaces it as a changeset error instead.
+      changeset =
+        Job.changeset(%Job{}, %{
+          id: "__ID_JOB_Envoyer-dans-DHIS2__",
+          name: "Test Job",
+          body: "fn(state => state)",
+          adaptor: "@openfn/language-common@latest"
+        })
+
+      refute changeset.valid?
+      assert changeset.errors[:id] == {"is not a valid UUID", []}
+    end
+
     test "accepts keychain_credential_id in changeset" do
       workflow = insert(:workflow)
 

test/lightning/workflows/trigger_test.exs

@@ -320,6 +320,39 @@ defmodule Lightning.Workflows.TriggerTest do
       assert get_field(changeset, :cron_cursor_job_id) == nil
     end
 
+    test "a malformed cron_cursor_job_id is a changeset error, not an Ecto.ChangeError on save" do
+      changeset =
+        Trigger.changeset(%Trigger{}, %{
+          type: :cron,
+          cron_expression: "* * * * *",
+          cron_cursor_job_id: "__ID_JOB_Envoyer-dans-DHIS2__"
+        })
+
+      refute changeset.valid?
+
+      assert changeset.errors[:cron_cursor_job_id] ==
+               {"is not a valid UUID", []}
+    end
+
+    test "cron_cursor_job_id pointing at a non-existent job is a changeset error, not a raise" do
+      workflow = insert(:workflow)
+
+      trigger =
+        insert(:trigger,
+          workflow: workflow,
+          type: :cron,
+          cron_expression: "* * * * *"
+        )
+
+      changeset =
+        Trigger.changeset(trigger, %{cron_cursor_job_id: Ecto.UUID.generate()})
+
+      assert {:error, changeset} = Lightning.Repo.update(changeset)
+
+      assert {"the referenced cursor job does not exist", _} =
+               changeset.errors[:cron_cursor_job_id]
+    end
+
     test "allows webhook_reply to be set for webhook triggers" do
       changeset =
         Trigger.changeset(%Trigger{}, %{