Phoenix Channel Integration

AshDispatch provides zero-configuration helpers to integrate real-time notifications and counter updates with Phoenix Channels. These helpers eliminate boilerplate and provide a best-practice implementation out of the box.

Quick Start: UserChannel Macro (Recommended)

The fastest way to set up real-time updates - just 3 lines of code:

# lib/my_app_web/channels/user_channel.ex
defmodule MyAppWeb.UserChannel do
  use AshDispatch.Phoenix.UserChannel,
    endpoint: MyAppWeb.Endpoint
end

Add to your socket:

# lib/my_app_web/user_socket.ex
channel "user:*", MyAppWeb.UserChannel

That's it! You get:

join/3 with authorization
handle_info(:after_join, ...) with initial state
handle_in("refresh_counters", ...) for client requests
broadcast_notification/2, broadcast_counter/4, broadcast_counters/2

All callbacks are defoverridable so you can customize as needed.

Customizing the Macro

Override any callback by defining it in your module:

defmodule MyAppWeb.UserChannel do
  use AshDispatch.Phoenix.UserChannel,
    endpoint: MyAppWeb.Endpoint

  # Custom join with logging
  def join("user:" <> user_id, payload, socket) do
    Logger.info("User #{user_id} joining channel")

    if socket.assigns.user_id == user_id do
      send(self(), :after_join)
      {:ok, socket}
    else
      {:error, %{reason: "unauthorized"}}
    end
  end

  # Add custom message handlers
  def handle_in("custom_action", payload, socket) do
    # Your custom logic
    {:reply, :ok, socket}
  end
end

What You Need to Write

Backend (Elixir)

File	Purpose	AshDispatch Provides
`user_channel.ex`	Phoenix channel for real-time updates	UserChannel macro (3 lines) or helpers
`user_socket.ex`	WebSocket authentication	- (standard Phoenix)

3 lines with UserChannel macro, or ~50-80 lines with manual helper usage.

Frontend (TypeScript/React)

File	Purpose	Generated/Manual
`ash-dispatch/`	Complete SDK	Generated by `mix ash_dispatch.gen`
`counters.ts`	Counter types & metadata	Generated by `mix ash_dispatch.gen`

Run mix ash_dispatch.gen to generate the complete TypeScript SDK with hooks, stores, and types. See Generator for details.

Architecture: Two Stores, One Source of Truth

┌─────────────────┐     WebSocket      ┌──────────────────┐
│  Counter Store  │ ←───────────────── │  AshDispatch     │
│ (all counters)  │   counter_updated  │  Broadcasting    │
└────────┬────────┘                    └──────────────────┘
         │
         │ read from
         ▼
┌─────────────────┐
│ useNotifications│ ← unreadCount from counter store
│ useCounters     │ ← all counters from counter store
└─────────────────┘

Key insight: The counter store is the single source of truth for all real-time counts. Feature hooks (like useNotifications) read counts from the counter store, not their own state.

See Counter Store as Single Source of Truth for details.

Quick Start

1. Configure Counter Broadcasting

Tell AshDispatch which function to call when broadcasting counter updates:

# config/config.exs
config :ash_dispatch,
  counter_broadcast_fn: {MyAppWeb.UserChannel, :broadcast_counter}

2. Setup Your UserChannel

Use the helper modules for zero-boilerplate channel implementation:

