Amplified.PubSub - a protocol-based PubSub abstraction for Phoenix and LiveView

Hi folks!

We’ve just published a module which we’ve built and have been using at Amplified for a few years: A protocol-based PubSub abstraction for Phoenix LiveView.

Why did we build this?

Amplified is a multi-user collaborative patent research platform that depends heavily on real-time updates both from background jobs and user interactions. We lean heavily on Phoenix.PubSub and before building this module, we repeated the same code smells all over the place:

• manually building PubSub channel strings like user:123 or tag:asdf, sometimes getting things wrong and causing hard-to-find bugs
• writing the same broadcast/2 helper in context modules
• writing the same handle_info/2 callbacks in LiveView modules, or forgetting them and wondering why views weren’t updating

Now, we just do this instead. Schema modules opt-in with 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

and contexts broadcast CRUD events on schema changes.

defmodule MyApp.Blog do
  alias Amplified.PubSub

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

We hook into LiveView’s handle_info/2 lifecycle callback chain in an on_mount hook:

attach_hook(:pubsub, :handle_info, &PubSub.handle_info/2)

and in either a LiveView or a mount hook, subscribe to events:

# Contrived example
123 |> Accounts.get_account!() |> PubSub.subscribe()

If you find yourself writing the same handle_info/2 callbacks across multiple LiveViews, you can put them in the schema module instead and that will run every time that thing changes:

defmodule MyApp.Accounts.User do
  use Amplified.PubSub do
    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
  end

  schema "users" do
    ...
  end
end

The ergonomics are wonderful - real-time updates just happen with near zero code and way fewer bugs.

Find it on Hex and

https://github.com/amplifiedai/amplified_pubsub