Phoenix LiveView Tutorial: Adding Phoenix PubSub and Pow Authentication to Messenger

We've already bootstrapped our Phoenix LiveView-based Messenger app's database structure and a first LiveView page.

This time, we're going to improve real-time communication between the app's users using Phoenix PubSub, and use the Pow library to add secure user authentication.

As of November 2020, the latest Phoenix LiveView version is 0.14.8 - and the series has been updated to match it!

To follow our Modern Talking with Elixir series, subscribe to our newsletter - new episodes are coming soon!

Live Updates: Phoenix PubSub to the rescue

Just as a reminder, this is a GitHub tag of where we start - if you're new to the Modern Talking with Elixir series, feel free to just start from here, or go back to the previous episode.

So far, we have this working view of a conversation between two or more users:

However, as mentioned at the end of the previous episode, while your "own" LiveView notifies its associated browser of conversation updates (and the page's DOM is then updated by LiveView's script), the issue is that the other LiveViews that see the conversation are not instantly updated.

And that shouldn't be surprising, since each LiveView process is, by default, only connected via a WebSocket connection to the browser.

Knowing that LiveViews are just Erlang processes, each of which has a PID, we could try to somehow store the PIDs of every conversation member's connected LiveViews, and when a new message is sent, broadcast it to all the PIDs - each of the processes would then handle it with the handle_info/2 callback.

This, however, would be rather troublesome, because we'd have to add schemas to our database or Erlang's native Mnesia to store the PIDs and maintain their cleanup on connection closing.

Phoenix PubSub is a much better solution. It is Phoenix's realization of the publish-subscribe pattern, in which one agent publishes messages, and others subscribe to get notified about them (just like you should subscribe to our newsletter if you haven't done it yet).

(In case you're wondering how it handles the storing of connected processes, it's driven by Erlang's pg2 mechanism or Redis - we'll stick to the former, but Redis could be interesting for persistence.)

This mechanism is used by Phoenix Channels (along with WebSocket connections to the frontend) which also drives LiveView.

In this case, we'd like to keep the frontend oblivious to whatever happens at the server side that makes the LiveView receive a message and render it into the new DOM fragment that's to be rendered by the browser. We'll make direct usage of PubSub's API to make respective users' LiveViews to notify each other about incoming messages.

Each LiveView that views a conversation will subscribe to the conversation's topic. Whenever someone sends a message, they'll broadcast a new_message event to that topic, to which other processes will react by updating state via handle_info, which will result in running the render callback.

PubSub implementation for Curious Messenger

The good news is that there's nothing to configure if you've followed our project setup instructions. As we said, LiveView uses Phoenix Channels which, in turn, use the PubSub module under the hood anyway.

In any case, some configuration is there in config/config.exs:

config :curious_messenger, CuriousMessengerWeb.Endpoint,
  # ...
  pubsub_server: CuriousMessenger.PubSub,
  # ...

You also need to add in the start function in lib/curious_messenger/application.ex:

def start(_type, _args) do
  children = [
    # ...,
    {Phoenix.PubSub, [name: CuriousMessenger.PubSub, adapter: Phoenix.PubSub.PG2]}
  ]
end

First, let's add subscribing to the conversation's specific channel to conversation_live.ex in handle_params:

def handle_params(%{"conversation_id" => conversation_id, "user_id" => user_id}, _uri, socket) do
  CuriousMessengerWeb.Endpoint.subscribe("conversation_#{conversation_id}")
  # ...
end

As you can see, the topic name is just a plain old string.

Then, right after we add user information to new_message:

CuriousMessengerWeb.Endpoint.broadcast_from!(self(), "conversation_#{conversation_id}", "new_message", new_message)

The semantic of broadcast_from!/4 is that the message will be broadcast to all subscribed process except for self(), because the handle_event callback updates socket assigns with new messages anyway. Other processes will do it via a new handle_info callback:

def handle_info(%{event: "new_message", payload: new_message}, socket) do
  updated_messages = socket.assigns[:messages] ++ [new_message]

  {:noreply, socket |> assign(:messages, updated_messages)}
end

This will work, but you can see we're duplicating the reassignment of newly updated messages to our socket.

