You are an expert in Phoenix Channels and PubSub for building real-time communication features in Elixir applications.

## Key Points

- **Socket**: The persistent connection between client and server. Handles authentication and connection-level state.
- **Channel**: A topic-scoped conversation over a socket. Handles joining, leaving, and message passing.
- **Topic**: A string identifier for a channel (e.g., `"room:lobby"`, `"user:42"`).
- **PubSub**: The backend broadcast system. Messages published to a topic reach all subscribers across the cluster.
- **Presence**: Built-in module for tracking which users are connected to which topics, with automatic conflict resolution across nodes.
- Authenticate at the socket level in `connect/3`, then authorize topic access in `join/3`.
- Use `broadcast_from!/3` instead of `broadcast!/3` when the sender should not receive their own message (e.g., typing indicators).
- Use Presence for tracking online users — it handles node failures and network partitions automatically via CRDTs.
- Keep channel payloads small. Send IDs and let clients fetch full data if needed, or use a separate API.
- Use `intercept/1` and `handle_out/3` sparingly — only when you need per-client message filtering (e.g., permission-based visibility).
- Rate-limit incoming messages in `handle_in/3` to prevent abuse.
- **Blocking in `handle_in`**: Long database queries or HTTP calls block the channel process. Spawn a Task and send results back via `push/3`.

## Quick Example

```elixir
defmodule MyAppWeb.Presence do
  use Phoenix.Presence,
    otp_app: :my_app,
    pubsub_server: MyApp.PubSub
end
```

Phoenix Channels and PubSub — Elixir/Phoenix

You are an expert in Phoenix Channels and PubSub for building real-time communication features in Elixir applications.

Overview

Phoenix Channels provide real-time, bidirectional communication between clients and servers over WebSockets (with long-polling fallback). Channels are built on top of Phoenix.PubSub, which provides a distributed publish-subscribe system. Together they power features like chat, notifications, live dashboards, and collaborative editing.

Core Philosophy

Phoenix Channels embrace the idea that real-time communication should be a first-class concern, not a bolted-on afterthought. The BEAM's lightweight process model makes it natural to maintain thousands of persistent WebSocket connections on a single node, each backed by its own process with isolated state. This is not a workaround — it is the architecture the platform was designed for.

The separation between sockets, channels, and PubSub reflects a deliberate layering of concerns. The socket handles connection-level authentication, the channel handles topic-level authorization and message routing, and PubSub handles cross-node distribution. Each layer has a single responsibility, and each can be reasoned about independently. When you push authorization into the channel's join/3 and keep business logic in your context modules, the channel becomes a thin routing layer that is easy to test and secure.

Presence builds on CRDTs to solve a genuinely hard distributed systems problem — tracking who is connected across multiple nodes without a single point of failure. Rather than implementing your own user-tracking state that drifts across nodes, Presence provides an eventually consistent view that handles node crashes, network partitions, and split-brain scenarios automatically.

Anti-Patterns

• Business Logic in Channels: Putting database queries, complex validations, and workflow orchestration directly in handle_in/3 callbacks. Channels should delegate to context modules and return the result. This keeps channels testable and prevents them from becoming god modules.

• Unbounded Broadcasts: Broadcasting large HTML payloads or full database records to every subscriber on every state change without considering the number of connected clients or message frequency. This can overwhelm clients and network bandwidth. Send minimal payloads and let clients fetch details if needed.

• Skipping Topic Validation: Allowing clients to join arbitrary topics by pattern-matching loosely on the topic string in join/3. Always validate that the connecting user has permission to access the specific resource identified in the topic.

• Blocking the Channel Process: Making synchronous HTTP calls or running expensive database queries inside handle_in/3. Since each channel is a process, blocking it prevents all other messages for that client from being processed. Spawn a Task for slow work and push results back asynchronously.

• Ignoring Reconnection Semantics: Designing channel interactions that assume a persistent, unbroken connection. Clients disconnect frequently due to network changes, device sleep, and server deploys. Design for idempotent rejoins and consider providing message history on reconnect.

Core Concepts

• Socket: The persistent connection between client and server. Handles authentication and connection-level state.
• Channel: A topic-scoped conversation over a socket. Handles joining, leaving, and message passing.
• Topic: A string identifier for a channel (e.g., "room:lobby", "user:42").
• PubSub: The backend broadcast system. Messages published to a topic reach all subscribers across the cluster.
• Presence: Built-in module for tracking which users are connected to which topics, with automatic conflict resolution across nodes.

Implementation Patterns

Socket Setup

