defmodule AshDispatch.Changes.BroadcastCounterUpdate do @moduledoc """ Broadcasts real-time counter updates for actions that don't dispatch events. This change module provides direct counter broadcasting for actions that need real-time UI updates but don't require full notification workflows. ## When to Use This Pattern Use `BroadcastCounterUpdate` for: - **Actions without events**: Cart add/remove, preference updates, mark as read, etc. - **Real-time counters only**: No email or in-app notification needed - **Simple state updates**: User adds item, counter updates immediately ## When to Use Event-Based Pattern Instead Use event `counters/2` callback for: - **Actions that dispatch events**: Order creation, ticket updates, etc. - **Notification workflows**: In-app + email delivery - **Multi-channel communication**: User and admin notifications ## Configuration The counter system requires two configured modules: config :ash_dispatch, counter_registry: MyApp.CounterRegistry, counter_broadcaster: MyApp.CounterBroadcaster See `AshDispatch.Behaviours.CounterRegistry` and `AshDispatch.Behaviours.CounterBroadcaster` for implementation details. ## Usage ### Simple User Counter # In resource action (e.g., Cart.add_item) change {AshDispatch.Changes.BroadcastCounterUpdate, counter: :cart_items} ### Admin-Only Counter # Broadcast to all admins instead of specific user change {AshDispatch.Changes.BroadcastCounterUpdate, counter: :pending_orders, broadcast_to: :all_admins } ### Custom User ID Field # If the user_id is stored in a different field change {AshDispatch.Changes.BroadcastCounterUpdate, counter: :cart_items, user_id_field: :owner_id } ## Event-Based Pattern (for actions with events) For actions that dispatch events, declare counters in the event module: # In event module (lib/my_app/events/orders/created.ex) defmodule MyApp.Events.Orders.Created do use AshDispatch.Event @impl true def counters(_context, %Channel{transport: :in_app, audience: :user}) do [:pending_orders, :cart_items] end def counters(_context, %Channel{transport: :in_app, audience: :admin}) do [:admin_pending_orders] end def counters(_context, _channel), do: [] end ## Options - `:counter_name` (required) - Counter name to broadcast - `:audience` (required) - Audience type (`:user`, `:admin`, or `:system`) - `:invalidates` (optional) - List of query keys to invalidate - `:user_id_field` (optional) - Field containing user ID (defaults to `:user_id`) ## Examples # User counter - broadcasts to specific user change {BroadcastCounterUpdate, counter_name: :pending_orders, audience: :user} # Admin counter - broadcasts to all admins change {BroadcastCounterUpdate, counter_name: :admin_pending_orders, audience: :admin, invalidates: ["admin_orders"]} ## How It Works The broadcaster implementation (configured via `:counter_broadcaster`) handles: - Query execution (how to count items) - Audience scoping (user-specific vs total counts) - Broadcasting to connected clients This change just declares WHAT to broadcast and to WHOM. """ use Ash.Resource.Change require Ash.Query require Logger @impl true def change(changeset, opts, _context) do Ash.Changeset.after_action(changeset, fn _changeset, result -> broadcast_counter(result, opts) {:ok, result} end) end defp broadcast_counter(record, opts) do counter_name = Keyword.fetch!(opts, :counter_name) resource = Keyword.fetch!(opts, :resource) query_filter = Keyword.fetch!(opts, :query_filter) audience = Keyword.fetch!(opts, :audience) invalidates = Keyword.get(opts, :invalidates, []) filter_by_record = Keyword.get(opts, :filter_by_record) Logger.debug( "[BroadcastCounterUpdate] Starting broadcast for #{counter_name}, audience: #{audience}" ) # Resolve recipients using unified system (same as events) recipients = resolve_recipients_for_counter(record, audience, opts) Logger.debug("[BroadcastCounterUpdate] Resolved #{length(recipients)} recipients") # For each recipient, execute query and broadcast Enum.each(recipients, fn recipient -> # Determine if query should be user-scoped user_scoped? = audience == :user # Execute count query count = try do if user_scoped? do # User-scoped: count items for this specific user query = resource |> Ash.Query.new() # Apply static filter only if not empty/nil query = apply_query_filter(query, query_filter) # Apply user_id filter based on configured path # Skip if filter_by_record is provided - it already scopes to user's data query = if filter_by_record do query else user_id_path = Keyword.get(opts, :user_id_path, [:user_id]) user_filter = build_user_filter(user_id_path, recipient.id) Ash.Query.filter(query, ^user_filter) end # Apply dynamic filter by record field (e.g., cart_id from Cart) query = apply_filter_by_record(query, record, filter_by_record) Ash.count!(query, authorize?: false) else # Global: count all items (admin/system counters) query = Ash.Query.new(resource) # Apply static filter only if not empty/nil query = apply_query_filter(query, query_filter) # Apply dynamic filter by record field query = apply_filter_by_record(query, record, filter_by_record) Ash.count!(query, authorize?: false) end rescue e -> Logger.error( "[BroadcastCounterUpdate] Failed to count #{counter_name}: #{inspect(e)}\n" <> "Resource: #{inspect(resource)}, Query filter: #{inspect(query_filter)}" ) 0 end # Broadcast to recipient Logger.debug( "[BroadcastCounterUpdate] Broadcasting #{counter_name}=#{count} to user #{recipient.id}" ) broadcast_to_user(recipient.id, counter_name, count, invalidates, audience) end) end # Resolve recipients for counter broadcasting using unified system defp resolve_recipients_for_counter(record, audience, opts) do case audience do :user -> # For :user audience, extract the user from the record case extract_user_from_record(record, opts) do nil -> [] user -> [user] end _ -> # For :admin, :system, or custom audiences, use event system channel = %{audience: audience} AshDispatch.Event.Helpers.resolve_recipients_for_audience(nil, channel) end end # Extract user from record (supports nested relationships) defp extract_user_from_record(record, opts) do user_id = extract_user_id(record, opts) if user_id do # Return minimal user struct with id (same format as Event.Helpers) %{id: user_id} else nil end end # Extract user_id from record, supporting nested relationship paths defp extract_user_id(record, opts) do case Keyword.get(opts, :user_id_path) do nil -> # Fallback to direct field lookup user_id_field = Keyword.get(opts, :user_id_field, :user_id) Map.get(record, user_id_field) path when is_list(path) -> # Load and traverse relationship path to get user_id # Example: [:cart, :user_id] loads cart, then gets user_id from cart case load_and_traverse_path(record, path) do {:ok, user_id} -> user_id {:error, reason} -> Logger.warning( "[BroadcastCounterUpdate] Failed to resolve user_id via path #{inspect(path)}: #{inspect(reason)}" ) nil end end end # Load relationships and traverse the path to extract the final value defp load_and_traverse_path(record, path) do # Split path into relationships to load and final field # Example: [:cart, :user_id] -> load [:cart], then get .cart.user_id {relationships, [_final_field]} = Enum.split(path, -1) # Load all relationships in the path case Ash.load(record, relationships, authorize?: false) do {:ok, loaded_record} -> # Traverse the path to get the final value value = get_nested_value(loaded_record, path) {:ok, value} {:error, reason} -> {:error, reason} end end # Recursively traverse a nested path to extract a value defp get_nested_value(record, [field]) do Map.get(record, field) end defp get_nested_value(record, [field | rest]) do case Map.get(record, field) do nil -> nil nested_record -> get_nested_value(nested_record, rest) end end # Apply query_filter keyword list using expr() macro defp apply_query_filter(query, query_filter) when is_list(query_filter) and query_filter != [] do # Convert keyword list to expr() filters # Example: [status: :pending] -> filter(query, status == :pending) Enum.reduce(query_filter, query, fn {field, value}, acc_query -> import Ash.Query import Ash.Expr filter(acc_query, ^ref(field) == ^value) end) end defp apply_query_filter(query, _query_filter), do: query # Apply filter by record field (e.g., filter CartItem by cart_id from Cart) defp apply_filter_by_record(query, _record, nil), do: query defp apply_filter_by_record(query, record, filter_config) do # Extract configuration filter_field = get_config_value(filter_config, :field) record_field = get_config_value(filter_config, :record_field, :id) if filter_field do # Get the value from the triggering record filter_value = Map.get(record, record_field) if filter_value do # Apply filter using expr macro for proper filter construction import Ash.Query import Ash.Expr filter(query, ^ref(filter_field) == ^filter_value) else Logger.warning( "[BroadcastCounterUpdate] Could not extract #{record_field} from record for filtering" ) query end else query end end # Helper to get value from keyword list or map defp get_config_value(config, key, default \\ nil) do cond do is_list(config) -> Keyword.get(config, key, default) is_map(config) -> Map.get(config, key, default) true -> default end end # Build a filter expression for user_id based on the path defp build_user_filter([field], user_id) do [{field, user_id}] end defp build_user_filter([relationship | rest], user_id) do # For nested paths like [:cart, :user_id], build nested filter nested_filter = build_user_filter(rest, user_id) [{relationship, nested_filter}] end defp broadcast_to_user(user_id, counter_name, count, invalidates, _audience) do # Get configured broadcast function case Application.get_env(:ash_dispatch, :counter_broadcast_fn) do nil -> Logger.warning( "[BroadcastCounterUpdate] No counter_broadcast_fn configured, skipping broadcast" ) broadcast_fn when is_function(broadcast_fn, 4) -> # Function capture: &Module.function/4 metadata = %{invalidate_queries: invalidates} broadcast_fn.(user_id, counter_name, count, metadata: metadata) {module, function} when is_atom(module) and is_atom(function) -> # MFA tuple: {Module, :function} metadata = %{invalidate_queries: invalidates} apply(module, function, [user_id, counter_name, count, [metadata: metadata]]) other -> Logger.error( "[BroadcastCounterUpdate] Invalid counter_broadcast_fn config: #{inspect(other)}" ) end end end