It is a common pattern in distributed programming to treat messages from self() just as if they were coming from anyone else. Let's replace the broadcast_from!/4 call with broadcast!/3:

CuriousMessengerWeb.Endpoint.broadcast!("conversation_#{conversation_id}", "new_message", new_message)

Remove the updated_messages construction and assignning it to the socket from handle_event, and just add {:noreply, socket} after the case statement - we won't do any socket assigns modification, this will all be handled by handle_info which receives the broadcast.

Let's also require Logger at the top of the ConversationLive module and add the following clause inside the case statement so that we don't "swallow" any error that might happen:

{:ok, new_message} ->
  # ...

{:error, err} ->
  Logger.error(inspect(err))

And that's it - when you repeat the test and run two browsers side-by-side with the same conversation, one view will react to the other sending messages!

LiveView template files

Let's do just one more thing: for further convenience, and to better separate concerns, let's create a separate file that will contain the template rendered by our LiveView's render function.

Put the contents of the ~L sigil into the lib/curious_messenger_web/templates/conversation/show.html.leex file. Create the very simple lib/curious_messenger_web/views/conversation_view.ex file to define a view (its render function will be automatically compiled from that .html.leex file we created):

defmodule CuriousMessengerWeb.ConversationView do
  use CuriousMessengerWeb, :view
end

...and refer to it in conversation_live.ex:

alias CuriousMessengerWeb.ConversationView

def render(assigns) do
  ConversationView.render("show.html", assigns)
end

We'll use a similar pattern in all of our subsequent LiveViews.

Authenticate with Pow

So far so good, but we still have to use seeds.exs or the console for creating users - this can't last for too long, so let's use Pow to handle user authentication.

According to Pow's docs,

Pow is a robust, modular, and extendable authentication and user management solution for Phoenix and Plug-based apps.

And that couldn't be more accurate. This is a field-tested library powering many production-grade apps, it has a modular design that allows you to easily enable or disable features such as password reset, email confirmations, persistent sessions or invitation-driven sign-up. It is also pretty amazing how simple it is to set it up.

First, add the following to deps function in mix.exs and run mix deps.get:

{:pow, "~> 1.0.21"}

Now, what the docs recommend that we do is run mix pow.install to generate a user schema and a migration creating a DB table for users. We'll do this, but here's where we need to be a bit cautious, because this command is intended for brand-new apps, and we've already got an auth_users table and a CuriousMessenger.Auth.User schema.

So after we've run the generator, we can see the following additions - starting from a generated CuriousMessenger.Users.User schema:

use Pow.Ecto.Schema

schema "users" do
  pow_user_fields()
  # ...
end

The pow_user_fields macro just adds email and password_hash fields to the schema. Let's move these new things over to our existing CuriousMessenger.Auth.User schema and ditch the new file whatsoever.

Let's also add the following to the changeset definition for the User schema:

def changeset(user, attrs) do
  user
  |> pow_changeset(attrs)
  |> # ...
end

pow_changeset is just a pipeline of functions ensuring that, given attrs contain an email and password, validates their presence and assigns an encrypted password_hash value that is then stored in the database.

There is also a new migration file, priv/repo/migrations/2019..._create_users.exs, creating a users table with email and password_hash. Let's rename the module to AddPowFieldsToUsers and the file name's suffix accordingly. Instead of creating a new table, we'll alter the existing table:

alter table(:auth_users) do
  add :email, :string, null: false
  add :password_hash, :string
end

create unique_index(:auth_users, [:email])

Since we're adding a not-null column to a table that already has data, before running migrations we'll need to mix ecto.drop and mix ecto.create again so that we've got a clean database - of course it wouldn't be so easy if we had already deployed the app to production, but at this early stage it's OK to do this locally. Afterwards, just run mix run ecto.migrate to create the new fields.

We'll need to notify Pow that we've changed its default setting, because out of the box it expects the user schema to be in Users.User, and we've got it in Auth.User - so put this in config/config.exs:

config :curious_messenger, :pow,
  user: CuriousMessenger.Auth.User,
  repo: CuriousMessenger.Repo,
  web_module: CuriousMessengerWeb

Sessions need to be stored in our app's session cookie, and, according to the docs,

