Optimize trust aggregation and discovery counts

config/test.exs

@@ -23,7 +23,8 @@ config :who_need_help, WhoNeedHelp.Repo,
   port: String.to_integer(System.get_env("DB_PORT", "5432")),
   database: "who_need_help_test#{System.get_env("MIX_TEST_PARTITION")}",
   pool: Ecto.Adapters.SQL.Sandbox,
-  pool_size: String.to_integer(System.get_env("TEST_POOL_SIZE", "10"))
+  pool_size: String.to_integer(System.get_env("TEST_POOL_SIZE", "10")),
+  ownership_timeout: String.to_integer(System.get_env("TEST_OWNERSHIP_TIMEOUT_MS", "120000"))
 
 # We don't run a server during test. If one is required,
 # you can enable the server option below.

lib/mix/tasks/wnh.db_scale_benchmark.ex

@@ -529,6 +529,32 @@ defmodule Mix.Tasks.Wnh.DbScaleBenchmark do
       WHERE help_assignments.status = 'completed'
         AND help_assignments.helper_id = md5('user-2')::uuid
       """,
+      "reputation_helper_pair_aggregate" => """
+      SELECT
+        sum(completed),
+        count(requester_id),
+        sum(verified_handovers),
+        sum(location_supported)
+      FROM (
+        SELECT
+          help_assignments.helper_id,
+          help_requests.requester_id,
+          count(help_assignments.id) AS completed,
+          count(help_assignments.id) FILTER (
+            WHERE help_assignments.handover_verified_at IS NOT NULL
+          ) AS verified_handovers,
+          count(help_assignments.id) FILTER (
+            WHERE help_assignments.proximity_observed_at IS NOT NULL
+              AND help_assignments.helper_movement_observed_at IS NOT NULL
+          ) AS location_supported
+        FROM help_assignments
+        INNER JOIN help_requests
+          ON help_requests.id = help_assignments.request_id
+        WHERE help_assignments.status = 'completed'
+          AND help_assignments.helper_id = md5('user-2')::uuid
+        GROUP BY help_assignments.helper_id, help_requests.requester_id
+      ) AS counterpart_stats
+      """,
       "leaderboard_aggregate" => """
       SELECT
         help_assignments.helper_id,
@@ -546,6 +572,33 @@ defmodule Mix.Tasks.Wnh.DbScaleBenchmark do
         ON help_requests.id = help_assignments.request_id
       WHERE help_assignments.status = 'completed'
       GROUP BY help_assignments.helper_id
