feat(aggregates): add multitenancy bypass option for aggregates

lib/ash/actions/aggregate.ex

@@ -84,29 +84,59 @@ defmodule Ash.Actions.Aggregate do
             with {:ok, query} <- Ash.Actions.Read.handle_multitenancy(query),
                  {:ok, %{valid?: true} = query} <-
                    authorize_query(query, opts, agg_authorize?),
-                 {:ok, aggregates} <- validate_aggregates(query, aggregates, opts),
-                 {:ok, data_layer_query} <-
-                   Ash.Query.data_layer_query(%Ash.Query{
-                     action: Ash.Resource.Info.action(query.resource, read_action),
-                     resource: query.resource,
-                     limit: query.limit,
-                     offset: query.offset,
-                     distinct: query.distinct,
-                     distinct_sort: query.distinct_sort,
-                     sort: query.sort,
-                     domain: query.domain,
-                     tenant: query.tenant,
-                     filter: query.filter,
-                     to_tenant: query.to_tenant,
-                     context: query.context
-                   }),
-                 {:ok, result} <-
-                   Ash.DataLayer.run_aggregate_query(
-                     data_layer_query,
-                     aggregates,
-                     query.resource
-                   ) do
-              {:cont, {:ok, Map.merge(acc, result)}}
+                 {:ok, aggregates} <- validate_aggregates(query, aggregates, opts) do
+              # Group aggregates by bypass vs tenant-specific
+              {bypass_aggs, tenant_aggs} =
+                Enum.split_with(aggregates, &(&1.multitenancy == :bypass))
+
+              # Run both groups and merge results
+              results =
+                Enum.reduce_while(
+                  [
+                    {bypass_aggs, query.tenant,
+                     Map.merge(query.context || %{}, %{
+                       shared: %{private: %{multitenancy: :bypass_all}}
+                     })},
+                    {tenant_aggs, query.tenant, query.context}
+                  ],
+                  {:ok, %{}},
+                  fn
+                    {[], _tenant, _context}, acc ->
+                      {:cont, acc}
+
+                    {aggs, tenant, context}, {:ok, results_acc} ->
+                      with {:ok, data_layer_query} <-
+                             Ash.Query.data_layer_query(%Ash.Query{
+                               action: Ash.Resource.Info.action(query.resource, read_action),
+                               resource: query.resource,
+                               limit: query.limit,
+                               offset: query.offset,
+                               distinct: query.distinct,
+                               distinct_sort: query.distinct_sort,
+                               sort: query.sort,
+                               domain: query.domain,
+                               tenant: tenant,
+                               filter: query.filter,
+                               to_tenant: tenant,
+                               context: context
+                             }),
+                           {:ok, group_results} <-
+                             Ash.DataLayer.run_aggregate_query(
+                               data_layer_query,
+                               aggs,
+                               query.resource
+                             ) do
+                        {:cont, {:ok, Map.merge(results_acc, group_results)}}
+                      else
+                        {:error, error} -> {:halt, {:error, error}}
+                      end
+                  end
+                )
+
+              case results do
+                {:ok, merged} -> {:cont, {:ok, Map.merge(acc, merged)}}
+                {:error, error} -> {:halt, {:error, error}}
+              end
             else
               {:ok, %Ash.Query{} = query} ->
                 {:halt, {:error, Ash.Error.to_error_class(query)}}

lib/ash/actions/read/read.ex

@@ -2787,17 +2787,14 @@ defmodule Ash.Actions.Read do
 
   defp handle_aggregate_multitenancy(query) do
     Enum.reduce_while(query.aggregates, {:ok, %{}}, fn {key, aggregate}, {:ok, acc} ->
-      case handle_multitenancy(
-             Ash.Query.set_tenant(aggregate.query, aggregate.query.tenant || query.tenant)
-           ) do
-        {:ok, %{valid?: true} = query} ->
-          {:cont, {:ok, Map.put(acc, key, %{aggregate | query: query})}}
-
-        {:ok, query} ->
-          {:halt, {:error, Ash.Error.set_path(query.errors, aggregate.name)}}
-
-        {:error, error} ->
-          {:halt, {:error, Ash.Error.set_path(error, aggregate.name)}}
+      with aggregate_query <-
+             apply_aggregate_tenant(aggregate.query, query.tenant, aggregate.multitenancy),
+           :ok <- validate_aggregate_multitenancy(aggregate),
+           {:ok, %{valid?: true} = q} <- handle_multitenancy(aggregate_query) do
+        {:cont, {:ok, Map.put(acc, key, %{aggregate | query: q})}}
+      else
+        {:ok, q} -> {:halt, {:error, Ash.Error.set_path(q.errors, aggregate.name)}}
+        {:error, error} -> {:halt, {:error, Ash.Error.set_path(error, aggregate.name)}}
       end
     end)
     |> case do
@@ -2806,6 +2803,60 @@ defmodule Ash.Actions.Read do
     end
   end
 
