A protocol-based PubSub abstraction for Phoenix LiveView.

defmodule MyApp.Blog do
  alias Amplified.PubSub

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

Amplified.PubSub wraps Phoenix.PubSub with a protocol layer so the same broadcast/2, subscribe/1, and handle_info/2 calls work whether you pass a struct, an {:ok, struct} tuple from a Repo operation, a list of structs, or a raw channel string. This lets you weave PubSub into your context functions as pipeline-friendly operations that chain naturally with Ecto.

Configuration

Configure the PubSub server name used for subscriptions and broadcasts:

# config/config.exs
config :amplified_pubsub, pubsub_server: :my_app

Setup

Schema modules opt in by adding use Amplified.PubSub:

defmodule MyApp.Blog.Post do
  use Ecto.Schema
  use Amplified.PubSub

  schema "posts" do
    field :title, :string
    field :body, :string
    timestamps()
  end
end

This generates an Amplified.PubSub.Protocol implementation with sensible defaults:

• channel/1 derives "post:<id>" from the module's last segment (snake_cased) and the struct's :id field
• subscribe/1 and unsubscribe/1 subscribe via the configured PubSub server
• broadcast/2 wraps atom/string events as {event, subject} and publishes to the subject's channel
• handle_info/2,3,4 return {:cont, socket} (pass-through) so unhandled messages don't crash

Broadcasting from context functions

Since broadcast/2 returns its first argument — and passes through {:ok, _} and {:error, _} tuples — it drops right into Ecto pipelines:

defmodule MyApp.Blog do
  alias Amplified.PubSub

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

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

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

On success, PubSub.broadcast({:ok, post}, :created) unwraps the tuple, broadcasts {:created, post} on the "post:<id>" channel, and returns {:ok, post}. On failure, {:error, changeset} passes through without broadcasting.

Subscribing in LiveViews

Subscribe a LiveView process to a subject's channel in mount/3. The subscription is idempotent — calling it twice won't produce duplicate messages:

def mount(%{"id" => id}, _session, socket) do
  post = Blog.get_post!(id)
  PubSub.subscribe(post)
  {:ok, assign(socket, post: post)}
end

Unsubscribe when you no longer want messages:

PubSub.unsubscribe(post)

You can also subscribe to a raw channel string:

PubSub.subscribe("posts:feed")

Handling messages

PubSub.handle_info/2 returns {:cont, socket} or {:halt, socket} — the same convention used by Phoenix.LiveView.attach_hook/4 — so you can wire it directly into the LiveView lifecycle as an on_mount hook. Every LiveView in the live session then gets PubSub handling automatically, with no per-view boilerplate.

Define a hooks module:

defmodule MyAppWeb.Hooks do
  import Phoenix.LiveView

  alias Amplified.PubSub

  defmodule Default do
    def on_mount(:default, _params, _session, socket) do
      {:cont, MyAppWeb.Hooks.attach_defaults(socket)}
    end
  end

  # This is the critical wiring step. attach_hook/4 registers
  # PubSub.handle_info/2 as a lifecycle hook that intercepts every
  # message the LiveView process receives. Without this, PubSub
  # messages will arrive but the protocol dispatch won't fire —
  # your schema handle_info/3 implementations won't be called.
  def attach_defaults(socket) do
    socket
    |> subscribe()
    |> attach_hook(:pubsub, :handle_info, &PubSub.handle_info/2)
  end

  # Subscribe based on whatever is assigned to the socket.
  # Unsubscribe first to prevent duplicates on reconnect.
  defp subscribe(socket) do
    if connected?(socket) do
      user = socket.assigns[:current_user]
      project = socket.assigns[:project]

      if user, do: PubSub.subscribe(user)
      if project, do: PubSub.subscribe(project)
    end

    socket
  end
end

Then attach the hook in the router:

live_session :authenticated, on_mount: MyAppWeb.Hooks.Default do
  live "/posts", PostLive.Index
  live "/posts/:id", PostLive.Show
end

With this in place, the hook subscribes every LiveView to the current user and project channels and dispatches all {action, subject} messages through the protocol. Individual LiveViews can still subscribe to additional channels in their own mount/3 — the hook returns {:cont, socket} for anything it doesn't handle, so the message continues to the view's handle_info/2.

Event handling in schemas

When a {action, subject} message arrives, the Tuple dispatcher looks up the subject's protocol implementation and calls its handle_info/3. This lets you define handlers in the schema's use Amplified.PubSub block.

A good example is keeping the current user up to date across all LiveViews — every view needs the same logic, so colocating it with the schema avoids repetition:

