Phoenix Pubsub Patterns

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

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

# 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}

@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

# 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

# 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

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