+  defp apply_aggregate_tenant(aggregate_query, fallback_tenant, :bypass) do
+    aggregate_query
+    |> Ash.Query.set_tenant(aggregate_query.tenant || fallback_tenant)
+    |> Ash.Query.set_context(%{shared: %{private: %{multitenancy: :bypass_all}}})
+  end
+
+  defp apply_aggregate_tenant(aggregate_query, fallback_tenant, _multitenancy) do
+    Ash.Query.set_tenant(aggregate_query, aggregate_query.tenant || fallback_tenant)
+  end
+
+  defp validate_aggregate_multitenancy(aggregate) do
+    if aggregate.multitenancy == :bypass do
+      with :ok <- validate_context_multitenancy_strategy(aggregate.resource, nil, aggregate.name) do
+        Enum.reduce_while(aggregate.relationship_path, {:ok, aggregate.resource}, fn
+          rel_name, {:ok, current_resource} ->
+            relationship = Ash.Resource.Info.relationship(current_resource, rel_name)
+
+            relationship.destination
+            |> validate_context_multitenancy_strategy(rel_name, aggregate.name)
+            |> case do
+              :ok -> {:cont, {:ok, relationship.destination}}
+              error -> {:halt, error}
+            end
+        end)
+        |> case do
+          {:ok, _} -> :ok
+          error -> error
+        end
+      end
+    else
+      :ok
+    end
+  end
+
+  # Validate context multitenancy and its relationships are not used with bypass
+  # Ref: https://github.com/ash-project/ash_postgres/pull/649#issuecomment-3536654583
+  defp validate_context_multitenancy_strategy(resource, relationship_name, aggregate_name) do
+    if Ash.Resource.Info.multitenancy_strategy(resource) == :context do
+      location = relationship_name && " in relationship `#{relationship_name}`"
+
+      {:error,
+       Ash.Error.Query.InvalidQuery.exception(
+         field: aggregate_name,
+         message: """
+         Aggregate `#{aggregate_name}` uses `multitenancy: :bypass` but resource \
+         `#{inspect(resource)}`#{location} uses `:context` multitenancy strategy. \
+         Multitenancy bypass only supports `:attribute` strategy.
+         """
+       )}
+    else
+      :ok
+    end
+  end
+
   defp add_tenant(data, query) do
     if Ash.Resource.Info.multitenancy_strategy(query.resource) do
       Enum.map(data, fn item ->
@@ -4886,6 +4937,10 @@ defmodule Ash.Actions.Read do
        when phase in ~w[preparing before_action after_action executing around_transaction]a,
        do: %{query | phase: phase}
 
+  defp get_shared_multitenancy(%{context: %{private: %{multitenancy: multitenancy}}}) do
+    multitenancy
+  end
+
   defp get_shared_multitenancy(%{context: %{multitenancy: multitenancy}}) do
     multitenancy
   end

lib/ash/query/aggregate.ex

@@ -18,6 +18,7 @@ defmodule Ash.Query.Aggregate do
     :load,
     :read_action,
     :agg_name,
+    :multitenancy,
     join_filters: %{},
     context: %{},
     authorize?: true,
@@ -143,6 +144,14 @@ defmodule Ash.Query.Aggregate do
       doc: "The tenant to use for the aggregate, if applicable.",
       default: nil
     ],
+    multitenancy: [
+      type: {:in, [:bypass]},
+      doc: """
+      Configures multitenancy behavior for the aggregate.
+
+      * `:bypass` - Aggregate data across all tenants, ignoring the tenant context even if it's set.
+      """
+    ],
     authorize?: [
       type: :boolean,
       default: true,
@@ -230,6 +239,16 @@ defmodule Ash.Query.Aggregate do
           build_opts -> build_query(target_resource, resource, build_opts)
         end
 
+      query =
+        if opts.multitenancy == :bypass do
+          query
+          |> Ash.Query.set_context(%{
+            shared: %{private: %{multitenancy: :bypass_all}}
+          })
+        else
+          query
+        end
+
       Enum.reduce_while(opts.join_filters, {:ok, %{}}, fn {path, filter}, {:ok, acc} ->
         case parse_join_filter(resource, path, filter) do
           {:ok, filter} ->
@@ -387,8 +406,9 @@ defmodule Ash.Query.Aggregate do
                {:ok, type, constraints} <-
                  get_type(kind, type, attribute_type, attribute_constraints, constraints),
                %{valid?: true} = query <- build_query(related, resource, query) do
+            # Tenant is already set for bypass earlier, just handle normal tenant case here
             query =
-              if opts.tenant do
+              if opts.multitenancy != :bypass && opts.tenant do
                 Ash.Query.set_tenant(query, opts.tenant)
               else
                 query
@@ -414,6 +434,7 @@ defmodule Ash.Query.Aggregate do
                sensitive?: sensitive?,
                authorize?: authorize?,
                read_action: read_action,
+               multitenancy: opts.multitenancy,
                join_filters:
                  Map.new(join_filters, fn {key, value} -> {List.wrap(key), value} end),
                related?: related?

lib/ash/query/query.ex

@@ -2471,6 +2471,7 @@ defmodule Ash.Query do
                  authorize?: aggregate.authorize?,
                  sortable?: aggregate.sortable?,
                  sensitive?: aggregate.sensitive?,
+                 multitenancy: aggregate.multitenancy,
                  join_filters:
                    Map.new(aggregate.join_filters, &{&1.relationship_path, &1.filter}),
                  resource: aggregate.resource,

lib/ash/resource/aggregate/aggregate.ex

@@ -20,6 +20,7 @@ defmodule Ash.Resource.Aggregate do
     :sort,
     :default,
     :uniq?,
+    :multitenancy,
     include_nil?: false,
     join_filters: [],
     authorize?: true,
@@ -112,6 +113,14 @@ defmodule Ash.Resource.Aggregate do
       doc: """
       Whether or not the aggregate query should authorize based on the target action, if the parent query is authorized. Requires filter checks on the target action.
       """
+    ],
+    multitenancy: [
+      type: {:in, [:bypass]},
+      doc: """
+      Configures multitenancy behavior for the aggregate.
+
+      * `:bypass` - Aggregate data across all tenants, ignoring the tenant context even if it's set.
+      """
     ]
   ]
 
@@ -132,6 +141,7 @@ defmodule Ash.Resource.Aggregate do
           sortable?: boolean,
           sensitive?: boolean,
           related?: boolean,
+          multitenancy: nil | :bypass,
           __spark_metadata__: Spark.Dsl.Entity.spark_meta()
         }