defmodule MyApp.Accounts.User do
  use Ecto.Schema
  use Amplified.PubSub do
    # Match the broadcast user's ID against the scope's user ID
    # to ensure we only update when the broadcast is for *this*
    # session's authenticated user.
    def handle_info(
          %User{id: id} = user,
          :updated,
          %{assigns: %{current_scope: %{user: %{id: id}} = scope}} = socket
        ) do
      {:cont, assign(socket, current_scope: %{scope | user: user})}
    end

    def handle_info(
          %User{id: id},
          :deleted,
          %{assigns: %{current_scope: %{user: %{id: id}}}} = socket
        ) do
      {:halt, redirect(socket, to: ~p"/sign-out")}
    end
  end

  schema "users" do
    field :email, :string
    field :name, :string
    timestamps()
  end
end

defmodule MyApp.Blog.Post do
  use Ecto.Schema
  use Amplified.PubSub do
    def handle_info(%Post{id: id} = post, :updated, %{assigns: %{post: %{id: id}}} = socket) do
      {:cont, assign(socket, post: post)}
    end

    def handle_info(%Post{id: id}, :deleted, %{assigns: %{post: %{id: id}}} = socket) do
      {:halt, push_navigate(socket, to: ~p"/posts")}
    end
  end

  schema "posts" do
    field :title, :string
    field :body, :string
    timestamps()
  end
end

With these implementations in place and a global attach_hook dispatching through PubSub.handle_info/2, a :updated broadcast for the current %User{} will automatically update the scope's user on every connected LiveView that belongs to that user — with no per-view code at all. Broadcasts for other users fall through as {:cont, socket} and are ignored.

The convention is to return {:halt, socket} when you've handled the message and you don't want other lifecycle hooks to run, and {:cont, socket} when you do. The defaults always return {:cont, socket}, so unmatched messages fall through safely.

Flash messages

The Tuple implementation recognises {:flash, level, message} tuples and calls Phoenix.LiveView.put_flash/3 automatically:

PubSub.broadcast("room:lobby", {:flash, :info, "Someone joined!"})

Custom channels

Override the channel derivation by passing a block to use:

use Amplified.PubSub do
  def channel(%Post{slug: slug}, _ns), do: "post:#{slug}"
end

Or implement the protocol externally with defimpl:

defimpl Amplified.PubSub.Protocol, for: MyApp.Blog.Post do
  use Amplified.PubSub, impl: true
  def channel(%{slug: slug}, _ns), do: "post:#{slug}"
end

When using impl: true, you get all the default function bodies injected into your defimpl block, so you only need to override the functions you want to customise.

Namespaced channels

All channel/2 functions accept an optional namespace for scoping. This is useful when different LiveViews care about different aspects of the same resource:

PubSub.channel(post)             # => "post:abc-123"
PubSub.channel(post, :comments)  # => "post:abc-123:comments"

Lists and streams

Broadcasting or subscribing to a list operates on each item individually:

PubSub.subscribe(posts)           # subscribes to each post's channel
PubSub.broadcast(posts, :archived) # broadcasts for each post

When broadcasting a list with more than one item, items are grouped by channel and sent as a single [{item, event}, ...] message per channel for efficiency. Streams are materialised to lists before operating.

Protocol implementations

Built-in protocol implementations handle the following types:

• BitString — treats the string as a literal channel name; broadcasts, subscribes, and unsubscribes via Phoenix.PubSub
• Atom — converts to a string channel (e.g. :users → "users"); broadcast is a no-op that returns the message
• Tuple — unwraps {:ok, subject} for broadcast/subscribe; passes {:error, _} through unchanged; dispatches {action, subject} messages in handle_info
• List — maps the operation across each element, grouping multi-item broadcasts by channel
• Stream — materialises to a list, then delegates to the List implementation
• Phoenix.LiveView.Socket — derives a channel from the socket's session ID ("socket:<id>")
• Structs via use Amplified.PubSub — derives channels from the module's last name segment and the struct's :id field

Telemetry

The following telemetry events are emitted:

• [:amplified, :pubsub, :broadcast] — fired on every broadcast. Measurements are empty (%{}). Metadata contains :topic and :message.

Attach a handler in your application to log broadcasts, collect metrics, or perform any other observation:

:telemetry.attach("my-app-pubsub-log", [:amplified, :pubsub, :broadcast], fn
  _event, _measurements, %{topic: topic, message: message}, _config ->
    Logger.debug("broadcast(\#{inspect(topic)}, \#{inspect(message)})")
end, nil)