How to build scalable and fault-tolerant applications with OTP?

git clone git@github.com:alukasz/todo_live_view.git && cd todo_live_view
git checkout part-two
mix deps.get && mix deps.compile
yarn install --cwd assets # or cd assets && npm install
mix phx.server

Before we dive into OTP we need one more thing. We have to remember a user between page reloads. To keep things as simple as possible we will skip authentication and remember user using a token stored in a session. We do that using a custom plug. Our plug will put a token in a session if it does not exist and then assign this token to the conn. Next, we add plug into the :browser pipeline in router file.

# lib/todo_web/plugs/todo_token_plug.ex
defmodule TodoWeb.TodoTokenPlug do
  import Plug.Conn

  def init(opts), do: opts

  def call(conn, _opts) do
    conn =
      case get_session(conn, :todo_token) do
        nil -> put_session(conn, :todo_token, Ecto.UUID.generate())
        _ -> conn
      end

    assign(conn, :todo_token, get_session(conn, :todo_token))
  end
end

# lib/todo_web/router.ex
  pipeline :browser do
    # add at the end
    plug TodoWeb.TodoTokenPlug
  end

The only thing left is to pass this token to the LiveView. Because todo_token is assigned to the conn, we can pass it directly to LiveView session with session: [:todo_token] added to the route. Then retrieve todo_token from session map passed into mount/2 and assign it to the LiveView socket.

# lib/todo_web/router.ex
  live "/", TodoLive, as: :todo, session: [:todo_token]
  live "/:filter", TodoLive, as: :todo, session: [:todo_token]

# lib/todo_web/live/todo_live.ex
  def mount(%{todo_token: todos_id}, socket) do
    {:ok, assign(socket, todos: @todos, filter: "all", todos_id: todos_id)}
  end

Processes (BEAM VM processes) are basics of concurrency of Elixir. There are multiple primitives for working with processes. You can start a new process with spawn/1,3 function, send/2 and receive/1 messages between processes, create links and monitors between them etc.
But it gets tedious very soon. send/2 is asynchronous so you need to implement synchronous messaging on your own. To keep a process alive, have state and respond to multiple messages you need a recursion with receive/1. Processes can be left forgotten only to consume memory or crash unexpectedly without anyone noticing.

• init/1 – initializes state of GenServer, should return tuple {:ok, state};
• handle_call/3 – for synchronous invocation GenServer.call/2,3. It receives the request, from tuple and GenServer state as arguments. from tuple contains the PID of the caller and unique reference. handle_call/3 usually returns {:reply, reply_value, new_state} tuple, but if the server cannot fulfil request immediately it can return {:noreply, new_state} and use from tuple to send the response later with GenServer.reply/2;
• handle_cast/2 – for asynchronous invocation GenServer.cast/2. It receives request and state as arguments. It should return {:noreply, new_state};
• handle_info/2 – similar to handle_cast/2, but it handles any standard message (send/2). Often used to send message to itself e.g. for scheduling and periodic tasks (Process.send_after(self(), :do_work, 1_000));
• terminate/2 – invoked when GenServer is terminated. Keep in mind this callback won’t be always invoked, e.g. when GenServer crashes abruptly;
• code_change/3 – invoked when upgrading application with hot code swapping. Yes, Erlang/Elixir can deploy new code without stopping application.

Back to the todo app. We will implement GenServer for storing todos for the user. We will keep public API and callbacks in a single module. use GenServer will bring some default functionality to the module. First, we need a start_link/1 function that will spawn new GenServer process. It is mostly for convenience but it will have an important role later. It simply calls GenServer.start_link/3 function with a __MODULE__ (__MODULE__ refers to the current module) as the callback module, arguments passed to the init/1 callback and :name as an option. name/1 function returns :via tuple that describes the mechanism used for naming registration. Here a Registry is used. Registry allows mapping arbitrary term (our todo token) to a process.

Time for the first callback. When GenServer process starts it invokes init/1 function. As the name suggests it is used to initialize GenServer and return initial state in {:ok, state} tuple. The state is an empty list that will hold todos.

# lib/todo/todo_server.ex
defmodule Todo.TodoServer do
  use GenServer

  # public API
  def start_link(id) do
    GenServer.start_link(__MODULE__, [], name: name(id))
  end

  defp name(id) do
    {:via, Registry, {Todo.TodoRegistry, id}}
  end

  # callbacks
  def init(_opts) do
    {:ok, []}
  end
end

get/1 is a public API function that invokes GenServer. We use synchronous GenServer.call/2 to retrieve todos from GenServer. name/1 helper refers to the server we want to call and :get is the request. Calls must implement corresponding handle_call/3 callback. As per convention, handle_call/3 returns tuple {:reply, return_value, new_state}. Similarly, add/2 and toggle/2 functions with their callbacks are added.

