fix(portal): Fix pagination issues with flows and activities, improve…

… error handling around live tables (#4330)

Fixes issues from logs.
Closes #4274 and similar issues for activities.
Simplifies error handling for live tables (we just reset filters with a
message when they are invalid because just showing an error 422 is not
actionable).

elixir/apps/domain/lib/domain/actors/actor/query.ex

@@ -159,6 +159,16 @@ defmodule Domain.Actors.Actor.Query do
         values: &Domain.Auth.all_providers!/1,
         fun: &filter_by_identity_provider_id/2
       },
+      %Domain.Repo.Filter{
+        name: :status,
+        title: "Status",
+        type: :string,
+        values: [
+          {"Enabled", "enabled"},
+          {"Disabled", "disabled"}
+        ],
+        fun: &filter_by_status/2
+      },
       %Domain.Repo.Filter{
         name: :type,
         title: "Type",
@@ -189,6 +199,14 @@ defmodule Domain.Actors.Actor.Query do
       }
     ]
 
+  def filter_by_status(queryable, "enabled") do
+    {queryable, dynamic([actors: actors], is_nil(actors.disabled_at))}
+  end
+
+  def filter_by_status(queryable, "disabled") do
+    {queryable, dynamic([actors: actors], not is_nil(actors.disabled_at))}
+  end
+
   def filter_by_type(queryable, type) do
     {queryable, dynamic([actors: actors], actors.type == ^type)}
   end

elixir/apps/domain/lib/domain/flows.ex

@@ -125,7 +125,6 @@ defmodule Domain.Flows do
     with :ok <- Auth.ensure_has_permissions(subject, Authorizer.manage_flows_permission()) do
       queryable
       |> Authorizer.for_subject(Flow, subject)
-      |> Ecto.Query.order_by([flows: flows], desc: flows.inserted_at, desc: flows.id)
       |> Repo.list(Flow.Query, opts)
     end
   end
@@ -145,39 +144,24 @@ defmodule Domain.Flows do
     end
   end
 
-  def list_flow_activities_for(assoc, ended_after, started_before, subject, opts \\ [])
+  def list_flow_activities_for(assoc, subject, opts \\ [])
 
-  def list_flow_activities_for(
-        %Flow{} = flow,
-        ended_after,
-        started_before,
-        %Auth.Subject{} = subject,
-        opts
-      ) do
+  def list_flow_activities_for(%Flow{} = flow, %Auth.Subject{} = subject, opts) do
     Activity.Query.all()
     |> Activity.Query.by_flow_id(flow.id)
-    |> list_activities(ended_after, started_before, subject, opts)
+    |> list_activities(subject, opts)
   end
 
-  def list_flow_activities_for(
-        %Accounts.Account{} = account,
-        ended_after,
-        started_before,
-        %Auth.Subject{} = subject,
-        opts
-      ) do
+  def list_flow_activities_for(%Accounts.Account{} = account, %Auth.Subject{} = subject, opts) do
     Activity.Query.all()
     |> Activity.Query.by_account_id(account.id)
-    |> list_activities(ended_after, started_before, subject, opts)
+    |> list_activities(subject, opts)
   end
 
-  defp list_activities(queryable, ended_after, started_before, subject, opts) do
+  defp list_activities(queryable, subject, opts) do
     with :ok <- Auth.ensure_has_permissions(subject, Authorizer.manage_flows_permission()) do
       queryable
-      |> Activity.Query.by_window_ended_at({:greater_than, ended_after})
-      |> Activity.Query.by_window_started_at({:less_than, started_before})
       |> Authorizer.for_subject(Activity, subject)
-      |> Ecto.Query.order_by([activities: activities], asc: activities.window_started_at)
       |> Repo.list(Activity.Query, opts)
     end
   end

elixir/apps/domain/lib/domain/flows/activity/query.ex

@@ -33,4 +33,31 @@ defmodule Domain.Flows.Activity.Query do
       {:activities, :asc, :window_started_at},
       {:activities, :asc, :id}
     ]
