Feat/gh 127 search run per params

apps/gust/lib/gust/flows.ex

@@ -577,6 +577,7 @@ defmodule Gust.Flows do
     * `:limit` - Required. The maximum number of runs to preload.
     * `:offset` - Required. The number of runs to skip before starting to preload.
     * `:status` - Optional. Filters preloaded runs by status when present.
+    * `:params_search` - Optional. Case-insensitively filters runs by parameter keys or values.
 
   ## Returns
 
@@ -589,14 +590,18 @@ defmodule Gust.Flows do
     limit = Keyword.fetch!(opts, :limit)
     offset = Keyword.fetch!(opts, :offset)
     status = Keyword.get(opts, :status)
+    params_search = Keyword.get(opts, :params_search)
 
     runs_q =
       from r in Run,
         order_by: [desc: r.inserted_at],
         limit: ^limit,
         offset: ^offset
 
-    runs_q = maybe_filter_run_status(runs_q, status)
+    runs_q =
+      runs_q
+      |> maybe_filter_run_status(status)
+      |> maybe_filter_run_params(params_search)
 
     Repo.one!(
       from d in Dag,
@@ -611,15 +616,18 @@ defmodule Gust.Flows do
   ## Parameters
 
     * `dag_id` - The identifier of the DAG whose runs should be counted.
+    * `status` - Optional status to filter by.
+    * `params_search` - Optional case-insensitive parameter key or value search.
 
   ## Returns
 
     * The integer count of runs associated with the specified DAG.
   """
-  def count_runs_on_dag(dag_id, status \\ nil) do
+  def count_runs_on_dag(dag_id, status \\ nil, params_search \\ nil) do
     Run
     |> where([r], r.dag_id == ^dag_id)
     |> maybe_filter_run_status(status)
+    |> maybe_filter_run_params(params_search)
     |> Repo.aggregate(:count)
   end
 
@@ -640,6 +648,22 @@ defmodule Gust.Flows do
     where(query, [r], r.status == ^status)
   end
 
+  defp maybe_filter_run_params(query, nil), do: query
+
+  defp maybe_filter_run_params(query, params_search) do
+    case String.trim(params_search) do
+      "" ->
+        query
+
+      params_search ->
+        where(
+          query,
+          [r],
+          fragment("strpos(lower(CAST(? AS text)), lower(?)) > 0", r.params, ^params_search)
+        )
+    end
+  end
+
   @doc """
   Lists all secrets.
   """

apps/gust/test/flows_test.exs

@@ -272,6 +272,99 @@ defmodule FlowsTest do
       assert [failed_run.id] == Enum.map(filtered_dag.runs, & &1.id)
     end
 
+    test "get_dag_by_name_with_runs!/1 searches run parameter keys and values" do
+      dag = dag_fixture(%{name: "params_filtered"})
+      other_dag = dag_fixture(%{name: "other_params_filtered"})
+
+      key_match = run_fixture(%{dag_id: dag.id, params: %{"CustomerReference" => "unrelated"}})
+      value_match = run_fixture(%{dag_id: dag.id, params: %{"reference" => "CUSTOMER-42"}})
+      _no_match = run_fixture(%{dag_id: dag.id, params: %{"reference" => "supplier-7"}})
+      _other_dag_match = run_fixture(%{dag_id: other_dag.id, params: %{"customer" => "42"}})
+
+      key_results =
+        Flows.get_dag_by_name_with_runs!(dag.name,
+          limit: 30,
+          offset: 0,
+          params_search: "customerreference"
+        )
+
+      value_results =
+        Flows.get_dag_by_name_with_runs!(dag.name,
+          limit: 30,
+          offset: 0,
+          params_search: "customer-42"
+        )
+
+      assert Enum.map(key_results.runs, & &1.id) == [key_match.id]
+      assert Enum.map(value_results.runs, & &1.id) == [value_match.id]
+    end
+
+    test "run parameter search works with status, pagination, and counts" do
+      dag = dag_fixture(%{name: "paginated_params_filtered"})
+
+      oldest_match =
+        run_fixture(%{
+          dag_id: dag.id,
+          status: :failed,
+          params: %{"customer" => "match"},
+          inserted_at: ~N[2021-01-01 00:00:00]
+        })
+
+      newest_match =
+        run_fixture(%{
+          dag_id: dag.id,
+          status: :failed,
+          params: %{"customer" => "MATCH"},
+          inserted_at: ~N[2022-01-01 00:00:00]
+        })
+
+      _wrong_status =
+        run_fixture(%{
+          dag_id: dag.id,
+          status: :succeeded,
+          params: %{"customer" => "match"}
+        })
+
+      _wrong_params =
+        run_fixture(%{dag_id: dag.id, status: :failed, params: %{"customer" => "other"}})
+
+      first_page =
+        Flows.get_dag_by_name_with_runs!(dag.name,
+          limit: 1,
+          offset: 0,
+          status: :failed,
+          params_search: "match"
+        )
+
+      second_page =
+        Flows.get_dag_by_name_with_runs!(dag.name,
+          limit: 1,
+          offset: 1,
+          status: :failed,
+          params_search: "match"
+        )
+
+      assert Enum.map(first_page.runs, & &1.id) == [newest_match.id]
+      assert Enum.map(second_page.runs, & &1.id) == [oldest_match.id]
+      assert Flows.count_runs_on_dag(dag.id, :failed, "match") == 2
+      assert Flows.count_runs_on_dag(dag.id, :failed, "missing") == 0
+    end
+
+    test "blank run parameter search does not filter runs" do
+      dag = dag_fixture(%{name: "blank_params_filtered"})
+      run = run_fixture(%{dag_id: dag.id, params: %{"customer" => "match"}})
+
+      loaded_dag =
+        Flows.get_dag_by_name_with_runs!(dag.name,
+          limit: 30,
+          offset: 0,
+          params_search: "  "
+        )
+
+      assert Enum.map(loaded_dag.runs, & &1.id) == [run.id]
+      assert Flows.count_runs_on_dag(dag.id, nil, "  ") == 1
+    end
+
     test "get_dag_by_name_with_runs!/1 raises when required pagination options are missing" do
       dag = dag_fixture(%{name: "missing_required_opts"})
 

apps/gust_web/assets/css/app.css

@@ -411,6 +411,14 @@
   .app-footer__value {
     @apply border border-slate-200 bg-white/80 px-2 py-1 text-[11px] font-medium text-slate-700 shadow-sm;
   }
+
+  .run-params__search-form {
+    @apply flex h-8 min-w-0 flex-1 items-center gap-2 [&_.fieldset]:mb-0 [&_.fieldset]:h-8 [&_.fieldset]:min-w-0 [&_.fieldset]:flex-1;
+  }
+
+  .run-status__filter-form {
+    @apply flex h-8 shrink-0 items-center [&_.fieldset]:mb-0 [&_.fieldset]:h-8;
+  }
 }
 
 @layer components {