# lib/todo/todo_server.ex
  # public API
  def get(id) do
    GenServer.call(name(id), :get)
  end

  def add(id, todo_params) do
    GenServer.call(name(id), {:add, todo_params})
  end

  def toggle(id, todo_id) do
    GenServer.call(name(id), {:toggle, todo_id})
  end

  # callbacks
  def handle_call(:get, _from, todos) do
    {:reply, todos, todos}
  end

  def handle_call({:add, todo_params}, _from, todos) do
    case Todo.create(todo_params) do
      {:ok, todo} ->
        new_todos = todos ++ [todo]
        {:reply, {:ok, new_todos}, new_todos}

      error ->
        {:reply, error, todos}
    end
  end

  def handle_call({:toggle, todo_id}, _from, todos) do
    todos = toggle_todo(todos, todo_id)
    {:reply, todos, todos}
  end

  defp toggle_todo(todos, todo_id) do
    Enum.map(todos, fn
      %Todo{id: ^todo_id, completed: completed} = todo ->
        %{todo | completed: !completed}

      todo ->
        todo
    end)
  end

Because we use a Registry, we need to add it to the supervision tree.

# lib/todo/application.ex
  def start(_type, _args) do
    children = [
      TodoWeb.Endpoint,
      # add this line
      {Registry, keys: :unique, name: Todo.TodoRegistry}
    ]

    opts = [strategy: :one_for_one, name: Todo.Supervisor]
    Supervisor.start_link(children, opts)
  end

Let’s take it for a spin. Fire up iex -S mix.

iex(1)> Todo.TodoServer.start_link(:foo)
{:ok, #PID<0.341.0>}
iex(2)> Todo.TodoServer.add(:foo, %{"title" => "example todo"})
{:ok,
 [
   %Todo{
     completed: false,
     id: "8ab685b2-4686-4cea-bdeb-f4b48df4b0ae",
     title: "example todo"
   }
 ]}
iex(3)> Todo.TodoServer.toggle(:foo, "8ab685b2-4686-4cea-bdeb-f4b48df4b0ae")
[
  %Todo{
    completed: true,
    id: "8ab685b2-4686-4cea-bdeb-f4b48df4b0ae",
    title: "example todo"
  }
]

Now it’s time to switch our LiveView implementation to use GenServer. First, we start a GenServer process in mount/2 with TodoServer.start_link/1 passing the token. If the server is already started this call returns error tuple that is ignored. With server started we can call TodoServer.get/1 to retrieve todos. Then replace remaining functionality with calls to the server.

# lib/todo_web/live/todo_live.ex
  # add alias to the beginning of the file
  alias Todo.TodoServer

  def mount(%{todo_token: todos_id}, socket) do
    TodoServer.start_link(todos_id)
    todos = TodoServer.get(todos_id)
    {:ok, assign(socket, todos: todos, filter: "all", todos_id: todos_id)}
  end

  def handle_event("add_todo", %{"todo" => todo_params}, socket) do
    case TodoServer.add(socket.assigns.todos_id, todo_params) do
      {:ok, todos} -> {:noreply, assign(socket, :todos, todos)}
      error -> {:noreply, socket}
    end
  end

  def handle_event("toggle_todo", %{"id" => todo_id}, socket) do
    todos = TodoServer.toggle(socket.assigns.todos_id, todo_id)
    {:noreply, assign(socket, :todos, todos)}
  end

Because start_link/1 is naming convention for spawning linked processes, we can pass only a module or two-element tuple with module and arguments to start_link/1.

# example child specification to start Registry
%{
  id: Registry,
  start: {Registry, :start_link, [keys: :unique, name: Todo.TodoRegistry]}
}
# which is equivalent to
{Registry, [keys: :unique, name: Todo.TodoRegistry]}~

We can apply our newly acquired knowledge to implement TodoServer supervisor. Because we intend to spawn the same GenServer for every user on demand we use DynamicSupervisor. Even if you don’t need a supervisor for its restarting capabilities it is still recommended to keep processes linked to the supervision tree.
Children are added with DynamicSupervisor.start_child/2 function that accepts child specification. Due to the fact we implemented Todo.TodoServer.start_link/1 we can use shorter child spec {Todo.TodoServer, id} in which id is the argument passed to Todo.TodoServer.start_link/1.

# example child specification to start Registry
%{
  id: Registry,
  start: {Registry, :start_link, [keys: :unique, name: Todo.TodoRegistry]}
}
# which is equivalent to
{Registry, [keys: :unique, name: Todo.TodoRegistry]}~

# lib/todo/application.ex
    children = [
      TodoWeb.Endpoint,
      # add Todo.TodoSupervisor
      Todo.TodoSupervisor,
      {Registry, keys: :unique, name: Todo.TodoRegistry}
    ]

The last step is to start TodoServer using supervisor when mounting LiveView.

# lib/todo_web/live/todo_live.ex
  def mount(%{todo_token: todos_id}, socket) do
    # replace TodoServer.start_link(todos_id)
    Todo.TodoSupervisor.start_child(todos_id)
    todos = TodoServer.get(todos_id)
    {:ok, assign(socket, todos: todos, filter: "all", todos_id: todos_id)}
  end

We can check the supervision tree with the Observer. Restart server with iex -S mix phx.server and type :observer.start().
Go to Applications tab, click on todo app from the list on the left and you can see the supervision tree. There will be a lot of processes from Phoenix itself but you should find the processes we added: Todo.TodoSupervisor and Todo.TodoRegistry. Once user visits the page, a new TodoServer process is spawned under TodoSupervisor.

• read/store todos in a file using DETS or in a database with Ecto;
• play with Process.send_after/3 and handle_info/2 to perform periodic tasks, e.g. deleting old todos;
• use GenServer timeout to terminate server after a period of inactivity.