---
name: phoenix-pubsub-patterns
description: Use when adding real-time updates via Phoenix.PubSub — subscribe/broadcast topology, topic naming, handle_info message handling.
file_patterns:
  - "**/*_live.ex"
  - "**/*_live/*.ex"
  - "**/contexts/**/*.ex"
auto_suggest: true
---

# Phoenix PubSub Patterns

## RULES — Follow these with no exceptions

1. **Always guard subscriptions with `if connected?(socket)`** — the disconnected render runs in a separate short-lived process; subscribing there is wasted work, not a duplicate (LiveView mounts twice: once static, once connected)
2. **Broadcast from contexts, not LiveViews** — keeps real-time logic in the business layer; LiveViews only subscribe and react
3. **Use consistent topic naming** — `"resource:id"` for specific resources, `"resource:action"` for collection-wide events
4. **Handle PubSub messages in `handle_info/2`** — never in `handle_event/3`; PubSub messages are process messages, not client events
5. **Prefer `update/3` when the new value derives from the old** — `update(socket, :posts, &[post | &1])` reads better than reaching into `socket.assigns` manually. Both are equivalent; this is style, not safety.
6. **Test PubSub by calling context functions and asserting LiveView updates** — don't test `PubSub.broadcast` directly; test the full cycle

---

## Subscription Pattern

Subscribe in `mount/3` only when connected. The static render doesn't need real-time updates.

```elixir
defmodule MyAppWeb.PostLive.Index do
  use MyAppWeb, :live_view

  @impl true
  def mount(_params, _session, socket) do
    if connected?(socket) do
      Phoenix.PubSub.subscribe(MyApp.PubSub, "posts")
    end

    {:ok, assign(socket, :posts, list_posts())}
  end

  @impl true
  def handle_info({:post_created, post}, socket) do
    {:noreply, update(socket, :posts, fn posts -> [post | posts] end)}
  end

  @impl true
  def handle_info({:post_updated, post}, socket) do
    {:noreply,
     update(socket, :posts, fn posts ->
       Enum.map(posts, fn
         p when p.id == post.id -> post
         p -> p
       end)
     end)}
  end

  @impl true
  def handle_info({:post_deleted, post}, socket) do
    {:noreply,
     update(socket, :posts, fn posts ->
       Enum.reject(posts, &(&1.id == post.id))
     end)}
  end
end
```

---

## Broadcasting from Contexts

Broadcast after successful database operations. The context owns the business logic — LiveViews are just subscribers.

```elixir
defmodule MyApp.Blog do
  alias MyApp.Blog.Post
  alias MyApp.Repo

  def create_post(attrs) do
    %Post{}
    |> Post.changeset(attrs)
    |> Repo.insert()
    |> broadcast(:post_created)
  end

  def update_post(%Post{} = post, attrs) do
    post
    |> Post.changeset(attrs)
    |> Repo.update()
    |> broadcast(:post_updated)
  end

  def delete_post(%Post{} = post) do
    post
    |> Repo.delete()
    |> broadcast(:post_deleted)
  end

  # Only broadcast on success
  defp broadcast({:ok, post}, event) do
    Phoenix.PubSub.broadcast(MyApp.PubSub, "posts", {event, post})
    {:ok, post}
  end

  defp broadcast({:error, changeset}, _event) do
    {:error, changeset}
  end
end
```

---

## Topic Naming Conventions

Use a consistent naming scheme so subscribers know what to expect.

```elixir
# Collection-wide — all posts
topic = "posts"
# Events: {:post_created, post}, {:post_updated, post}, {:post_deleted, post}

# Specific resource — one post
topic = "posts:#{post.id}"
# Events: {:post_updated, post}, {:post_deleted, post}, {:comment_added, comment}

# User-scoped — all activity for a user
topic = "users:#{user.id}"
# Events: {:notification, notification}, {:message_received, message}
```

### Subscribing to Specific Resources

```elixir
@impl true
def mount(%{"id" => id}, _session, socket) do
  post = Blog.get_post!(id)

  if connected?(socket) do
    Phoenix.PubSub.subscribe(MyApp.PubSub, "posts:#{post.id}")
  end

  {:ok, assign(socket, :post, post)}
end
```

---

## Scoped Broadcasting

When events should only reach specific users or resources:

```elixir
# In context — broadcast to resource-specific topic
defp broadcast({:ok, comment}, :comment_added) do
  Phoenix.PubSub.broadcast(
    MyApp.PubSub,
    "posts:#{comment.post_id}",
    {:comment_added, comment}
  )
  {:ok, comment}
end

# In context — broadcast to user-specific topic
defp broadcast({:ok, notification}, :new_notification) do
  Phoenix.PubSub.broadcast(
    MyApp.PubSub,
    "users:#{notification.user_id}",
    {:new_notification, notification}
  )
  {:ok, notification}
end
```

---

## Deriving New Assigns with `update/3`

Use `update/3` when the new value derives from the old one — it reads better
than reaching into `socket.assigns` manually. Functionally, `assign/3` and
`update/3` produce identical results here; this is a style preference, not a
correctness rule.

```elixir
# Equivalent — reaches into socket.assigns directly
def handle_info({:post_created, post}, socket) do
  {:noreply, assign(socket, :posts, [post | socket.assigns.posts])}
end

# Preferred — update/3 reads better when deriving from the old value
def handle_info({:post_created, post}, socket) do
  {:noreply, update(socket, :posts, fn posts -> [post | posts] end)}
end

# Good — update a specific item in the list
def handle_info({:post_updated, updated_post}, socket) do
  {:noreply,
   update(socket, :posts, fn posts ->
     Enum.map(posts, fn
       post when post.id == updated_post.id -> updated_post
       post -> post
     end)
   end)}
end

# Good — remove an item from the list
def handle_info({:post_deleted, deleted_post}, socket) do
  {:noreply,
   update(socket, :posts, fn posts ->
     Enum.reject(posts, &(&1.id == deleted_post.id))
   end)}
end
```

---

## Testing PubSub

Test the full cycle: call a context function, assert the LiveView updates. Don't test `PubSub.broadcast` in isolation.

```elixir
describe "real-time updates" do
  test "new post appears in list", %{conn: conn} do
    user = user_fixture()
    conn = log_in_user(conn, user)

    {:ok, lv, _html} = live(conn, ~p"/posts")

    # Create a post through the context (triggers broadcast)
    {:ok, post} = Blog.create_post(%{title: "New Post", user_id: user.id})

    # Assert the LiveView received and rendered the update
    assert render(lv) =~ "New Post"
  end

  test "updated post reflects changes", %{conn: conn} do
    user = user_fixture()
    post = post_fixture(user_id: user.id, title: "Original")
    conn = log_in_user(conn, user)

    {:ok, lv, _html} = live(conn, ~p"/posts")
    assert render(lv) =~ "Original"

    {:ok, _post} = Blog.update_post(post, %{title: "Updated"})

    assert render(lv) =~ "Updated"
    refute render(lv) =~ "Original"
  end

  test "deleted post disappears from list", %{conn: conn} do
    user = user_fixture()
    post = post_fixture(user_id: user.id, title: "To Delete")
    conn = log_in_user(conn, user)

    {:ok, lv, _html} = live(conn, ~p"/posts")
    assert render(lv) =~ "To Delete"

    {:ok, _post} = Blog.delete_post(post)

    refute render(lv) =~ "To Delete"
  end
end
```

---

See `phoenix-liveview-essentials` skill for LiveView lifecycle patterns.
See `testing-essentials` skill for comprehensive testing patterns.