Audit Enabling / Disabling Workflows

CHANGELOG.md

@@ -17,6 +17,8 @@ and this project adheres to
 
 ### Added
 
+- Audit Enabling / Disabling Workflows
+  [#2697](https://github.com/OpenFn/lightning/issues/2697)
 - Enable / Disable Workflows UI
   [#2698](https://github.com/OpenFn/lightning/issues/2698)
 

lib/lightning/workflows.ex

@@ -106,6 +106,7 @@ defmodule Lightning.Workflows do
         multi |> capture_snapshot()
       end
     end)
+    |> maybe_audit_workflow_state_changes(changeset)
     |> Repo.transaction()
     |> case do
       {:ok, %{workflow: workflow}} ->
@@ -209,13 +210,44 @@ defmodule Lightning.Workflows do
       &(Map.get(&1, :workflow) |> Snapshot.build()),
       returning: false
     )
-    |> Multi.insert(:audit, fn changes ->
+    |> Multi.insert(:audit_snapshot_creation, fn changes ->
       %{snapshot: %{id: snapshot_id}, workflow: %{id: workflow_id}} = changes
 
       Audit.snapshot_created(workflow_id, snapshot_id, changes.actor)
     end)
   end
 
+  defp maybe_audit_workflow_state_changes(multi, changeset) do
+    with triggers when not is_nil(triggers) <- get_change(changeset, :triggers),
+         trigger when not is_nil(trigger) <- find_changed(triggers, :enabled),
+         value <- get_change(trigger, :enabled) do
+      Ecto.Multi.insert(
+        multi,
+        :audit_workflow_state_change,
+        fn %{workflow: %{id: workflow_id}, actor: actor} ->
+          Audit.workflow_state_changed(
+            if(value, do: "enabled", else: "disabled"),
+            workflow_id,
+            actor,
+            %{before: %{enabled: !value}, after: %{enabled: value}}
+          )
+        end
+      )
+    else
+      _ -> multi
+    end
+  end
+
+  defp find_changed(changesets, changed_key) do
+    Enum.find(changesets, fn changeset ->
+      Ecto.Changeset.changed?(changeset, changed_key)
+    end)
+  end
+
+  defp get_change(changeset, change_key) do
+    Ecto.Changeset.get_change(changeset, change_key)
+  end
+
   # Helper to preload associations only if they are present in the attributes
   defp maybe_preload(workflow, assoc, attrs) do
     List.wrap(assoc)

lib/lightning/workflows/audit.ex

@@ -6,7 +6,9 @@ defmodule Lightning.Workflows.Audit do
     repo: Lightning.Repo,
     item: "workflow",
     events: [
-      "snapshot_created"
+      "snapshot_created",
+      "enabled",
+      "disabled"
     ]
 
   def snapshot_created(workflow_id, snapshot_id, actor) do
@@ -19,4 +21,13 @@ defmodule Lightning.Workflows.Audit do
       }
     )
   end
+
+  def workflow_state_changed(event, workflow_id, actor, changes) do
+    event(
+      event,
+      workflow_id,
+      actor,
+      changes
+    )
+  end
 end

test/lightning/workflows_test.exs

@@ -125,6 +125,75 @@ defmodule Lightning.WorkflowsTest do
              } = Repo.one(Audit)
     end
 
+    test "save_workflow/1 audits when a trigger is enabled" do
+      %{id: user_id} = user = insert(:user)
+      workflow = create_workflow()
+
+      {:ok, _workflow} =
+        workflow
+        |> Workflows.update_triggers_enabled_state(true)
+        |> Workflows.save_workflow(user)
+
+      assert_trigger_state_audit(workflow.id, user_id, false, true, "enabled")
+    end
+
+    test "save_workflow/1 audits when a trigger is disabled" do
+      %{id: user_id} = user = insert(:user)
+      workflow = create_workflow(enabled: true)
+
+      {:ok, _workflow} =
+        workflow
+        |> Workflows.update_triggers_enabled_state(false)
+        |> Workflows.save_workflow(user)
+
+      assert_trigger_state_audit(workflow.id, user_id, true, false, "disabled")
+    end
+
+    test "save_workflow/1 does not audit when trigger enabled state doesn't change" do
+      user = insert(:user)
+      workflow = create_workflow(enabled: true)
+
+      {:ok, _workflow} =
+        workflow
+        |> Workflows.update_triggers_enabled_state(true)
+        |> Workflows.save_workflow(user)
+
+      assert Repo.aggregate(Audit, :count) == 0
+    end
+
+    test "save_workflow/1 does not audit when updating other workflow attributes" do
+      user = insert(:user)
+      workflow = create_workflow(enabled: true)
+
+      {:ok, _workflow} =
+        workflow
+        |> Workflows.change_workflow(%{name: "updated name"})
+        |> Workflows.save_workflow(user)
+
+      assert Repo.aggregate(
+               from(a in Audit, where: a.event in ["enabled", "disabled"]),
+               :count
+             ) == 0
+    end
+
+    test "save_workflow/1 with simultaneous trigger and name changes only audits trigger" do
+      %{id: user_id} = user = insert(:user)
+      workflow = create_workflow(enabled: true)
+
+      {:ok, _workflow} =
+        workflow
+        |> Workflows.change_workflow(%{name: "new name"})
+        |> Workflows.update_triggers_enabled_state(false)
+        |> Workflows.save_workflow(user)
+
+      assert_trigger_state_audit(workflow.id, user_id, true, false, "disabled")
+
+      assert Repo.aggregate(
+               from(a in Audit, where: a.event in ["enabled", "disabled"]),
+               :count
+             ) == 1
+    end
+
     test "save_workflow/1 publishes event for updated Kafka triggers" do
       kafka_configuration = build(:triggers_kafka_configuration)
 
@@ -676,4 +745,38 @@ defmodule Lightning.WorkflowsTest do
       assert_received %KafkaTriggerUpdated{trigger_id: ^kafka_trigger_1_id}
     end
   end
+
+  defp assert_trigger_state_audit(
+         workflow_id,
+         user_id,
+         before_state,
+         after_state,
+         event
+       ) do
+    audit =
+      from(a in Audit, where: a.event in ["enabled", "disabled"]) |> Repo.one!()
+
+    assert %{
+             event: ^event,
+             item_type: "workflow",
+             item_id: ^workflow_id,
+             actor_id: ^user_id,
+             changes: %{
+               before: %{"enabled" => ^before_state},
+               after: %{"enabled" => ^after_state}
+             }
+           } = audit
+  end
+
+  defp create_workflow(opts \\ []) do
+    enabled = Keyword.get(opts, :enabled, false)
+    trigger = build(:trigger, type: :cron, enabled: enabled)
+    job = build(:job)
+
+    build(:workflow)
+    |> with_job(job)
+    |> with_trigger(trigger)
+    |> with_edge({trigger, job})
+    |> insert()
+  end
 end