+      """,
+      "leaderboard_pair_aggregate" => """
+      SELECT
+        helper_id,
+        sum(completed),
+        count(requester_id),
+        count(requester_id) FILTER (WHERE location_supported > 0),
+        count(requester_id) FILTER (WHERE verified > 0)
+      FROM (
+        SELECT
+          help_assignments.helper_id,
+          help_requests.requester_id,
+          count(help_assignments.id) AS completed,
+          count(help_assignments.id) FILTER (
+            WHERE help_assignments.proximity_observed_at IS NOT NULL
+              AND help_assignments.helper_movement_observed_at IS NOT NULL
+          ) AS location_supported,
+          count(help_assignments.id) FILTER (
+            WHERE help_assignments.handover_verified_at IS NOT NULL
+          ) AS verified
+        FROM help_assignments
+        INNER JOIN help_requests
+          ON help_requests.id = help_assignments.request_id
+        WHERE help_assignments.status = 'completed'
+        GROUP BY help_assignments.helper_id, help_requests.requester_id
+      ) AS counterpart_stats
+      GROUP BY helper_id
       """
     }
   end

lib/who_need_help/trust.ex

@@ -235,19 +235,31 @@ defmodule WhoNeedHelp.Trust do
         location_supported: not is_nil(assignment.proximity_observed_at)
       })
 
-    aggregates =
+    counterpart_aggregates =
       helper_rows
       |> union_all(^requester_rows)
       |> subquery()
-      |> group_by([row], row.user_id)
+      |> group_by([row], [row.user_id, row.counterpart_id])
       |> select([row], %{
         user_id: row.user_id,
+        counterpart_id: row.counterpart_id,
         completed: count(row.id),
-        unique_people: count(row.counterpart_id, :distinct),
         verified_handovers: filter(count(row.id), row.verified),
         location_supported: filter(count(row.id), row.location_supported)
       })
 
+    aggregates =
+      counterpart_aggregates
+      |> subquery()
+      |> group_by([counterpart], counterpart.user_id)
+      |> select([counterpart], %{
+        user_id: counterpart.user_id,
+        completed: type(sum(counterpart.completed), :integer),
+        unique_people: count(counterpart.counterpart_id),
+        verified_handovers: type(sum(counterpart.verified_handovers), :integer),
+        location_supported: type(sum(counterpart.location_supported), :integer)
+      })
+
     ratings =
       Review
       |> where(
@@ -1052,30 +1064,43 @@ defmodule WhoNeedHelp.Trust do
   end
 
   defp leaderboard_rows(cursor, limit) do
-    stats =
+    counterpart_stats =
       Assignment
       |> join(:inner, [assignment], request in HelpRequest,
         on: request.id == assignment.request_id
       )
       |> where([assignment], assignment.status == :completed)
-      |> group_by([assignment], assignment.helper_id)
+      |> group_by([assignment, request], [assignment.helper_id, request.requester_id])
       |> select([assignment, request], %{
         helper_id: assignment.helper_id,
+        requester_id: request.requester_id,
         completed: count(assignment.id),
-        unique_people: count(request.requester_id, :distinct),
-        location_supported_people:
+        location_supported:
           filter(
-            count(request.requester_id, :distinct),
+            count(assignment.id),
             not is_nil(assignment.proximity_observed_at) and
               not is_nil(assignment.helper_movement_observed_at)
           ),
-        verified_people:
+        verified:
           filter(
-            count(request.requester_id, :distinct),
+            count(assignment.id),
             not is_nil(assignment.handover_verified_at)
           )
       })
 
+    stats =
+      counterpart_stats
+      |> subquery()
+      |> group_by([counterpart], counterpart.helper_id)
+      |> select([counterpart], %{
+        helper_id: counterpart.helper_id,
+        completed: type(sum(counterpart.completed), :integer),
+        unique_people: count(counterpart.requester_id),
+        location_supported_people:
+          filter(count(counterpart.requester_id), counterpart.location_supported > 0),
+        verified_people: filter(count(counterpart.requester_id), counterpart.verified > 0)
+      })
+
     ratings =
       Review
       |> where([review], not is_nil(review.revealed_at))

lib/who_need_help_web/live/activity_live/index.ex

@@ -263,10 +263,17 @@ defmodule WhoNeedHelpWeb.ActivityLive.Index do
               Activities.map_discovery_items(scope, filters, viewport)
             end
 
+          map_result_count = Enum.reduce(items, 0, &(map_item_count(&1) + &2))
+
           list_result_count =
-            if view_mode == "map" do
+            cond do
+              view_mode == "map" ->
                 0
-            else
+
+              view_mode != "list" and map_and_list_counts_equivalent?(filters) ->
+                map_result_count
+
+              true ->
                 Activities.count_open_activities(scope, filters, viewport: viewport)
             end
 
@@ -346,6 +353,17 @@ defmodule WhoNeedHelpWeb.ActivityLive.Index do
   defp map_item_count(%{type: "cluster", count: count}), do: count
   defp map_item_count(_activity), do: 1
 
+  # Hidden activities can only join a viewport result through an explicit area
+  # search. Without one the map aggregation and list query describe the same
+  # records, so the cluster counts are already the exact list total.
+  defp map_and_list_counts_equivalent?(filters) do
+    case filters["area"] || filters[:area] do
+      nil -> true
+      area when is_binary(area) -> String.trim(area) == ""
+      _invalid -> false
+    end
+  end
+
   defp invalidate_discovery_cache(socket) do
     :ok =
       DiscoveryQueryCache.invalidate(

lib/who_need_help_web/live/request_live/index.ex

@@ -277,10 +277,17 @@ defmodule WhoNeedHelpWeb.RequestLive.Index do
               Help.map_discovery_items(scope, filters, viewport)
             end
 
+          map_result_count = Enum.reduce(items, 0, &(map_item_count(&1) + &2))
+
           list_result_count =
-            if view_mode == "map" do
+            cond do
+              view_mode == "map" ->
                 0
-            else
+
+              view_mode != "list" and map_and_list_counts_equivalent?(filters) ->
+                map_result_count
+
+              true ->
                 Help.count_open_requests(scope, filters, viewport: viewport)
             end
 
@@ -348,6 +355,18 @@ defmodule WhoNeedHelpWeb.RequestLive.Index do
   defp map_item_count(%{type: "cluster", count: count}), do: count
   defp map_item_count(_request), do: 1
 
+  # Hidden or location-less requests can only join a viewport result through
+  # an explicit area search. Without that search the map aggregation and list
+  # query describe the same records, so the cluster counts are already the
+  # exact list total and a second broad COUNT(*) would repeat the scan.
+  defp map_and_list_counts_equivalent?(filters) do
+    case filters["area"] || filters[:area] do
+      nil -> true
+      area when is_binary(area) -> String.trim(area) == ""
+      _invalid -> false
+    end
+  end
+
   defp invalidate_discovery_cache(socket) do
     :ok =
       DiscoveryQueryCache.invalidate(

test/who_need_help/trust_safety_test.exs

@@ -303,6 +303,18 @@ defmodule WhoNeedHelp.TrustSafetyTest do
     assert is_nil(entry.user.email)
     assert is_nil(entry.user.hashed_password)
 
+    helper_reputation = Trust.reputation(context.helper.id)
+    requester_reputation = Trust.reputation(context.requester.id)
+
+    assert helper_reputation.completed == 2
+    assert helper_reputation.unique_people == 1
+    assert helper_reputation.verified_handovers == 2
+    assert helper_reputation.location_supported == 2
+    assert requester_reputation.completed == 2
+    assert requester_reputation.unique_people == 1
+    assert requester_reputation.verified_handovers == 2
+    assert requester_reputation.location_supported == 2
+
     assert Repo.exists?(
              from signal in AbuseSignal,
                where: signal.assignment_id == ^second.id and signal.kind == :repeated_pair