defmodule MyAppWeb.UserSocket do
  use Phoenix.Socket

  channel "room:*", MyAppWeb.RoomChannel
  channel "notifications:*", MyAppWeb.NotificationChannel

  @impl true
  def connect(%{"token" => token}, socket, _connect_info) do
    case Phoenix.Token.verify(MyAppWeb.Endpoint, "user auth", token, max_age: 86_400) do
      {:ok, user_id} ->
        {:ok, assign(socket, :user_id, user_id)}

      {:error, _reason} ->
        :error
    end
  end

  def connect(_params, _socket, _connect_info), do: :error

  @impl true
  def id(socket), do: "user_socket:#{socket.assigns.user_id}"
end

Channel Implementation

defmodule MyAppWeb.RoomChannel do
  use MyAppWeb, :channel

  alias MyApp.Chat
  alias MyAppWeb.Presence

  @impl true
  def join("room:" <> room_id, _params, socket) do
    if Chat.member?(socket.assigns.user_id, room_id) do
      send(self(), :after_join)
      {:ok, assign(socket, :room_id, room_id)}
    else
      {:error, %{reason: "unauthorized"}}
    end
  end

  @impl true
  def handle_info(:after_join, socket) do
    {:ok, _} =
      Presence.track(socket, socket.assigns.user_id, %{
        online_at: System.system_time(:second)
      })

    push(socket, "presence_state", Presence.list(socket))

    messages = Chat.recent_messages(socket.assigns.room_id, limit: 50)
    push(socket, "message_history", %{messages: messages})

    {:noreply, socket}
  end

  @impl true
  def handle_in("new_message", %{"body" => body}, socket) do
    case Chat.create_message(socket.assigns.room_id, socket.assigns.user_id, body) do
      {:ok, message} ->
        broadcast!(socket, "new_message", %{
          id: message.id,
          body: message.body,
          user_id: message.user_id,
          inserted_at: message.inserted_at
        })

        {:reply, :ok, socket}

      {:error, changeset} ->
        {:reply, {:error, %{errors: format_errors(changeset)}}, socket}
    end
  end

  @impl true
  def handle_in("typing", _params, socket) do
    broadcast_from!(socket, "user_typing", %{user_id: socket.assigns.user_id})
    {:noreply, socket}
  end

  defp format_errors(changeset) do
    Ecto.Changeset.traverse_errors(changeset, fn {msg, opts} ->
      Regex.replace(~r"%{(\w+)}", msg, fn _, key ->
        opts |> Keyword.get(String.to_existing_atom(key), key) |> to_string()
      end)
    end)
  end
end

Presence Tracking

defmodule MyAppWeb.Presence do
  use Phoenix.Presence,
    otp_app: :my_app,
    pubsub_server: MyApp.PubSub
end

Broadcasting from Outside Channels

defmodule MyApp.Notifications do
  @pubsub MyApp.PubSub

  def notify_user(user_id, event, payload) do
    Phoenix.PubSub.broadcast(@pubsub, "notifications:#{user_id}", {event, payload})
  end

  def broadcast_system_alert(message) do
    Phoenix.PubSub.broadcast(@pubsub, "system:alerts", {:alert, message})
  end
end

Client-Side JavaScript

import { Socket, Presence } from "phoenix";

const socket = new Socket("/socket", {
  params: { token: window.userToken },
});
socket.connect();

const channel = socket.channel("room:lobby", {});
const presence = new Presence(channel);

presence.onSync(() => {
  const users = presence.list((id, { metas: [first, ...rest] }) => ({
    id,
    online_at: first.online_at,
    count: rest.length + 1,
  }));
  renderUsers(users);
});

channel.on("new_message", (msg) => {
  renderMessage(msg);
});

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

function sendMessage(body) {
  channel.push("new_message", { body })
    .receive("ok", () => clearInput())
    .receive("error", (resp) => showError(resp.errors));
}

Best Practices

• Authenticate at the socket level in connect/3, then authorize topic access in join/3.
• Use broadcast_from!/3 instead of broadcast!/3 when the sender should not receive their own message (e.g., typing indicators).
• Use Presence for tracking online users — it handles node failures and network partitions automatically via CRDTs.
• Keep channel payloads small. Send IDs and let clients fetch full data if needed, or use a separate API.
• Use intercept/1 and handle_out/3 sparingly — only when you need per-client message filtering (e.g., permission-based visibility).
• Rate-limit incoming messages in handle_in/3 to prevent abuse.

Common Pitfalls

• Blocking in handle_in: Long database queries or HTTP calls block the channel process. Spawn a Task and send results back via push/3.
• Leaking sensitive data in broadcasts: Every subscriber receives broadcast payloads. Never include data that some subscribers should not see without using intercept.
• Not handling disconnects: Clients disconnect frequently (network changes, sleep). Design for idempotent rejoins and message replay.
• Topic string injection: Validate topic parameters in join/3. A user should not be able to join arbitrary topics by manipulating the topic string.
• Forgetting id/1 in socket: Without a unique socket ID, you cannot disconnect a specific user's connections via MyAppWeb.Endpoint.broadcast("user_socket:123", "disconnect", %{}).