defmodule MyAppWeb.UserChannel do
  use MyAppWeb, :channel

  # Import all helper modules
  alias AshDispatch.Helpers.{ChannelState, CounterLoader, NotificationLoader}

  @impl true
  def join("user:" <> user_id, _payload, socket) do
    if socket.assigns.user_id == user_id do
      send(self(), :after_join)
      {:ok, socket}
    else
      {:error, %{reason: "unauthorized"}}
    end
  end

  @impl true
  def handle_info(:after_join, socket) do
    # Single line to load complete initial state (counters + notifications in parallel!)
    initial_state = ChannelState.build(socket.assigns.user_id)

    push(socket, "initial_state", initial_state)
    {:noreply, socket}
  end

  # Client requests to refresh counters
  @impl true
  def handle_in("refresh_counters", _payload, socket) do
    counters = CounterLoader.load_counters_for_user(socket.assigns.user_id)
    {:reply, {:ok, %{counters: counters}}, socket}
  end

  # Client marks notification as read
  @impl true
  def handle_in("mark_notification_read", %{"id" => id}, socket) do
    case NotificationLoader.mark_as_read(id, actor: socket.assigns.current_user) do
      {:ok, _} -> {:reply, :ok, socket}
      {:error, %Ash.Error.Forbidden{}} -> {:reply, {:error, %{reason: "unauthorized"}}, socket}
      {:error, error} -> {:reply, {:error, %{reason: inspect(error)}}, socket}
    end
  end

  # Client marks all notifications as read
  @impl true
  def handle_in("mark_all_notifications_read", _payload, socket) do
    user_id = socket.assigns.user_id

    case NotificationLoader.mark_all_as_read(user_id, actor: socket.assigns.current_user) do
      {:ok, %{marked_count: _count}} ->
        broadcast_all_notifications_read(user_id)
        broadcast_counter(user_id, :unread_notifications, 0)
        {:reply, :ok, socket}

      {:error, reason} ->
        {:reply, {:error, %{reason: "failed", details: inspect(reason)}}, socket}
    end
  end

  ## Broadcaster functions (called by AshDispatch)

  def broadcast_notification(user_id, notification) do
    MyAppWeb.Endpoint.broadcast("user:#{user_id}", "new_notification", notification)
  end

  def broadcast_counter(user_id, counter_name, value, opts \\ []) do
    metadata = Keyword.get(opts, :metadata, %{})

    MyAppWeb.Endpoint.broadcast("user:#{user_id}", "counter_updated", %{
      counter: counter_name,
      value: value,
      metadata: metadata
    })
  end

  defp broadcast_all_notifications_read(user_id) do
    MyAppWeb.Endpoint.broadcast("user:#{user_id}", "all_notifications_read", %{})
  end
end

That's it! About 50 lines of code for a complete real-time notification and counter system.

Helper Modules

ChannelState - Complete Initial State

Module: AshDispatch.Helpers.ChannelState

Builds the complete initial state when users connect to your channel. Loads counters and notifications in parallel for optimal performance.

Basic Usage

def handle_info(:after_join, socket) do
  user_id = socket.assigns.user_id

  # Single call loads everything
  initial_state = ChannelState.build(user_id)
  # => %{
  #   "counters" => %{"pending_orders" => 5, "cart_items" => 3},
  #   "notifications" => [%{id: "...", title: "...", ...}, ...]
  # }

  push(socket, "initial_state", initial_state)
  {:noreply, socket}
end

Custom Options

# Limit number of notifications
ChannelState.build(user_id, notification_limit: 10)

# Custom notification serializer
ChannelState.build(user_id,
  notification_serializer: &MyApp.serialize_notification/1
)

# Load from specific domains only
ChannelState.build(user_id,
  counter_domains: [MyApp.Orders, MyApp.Tickets]
)

# Disable parallel loading (for debugging)
ChannelState.build(user_id, parallel: false)

Loading Only Counters or Notifications

Use the underlying loaders directly when you only need one type:

alias AshDispatch.Helpers.{CounterLoader, NotificationLoader}

# Counters only
counters = CounterLoader.load_counters_for_user(user_id)
# => %{pending_orders: 5, cart_items: 3}

# Notifications only
notifications = NotificationLoader.load_recent(user_id, limit: 20)
# => [%{id: "...", ...}, ...]

Performance

Parallel Loading: Uses Task.async to load counters and notifications simultaneously
Auto-Discovery: Counters are discovered from DSL, no manual configuration
Efficient Queries: Only loads what's needed for the specific user

CounterLoader - Auto-Discovery Counter Loading

Module: AshDispatch.Helpers.CounterLoader

