feat: add show detailed trigger history (wraft#176)

* feat: add show detailed trigger history

* feat: add pipeline_id and trigger_Id to form entry response

* chore: update swagger doc

* chore: update dependencies in mix.exs and mix.lock

---------

Co-authored-by: Salsabeel <[email protected]>

Author: Faheemvc

lib/wraft_doc/forms/forms.ex

@@ -543,8 +543,8 @@ defmodule WraftDoc.Forms do
     end)
     |> Repo.transaction()
     |> case do
-      {:ok, _} ->
-        :ok
+      {:ok, %{trigger_history: trigger_history}} ->
+        {:ok, %{pipeline_id: pipeline_id, trigger_id: trigger_history.id}}
 
       {:error, step, error, _} ->
         Logger.error("Pipeline failed in step #{inspect(step)}", error: error)

lib/wraft_doc/pipelines/trigger_histories/trigger_histories.ex

@@ -123,4 +123,19 @@ defmodule WraftDoc.Pipelines.TriggerHistories do
   end
 
   def create_pipeline_job(_, _), do: nil
+
+  @doc """
+  Get a trigger history by id.
+  """
+  @spec get_trigger_history(User.t(), Ecto.UUID.t()) :: TriggerHistory.t() | nil
+  def get_trigger_history(%User{current_org_id: org_id}, <<_::288>> = id) do
+    TriggerHistory
+    |> join(:inner, [t], p in Pipeline, on: t.pipeline_id == p.id)
+    |> where([t, p], p.organisation_id == ^org_id)
+    |> where([t], t.id == ^id)
+    |> preload([:creator])
+    |> Repo.one()
+  end
+
+  def get_trigger_history(_, _), do: nil
 end

lib/wraft_doc/pipelines/trigger_histories/trigger_history.ex