+
+  @impl Domain.Repo.Query
+  def filters,
+    do: [
+      %Domain.Repo.Filter{
+        name: :window_within,
+        title: "Window",
+        type: {:range, :datetime},
+        fun: &filter_by_window/2
+      }
+    ]
+
+  def filter_by_window(queryable, %Domain.Repo.Filter.Range{from: from, to: nil}) do
+    {queryable, dynamic([activities: activities], ^from <= activities.window_started_at)}
+  end
+
+  def filter_by_window(queryable, %Domain.Repo.Filter.Range{from: nil, to: to}) do
+    {queryable, dynamic([activities: activities], activities.window_ended_at <= ^to)}
+  end
+
+  def filter_by_window(queryable, %Domain.Repo.Filter.Range{from: from, to: to}) do
+    {queryable,
+     dynamic(
+       [activities: activities],
+       ^from <= activities.window_started_at and activities.window_ended_at <= ^to
+     )}
+  end
 end

elixir/apps/domain/lib/domain/repo/query.ex

@@ -87,12 +87,6 @@ defmodule Domain.Repo.Query do
   def by_range(%Filter.Range{from: from, to: to}, fragment),
     do: dynamic(^from <= ^fragment and ^fragment <= ^to)
 
-  def by_range(%Filter.Range{to: to}, fragment),
-    do: dynamic(^fragment <= ^to)
-
-  def by_range(%Filter.Range{from: from}, fragment),
-    do: dynamic(^fragment >= ^from)
-
   @doc """
   This function is to allow reuse of the filter function in the regular query helpers,
   it takes a return of a filter function (`{queryable, dynamic}`) and applies it to the queryable.

elixir/apps/domain/test/domain/flows_test.exs

@@ -571,15 +571,11 @@ defmodule Domain.FlowsTest do
       flow: flow,
       subject: subject
     } do
-      now = DateTime.utc_now()
-      ended_after = DateTime.add(now, -30, :minute)
-      started_before = DateTime.add(now, 30, :minute)
-
       assert {:ok, [], _metadata} =
-               list_flow_activities_for(account, ended_after, started_before, subject)
+               list_flow_activities_for(account, subject)
 
       assert {:ok, [], _metadata} =
-               list_flow_activities_for(flow, ended_after, started_before, subject)
+               list_flow_activities_for(flow, subject)
     end
 
     test "does not list flow activities from other accounts", %{
@@ -589,15 +585,11 @@ defmodule Domain.FlowsTest do
       flow = Fixtures.Flows.create_flow()
       Fixtures.Flows.create_activity(flow: flow)
 
-      now = DateTime.utc_now()
-      ended_after = DateTime.add(now, -30, :minute)
-      started_before = DateTime.add(now, 30, :minute)
-
       assert {:ok, [], _metadata} =
-               list_flow_activities_for(account, ended_after, started_before, subject)
+               list_flow_activities_for(account, subject)
 
       assert {:ok, [], _metadata} =
-               list_flow_activities_for(flow, ended_after, started_before, subject)
+               list_flow_activities_for(flow, subject)
     end
 
     test "returns ordered by window start time flow activities within a time window", %{
@@ -623,49 +615,73 @@ defmodule Domain.FlowsTest do
       assert {:ok, [], _metadata} =
                list_flow_activities_for(
                  account,
-                 thirty_minutes_in_future,
-                 sixty_minutes_in_future,
-                 subject
+                 subject,
+                 filter: [
+                   window_within: %Domain.Repo.Filter.Range{
+                     from: thirty_minutes_in_future,
+                     to: sixty_minutes_in_future
+                   }
+                 ]
                )
 
       assert {:ok, [], _metadata} =
                list_flow_activities_for(
                  flow,
-                 thirty_minutes_in_future,
-                 sixty_minutes_in_future,
-                 subject
+                 subject,
+                 filter: [
+                   window_within: %Domain.Repo.Filter.Range{
+                     from: thirty_minutes_in_future,
+                     to: sixty_minutes_in_future
+                   }
+                 ]
                )
 
       assert {:ok, [], _metadata} =
                list_flow_activities_for(
                  account,
-                 thirty_minutes_ago,
-                 five_minutes_ago,
-                 subject
+                 subject,
+                 filter: [
+                   window_within: %Domain.Repo.Filter.Range{
+                     from: thirty_minutes_ago,
+                     to: five_minutes_ago
+                   }
+                 ]
                )
 
       assert {:ok, [], _metadata} =
                list_flow_activities_for(
                  flow,
-                 thirty_minutes_ago,
-                 five_minutes_ago,
-                 subject
+                 subject,
+                 filter: [
+                   window_within: %Domain.Repo.Filter.Range{
+                     from: thirty_minutes_ago,
+                     to: five_minutes_ago
+                   }
+                 ]
                )
 
       assert {:ok, [^activity1], _metadata} =
                list_flow_activities_for(
                  account,
-                 five_minutes_ago,
-                 now,
-                 subject
+                 subject,
+                 filter: [
+                   window_within: %Domain.Repo.Filter.Range{
+                     from: five_minutes_ago,
+                     to: now
+                   }
+                 ]
                )
 
       assert {:ok, [^activity1], _metadata} =
                list_flow_activities_for(
                  flow,
-                 five_minutes_ago,
-                 now,
-                 subject
+                 subject,
+                 filter: [
+                   window_within: %Domain.Repo.Filter.Range{
+                     from: five_minutes_ago,
+                     to: now
+                   }
+                 ]
                )
 
       activity2 =
@@ -678,17 +694,25 @@ defmodule Domain.FlowsTest do
       assert {:ok, [^activity2, ^activity1], _metadata} =
                list_flow_activities_for(
                  account,
-                 thirty_minutes_ago,
-                 now,
-                 subject
+                 subject,
+                 filter: [
+                   window_within: %Domain.Repo.Filter.Range{
+                     from: thirty_minutes_ago,
+                     to: now
+                   }
+                 ]
                )
 
       assert {:ok, [^activity2, ^activity1], _metadata} =
                list_flow_activities_for(
                  flow,
-                 thirty_minutes_ago,
-                 now,
-                 subject
+                 subject,
+                 filter: [
+                   window_within: %Domain.Repo.Filter.Range{
+                     from: thirty_minutes_ago,
+                     to: now
+                   }
+                 ]
                )
     end
 
@@ -697,19 +721,15 @@ defmodule Domain.FlowsTest do
       flow: flow,
       subject: subject
     } do
-      now = DateTime.utc_now()
-      ended_after = DateTime.add(now, -30, :minute)
-      started_before = DateTime.add(now, 30, :minute)
-
       subject = Fixtures.Auth.remove_permissions(subject)
 
-      assert list_flow_activities_for(account, ended_after, started_before, subject) ==
+      assert list_flow_activities_for(account, subject) ==
                {:error,
                 {:unauthorized,
                  reason: :missing_permissions,
                  missing_permissions: [Flows.Authorizer.manage_flows_permission()]}}
 
-      assert list_flow_activities_for(flow, ended_after, started_before, subject) ==
+      assert list_flow_activities_for(flow, subject) ==
                {:error,
                 {:unauthorized,
                  reason: :missing_permissions,

elixir/apps/domain/test/domain/jobs/executors/global_test.exs

@@ -30,7 +30,7 @@ defmodule Domain.Jobs.Executors.GlobalTest do
 
   test "registers itself as a leader if there is no global name registered" do
     assert {:ok, pid} = start_link({{__MODULE__, :send_test_message}, 25, test_pid: self()})
-    assert_receive {:executed, ^pid, _time}, 500
+    assert_receive {:executed, ^pid, _time}, 1000
     name = {Domain.Jobs.Executors.Global, __MODULE__, :send_test_message}
     assert :global.whereis_name(name) == pid
 

elixir/apps/web/lib/web/controllers/error_controller.ex

@@ -4,7 +4,6 @@ defmodule Web.ErrorController do
   def show(_conn, params) do
     case params["code"] do
       "404" -> raise Web.LiveErrors.NotFoundError
-      "422" -> raise Web.LiveErrors.InvalidRequestError
       "500" -> raise "internal server error"
     end
 

elixir/apps/web/lib/web/live/actors/index.ex

@@ -9,6 +9,8 @@ defmodule Web.Actors.Index do
       |> assign(page_title: "Actors")
       |> assign_live_table("actors",
         query_module: Actors.Actor.Query,
+        # TODO[bmanifold]: Enable this filter once we start collapsing them
+        hide_filters: [:type],
         sortable_fields: [
           {:actors, :name}
         ],

elixir/apps/web/lib/web/live/flows/download_activities.ex

@@ -36,8 +36,6 @@ defmodule Web.Flows.DownloadActivities do
     with {:ok, activities, activities_metadata} <-
            Flows.list_flow_activities_for(
              flow,
-             flow.inserted_at,
-             flow.expires_at,
              conn.assigns.subject,
              page: [cursor: cursor, limit: 100]
            ),