Automatically discovers counter definitions from your resource DSL and loads their current values.

Basic Usage

# Load all counters for a user
counters = CounterLoader.load_counters_for_user(user_id)
# => %{
#   pending_orders: 5,
#   cart_items: 3,
#   active_tickets: 2
# }

How It Works

Discovers Resources: Scans all configured Ash domains
Reads DSL: Finds all counters do blocks in resources
Determines Audiences: Checks which counters apply to this user (:user, :admin, etc.)
Executes Queries: Runs counter queries and returns results

Audience Filtering

Automatically filters counters based on user's audiences:

# Regular user - gets only :user counters
CounterLoader.load_counters_for_user("user-123")
# => %{pending_orders: 5, cart_items: 3}

# Admin user - gets :user + :admin counters
CounterLoader.load_counters_for_user("admin-456")
# => %{
#   pending_orders: 12,           # :user counter
#   cart_items: 0,                # :user counter
#   admin_pending_reseller_requests: 3  # :admin counter
# }

Configuration

# config/config.exs
config :ash_dispatch,
  domains: [MyApp.Orders, MyApp.Tickets, MyApp.Catalog],
  user_module: MyApp.Accounts.User,
  recipient_filters: [
    audiences: [
      admin: [admin: true],
      partner: [role: :partner]
    ]
  ]

NotificationLoader - Notification Management

Module: AshDispatch.Helpers.NotificationLoader

Handles loading, serializing, and updating notifications with sensible defaults.

Load Recent Notifications

# Default: 50 most recent
notifications = NotificationLoader.load_recent(user_id)
# => [%{id: "...", title: "...", message: "...", read: false, ...}]

# Custom limit
notifications = NotificationLoader.load_recent(user_id, limit: 10)

# Custom serializer
notifications = NotificationLoader.load_recent(user_id,
  serializer: &MyApp.serialize_notification/1
)

Mark as Read

# Mark single notification as read
case NotificationLoader.mark_as_read(notification_id, actor: current_user) do
  {:ok, notification} -> # Success
  {:error, %Ash.Error.Forbidden{}} -> # Unauthorized
  {:error, reason} -> # Other error
end

# Mark all as read
case NotificationLoader.mark_all_as_read(user_id, actor: current_user) do
  {:ok, %{marked_count: count}} -> # Success, count notifications marked
  {:error, reason} -> # Failed
end

Default Serialization

The default serializer uses camelCase keys for JavaScript compatibility:

%{
  id: "abc-123",
  type: :success,
  title: "Order Created",
  message: "Your order #1234 has been created",
  read: false,
  timestamp: ~U[2025-01-16 10:00:00Z],
  metadata: %{order_id: "1234"},
  actionLabel: "View Order",  # camelCase!
  actionUrl: "/orders/1234"   # camelCase!
}

Custom Serializer

Provide your own serializer for custom fields or format:

defmodule MyApp.NotificationSerializer do
  def serialize(notification) do
    %{
      id: notification.id,
      title: notification.title,
      message: notification.message,
      read: notification.read,
      # Custom fields
      priority: notification.metadata[:priority] || :normal,
      icon: notification_icon(notification.type),
      timestamp: format_timestamp(notification.inserted_at)
    }
  end

  defp notification_icon(:success), do: "check-circle"
  defp notification_icon(:error), do: "alert-circle"
  defp notification_icon(_), do: "info-circle"

  defp format_timestamp(dt), do: DateTime.to_unix(dt)
end

# Use in channel
NotificationLoader.load_recent(user_id,
  serializer: &MyApp.NotificationSerializer.serialize/1
)

Complete Example: Production-Ready UserChannel

Here's a complete, production-ready Phoenix Channel using all helpers:

defmodule MyAppWeb.UserChannel do
  @moduledoc """
  Central channel for user-specific real-time updates:
  - Notifications (in-app)
  - Real-time counters (cart, tickets, orders, etc.)
  - Other user-specific events

  Uses AshDispatch helpers for zero-boilerplate implementation.
  """
  use MyAppWeb, :channel

  alias AshDispatch.Helpers.{ChannelState, CounterLoader, NotificationLoader}
  require Logger

  @impl true
  def join("user:" <> user_id, _payload, socket) do
    # Verify authorization (token verification in UserSocket)
    if socket.assigns.user_id == user_id do
      # Send initial state after successful join
      send(self(), :after_join)
      {:ok, socket}
    else
      Logger.warning("[UserChannel] Unauthorized join attempt for user:#{user_id}")
      {:error, %{reason: "unauthorized"}}
    end
  end

  @impl true
  def handle_info(:after_join, socket) do
    user_id = socket.assigns.user_id
    Logger.debug("[UserChannel] Loading initial state for user:#{user_id}")

    # Build complete initial state (counters + notifications in parallel)
    initial_state = ChannelState.build(user_id)

    push(socket, "initial_state", initial_state)
    {:noreply, socket}
  end

  ## Client Requests

  @impl true
  def handle_in("refresh_counters", _payload, socket) do
    user_id = socket.assigns.user_id
    counters = CounterLoader.load_counters_for_user(user_id)
    {:reply, {:ok, %{counters: counters}}, socket}
  end

  @impl true
  def handle_in("mark_notification_read", %{"id" => id}, socket) do
    case NotificationLoader.mark_as_read(id, actor: socket.assigns.current_user) do
      {:ok, _notification} ->
        {:reply, :ok, socket}

      {:error, %Ash.Error.Forbidden{}} ->
        Logger.warning("[UserChannel] Unauthorized mark_as_read attempt: #{id}")
        {:reply, {:error, %{reason: "unauthorized"}}, socket}

      {:error, error} ->
        Logger.error("[UserChannel] Failed to mark notification as read: #{inspect(error)}")
        {:reply, {:error, %{reason: "failed"}}, socket}
    end
  end

  @impl true
  def handle_in("mark_all_notifications_read", _payload, socket) do
    user_id = socket.assigns.user_id

    case NotificationLoader.mark_all_as_read(user_id, actor: socket.assigns.current_user) do
      {:ok, %{marked_count: _count}} ->
        # Broadcast to all user's connected clients
        broadcast_all_notifications_read(user_id)
        # Update unread counter
        broadcast_counter(user_id, :unread_notifications, 0)
        {:reply, :ok, socket}

      {:error, reason} ->
        Logger.error("[UserChannel] Failed to mark all as read: #{inspect(reason)}")
        {:reply, {:error, %{reason: "failed"}}, socket}
    end
  end

  ## Broadcaster Functions (called from AshDispatch)

  @doc """
  Broadcast a new notification to a user.
  Called automatically by AshDispatch when notifications are created.
  """
  def broadcast_notification(user_id, notification) do
    MyAppWeb.Endpoint.broadcast("user:#{user_id}", "new_notification", notification)
  end

  @doc """
  Broadcast counter update to a user.
  Called automatically by AshDispatch when counters change.

  Options:
  - `:metadata` - Map with optional `invalidate_queries` list
  """
  def broadcast_counter(user_id, counter_name, value, opts \\ []) do
    metadata = Keyword.get(opts, :metadata, %{})

    MyAppWeb.Endpoint.broadcast("user:#{user_id}", "counter_updated", %{
      counter: counter_name,
      value: value,
      metadata: metadata
    })
  end

  ## Private Helpers

  defp broadcast_all_notifications_read(user_id) do
    MyAppWeb.Endpoint.broadcast("user:#{user_id}", "all_notifications_read", %{})
  end
end

Frontend Integration

React/TypeScript Example

import { Socket, Channel } from "phoenix";

interface Counters {
  pending_orders: number;
  cart_items: number;
  active_tickets: number;
  [key: string]: number;
}

interface Notification {
  id: string;
  type: 'info' | 'success' | 'warning' | 'error';
  title: string;
  message: string;
  read: boolean;
  timestamp: string;
  actionLabel?: string;
  actionUrl?: string;
}