@@ -11,6 +11,7 @@ defmodule WraftDoc.Pipelines.TriggerHistories.TriggerHistory do
            only: [
              :id,
              :data,
+             :response,
              :error,
              :state,
              :pipeline_id,
@@ -24,6 +25,7 @@ defmodule WraftDoc.Pipelines.TriggerHistories.TriggerHistory do
 
   schema "trigger_history" do
     field(:data, :map)
+    field(:response, :map)
     field(:error, :map, default: %{})
     field(:state, :integer)
     field(:start_time, :naive_datetime)
@@ -73,8 +75,8 @@ defmodule WraftDoc.Pipelines.TriggerHistories.TriggerHistory do
 
   def update_changeset(%TriggerHistory{} = trigger, attrs \\ %{}) do
     trigger
-    |> cast(attrs, [:error, :state, :start_time, :zip_file])
-    |> validate_required([:state, :error])
+    |> cast(attrs, [:response, :error, :state, :start_time, :zip_file])
+    |> validate_required([:state])
   end
 
   def trigger_end_changeset(%TriggerHistory{} = trigger, attrs \\ %{}) do

lib/wraft_doc/webhooks/event_trigger.ex

@@ -145,7 +145,7 @@ defmodule WraftDoc.Webhooks.EventTrigger do
   @doc """
   Trigger pipeline.completed event when a pipeline execution completes successfully.
   """
-  @spec trigger_pipeline_completed(TriggerHistory.t(), map()) :: :ok
+  @spec trigger_pipeline_completed(TriggerHistory.t(), map()) :: map() | :ok
   def trigger_pipeline_completed(
         %TriggerHistory{pipeline: %{organisation_id: org_id}} = trigger_history,
         pipeline_result
@@ -156,6 +156,8 @@ defmodule WraftDoc.Webhooks.EventTrigger do
     Logger.info(
       "Triggered pipeline.completed webhook for pipeline id #{trigger_history.pipeline_id}"
     )
+
+    payload.pipeline
   end
 
   def trigger_pipeline_completed(%TriggerHistory{} = trigger_history, _pipeline_result) do
@@ -169,7 +171,7 @@ defmodule WraftDoc.Webhooks.EventTrigger do
   @doc """
   Trigger pipeline.failed event when a pipeline execution fails.
   """
-  @spec trigger_pipeline_failed(TriggerHistory.t(), map()) :: :ok
+  @spec trigger_pipeline_failed(TriggerHistory.t(), map()) :: map() | :ok
   def trigger_pipeline_failed(
         %TriggerHistory{pipeline: %{organisation_id: org_id}} = trigger_history,
         error_data
@@ -180,6 +182,8 @@ defmodule WraftDoc.Webhooks.EventTrigger do
     Logger.info(
       "Triggered pipeline.failed webhook for pipeline id #{trigger_history.pipeline_id}"
     )
+
+    payload.pipeline
   end
 
   def trigger_pipeline_failed(%TriggerHistory{} = trigger_history, _error_data) do
@@ -193,7 +197,7 @@ defmodule WraftDoc.Webhooks.EventTrigger do
   @doc """
   Trigger pipeline.partially_completed event when a pipeline execution partially completes.
   """
-  @spec trigger_pipeline_partially_completed(TriggerHistory.t(), map()) :: :ok
+  @spec trigger_pipeline_partially_completed(TriggerHistory.t(), map()) :: map() | :ok
   def trigger_pipeline_partially_completed(
         %TriggerHistory{pipeline: %{organisation_id: org_id}} = trigger_history,
         pipeline_result
@@ -204,6 +208,8 @@ defmodule WraftDoc.Webhooks.EventTrigger do
     Logger.info(
       "Triggered pipeline.partially_completed webhook for pipeline id #{trigger_history.pipeline_id}"
     )
+
+    payload.pipeline
   end
 
   def trigger_pipeline_partially_completed(%TriggerHistory{} = trigger_history, _pipeline_result) do

lib/wraft_doc/workers/bulk_worker.ex

@@ -117,13 +117,22 @@ defmodule WraftDoc.Workers.BulkWorker do
       stage: stage
     }
 
-    trigger =
-      update_trigger_history_state_and_error(trigger, state, error_data)
-
     # Trigger webhook for pipeline failure
     Task.start(fn ->
       trigger = Repo.preload(trigger, :pipeline)
-      EventTrigger.trigger_pipeline_failed(trigger, error_data)
+
+      payload =
+        trigger
+        |> EventTrigger.trigger_pipeline_failed(error_data)
+        |> case do
+          payload when is_map(payload) ->
+            payload
+
+          _ ->
+            %{}
+        end
+
+      update_trigger_history_state_and_error(trigger, state, error_data, payload)
     end)
 
     Logger.error("Form mapping not complete. Pipeline execution failed.")
@@ -143,13 +152,22 @@ defmodule WraftDoc.Workers.BulkWorker do
       stage: stage
     }
 
-    trigger =
-      update_trigger_history_state_and_error(trigger, state, error_data)
-
     # Trigger webhook for pipeline failure
     Task.start(fn ->
       trigger = Repo.preload(trigger, :pipeline)
-      EventTrigger.trigger_pipeline_failed(trigger, error_data)
+
+      payload =
+        trigger
+        |> EventTrigger.trigger_pipeline_failed(error_data)
+        |> case do
+          payload when is_map(payload) ->
+            payload
+
+          _ ->
+            %{}
+        end
+
+      update_trigger_history_state_and_error(trigger, state, error_data, payload)
     end)
 
     Logger.error("Pipeline not found. Pipeline execution failed.")
@@ -169,13 +187,21 @@ defmodule WraftDoc.Workers.BulkWorker do
       stage: stage
     }
 
-    trigger =
-      update_trigger_history_state_and_error(trigger, state, error_data)
-
     # Trigger webhook for pipeline failure
     Task.start(fn ->
-      trigger = Repo.preload(trigger, :pipeline)
-      EventTrigger.trigger_pipeline_failed(trigger, error_data)
+      payload =
+        trigger
+        |> Repo.preload(:pipeline)
+        |> EventTrigger.trigger_pipeline_failed(error_data)
+        |> case do
+          payload when is_map(payload) ->
+            payload
+
+          _ ->
+            %{}
+        end
+
+      update_trigger_history_state_and_error(trigger, state, error_data, payload)
     end)
 
     Logger.error("Instance creation failed. Pipeline execution failed.")
@@ -195,13 +221,22 @@ defmodule WraftDoc.Workers.BulkWorker do
       stage: stage
     }
 
-    trigger =
-      update_trigger_history_state_and_error(trigger, state, error_data)
-
     # Trigger webhook for pipeline failure
     Task.start(fn ->
       trigger = Repo.preload(trigger, :pipeline)
-      EventTrigger.trigger_pipeline_failed(trigger, error_data)
+
+      payload =
+        trigger
+        |> EventTrigger.trigger_pipeline_failed(error_data)
+        |> case do
+          payload when is_map(payload) ->
+            payload
+
+          _ ->
+            %{}
+        end
+
+      update_trigger_history_state_and_error(trigger, state, error_data, payload)
     end)
 
     Logger.error("Instance creation failed. Pipeline execution failed.")
@@ -221,12 +256,21 @@ defmodule WraftDoc.Workers.BulkWorker do
       stage: stage
     }
 
-    trigger =
-      update_trigger_history_state_and_error(trigger, state, error_data)
-
     Task.start(fn ->
       trigger = Repo.preload(trigger, :pipeline)
-      EventTrigger.trigger_pipeline_failed(trigger, error_data)
+
+      payload =
+        trigger
+        |> EventTrigger.trigger_pipeline_failed(error_data)
+        |> case do
+          payload when is_map(payload) ->
+            payload
+
+          _ ->
+            %{}
+        end
+
+      update_trigger_history_state_and_error(trigger, state, error_data, payload)
     end)
 
     Logger.error("Invalid JSON error. Pipeline execution failed.")