The user struct will be collected from a cache store through a GenServer using a unique token generated for the session.

The cache store is set to use the in-memory Erlang Term Storage by default, which is about OK for development, but will fail you in production because it goes away when you restart the Erlang VM. We need a persistent storage solution then, and thankfully there is a Mnesia cache module driven by Erlang's built-in Mnesia DBMS.

(It's worth noting that similar authentication libraries in other languages, such as Ruby's Devise, would add new columns to the user table in the main database instead of using external storage. The upside of Pow's solution, though, is that this concern remains opaque to our schemas and we can focus on the business.)

Let's use the Mnesia storage instead of the default ETS - I believe it's rarely a good idea to use different stacks in production and development (which reminds me of Ruby on Rails still defaulting to SQLite in 2019...). Go to lib/curious_messenger_web/endpoint.ex and add this after the Plug.Session plug:

plug Pow.Plug.Session,
  otp_app: :curious_messenger,
  cache_store_backend: Pow.Store.Backend.MnesiaCache

Mnesia will run as an Erlang process along with your app's supervision tree, but we need to add it to lib/curious_messenger/application.ex's start/2 function for it to start:

def start(_type, _args) do
  children = [
    CuriousMessenger.Repo,
    CuriousMessengerWeb.Endpoint,
    Pow.Store.Backend.MnesiaCache
  ]

  # ...
end

It will automatically create a Mnesia.nonode@nohost directory as its storage, so it's a good idea to add /Mnesia.* to your .gitignore file.

Pow routing and pages

We'll need a number of new pages to handle registration and sign-in, for which Pow has default controller and view code. We'll do the minimal amount of work needed to get our setup up and running, so we won't be customizing all of them for now. Let's start from the router - go to router.ex and add the following to add Pow routes and ensure that certain pages require authentication:

defmodule CuriousMessengerWeb.Router do
  use CuriousMessengerWeb, :router
  use Pow.Phoenix.Router # Pow route macros

  # ...

  pipeline :protected do
    plug Pow.Plug.RequireAuthenticated,
      error_handler: Pow.Phoenix.PlugErrorHandler
  end

  # Pow browser routes - remember not to add them to the existing scope referencing the
  # CuriousMessengerWeb module, but add a new block instead like this:
  scope "/" do
    pipe_through :browser

    pow_routes()
  end

  scope "/", CuriousMessengerWeb do
    pipe_through :browser

    get "/", PageController, :index
  end

  # Make conversation routes protected by requiring authentication
  scope "/", CuriousMessengerWeb do
    pipe_through [:browser, :protected]

    resources "/conversations", ConversationController

    live "/conversations/:conversation_id/users/:user_id", ConversationLive, as: :conversation
  end

  # ...
end

To get an overview of what new routes have been generated by Pow, run mix phx.routes | grep Pow. It's going to look like this:

     pow_session_path  GET     /session/new        Pow.Phoenix.SessionController :new
     pow_session_path  POST    /session            Pow.Phoenix.SessionController :create
     pow_session_path  DELETE  /session            Pow.Phoenix.SessionController :delete
pow_registration_path  GET     /registration/edit  Pow.Phoenix.RegistrationController :edit
pow_registration_path  GET     /registration/new   Pow.Phoenix.RegistrationController :new
pow_registration_path  POST    /registration       Pow.Phoenix.RegistrationController :create
pow_registration_path  PATCH   /registration       Pow.Phoenix.RegistrationController :update
                       PUT     /registration       Pow.Phoenix.RegistrationController :update
pow_registration_path  DELETE  /registration       Pow.Phoenix.RegistrationController :delete

You can now replace the Get Started link in app.html.eex template with actual links to registration, sign-in, profile edit and logout:

<%= if Pow.Plug.current_user(@conn) do %>
  <li><%= link "Profile", to: Routes.pow_registration_path(@conn, :edit) %></li>
  <li><%= link "Sign out", to: Routes.pow_session_path(@conn, :delete), method: :delete %></li>
<% else %>
  <li><%= link "Register", to: Routes.pow_registration_path(@conn, :new) %></li>
  <li><%= link "Sign in", to: Routes.pow_session_path(@conn, :new) %></li>
<% end %>

Pow's default template doesn't contain the nickname field for users, which is required in our setup - so it won't work. We'll need to customize the form - we've already set web_module: CuriousMessengerWeb in the config, so let's run mix pow.phoenix.gen.templates to generate view and template files.

We'll leave most of the files untouched (but not delete them), and focus mostly on lib/curious_messenger_web/templates/pow/registration/new.html.eex and lib/curious_messenger_web/templates/pow/registration/edit.html.eex to modify templates for registration and profile edit. Let's just add an additional nickname field to both:

<%= label f, :nickname %>
<%= text_input f, :nickname %>
<%= error_tag f, :nickname %>

We're good to go, and the registration functionality should now be working - here's the GitHub revision at the current stage, if you'd like to start from here.

Create your group conversations

Our Messenger app is useless when users can't create conversations on their own, and we'll now deal with this.

Here's what it's going to look like - just for a start, we'd like to be able to display a list of ongoing conversations, and of course also initiate them:

Schema modifications

Since we'd also like to exercise using Phoenix's form helpers with changesets and nesting association data, we're going to fix up a few of our schema definitions.

Let's add the following to user.ex:

alias CuriousMessenger.Chat.ConversationMember

schema "auth_users" do
  # ...

  has_many :conversation_members, ConversationMember
  has_many :conversations, through: [:conversation_members, :conversation]
end

This will ensure that conversation members and conversations can be understood by Ecto as associations, e.g. for the purpose of preloading.

Conversely, in conversation.ex we need to add the following:

schema "chat_conversations" do
  has_many :conversation_members, ConversationMember
  has_many :conversations, through: [:conversation_members, :conversation]
end

def changeset(conversation, attrs) do
  |> cast(attrs, [:title])
  |> cast_assoc(:conversation_members)
  |> validate_required([:title])
end

This is just the other side of the ConversationMember many-to-many relationship (note, though, we don't use Ecto's many_to_many declaration, because we have an additional owner column in the intermediate table). The cast_assoc thing is needed for us to make us able to use Chat.create_conversation/1 with the following argument:

%{
  "conversation_members" => %{
    "0" => %{"user_id" => "3"},
    "1" => %{"user_id" => "2"},
    "2" => %{"user_id" => "1"},
    "3" => %{"user_id" => "4"}
  },
  "title" => "Curious Conversation"
}

Lastly, let's make the ConversationMember's changeset look like the following:

def changeset(conversation_member, attrs) do
  conversation_member
  |> cast(attrs, [:owner, :user_id])
  |> validate_required([:owner, :user_id])
  |> unique_constraint(:user, name: :chat_conversation_members_conversation_id_user_id_index)
  |> unique_constraint(:conversation_id, name: :chat_conversation_members_owner)
end

Notice how we don't validate the requirement for conversation_id to be present: we can't do this because we want these records to be created along with a conversation.

Designing the messenger dashboard LiveView

The dashboard will be driven by a LiveView module that will manage the list of currently available conversations and adding a new conversation.

At all times, it will need to know about the following - which will constitute its state:

• who the current_user is (and what are it's available associated conversations),
• what are the contacts (let's assume that, for simplicity, it's just all of the app's registered users),
• for the purpose of creating a new conversation - the users that the current user currently wants to add to a new conversation.

Create the lib/curious_messenger_web/live/dashboard_live.ex file. Let's start with declaring a very basic setup. In this example, we'll want to define the container element, in which the LiveView will be rendered (the purpose of which will get clear when we get to the template code). We'll also use Phoenix's HTML helpers, and alias a few of the modules we're going to use. Last but not least, we'll mount the component, having ensured that current_user is provided.

defmodule CuriousMessengerWeb.DashboardLive do
  require Logger

  use Phoenix.LiveView, container: {:div, [class: "row"]}
  use Phoenix.HTML

  alias CuriousMessenger.{Auth, Chat}
  alias CuriousMessenger.Chat.Conversation
  alias CuriousMessengerWeb.DashboardView
  alias CuriousMessenger.Repo
  alias Ecto.Changeset

  def render(assigns) do
    DashboardView.render("show.html", assigns)
  end

  def mount(_params, %{"current_user" => current_user}, socket) do
    {:ok,
     socket
     |> assign(current_user: current_user)
     |> assign_new_conversation_changeset()
     |> assign_contacts(current_user)}
  end

  # Build a changeset for the newly created conversation, initially nesting a single conversation
  # member record - the current user - as the conversation's owner.
  #
  # We'll use the changeset to drive a form to be displayed in the rendered template.
  defp assign_new_conversation_changeset(socket) do
    changeset =
      %Conversation{}
      |> Conversation.changeset(%{
        "conversation_members" => [%{owner: true, user_id: socket.assigns[:current_user].id}]
      })

    assign(socket, :conversation_changeset, changeset)
  end

  # Assign all users as the contact list.
  defp assign_contacts(socket, current_user) do
    users = Auth.list_auth_users()

    assign(socket, :contacts, users)
  end
end

Notice that we don't implement handle_params - we're going to exercise including this LiveView inside a different template, as opposed to what we did with conversations (a standalone route).

Create the DashboardView module in lib/curious_messenger_web/views/dashboard_view.ex. Let's define a number of decorator functions that we'll use in the corresponding template for clarity and readability. Notice how we use phx_click to make links emit the LiveView add_member and remove_member events, and phx_value_user_id to make these events carry additional values retrievable under the user-id key.

defmodule CuriousMessengerWeb.DashboardView do
  use CuriousMessengerWeb, :view

  def remove_member_link(contacts, user_id, current_user_id) do
    nickname = contacts |> Enum.find(&(&1.id == user_id)) |> Map.get(:nickname)

    link("#{nickname} #{if user_id == current_user_id, do: "(me)", else: "✖"} ",
      to: "#!",
      phx_click: unless(user_id == current_user_id, do: "remove_member"),
      phx_value_user_id: user_id
    )
  end

  def add_member_link(user) do
    link(user.nickname,
      to: "#!",
      phx_click: "add_member",
      phx_value_user_id: user.id
    )
  end

  def contacts_except(contacts, current_user) do
    Enum.reject(contacts, &(&1.id == current_user.id))
  end

  def disable_create_button?(assigns) do
    Enum.count(assigns[:conversation_changeset].changes[:conversation_members]) < 2
  end
end

We'll need a lib/curious_messenger_web/templates/dashboard/show.html.leex file as the template for our dashboard. Thanks to the functions defined above, it's going to look simple and clear.

It's got two class="column" elements (which is why we needed the LiveView container to have the row class). In the first column, we display all the ongoing conversations, with links using the simple route we created in the previous episode.

The second column contains a form driven by the LiveView's assigned @conversation_changeset. It repeats the changeset structure using form_for @conversation_changeset, ... and its nested inputs_for f, :conversation_members declaration, inside which there are removal links for each added member and hidden inputs that ensure the form's encoded data will match the structure that we expect to be given to changesets. There are also member add links for each user, and a field to define the conversation's title.

On form submit, the create_conversation event is being emitted, and the encoded form data is passed in the event's payload, allowing us to use it in a Chat.create_conversation/1 call.

<article class="column">
  <h2>Ongoing Conversations</h2>
  <%= for conversation <- @current_user.conversations do %>
    <div>
      <%= link conversation.title,
               to: Routes.conversation_path(@socket,
                                            CuriousMessengerWeb.ConversationLive,
                                            conversation.id,
                                            @current_user.id) %>
    </div>
  <% end %>
</article>

<article class="column">
  <h2>Create Conversation</h2>

  <%= form_for @conversation_changeset, "", [phx_submit: :create_conversation], fn f -> %>
    <p>
      <%= inputs_for f, :conversation_members, fn cmf -> %>
        <%= remove_member_link(@contacts, cmf.source.changes[:user_id], @current_user.id) %>

        <%= hidden_input cmf, :user_id, value: cmf.source.changes[:user_id] %>
      <% end %>
    </p>

    <p>
      <%= text_input f, :title, placeholder: "Title (optional)" %>
      <%= submit "Create", disabled: disable_create_button?(assigns) %>
    </p>

    <ul>
      <%= for user <- contacts_except(@contacts, @current_user) do %>
        <li>
          <%= add_member_link(user) %>
        </li>
      <% end %>
    </ul>
  <% end %>
</article>

We have three events to handle: create_conversation, add_member and remove_member. Define the following clauses for handle_event/3 in dashboard_live.ex along with a helper function:

# Create a conversation based on the payload that comes from the form (matched as `conversation_form`).
# If its title is blank, build a title based on the nicknames of conversation members.
# Finally, reload the current user's `conversations` association, and re-assign it to the socket,
# so the template will be re-rendered.
def handle_event(
      "create_conversation",
      %{"conversation" => conversation_form},
      %{
        assigns: %{
          conversation_changeset: changeset,
          current_user: current_user,
          contacts: contacts
        }
      } = socket
    ) do
  conversation_form =
    Map.put(
      conversation_form,
      "title",
      if(conversation_form["title"] == "",
        do: build_title(changeset, contacts),
        else: conversation_form["title"]
      )
    )

  case Chat.create_conversation(conversation_form) do
    {:ok, _} ->
      {:noreply,
        assign(
          socket,
          :current_user,
          Repo.preload(current_user, :conversations, force: true)
        )}

    {:error, err} ->
      Logger.error(inspect(err))
  end
end

# Add a new member to the newly created conversation.
# "user-id" is passed from the link's "phx_value_user_id" attribute.
# Finally, assign the changeset containing the new member's definition to the socket,
# so the template can be re-rendered.
def handle_event(
      "add_member",
      %{"user-id" => new_member_id},
      %{assigns: %{conversation_changeset: changeset}} = socket
    ) do
  {:ok, new_member_id} = Ecto.Type.cast(:integer, new_member_id)

  old_members = socket.assigns[:conversation_changeset].changes.conversation_members
  existing_ids = old_members |> Enum.map(&(&1.changes.user_id))

  cond do
    new_member_id not in existing_ids ->
      new_members = [%{user_id: new_member_id} | old_members]

      new_changeset = Changeset.put_change(changeset, :conversation_members, new_members)

      {:noreply, assign(socket, :conversation_changeset, new_changeset)}

    true ->
      {:noreply, socket}
  end
end

# Remove a member from the newly create conversation and handle it similarly to
# when a member is added.
def handle_event(
      "remove_member",
      %{"user-id" => removed_member_id},
      %{assigns: %{conversation_changeset: changeset}} = socket
    ) do
  {:ok, removed_member_id} = Ecto.Type.cast(:integer, removed_member_id)

  old_members = socket.assigns[:conversation_changeset].changes.conversation_members
  new_members = old_members |> Enum.reject(&(&1.changes[:user_id] == removed_member_id))

  new_changeset = Changeset.put_change(changeset, :conversation_members, new_members)

  {:noreply, assign(socket, :conversation_changeset, new_changeset)}
end

defp build_title(changeset, contacts) do
  user_ids = Enum.map(changeset.changes.conversation_members, &(&1.changes.user_id))

  contacts
  |> Enum.filter(&(&1.id in user_ids))
  |> Enum.map(&(&1.nickname))
  |> Enum.join(", ")
end

To render the DashboardLive component on the landing page, go to index.html.eex and insert right after the phx-hero section:

<%= if @current_user do %>
  <%= live_render(@conn,
                  CuriousMessengerWeb.DashboardLive,
                  session: %{"current_user" => @current_user}) %>
<% end %>

That's it! You're now able to sign up, log in and manage your conversations with different users of the app.

Wrapping up: Additional exercise & further steps

Here's the repository of our app, current as of the end of this episode.

The app has now started to take shape! As an exercise, you can use Phoenix PubSub to notify different users of the app about created conversations that involve them.

In the next episodes of the Modern Talking with Elixir series, we'll show you what kinds problems can be encountered when using Phoenix LiveView, and how to further improve the application with settings, presence detection and push notifications.

Keep #BusyBeingCurious and subscribe to our newsletter!

Check out other articles in this series:

• Modern Talking with Elixir: Messenger App Tutorial with Phoenix LiveView
• Phoenix LiveView Tutorial: Bootstrap Your Messenger App
• Phoenix LiveView Tutorial: Handling Connection Errors And Push Notifications With JS Hooks