interface InitialState {
  counters: Counters;
  notifications: Notification[];
}

export function useUserChannel(userId: string) {
  const [counters, setCounters] = useState<Counters>({});
  const [notifications, setNotifications] = useState<Notification[]>([]);
  const [channel, setChannel] = useState<Channel | null>(null);

  useEffect(() => {
    // Connect to channel
    const socket = new Socket("/socket", {
      params: { token: getAuthToken() }
    });
    socket.connect();

    const channel = socket.channel(`user:${userId}`, {});

    // Receive initial state
    channel.on("initial_state", (payload: InitialState) => {
      setCounters(payload.counters);
      setNotifications(payload.notifications);
    });

    // Counter updates
    channel.on("counter_updated", (payload: {
      counter: string;
      value: number;
      metadata: { invalidate_queries?: string[] };
    }) => {
      setCounters(prev => ({ ...prev, [payload.counter]: payload.value }));

      // Invalidate related queries
      payload.metadata.invalidate_queries?.forEach(queryKey => {
        queryClient.invalidateQueries([queryKey]);
      });
    });

    // New notifications
    channel.on("new_notification", (notification: Notification) => {
      setNotifications(prev => [notification, ...prev]);

      // Show toast
      toast.success(notification.title, {
        description: notification.message
      });
    });

    // All notifications marked as read
    channel.on("all_notifications_read", () => {
      setNotifications(prev => prev.map(n => ({ ...n, read: true })));
    });

    channel.join()
      .receive("ok", () => console.log("Joined user channel"))
      .receive("error", (err) => console.error("Failed to join", err));

    setChannel(channel);

    return () => {
      channel.leave();
      socket.disconnect();
    };
  }, [userId]);

  // Actions
  const markAsRead = (notificationId: string) => {
    channel?.push("mark_notification_read", { id: notificationId })
      .receive("ok", () => {
        setNotifications(prev =>
          prev.map(n => n.id === notificationId ? { ...n, read: true } : n)
        );
      });
  };

  const markAllAsRead = () => {
    channel?.push("mark_all_notifications_read", {})
      .receive("ok", () => {
        setNotifications(prev => prev.map(n => ({ ...n, read: true })));
      });
  };

  const refreshCounters = () => {
    channel?.push("refresh_counters", {})
      .receive("ok", (resp) => {
        setCounters(resp.counters);
      });
  };

  return {
    counters,
    notifications,
    markAsRead,
    markAllAsRead,
    refreshCounters
  };
}

Usage in Components

function Dashboard() {
  const { counters, notifications, markAsRead, markAllAsRead } =
    useUserChannel(currentUser.id);

  return (
    <div>
      {/* Show counters */}
      <Badge>Pending Orders: {counters.pending_orders || 0}</Badge>
      <Badge>Cart Items: {counters.cart_items || 0}</Badge>

      {/* Notification dropdown */}
      <NotificationDropdown
        notifications={notifications}
        onMarkAsRead={markAsRead}
        onMarkAllAsRead={markAllAsRead}
      />
    </div>
  );
}

Configuration

Counter Broadcasting

Configure the function to call when broadcasting counter updates:

# config/config.exs
config :ash_dispatch,
  # MFA tuple (recommended - easier to test)
  counter_broadcast_fn: {MyAppWeb.UserChannel, :broadcast_counter}

  # Or function capture
  # counter_broadcast_fn: &MyAppWeb.UserChannel.broadcast_counter/4

Full Configuration Example

# config/config.exs
config :ash_dispatch,
  # Required: Ash domains to scan for counters
  domains: [MyApp.Orders, MyApp.Tickets, MyApp.Catalog, MyApp.Accounts],

  # Required: User module for audience checking
  user_module: MyApp.Accounts.User,

  # Required: Counter broadcasting
  counter_broadcast_fn: {MyAppWeb.UserChannel, :broadcast_counter},

  # Optional: Audience filters for counter visibility
  recipient_filters: [
    audiences: [
      admin: [admin: true],
      partner: [role: :partner],
      user: []  # All authenticated users
    ]
  ]