@@ -238,7 +282,6 @@ defmodule WraftDoc.Workers.BulkWorker do
          current_user
        ) do
     state = TriggerHistory.states()[:success]
-    trigger = update_trigger_history(trigger, %{state: state, zip_file: zip_file})
 
     instances = Map.get(result, :instances, [])
 
@@ -249,7 +292,20 @@ defmodule WraftDoc.Workers.BulkWorker do
 
     Task.start(fn ->
       trigger = Repo.preload(trigger, :pipeline)
-      EventTrigger.trigger_pipeline_completed(trigger, pipeline_result)
+
+      payload =
+        trigger
+        |> EventTrigger.trigger_pipeline_completed(pipeline_result)
+        |> case do
+          payload when is_map(payload) -> payload
+          _ -> %{}
+        end
+
+      update_trigger_history(trigger, %{
+        state: state,
+        response: payload,
+        zip_file: zip_file
+      })
     end)
 
     Task.start(fn ->
@@ -310,11 +366,18 @@ defmodule WraftDoc.Workers.BulkWorker do
       documents: successful_instances
     }
 
-    trigger = update_trigger_history_state_and_error(trigger, state, error_data_for_db)
-
     Task.start(fn ->
       trigger_with_pipeline = Repo.preload(trigger, :pipeline)
-      EventTrigger.trigger_pipeline_partially_completed(trigger_with_pipeline, pipeline_result)
+
+      payload =
+        trigger_with_pipeline
+        |> EventTrigger.trigger_pipeline_partially_completed(pipeline_result)
+        |> case do
+          payload when is_map(payload) -> payload
+          _ -> %{}
+        end
+
+      update_trigger_history_state_and_error(trigger, state, error_data_for_db, payload)
     end)
 
     Task.start(fn ->
@@ -353,12 +416,12 @@ defmodule WraftDoc.Workers.BulkWorker do
   end
 
   # Update state and error of a trigger history
-  @spec update_trigger_history_state_and_error(TriggerHistory.t(), integer, map) ::
+  @spec update_trigger_history_state_and_error(TriggerHistory.t(), integer(), map(), map()) ::
           TriggerHistory.t()
-  defp update_trigger_history_state_and_error(trigger, state, error) do
+  defp update_trigger_history_state_and_error(trigger, state, error, response \\ %{}) do
     failure_time = DateTime.to_iso8601(Timex.now())
     error = Map.put(error, :failure_time, failure_time)
-    update_trigger_history(trigger, %{state: state, error: error})
+    update_trigger_history(trigger, %{state: state, error: error, response: response})
   end
 
   # Update trigger history on start

lib/wraft_doc_web/controllers/form_entry_controller.ex

@@ -73,6 +73,8 @@ defmodule WraftDocWeb.Api.V1.FormEntryController do
           properties do
             form_id(:string, "Form ID")
             id(:string, "Form Entry ID")
+            trigger_id(:string, "Trigger ID")
+            pipeline_id(:string, "Pipeline ID")
             inserted_at(:string, "When was the form entry inserted", format: "ISO-8601")
             status(:string, "Status of the form entry")
             updated_at(:string, "When was the form entry last updated", format: "ISO-8601")
@@ -87,6 +89,8 @@ defmodule WraftDocWeb.Api.V1.FormEntryController do
             },
             form_id: "aa18afe1-3383-4653-bc0e-505ec3bbfc19",
             id: "f507ca98-9848-49af-89f8-a21f12202ec0",
+            pipeline_id: "12345678-9abc-def0-1234-56789abcdef0",
+            trigger_id: "af2cf1c6-f342-4042-8425-6346e9fd6c44",
             inserted_at: "2024-04-17T07:10:17",
             status: "draft",
             updated_at: "2024-04-17T07:10:17",
@@ -194,8 +198,11 @@ defmodule WraftDocWeb.Api.V1.FormEntryController do
          %FormPipeline{} <- Forms.get_form_pipeline(form, pipeline_id),
          {:ok, %FormEntry{data: data} = form_entry} <-
            Forms.create_form_entry(current_user, form, params),
-         :ok <- Forms.trigger_pipeline(current_user, pipeline_id, data, 0) do
-      render(conn, "form_entry.json", %{form_entry: form_entry})
+         {:ok, trigger_response} <- Forms.trigger_pipeline(current_user, pipeline_id, data, 0) do
+      render(conn, "form_entry.json", %{
+        form_entry: form_entry,
+        trigger_response: trigger_response
+      })
     end
   end