Testing

Test Channel with Helpers

defmodule MyAppWeb.UserChannelTest do
  use MyAppWeb.ChannelCase

  alias AshDispatch.Helpers.{ChannelState, CounterLoader, NotificationLoader}

  setup do
    user = build(:user) |> create!()
    {:ok, socket} = connect(MyAppWeb.UserSocket, %{"token" => user.token})
    {:ok, _, socket} = subscribe_and_join(socket, MyAppWeb.UserChannel, "user:#{user.id}")

    %{socket: socket, user: user}
  end

  test "sends initial state on join", %{user: user} do
    # Initial state pushed automatically
    assert_push "initial_state", %{
      "counters" => counters,
      "notifications" => notifications
    }

    assert is_map(counters)
    assert is_list(notifications)
  end

  test "refreshes counters on request", %{socket: socket} do
    ref = push(socket, "refresh_counters", %{})
    assert_reply ref, :ok, %{counters: counters}

    assert is_map(counters)
  end

  test "marks notification as read", %{socket: socket, user: user} do
    # Create notification
    notification = build(:notification, %{user_id: user.id}) |> create!()

    ref = push(socket, "mark_notification_read", %{"id" => notification.id})
    assert_reply ref, :ok, %{}

    # Verify marked as read
    notification = reload!(notification)
    assert notification.read == true
  end
end

Mock Counter Broadcast in Tests

# config/test.exs
config :ash_dispatch,
  counter_broadcast_fn: {MyAppTest.MockCounterBroadcaster, :broadcast}

# test/support/mock_counter_broadcaster.ex
defmodule MyAppTest.MockCounterBroadcaster do
  def broadcast(user_id, counter_name, value, opts) do
    # Send to test process for assertions
    send(self(), {:counter_broadcast, user_id, counter_name, value, opts})
    :ok
  end
end

# In tests
test "broadcasts counter update" do
  # Trigger action that updates counter...

  assert_received {:counter_broadcast, user_id, :pending_orders, 5, _opts}
end

Performance Optimization

Parallel Loading

The ChannelState module loads counters and notifications in parallel by default:

# Automatic parallel loading
ChannelState.build(user_id)

# Disable for debugging
ChannelState.build(user_id, parallel: false)

Counter Caching

For frequently-accessed counters, consider caching at the database level:

# In resource
counters do
  counter :pending_orders,
    trigger_on: [:create, :complete, :cancel],
    counter_name: :pending_orders,
    query_filter: [status: :pending],
    audience: :user,
    invalidates: ["orders"]
end

# Add database index for fast counting
create index(:orders, [:user_id, :status])

Reduce Notification Payload

Load fewer notifications on initial join:

# Default: 50 notifications
ChannelState.build(user_id)

# Reduced: 10 notifications
ChannelState.build(user_id, notification_limit: 10)

Troubleshooting

"No counter_broadcast_fn configured"

Problem: Warning logged when counter updates.

Solution:

config :ash_dispatch,
  counter_broadcast_fn: {MyAppWeb.UserChannel, :broadcast_counter}

"Failed to load counters"

Problem: Error when loading counters.

Causes:

Missing :domains configuration
Missing :user_module configuration
Counter DSL error in resource

Solution:

config :ash_dispatch,
  domains: [MyApp.Orders, MyApp.Tickets],
  user_module: MyApp.Accounts.User

Counters Not Updating in Real-Time

Problem: Counter changes don't trigger broadcasts.

Causes:

trigger_on doesn't match action name
Counter DSL not properly configured
Broadcast function not configured

Debug:

# Check counter definitions
AshDispatch.Dsl.Info.counters(MyApp.Orders.ProductOrder)

# Verify broadcast function
Application.get_env(:ash_dispatch, :counter_broadcast_fn)

Next Steps

Counter Broadcasting - Define counters in your resources
Configuration - Complete configuration reference

← Previous Page Integrating AshDispatch into Your Application

Next Page → Counter Broadcasting