DEV Community: Christian Alexander

Async Elixir Telemetry

Christian Alexander — Thu, 22 Feb 2024 13:17:54 +0000

Learn how to use Telemetry and GenServer in Elixir to wire up analytics without sacrificing app responsiveness.

One of the most fantastic abstractions used in many Elixir applications is provided by the Telemetry library. It allows libraries to emit arbitrary events that other code can subscribe to. This is similar to a Node.js event emitter, but make it global.

While this doesn't seem very important at first glance, the power of Telemetry comes from its vast adoption. Packages such as Phoenix (the popular web framework), Oban (the popular background job framework), and Finch (the HTTP request library used by Req) all emit events through Telemetry.

Typically, Telemetry is used to publish metrics through systems like Prometheus, StatsD, and CloudWatch. These solutions often rely on the telemetry_metrics library and consume numeric values included in many telemetry events.

Telemetry also tends to be used for request logging, since every Phoenix request emits a series of events containing the requested path, method, IP, and response status code. It's a really convenient way to plug in logging services without altering router and controller code.

The most important limitation to consider when using telemetry can be found in the readme:

The handle_event callback of each handler is invoked synchronously on each telemetry:execute call. Therefore, it is extremely important to avoid blocking operations. If you need to perform any action that is not immediate, consider offloading the work to a separate process (or a pool of processes) by sending a message.

This brings me to the topic of today's post, but first I want to set some context.

Intent: Capture Product Analytics

Recently I've been working on a lightweight sprint point estimation tool using Phoenix LiveView. This tool is called Sprinty Points, and I think most agile teams will like it.

Sprinty Points doesn't yet have a database. Every user is anonymous and session state lives in GenServers. This has allowed me to build a very performant prototype really quickly without worrying about schema evolution.

Unfortunately, this architecture makes it hard for me to keep track of user metrics and the growth of the application. Without a users table, it's hard for me to know how many users I have. Without an events or sessions table, it's hard for me to get an estimate on usage.

To get some very basic metrics, I looked around and found an analytics tool with a generous free tier. It's called PostHog (not sponsored, just a fan). They have a very simple HTTP API through which I can post user registrations and custom events. These events can be turned into dashboards.

Since I wasn't very attached to any specific analytics application, I decided to not directly call their API from my handler. Instead, I added some custom telemetry events to a few key controllers and wired up a custom Telemetry handler to send the events out.

Sample Code, Synchronous Approach

The first approach I took involved three parts: create a simple PostHog client module, wire up a simple Telemetry handler, and start emitting events from Phoenix controllers.

PostHog Client Module

The client module is a basic wrapper around their capture endpoint, using Req for HTTP and an environment variable for the API key.

defmodule Posthog do
  require Logger

  @base_url "https://app.posthog.com"

  def capture(event, distinct_id, properties \\ %{}) do
    api_key = api_key()
    body = %{event: event, distinct_id: distinct_id, api_key: api_key, properties: properties}

    if is_nil(api_key) do
      Logger.info("Posthog API key not set, skipping event: #{Jason.encode!(body)}")
      {:ok, nil}
    else
      base_request()
      |> Req.post(url: "/capture", json: body)
      |> transform_response()
    end
  end

  def capture_batch(entries) do
    api_key = api_key()
    body = %{batch: entries, api_key: api_key}

    if is_nil(api_key) do
      Logger.info("Posthog API key not set, skipping batch: #{Jason.encode!(body)}")
      {:ok, nil}
    else
      base_request()
      |> Req.post(url: "/capture", json: body)
      |> transform_response()
    end
  end

  defp base_request do
    Req.new(base_url: base_url())
  end

  defp transform_response({:ok, %{status: status} = response}) when status >= 400 do
    {:error, response}
  end

  defp transform_response(response), do: response

  defp api_key do
    Application.get_env(:points, :posthog_api_key)
  end

  defp base_url do
    case Application.get_env(:points, :posthog_api_url) do
      nil -> @base_url
      url -> url
    end
  end
end

The Synchronous Telemetry Handler

This handler just passes whatever is sent under the [:posthog, :event] key directly to the capture method.

defmodule Points.TelemetryHandlers.SimplePosthog do
  def attach() do
    :telemetry.attach("posthog-events", [:posthog, :event], &__MODULE__.handle_event/4)
  end

  def detach() do
    :telemetry.detach("posthog-events")
  end

  def handle_event([:posthog, :event], measure, _meta, config) do
    Posthog.capture(measure)
  end
end

To register the handler, I just called Points.TelemetryHandlers.SimplePosthog.attach() in the initialization method of my application.

Sending an Event

With the plumbing in place, all I had to do was send a Telemetry event under the expected key.

Here's a snippet from the controller that creates a new session:

defmodule PointsWeb.SessionController do
  alias Points.SessionServer
  use PointsWeb, :controller

  def new(conn, params) do
    session_id = generate_new_session_id()
    SessionServer.ensure_started(session_id)

    :telemetry.execute([:posthog, :event], %{
      event: "session_created",
      distinct_id: conn.assigns.current_user.id,
      properties: %{
        "session_id" => session_id
      }
    })

    redirect(conn, to: ~p"/sessions/#{session_id}")
  end
end

The value passed as the second argument to :telemetry.execute makes it to PostHog.

Reflecting on the Approach

The biggest problem with this naive solution is that the :telemetry.execute call caused a meaningful, inconsistent delay every time a user created a session. This is because the application had to wait for the registered event handler to execute before the application could proceed, and the event handler made a blocking call to an external service.

Clearly, there must be a better way.

Making it Better: Async

As with many other problems faced in Elixir, the solution may involve a GenServer. Before I get into that, I want to make an important note: I didn't reach for a GenServer until I found that I had a need for a background process with persistent state. It's valuable to try a simple approach before introducing another process to the supervision tree.

Knowing that the synchronous event handler caused a real delay in the application, I chose to follow the advice from the Telemetry readme:

If you need to perform any action that is not immediate, consider offloading the work to a separate process (or a pool of processes) by sending a message.

The new architecture includes a handler that stores the event in memory and periodically flushes its received events to PostHog. Initially, I was going to directly use GenServer state to temporarily store events, but the handle_event method is called by the process that invokes :telemetry.execute. To get this event into the GenServer state, I'd have to use GenServer.cast and cause a bottleneck in the GenServer's mailbox.

Instead of GenServer state, I realized a write-optimized Ets table would be a more appropriate temporary storage location. This is natively implemented code within the BEAM virtual machine and scales much better than a single process.

The new Telemetry handler can be found below. To use it, I added Points.TelemetryReporters.Posthog to my application supervision tree. When it starts up, it attaches to the existing Telemetry event key and goes to work.

defmodule Points.TelemetryReporters.Posthog do
  use GenServer, shutdown: 2_000

  require Logger

  @default_publish_interval 10_000

  def start_link(opts) when is_list(opts) do
    GenServer.start_link(__MODULE__, opts)
  end

  @impl GenServer
  def init(opts) do
    name = opts[:name]
    table_id = create_table(String.to_atom("#{name}.Ets"))

    Process.flag(:trap_exit, true)

    attach(table_id)

    Logger.info("#{name} Started")

    {:ok, {name, table_id}, @default_publish_interval}
  end

  @impl GenServer
  def handle_info(:timeout, {name, table_id} = state) do
    now = DateTime.to_unix(DateTime.utc_now(), :millisecond)

    events = :ets.select(table_id, [{{:"$1", :"$2"}, [{:<, :"$1", now}], [:"$2"]}])

    if length(events) > 0 do
      case Posthog.capture_batch(events) do
        {:ok, _} ->
          :ets.select_delete(table_id, [{{:"$1", :"$2"}, [{:<, :"$1", now}], [true]}])

        {:error, error} ->
          Logger.error("[#{name}] Failed to flush events to Posthog: #{inspect(error)}")
      end
    end

    {:noreply, state, @default_publish_interval}
  end

  @impl GenServer
  def terminate(reason, {name, table_id}) do
    Logger.warning("[#{name}] Stopped with reason #{inspect(reason)}")
    detach()

    Logger.info("[#{name}] Flushing final events to Posthog")
    events = :ets.select(table_id, [{{:"$1", :"$2"}, [], [:"$2"]}])

    if length(events) > 0 do
      case Posthog.capture_batch(events) do
        {:ok, _} -> Logger.info("[#{name}] Final events flushed to Posthog")
        {:error, _} -> Logger.error("[#{name}] Failed to flush final events to Posthog")
      end
    end

    :ets.delete(table_id)
  end

  def attach(table_id) do
    :telemetry.attach("posthog-events", [:posthog, :event], &__MODULE__.handle_event/4,
      table_id: table_id
    )
  end

  def detach() do
    :telemetry.detach("posthog-events")
  end

  def handle_event([:posthog, :event], measure, _meta, config) do
    table_id = config[:table_id]
    :ets.insert(table_id, {DateTime.to_unix(DateTime.utc_now(), :millisecond), measure})
  end

  defp create_table(name) do
    :ets.new(name, [:named_table, :duplicate_bag, :public, {:write_concurrency, true}])
  end
end

Every ten seconds, this process checks the Ets table for new events and conditionally posts them to Posthog as a batch operation. This makes heavy use of the third response value of a GenServer lifecycle method: the timeout. After some specified number of milliseconds passes, if no other event is processed by the GenServer, the handle_info method is called with a message of :timeout. It's a great way to create a background loop in an Elixir application.

To avoid losing events when the server shuts down, exits are trapped and the terminate callback makes a final attempt to send events out. The shutdown value of 2_000 indicates that the process wants two seconds to perform its cleanup task. This behavior is described well in the Elixir docs.

Outcome

Now that events are buffered in memory through a write-optimized Ets table, I don't have to worry about the latency penalty of a blocking network call. The real beauty of this approach is that I didn't have to change any of my :telemetry.execute calls to get this performance improvement. Thanks to the Telemetry abstraction, it just works.

I'm able to measure the growth of the application I'm working on and may be motivated to continue working on it—knowing that it has a healthy user base that comes back sprint after sprint.

Random Cut Forests

Christian Alexander — Mon, 07 Aug 2023 13:00:00 +0000

Anomaly detection is a complicated and well-researched area of computer science. With all of the machine learning hype of recent months, I wanted to build something using my favorite programming language, Elixir.

In 2016, a paper called Robust Random Cut Forest Algorithm was published.

The Random Cut Forest Algorithm is an unsupervised anomaly detection algorithm that is typically used in Amazon Sagemaker (docs). It aims to improve upon the Isolation Forest Algorithm. I could have tried my hand at isolation forests, but they’re so 2008.

What makes the random cut forest algorithm different is that it measures the collusive displacement rather than just the depth of a leaf node. This measure represents how many leaves in the tree would be displaced as a result of its insertion or removal. In datasets with irrelevant features, this produces much better results.

In both algorithms, an ensemble of trees (a forest) is constructed. Each tree provides a value for a given point and they are combined to produce the result.

The solution in the LiveBook document above is written entirely in Elixir, allowing it to be invoked without NIFs. A production-grade solution might involve calling a native implementation, but it’s likely that a more efficient version could still be built in Elixir.

How RCF Works

Making a tree

A set of multi-dimensional points are introduced to the tree. For simplicity, they will be depicted in two dimensions. In reality, any number of dimensions will do.

A random cut is made along a random axis, weighted by the span of each dimension’s bounding box. In this case, we randomly choose to make a vertical cut, isolating one point.

This can also be represented visually as a decision tree.

We then repeat this process until each point (leaf node) is isolated.

The Forest

A random cut forest is an ensemble of trees. Each tree is unique and learns something different about the dataset by the points it’s fed and the cuts it chooses.

Evaluation

To determine how much of anomaly a datapoint is, we insert it into the tree and calculate a collusive displacement score. This score is calculated across all trees in the forest, and the arithmetic mean is used as the result.

For example, if we were looking for the orange dot in the tree above, its score from the tree would be 6—the number of nodes that would be displaced if we were to remove it. They would be displaced because the root node would be replaced, and all branches below would be shifted up.

The displacement score of the red dot in this tree would be 4—the number of leaf nodes under its sibling. Visually, this makes sense. It’s closer to the cluster than the orange point.

Another tree might have made a different observation, deciding to cut out the red dot before the black one. This would result in a higher score.

For this reason, it’s important to consider the perspectives of the full forest.

Sine Wave Anomaly Exercise

In the research paper, the team constructs a sine wave and inserts an anomalous flat section in the middle. They then train a forest on the normal section and present the calculated anomaly scores for each point.

The linked LiveBook does the same, with some fuzziness applied to the wave for added realism.

Shingling the Data

Shingling is used to turn the time series into something periodic. Adjacent values are inserted in the tree together as multiple dimensions of the same datapoint.

For example, given a series [1, 2, 3, 4, 5] and a shingle factor of 3, the tree would receive the following multi-dimensional data points:

[1, 2, 3]
[2, 3, 4]
[3, 4, 5]

Shingling is used to represent patterns within the larger dataset. A peak is typically surrounded by lower values, and downward values typically only increase after a valley.

Final Results

With 1,000 data points, an anomaly lasting 20 steps, 50 trees containing 256 4-shingled points each, the following result is achieved:

The outlier score spikes at the beginning and end of the anomaly. There are also some smaller spikes as a result of the noise applied to the wave.

A histogram of the outlier scores shows that most scores hover around the 4–4.5 region, while a very small number of scores exceed 20—let alone 30. This analysis can be helpful in tuning the parameters of the forest as well as determining an alerting threshold.

Summary

The linked LiveBook explores a partial implementation of the approach described in the RCF paper. Along with some visuals, it has helped me build an intuition for how some random choices and a few trees can result in unsupervised anomaly detection.

If you haven’t already, try out the LiveBook locally and check out the implementation details that didn’t fit in this post:

Citations

S. Guha, N. Mishra, G. Roy, & O. Schrijvers, Robust random cut forest based anomaly detection on streams, in Proceedings of the 33rd International conference on machine learning, New York, NY, 2016 (pp. 2712-2721).

F. T. Liu, K. M. Ting and Z. -H. Zhou, "Isolation Forest," 2008 Eighth IEEE International Conference on Data Mining, Pisa, Italy, 2008, pp. 413-422, doi: 10.1109/ICDM.2008.17.

Mitigating Server-Side Request Forgery

Christian Alexander — Wed, 26 Apr 2023 13:30:00 +0000

Server-Side Request Forgery (SSRF) vulnerabilities allow an attacker to cause a server application to perform an unintended request. When exploited, the server could leak sensitive internal information or perform dangerous actions. Because this vulnerability depends on the capabilities of the server application, the potential impact of an attack can vary.

Webhooks are among the most common features that introduce SSRF vulnerabilities to applications. They combine arbitrary user input (the webhook URL) with the ability to make requests from the backend. It’s important to consider this threat when building and operating webhook systems.

The attack

For the purposes of this post, imagine we have a web application that is able to perform outbound requests to a user-configured endpoint.

Cloud credential leak

Every AWS compute resource has a special internal service, called the Instance Metadata Service (IMDS). This endpoint provides network information, initialization scripts, and even temporary AWS access keys. All of this is conveniently hosted at http://169.254.169.254 . The feature is great for avoiding hard-coded keys but comes at a cost—by default, it is available to every process on the server.

The scenario for exploiting this is simple: the user provides a URL of http://169.254.169.254/latest/meta-data/iam/security-credentials/role-name to the application, then look at the response. It should look something like this:

{
  "Code" : "Success",
  "LastUpdated" : "2012-04-26T16:39:16Z",
  "Type" : "AWS-HMAC",
  "AccessKeyId" : "ASIAIOSFODNN7EXAMPLE",
  "SecretAccessKey" : "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
  "Token" : "token",
  "Expiration" : "2017-05-17T15:09:54Z"
}

With the key in hand, the attacker can perform any action that the compute instance is permitted to do. Because AWS IAM can be complicated, some organizations provide more permission than is the server needs—using * wildcards. Occasionally, the same IAM role will be used across staging and production environments.

Even if the server isn’t hosted in AWS, there are similar endpoints in all major cloud providers. This includes Google Cloud, Azure, and DigitalOcean.

Internal endpoints

Mature organizations might rely on private networking to protect internal resources, asserting that a VPN connection is required for access. Unfortunately, web servers tend to sit on the same network.

Imagine there’s an intranet site containing trade secrets sitting on the internal network, available for all employees to access. By convincing an application server to perform requests against the site, an attacker can steal all of those secrets.

Filesystem access

Web application servers tend to contain secrets of their own: source code, configuration files, and user uploads. An attacker providing a request URL with the file:// scheme may be able to access any file visible to the web server process. This might already be locked down depending on the library being used to perform requests from the server.

Mitigating the vulnerability

Deliver, the webhook monitoring system I've been working on, is written in Elixir. Naturally, the examples will also use Elixir. The same mitigations will likely be available in any technology, but specific package recommendations may not apply.

Preventing non-HTTP(S) URIs

The following example shows how the URI.parse/1 method breaks a URI into a struct.

iex> URI.parse("https://elixir-lang.org/")

%URI{
    authority: "elixir-lang.org",
    fragment: nil,
    host: "elixir-lang.org",
    path: "/",
    port: 443,
    query: nil,
    scheme: "https",
    userinfo: nil
}

A first pass at SSRF protection might look like this:

def permit_uri?(uri) do
    URI.parse(uri).scheme in ["http", "https"]
end

By locking out unacceptable schemes, we ensure that the filesystem can't directly be accessed.

Preventing the instance metadata IP address

Another layer of protection that could be valuable is excluding 169.254.169.254, as follows:

def permit_uri?(uri) do
  parsed_uri = URI.parse(uri)

  parsed_uri.scheme in ["http", "https"] and
    parsed_uri.host != "169.254.169.254"
end

One interesting problem with this solution is that public DNS endpoints can return any IP address they'd like in response to an A record query. For example, this would be completely legal in DNS:

> nslookup malicious.example.com
Server:     1.1.1.1
Address:    1.1.1.1#53

Non-authoritative answer:
Name:   malicious.example.com
Address: 169.254.169.254

If an attacker arranged for a request to http://malicious.example.com/latest/meta-data/iam/security-credentials/role-name, it would be treated as though the IMDS IP address were provided to begin with.

Looking up DNS hostnames

Erlang's built-in :inet_dns module can be used directly in order to perform DNS lookups. On the Erlang side, IPv4 addresses are expressed as tuples with four elements, and strings used for networking have to be of type charlist instead of binary. Clearly, this implementation is getting complicated.

defmodule SSRFProtection do
  @disallowed_hosts [
    {169, 254, 169, 254}
  ]

  defp resolve_host(host) do
    host = to_charlist(host)
    case :inet.parse_address(host) do
      {:ok, addr} ->
        [addr]
      _ ->
        :inet_res.lookup(host, :in, :a)
    end
  end

  def permit_uri?(uri) do
    parsed_uri = URI.parse(uri)

    parsed_uri.scheme in ["http", "https"] and
      not Enum.any?(
        resolve_host(parsed_uri.host),
        &(&1 in @disallowed_hosts)
      )
  end
end

By running the user-provided URIs through the permit_uri?/1 method of SSRFProtection, we can prevent this class of requests.

iex> SSRFProtection.permit_uri?("https://google.com/something")
true

iex> SSRFProtection.permit_uri?("https://169.254.169.254/something")
false

iex> SSRFProtection.permit_uri?("https://malicious.example.com/something")
false

iex> SSRFProtection.permit_uri?("file://C:\\inetpub\\wwwroot\\secret.html")
false

Perform validation on every request

While there is some level of protection provided by validating a user-provided URI when it's first submitted, it's not the best we can do. DNS is not static. An attacker could submit a URI that previously resolved to 123.123.123.123, but update the record to 169.254.169.254 after the application validates it.

This is why it's important to run permit_uri?/1 (or its equivalent) for each request the server makes.

Using SafeURL

As indicated above, this implementation is messy and not great. It describes the concepts, but it's probably not the most production-ready solution. Luckily, there's an open-source Elixir package called SafeURL that encapsulates SSRF-protecting validation into a single module.

Wherever SSRFProtection.permit_uri?/1 was used, we can instead use SafeURL.allowed?/1. This library has a built-in list of IP addresses, ranges, and schemes that are a great default. Depending on your use case, it can even perform GET requests for your application through an optional HTTPoison integration.

Similar libraries

There are similar libraries for SSRF protection in most programming languages. A couple that appear to be reputable are:

Ruby: ssrf_filter
Python: advocate

Taking it further

Don't follow redirects

Some request libraries automatically follow HTTP 301 and 302 redirects. Ensure this behavior is disabled when performing requests against user-provided URLs. A malicious actor could host https://happy-looking-site.com, where it redirects to 169.254.169.254. If the web application follows the redirect, the attack will succeed.

Security groups

Make sure to configure security groups or the cloud firewall for the hosting provider of the web application in a way that limits access to what's strictly necessary. If it doesn't make sense for the application to reach a specific internal service, don't allow it.

Proxy

An additional layer of security that some companies choose to implement is an HTTP egress proxy. These typically are instances of Squid, Envoy, or a cloud provider hosted solution. When outbound requests are made by the web application, they go through the proxy. This proxy should be configured with no permission to reach back into the private network and ideally will prevent requests to bad IPs.

Wrapping up

Security is tricky to get right and can have disastrous side-effects when done incorrectly. Attackers will poke holes in every line of defense, which is why it's important to consider your threat model and establish a defense-in-depth strategy.

While writing a solution in-house can seem reasonable, there are tricky edge cases that might be solved in a well-accepted open-source package. Look to the community for the best defense.

Go and protect your application from SSRF vulnerabilities!

Efficient bidirectional infinite scroll in Phoenix LiveView

Christian Alexander — Fri, 14 Oct 2022 13:35:30 +0000

In this post, I'll share the progression of my implementation of an infinite scrolling UI in Phoenix LiveView—from naive to efficient. Follow along with the example implementation in this repo.

Context

Phoenix LiveView makes it easy to create interactive web apps without having to write much (if any) frontend code. The server sends diffs over a websocket to update the client's UI. Data can be presented to the user without page refreshes or polling.

A few weeks ago, I set out to add an alert list page to Deliver, a webhook monitoring system I've been building (check it out here).

As the page is scrolled, older alerts are loaded into the table. Critically, new alerts are added to the top of the page as they occur, backed by a Phoenix PubSub subscription.

Setup

To demonstrate the implementation of this project, I've created a new LiveView application that contains a simplified data model that simulates what had to be built in Deliver. It uses a background process (a GenServer) to create events and publish them over Phoenix PubSub, storing the events in a database for later retrieval.

Attempt 1: The Memory Hog

The first implementation of the alerts page was really easy to implement, but it kept on using more and more memory during sustained sessions.

First, I created a simple table view that loaded five alerts from the alerts table (link to related commit).



defmodule BidirectionalScrollWeb.Live.ScrollDemoLive do
  use BidirectionalScrollWeb, :live_view

  alias BidirectionalScroll.Alerts

  def mount(_params, _session, socket) do
    alerts = Alerts.list_alerts(limit: 5)

    socket = assign(socket, alerts: alerts)

    {:ok, socket}
  end

  def render(assigns) do
    ~H"""
    <h1>Alerts</h1>
    <table>
      <thead>
        <tr>
          <th>ID</th>
          <th>Started At</th>
          <th>Resolved At</th>
        </tr>
      </thead>
      <tbody>
        <%= for alert <- @alerts do %>
          <tr>
            <td><%= alert.id %></td>
            <td><%= alert.started_at %></td>
            <td><%= alert.resolved_at %></td>
          </tr>
        <% end %>
      </tbody>
    </table>
    """
  end
end

Next, I added a "Load More" button to the bottom of the page, using phx-click and a handle_event callback (link to related commit). Additional alerts are added to the existing ones in a large list, assigned to the socket and rendered in the table.



--- a/lib/bidirectional_scroll_web/live/scroll_demo_live.ex
+++ b/lib/bidirectional_scroll_web/live/scroll_demo_live.ex
@@ -11,6 +11,18 @@ defmodule BidirectionalScrollWeb.Live.ScrollDemoLive do
     {:ok, socket}
   end

+  def handle_event("load-more", _value, socket) do
+    alerts = socket.assigns.alerts
+
+    oldest_loaded_alert = Enum.min_by(alerts, & &1.started_at, NaiveDateTime)
+    older_alerts = Alerts.list_alerts(started_before: oldest_loaded_alert.started_at, limit: 5)
+
+    alerts = alerts ++ older_alerts
+    socket = assign(socket, alerts: alerts)
+
+    {:noreply, socket}
+  end
+
   def render(assigns) do
     ~H"""
     <h1>Alerts</h1>
@@ -32,6 +44,7 @@ defmodule BidirectionalScrollWeb.Live.ScrollDemoLive do
         <% end %>
       </tbody>
     </table>
+    <button phx-click="load-more">Load More</button>
     """
   end
 end

Finally, I wired up Phoenix PubSub to broadcast when alerts are created and updated. The subscription in the scrolling UI causes two handle_info callbacks to be invoked. When :alert_created is received, the new alert is prepended to the list that is assigned to the socket. :alert_updated uses Enum.map to update the matching alert by ID. (Link to related commit)



def handle_info({:alert_created, alert}, socket) do
    alerts = socket.assigns.alerts

    alerts = [alert | alerts]
    socket = assign(socket, alerts: alerts)

    {:noreply, socket}
  end

  def handle_info({:alert_updated, %{id: alert_id} = alert}, socket) do
    alerts = socket.assigns.alerts

    alerts =
      Enum.map(alerts, fn
        %{id: ^alert_id} -> alert
        a -> a
      end)

    socket = assign(socket, alerts: alerts)

    {:noreply, socket}
  end

At this point, the alerts page technically works. However, in production this will end up consuming tons of memory. Every connected session will retain a copy of all alerts in an ever-increasing list, never being garbage collected.

This is one of the big downsides of naive Phoenix LiveView implementations: in order to automatically produce efficient diffs to send to the browser, the server must keep its assigned values in the socket so it can re-render.

Luckily, this is not the only way to build a list view.

Attempt 2: Out of Order

Phoenix LiveView has a concept called temporary assigns to solve the memory consumption issue encountered in the first attempt.

To use temporary assigns, we have to add a phx-update attribute to the list's container, along with providing a DOM ID to uniquely identify the container on the page. This allows the LiveView client script to prepend and append to the container's contents. Each child element also requires an ID to perform updates on items that have already been rendered. (Link to related commit)



@@ -61,9 +48,9 @@ defmodule BidirectionalScrollWeb.Live.ScrollDemoLive do
           <th>Resolved At</th>
         </tr>
       </thead>
-      <tbody>
+      <tbody id="alert-list" phx-update="prepend">
         <%= for alert <- @alerts do %>
-          <tr>
+          <tr id={"alert-#{alert.id}"} >
             <td><%= alert.id %></td>
             <td><%= alert.started_at %></td>
             <td><%= alert.resolved_at %></td>

For now, we are hard-coding phx-update to prepend. Any alert that hasn't yet been received by the frontend will be added to the top of the list.

Along with this render update, we have to start using temporary assigns:



@@ -6,33 +6,20 @@ defmodule BidirectionalScrollWeb.Live.ScrollDemoLive do  
   def mount(_params, _session, socket) do
     if connected?(socket) do
       Phoenix.PubSub.subscribe(BidirectionalScroll.PubSub, "alerts")
     end
     alerts = Alerts.list_alerts(limit: 5)

     socket = assign(socket, alerts: alerts)

-    {:ok, socket}
+    {:ok, socket, temporary_assigns: [alerts: []]}
   end

   def handle_info({:alert_created, alert}, socket) do
-    alerts = socket.assigns.alerts
-
-    alerts = [alert | alerts]
-    socket = assign(socket, alerts: alerts)
-
+    socket = assign(socket, alerts: [alert])
     {:noreply, socket}
   end

-  def handle_info({:alert_updated, %{id: alert_id} = alert}, socket) do
-    alerts = socket.assigns.alerts
-
-    alerts =
-      Enum.map(alerts, fn
-        %{id: ^alert_id} -> alert
-        a -> a
-      end)
-
-    socket = assign(socket, alerts: alerts)
-
+  def handle_info({:alert_updated, alert}, socket) do
+    socket = assign(socket, alerts: [alert])
     {:noreply, socket}
   end

Adding the temporary_assigns to the mount response causes the assigns with corresponding keys to be reset to the default value after every render. In this case, we reset socket.assigns.alerts to [] after every render. By resetting the value, we allow the runtime to garbage collect the Alert instances that were being referenced as members of the alerts list.

This update ends up simplifying the handle_info callback implementations. Since a render runs every time a callback returns, we can set the alerts list to a list only containing the new or updated value. In the case of :alert_created, there will be no element with a matching ID and it will follow the phx-update behavior—in this case prepending to the list. As for :alert_updated, a matching element should be found in the DOM. The matching element will simply be replaced.

The "Load More" button causes the LiveView process to crash now, since its previous implementation looked at the list of assigned alerts to figure out the started_at value of the earliest alert on the page. Since the list is empty, we have no record of the oldest loaded alert. To fix this, we can add (and maintain) a single datetime in the socket. (Link to related commit)

At this point, the page prepends items as expected, but the "Load More" button ends up prepending values that should be appended.

Attempt 3: It Works Again

At the end of the second attempt, we had an implementation that was close to feature parity with the naive memory hog, but it had one major flaw: we were only prepending to the list.

The update behavior is controlled by the phx-update attribute of the container element. There's nothing keeping us from setting this to append or prepend dynamically. It turns out that the solution to this problem is quite simple:
(Link to relevant commit)



@@ -6,28 +6,39 @@ defmodule BidirectionalScrollWeb.Live.ScrollDemoLive do  
   def mount(_params, _session, socket) do
     if connected?(socket) do
       Phoenix.PubSub.subscribe(BidirectionalScroll.PubSub, "alerts")
     end
     alerts = Alerts.list_alerts(limit: 5)
     oldest_loaded_alert = Enum.min_by(alerts, & &1.started_at, NaiveDateTime)

     socket =
       assign(socket,
         alerts: alerts,
-        oldest_alert_started_at: oldest_loaded_alert.started_at
+        oldest_alert_started_at: oldest_loaded_alert.started_at,
+        update_direction: "append"
       )

     {:ok, socket, temporary_assigns: [alerts: []]}
   end

   def handle_info({:alert_created, alert}, socket) do
-    socket = assign(socket, alerts: [alert])
+    socket =
+      assign(socket,
+        alerts: [alert],
+        update_direction: "prepend"
+      )
+
     {:noreply, socket}
   end

   def handle_info({:alert_updated, alert}, socket) do
-    socket = assign(socket, alerts: [alert])
+    socket =
+      assign(socket,
+        alerts: [alert],
+        update_direction: "prepend"
+      )
+
     {:noreply, socket}
   end

@@ -34,13 +45,14 @@ defmodule BidirectionalScrollWeb.Live.ScrollDemoLive do
   def handle_event("load-more", _value, socket) do
     oldest_alert_started_at = socket.assigns.oldest_alert_started_at
     alerts = Alerts.list_alerts(started_before: oldest_alert_started_at, limit: 5)
     oldest_loaded_alert = Enum.min_by(alerts, & &1.started_at, NaiveDateTime)

     socket =
       assign(socket,
         alerts: alerts,
-        oldest_alert_started_at: oldest_loaded_alert.started_at
+        oldest_alert_started_at: oldest_loaded_alert.started_at,
+        update_direction: "append"
       )

     {:noreply, socket}
@@ -57,7 +69,7 @@ defmodule BidirectionalScrollWeb.Live.ScrollDemoLive do
           <th>Resolved At</th>
         </tr>
       </thead>
-      <tbody id="alert-list" phx-update="prepend">
+      <tbody id="alert-list" phx-update={@update_direction}>
         <%= for alert <- @alerts do %>
           <tr id={"alert-#{alert.id}"} >
             <td><%= alert.id %></td>

In this change, we bind the phx-update attribute of the alert-list container element to the assigns value of update_direction. Whenever we receive a new alert, we set the direction to append. Whenever the "Load More" button is clicked, we set it to append.

We now have a working implementation that matches the first attempt, but with substantially less per-connection memory consumption on the server side.

Bonus: Automatically Load More

Most infinite scrolling implementations don't require the user to click a button to load more content. We can do the same using LiveView client hooks.

Hooks allow the client to execute JavaScript functions during the lifecycle of any element with a corresponding phx-hook attribute. (Link to related commit)

In this case, we will add a hook named InfiniteScrollButton that will simulate a click of the button whenever it enters the browser's viewport. This code replaces the existing new LiveSocket call in assets/js/app.js.



let liveSocket = new LiveSocket("/live", Socket, {
  params: { _csrf_token: csrfToken },
  hooks: {
    InfiniteScrollButton: {
      loadMore(entries) {
        const target = entries[0];
        if (target.isIntersecting && !this.el.disabled) {
          this.el.click();
        }
      },
      mounted() {
        this.observer = new IntersectionObserver(
          (entries) => this.loadMore(entries),
          {
            root: null, // window by default
            rootMargin: "0px",
            threshold: 1.0,
          }
        );
        this.observer.observe(this.el);
      },
      beforeDestroy() {
        this.observer.unobserve(this.el);
      },
    },
  },
});

Finally, we have to add the phx-hook attribute (and a DOM ID to help with tracking) to the rendered "Load More" button to cause it to click itself whenever it enters the page:



@@ -79,7 +79,7 @@ defmodule BidirectionalScrollWeb.Live.ScrollDemoLive do
         <% end %>
       </tbody>
     </table>
-    <button phx-click="load-more">Load More</button>
+    <button id="alerts-load-more" phx-hook="InfiniteScrollButton" phx-click="load-more">Load More</button>
     """
   end
 end

That's it! We now have a memory-efficient, bidirectional infinite scrolling view that automatically keeps elements up-to-date whenever they are sent over PubSub.

Conclusion

Phoenix LiveView makes it simple to build applications that work. When performance and scaling issues come up, escape hatches such as temporary assigns and the client library can save the day. As a final step, interactivity can be added through hooks and (not seen here) JS commands.

LiveView is easily the most productive full-stack solution I've ever encountered, and it's a joy to get to use it in projects like Deliver (which, again, you should check out if you have a webhook system).

Adding soft delete to a Phoenix Commanded (CQRS) API

Christian Alexander — Wed, 01 Jun 2022 22:41:41 +0000

Goal

Implement a soft delete in the API from part one, allowing items to be restored after deletion.

Follow along with what I learned while iterating on a project named todo_backend_commanded. Its git history shows the steps involved to implement soft delete and restore functionality.

Context

In the previous post, I converted a vanilla Phoenix API to CQRS with Commanded.

This application writes to the database using the commanded_ecto_projections hex package, which subscribes to events and projects their state into tables managed by the Ecto database library.

Since the core data model of this application is an append-only (immutable) log, the events can be replayed and the read model can be dramatically changed using existing data.

Updating the aggregate

(Link to relevant commit)

The first step is to add a deleted_at field to the Todo aggregate.

 defmodule TodoBackend.Todos.Aggregates.Todo do
   defstruct [
     :uuid,
     :title,
     :completed,
-    :order
+    :order,
+    deleted_at: nil
   ]

   ...
-  def apply(%Todo{uuid: uuid}, %TodoDeleted{uuid: uuid}) do
-    nil
-  end
+  def apply(%Todo{uuid: uuid} = todo, %TodoDeleted{uuid: uuid}) do
+    %Todo{todo | deleted_at: DateTime.utc_now()}
+  end

An aggregate is a struct representing the state of an object in an application. In Commanded, aggregates are hydrated in-memory from dispatched events. The aggregate struct is used to determine the side-effects of a command, if the command is to be accepted at all.

This diff causes the aggregate's deleted_at field to be set as the datetime when the event log is hydrated. We'll come back to this, since it would be preferable to capture the time the todo item was deleted.

Aside: aggregates sound slow

If an aggregate has to be hydrated from its events every time it is used, how is this not a terrible performance burden?

In practice, an individual aggregate shouldn't have many events. How many interactions could one todo item possibly have? If it turns out certain items have large event logs, there are a couple of levers that can be pulled in Commanded.

Lifespans

(Link to relevant commit)

So, I ~~lied~~ told a half-truth about aggregates. They are not hydrated in-memory for every command / event. In reality, aggregates are implemented with GenServer each caching their state and being managed under the commanded application's supervision tree (ultimately by a DynamicSupervisor called Commanded.Aggregates.Supervisor, to be specific).

Since aggregates are cached as long-running processes, we only have to fetch each event once.

The timeout of an aggregate process can be configured via an implementation of the AggregateLifespan behaviour. The default lifespan, DefaultLifespan, sets an infinite timeout unless an exception is raised.

Depending on the number of aggregate instances in a live system, this could introduce another problem—these processes could stay alive for the life of the server. That sounds like an OOM waiting to happen.

To address this, I implemented a lifespan for the Todo aggregate that keeps aggregates around for a minute—unless they receive the TodoDeleted event. Being able to pattern match and implement this within the context of an application means the timeouts can reflect the specific details of your domain.

Snapshots

For aggregates with many events, Commanded has aggregate state snapshots. These can be configured by adding a snapshotting section to the Commanded application in config/config.exs.

There are two options per aggregate: snapshot_every and snapshot_version. snapshot_every is used to cause snapshots to be created after a certain number of events. This value acts as an upper bound of events that must be evaluated when hydrating an aggregate state. snapshot_version is a non-negative integer that should be updated every time the aggregate is updated. It is used to invalidate old snapshots that may be missing new fields or were created with old apply methods.

Snapshots can be valuable in applications with long event logs, but didn't seem to be worthwhile in this todo API. If we were using snapshots, the addition of the deleted_at field would have required an increment of snapshot_version.

To learn more about aggregate performance, check out this blog post where an organization dramatically improved the performance and footprint of their Commanded application through lifespans and snapshots.

Effective dating the delete event

(Link to relevant commit)

As I noted when adding the deleted_at field to the aggregate, it's not appropriate to use the datetime when the event is applied. This would cause the deleted_at value to be different every time the service is deployed or the aggregate genserver is restarted. It would be much better to capture the datetime when the event is created, so it can be persisted in the event log.

First, I added a datetime field to the TodoDeleted event struct:

 defmodule TodoBackend.Todos.Events.TodoDeleted do
   @derive Jason.Encoder
   defstruct [
-    :uuid
+    :uuid,
+    :datetime
   ]
end

Next, I populated this value in the execute method of the todo aggregate, when the DeleteTodo command received. Since execute is called when a command is dispatched, this will be set to the datetime when the delete endpoint is called.

I also updated the apply method to use the datetime from the event, instead of the current datetime. This value will come from the event log.

 defmodule TodoBackend.Todos.Aggregates.Todo do
   ...
   def execute(%Todo{uuid: uuid}, %DeleteTodo{uuid: uuid}) do
-    %TodoDeleted{uuid: uuid}
+    %TodoDeleted{uuid: uuid, datetime: DateTime.utc_now()}
   end

   ...
-  def apply(%Todo{uuid: uuid} = todo, %TodoDeleted{uuid: uuid}) do
-    %Todo{todo | deleted_at: DateTime.utc_now()}
+  def apply(%Todo{uuid: uuid} = todo, %TodoDeleted{uuid: uuid, datetime: effective_datetime}) do
+    %Todo{todo | deleted_at: effective_datetime}
   end

Read your writes

The log stores events in JSON format. Since JSON does not have a native datetime format, the Commanded.Serialization.JsonDecoder protocol must be implemented on the event struct.

defmodule TodoBackend.Todos.Events.TodoDeleted do
  ...

  defimpl Commanded.Serialization.JsonDecoder, for: TodoDeleted do
    @doc """
    Parse the datetime included in the aggregate state
    """
    def decode(%TodoDeleted{} = state) do
      %TodoDeleted{datetime: datetime} = state

      {:ok, dt, _} = DateTime.from_iso8601(datetime)

      %TodoDeleted{state | datetime: dt}
    end
  end
end

What about existing events?

This is fine, but what about all of those events that are already in the log without datetimes?

Since the log is append-only and immutable, the way to extend an event definition is to add a fallback value for the new field.

In this case, I modified the JSON decoder method—substituting a nil value for the current datetime. Since the value was never written to old events, the actual deletion time is lost. The best thing we can do is provide some value and ensure we write a datetime on all future events.

 defmodule TodoBackend.Todos.Events.TodoDeleted do
   ...

   defimpl Commanded.Serialization.JsonDecoder, for: TodoDeleted do
     @doc """
     Parse the datetime included in the aggregate state
     """
     def decode(%TodoDeleted{} = state) do
       %TodoDeleted{datetime: datetime} = state
-
-      {:ok, dt, _} = DateTime.from_iso8601(datetime)
-
-      %TodoDeleted{state | datetime: dt}
+
+      if datetime == nil do
+        %TodoDeleted{state | datetime: DateTime.utc_now()}
+      else
+        {:ok, dt, _} = DateTime.from_iso8601(datetime)
+
+        %TodoDeleted{state | datetime: dt}
+      end
+    end
   end
 end

Updating the read model

(Link to relevant commit)

In a production environment it would be best to create a new database table and projector containing a deleted_at field, deploy the application to build the table, update callers to use the new table, and then delete the old table. This is safe to do because all writes happen in the event layer—the read model is read-only.

In some cases, it could also be reasonable to update the projector such that the data can be migrated in-place. This seemed like more effort than it was worth for a blog post, so I opted for the dangerous option: blow away the table and rebuild it with an updated projector.

Preparing the database

First, I generated a migration to add the deleted_at column to the todos table:

mix ecto.gen.migration add_deleted_at_to_todo

In the generated migration file, I added the column and an index:

defmodule TodoBackend.Repo.Migrations.AddDeletedAtToTodo do
  use Ecto.Migration

  def change do
    alter table(:todos) do
      add :deleted_at, :naive_datetime_usec
    end

    create index(:todos, [:deleted_at])
  end
end

Finally, I added the field to the projection:

 defmodule TodoBackend.Todos.Projections.Todo do
   ...
   schema "todos" do
     field :completed, :boolean, default: false
     field :title, :string
     field :order, :integer, default: 0
+    field :deleted_at, :naive_datetime_usec, default: nil

     timestamps()
   end
 end

After running mix ecto.migrate, the database side is ready to go.

Updating the projector

The projector update is very simple: instead of deleting the row from the table, just update the deleted_at column.

 defmodule TodoBackend.Todos.Projectors.Todo do
   ...
-  project(%TodoDeleted{uuid: uuid}, _, fn multi ->
-    Ecto.Multi.delete(multi, :todo, fn _ -> %Todo{uuid: uuid} end)
+  project(%TodoDeleted{uuid: uuid, datetime: effective_datetime}, _, fn multi ->
+    case Repo.get(Todo, uuid) do
+      nil ->
+        multi
+
+      todo ->
+        Ecto.Multi.update(
+          multi,
+          :todo,
+          Todo.delete_changeset(todo, %{deleted_at: effective_datetime})
+        )
+    end
+  end)
   ...
 end

I added a delete_changeset to the Todo projection to represent a specific desired change. It ensures we only update the deleted_at column. This is referenced in the new TodoDeleted projector method.

defmodule TodoBackend.Todos.Projections.Todo do
  ...

  def delete_changeset(todo, attrs \\ %{}) do
    todo
    |> cast(attrs, [:deleted_at])
  end
end

Forcefully re-populating the database

Like I mentioned before, it's a terrible idea to truncate a database in production. Here's how I did it locally to explore Commanded. It's possible this could also be done by updating the name value provided to the use Commanded.Projections.Ecto call in TodoBackend.Todos.Projections.Todo and restarting the server.

If there is no existing data in the log, this step can be skipped.

In an iex shell (iex -S mix phx.server):

# Delete the projected values
TodoBackend.Repo.delete_all(TodoBackend.Todos.Projections.Todo)

# Remove the version tracking entry for the ecto projection
TodoBackend.Repo.delete(%TodoBackend.Todos.Projectors.Todo.ProjectionVersion{projection_name: "Todos.Projectors.Todo"})

alias Commanded.Event.Handler
alias Commanded.Registration

# Trigger a reset of the projector
registry_name = Handler.name(TodoBackend.App, "Todos.Projectors.Todo")
projector = Registration.whereis_name(TodoBackend.App, registry_name)
send(projector, :reset)

Hiding deleted items

Now that deleted items are present in the todos table, it's important to update the Todos context module to filter out deleted items. This can be done in the context without needing any updates to the controller, since all access is encapsulated. As I noted in the previous post, this is a really valuable property of the DDD-inspired approach of modern Phoenix applications.

 defmodule TodoBackend.Todos do
   ...

   def list_todos do
-    Repo.all(Todo)
+    from(t in Todo,
+      where: is_nil(t.deleted_at)
+    )
+    |> Repo.all()
   end

   ...
-  def get_todo!(uuid), do: Repo.get_by!(Todo, uuid: uuid)
+  def get_todo!(uuid) do
+    from(t in Todo,
+      where: is_nil(t.deleted_at)
+    )
+    |> Repo.get_by!(uuid: uuid)
+  end
 end

Where are we, now?

Soft deletes have been implemented. The read model has been reconstructed from existing data, including previously-deleted items.

In CQRS applications, data never is really deleted—at most, it just gets removed from the read model.

Not legal advice on handling GDPR: some applications may choose to encrypt log entries with a per-entity encryption key in order to satisfy permanent data deletion requests. To fully delete data, the encryption key could be discarded. This would cause the events to be unreadable if they were ever replayed. Some special handling would be necessary to prevent the application from crashing on GDPR-deleted entities. This is very clearly out of scope for this blog post, but a very interesting concept.

Michiel Rook wrote two posts on this topic, which can be found here and here.

New feature: restore deleted items

Now that soft deletes are implemented, it's time to write the new feature: un-delete!

Implementation

(Link to relevant commit)

Let's quickly walk through the code changes required to implement the restoration of deleted todo items:

Create a `RestoreTodo` command

defmodule TodoBackend.Todos.Commands.RestoreTodo do
  defstruct [
    :uuid
  ]
end

Create a `TodoRestored` event

defmodule TodoBackend.Todos.Events.TodoRestored do
  @derive Jason.Encoder
  defstruct [
    :uuid
  ]
end

Update the aggregate

 defmodule TodoBackend.Todos.Aggregates.Todo do
   ...
+  def execute(%Todo{uuid: uuid}, %RestoreTodo{uuid: uuid}) do
+    %TodoRestored{uuid: uuid}
+  end

   ...
+  def apply(%Todo{uuid: uuid} = todo, %TodoRestored{uuid: uuid}) do
+    %Todo{todo | deleted_at: nil}
+  end

   ...
end

Route the command to the aggregate

 defmodule TodoBackend.Router do
   ...
-  dispatch([CreateTodo, DeleteTodo, UpdateTodo],
+  dispatch([CreateTodo, DeleteTodo, UpdateTodo, RestoreTodo],
    to: Todo,
    identity: :uuid,
    lifespan: Todo
  )
end

Update the projector

 defmodule TodoBackend.Todos.Projectors.Todo do
   ...
+  project(%TodoRestored{uuid: uuid}, _, fn multi ->
+    case Repo.get(Todo, uuid) do
+      nil ->
+        multi
+
+      todo ->
+        Ecto.Multi.update(multi, :todo, Todo.delete_changeset(todo, %{deleted_at: nil}))
+    end
+  end)
 end

At this point, we have implemented everything required for restoring deleted todo items, except for actually dispatching the RestoreTodo command.

Adding the API method

(Link to relevant commit)

To dispatch the RestoreTodo command, I added PUT /api/todos/:id/restore to the API.

Add a method to the context module

defmodule TodoBackend.Todos do
  ...

  def restore_todo(id) do
    command = %RestoreTodo{uuid: id}

    with :ok <- App.dispatch(command, consistency: :strong) do
      {:ok, get_todo!(id)}
    else
      reply -> reply
    end
  end
end

Add a method to the controller

defmodule TodoBackendWeb.TodoController do
  ...

  def restore(conn, %{"id" => id}) do
    with {:ok, %Todo{} = todo} <- Todos.restore_todo(id) do
      render(conn, "show.json", todo: todo)
    end
  end
end

Register the route

 defmodule TodoBackendWeb.Router do
   ...

   scope "/api", TodoBackendWeb do
     pipe_through :api

     resources "/todos", TodoController
     delete "/todos", TodoController, :delete_all
+    put "/todos/:id/restore", TodoController, :restore
   end
 end

Validation

For the first time in this system, let's look at how we can use aggregate state and the execute command handler to prevent deleting already-deleted items (and restoring non-deleted items).

Command validation is a core property of CQRS designs. Once events are emitted, they are never changed. The only way to enforce validations is to accept or reject commands.

The first validation I'll add is that items which are currently deleted can not be deleted:

 defmodule TodoBackend.Todos.Aggregates.Todo do
   ...
-  def execute(%Todo{uuid: uuid}, %DeleteTodo{uuid: uuid}) do
+  def execute(%Todo{uuid: uuid, deleted_at: nil}, %DeleteTodo{uuid: uuid}) do
     %TodoDeleted{uuid: uuid, datetime: DateTime.utc_now()}
   end

   ...
+  def execute(%Todo{}, %DeleteTodo{}) do
+    {:error, "Can not delete todo that is already deleted"}
+  end
 end

This pattern-matched method will only emit a TodoDeleted event when the existing aggregate state contains a deleted_at value of nil. In all other cases, it will fall through to the implementation that returns an {:error, message} tuple.

In a production system, structured errors should be returned in order to provide API clients with useful responses. Currently, this implementation just causes a 500.

The other validation will prevent restoring non-deleted items. It reads nearly identically to the first one.

 defmodule TodoBackend.Todos.Aggregates.Todo do
   ...
+  def execute(%Todo{deleted_at: nil}, %RestoreTodo{}) do
+    {:error, "Can only restore deleted todos"}
+  end
+
  def execute(%Todo{uuid: uuid}, %RestoreTodo{uuid: uuid}) do
    %TodoRestored{uuid: uuid}
  end
end

Summary

In this post, I shared how the CQRS implementation of a todo list API made it simple to add soft-delete and restore functionality.

I added validations and showed how to up-convert existing events to ensure they are compatible with the evolution of a Commanded application.

In most designs, this would probably not be possible unless a table tracking extension is being used in an ORM. Even with change tracking enabled through extensions like paper trail or Django simple history, it can be tricky to restore deleted entities. Object tracking would need to have been enabled before it is needed to ensure the data is still around to be restored.

When an immutable, append-only log is the source of truth for an application, it can be updated to meet requirements that were not known at the onset of the project.

Using CQRS in a simple Phoenix API with Commanded

Christian Alexander — Tue, 10 May 2022 14:01:49 +0000

Despite being a fan of event sourcing and seeing the clear benefits of the approach, I never built anything from scratch. This weekend, I finally decided to break this study cycle and do something practical.

Follow along with what I learned while implementing a project named todo_backend_commanded.
Its git history reflects the process of migrating from a vanilla Phoenix API to an event sourced solution.

Context

I have been curious about the concepts of event sourcing and CQRS for a while—obsessively reading books like Practical Microservices (Garofolo) and Architecture Patterns with Python (Percival, Gregory), along with documentation for libraries like Sequent (Ruby), Commanded (Elixir).

Whenever the topic comes up in conversation—something that happens more often than one might expect—I share a link to Kickstarter's post on how they used event sourcing to launch d.rip and suggest it wouldn't be too hard to follow their example and give it a try.

What did I do?

As noted in my ASP.NET Core post from 2017 (wow, I should write more often), there is a project called Todo-Backend that provides tests gradually guiding its users toward a full implementation of a backend API for a todo list app.

It's a very familiar API that can be implemented in a few minutes with the Phoenix Framework.

In fact, the first commit of this post's repository does just that. Most of it is the result of just two commands:

mix phx.new todo_backend_commanded --app todo_backend --no-assets --no-html --no-gettext --no-dashboard --no-live --no-mailer

mix phx.gen.json Todos Todo todos title:string completed:boolean

This is a testiment to the value and productivity of Phoenix, but the resulting code is just basic CRUD. The views are tied 1:1 with their database-backed Ecto schemas. One thing to note is that Phoenix generates DDD-style contexts. This is unlike Rails, which would produce a typical ActiveRecord sprawl: bloated models directly being accessed and lazily queried across the entire application.

Phoenix produces a single interface for each bounded context, making it a great framework for experimenting with a swap of the persistence layer.

Commanded

The Commanded hex package is a fabulous CQRS library used by some real companies in production, but it doesn't have a great on-ramp.

There's an example application called Conduit, which is the source code for an eBook's project, but its guidance is not up to date and the book itself starts with account / user management (a pretty advanced domain). The other resource outside of the package documentation is a 20 minute conference talk from 2018.

Initializing

To get started with Commanded, I installed the hex package and created two modules within the TodoBackend application, App and EventStore:

defmodule TodoBackend.App do
  use Commanded.Application, otp_app: :todo_backend
end

defmodule TodoBackend.EventStore do
  use EventStore, otp_app: :todo_backend
end

Both of these modules rely on macros exposed by Commanded and EventStore. The Commanded documentation uses the name Application, but this didn't make much sense to me, since TodoBackend already had an Application module for the supervision tree. Initially, I even thought that I was supposed to add the Commanded.Application macro to the supervisor. Once that didn't work, I renamed it to App and decided that would be the module to house my dispatching and routing.

Since I already had a Postgres database running, I decided to use EventStore rather than installing and babysitting EventStoreDB. To initialize the database and tables, I ran mix event_store.init and mix event_store.create.

First command: create todo

(Link to relevant commit)

CQRS libraries such as Commanded have a concept of aggregates, which are the core objects of a domain. With the simple data model of this API, I created a single aggregate to represent a todo. Its initial definition looks like this:

defmodule TodoBackend.Todos.Aggregates.Todo do
  defstruct [
    :uuid,
    :title,
    :completed,
    :order
  ]
end

An aggregate starts out as a plain struct. Once we have an aggregate, we can layer on commands and events.

A command represents a caller's intent to have the system respond to some proposed action. It can be accepted or rejected. When accepted, a command produces one or more events. These represent the fact that something did happen. They tend to be named in the past tense.

The first action I created in this system is CreateTodo, which is defined as a struct with the same fields as the Todo aggregate. I also created a TodoCreated event struct—also containing the same fields.

To decide how to process a command, the execute/2 method is called on the aggregate module, where the first argument is the previous aggregate state (if any exists), and the second argument is the command. If this returns a value that is not an {:error, something} tuple, the return value will be interpreted as one or more events to be committed to the log.

In action, the the Todo aggregate implementation includes the following two methods:

def execute(%Todo{uuid: nil}, %CreateTodo{} = create) do
    %TodoCreated{
        uuid: create.uuid,
        title: create.title,
        completed: create.completed,
        order: create.order
    }
end

def apply(%Todo{} = todo, %TodoCreated{} = created) do
    %Todo{
        todo
        | uuid: created.uuid,
        title: created.title,
        completed: created.completed,
        order: created.order
    }
end

At this point, Commanded seems to be a complicated system where the same struct has to be defined many times. The initial boilerplate is quite verbose, but the benefits of centering the application around an append-only log will come soon.

Once a command exists, a router is needed to determine which aggregate a command belongs to. This basic router gets the CreateTodo command wired to the Todo aggregate, using the uuid property of the command to determine which Todo instance is being referred to:

defmodule TodoBackend.Router do
  use Commanded.Commands.Router

  alias TodoBackend.Todos.Aggregates.Todo
  alias TodoBackend.Todos.Commands.CreateTodo

  dispatch([CreateTodo], to: Todo, identity: :uuid)
end

To test out the first aggregate and command, the following can be executed in an iex session (just run these commands in iex -S mix)

alias TodoBackend.App
alias TodoBackend.Todos.Aggregates.Todo
alias TodoBackend.Todos.Commands.CreateTodo

# Generate an ID for the todo item
uuid = Ecto.UUID.generate()

# Create a command instance
command = %CreateTodo{uuid: uuid, title: "Hello, world!", completed: false, order: 66}

# Run the command, which is dispatched to the aggregate via the router
App.dispatch(command)

# Query for the aggregate state
App.aggregate_state(Todo, uuid)

# Result
%Todo{
    uuid: "51004ff5-5a73-4681-87bb-1b1ffbf03fe0",
    title: "Hello, world!",
    completed: false,
    order: 66
}

Second command: delete todo

(Link to relevant commit)

Since CQRS applications rely on append-only logs, there is no way events can be deleted directly. This is problematic when the requirement to delete todo items comes around.

Luckily, there is a solution to this problem: a TodoDeleted event. This can act as a tombstone record describing that the todo stopped existing at a certain point.

With the aggregate boilerplate out of the way, this is quite easy to implement.

Create the command

defmodule TodoBackend.Todos.Commands.DeleteTodo do
  defstruct [
    :uuid
  ]
end

Define the event

defmodule TodoBackend.Todos.Events.TodoDeleted do
  @derive Jason.Encoder
  defstruct [
    :uuid
  ]
end

Add the `execute` and `apply` methods to the aggregate

In this case, we are turning the aggregate state into nil when receiving a TodoDeleted event.

def execute(%Todo{uuid: uuid}, %DeleteTodo{uuid: uuid}) do
    %TodoDeleted{uuid: uuid}
end

def apply(%Todo{uuid: uuid}, %TodoDeleted{uuid: uuid}) do
    nil
end

Update the router

dispatch([DeleteTodo], to: Todo, identity: :uuid)

Now, the DeleteTodo command can be handled by the App!

Many events per command: update todo

(Link to relevant commit)

Up to this point, there has been a one-to-one correlation between commands and events. No decisions have been made in execute/2 methods.

One important thing about events is that they can (and should) represent something meaningful happening in the domain of the application. For example, it would be tempting to create a TodoUpdated event containing title, completed, and order values.

Imagine this todo list app becomes the product of a company, and that company has a team that wants to collect metrics on how often items are completed. An analytics pipeline might need to consume all of the TodoUpdated events to determine if any of them changed the completed value. This would require knowledge of the state prior to the event. "Updated" lacks domain context and doesn't have as much utility as it could have.

Instead, we should choose to break down the update into many possible events:

Mark an item as completed
Update the title of an item
Mark an item as un-completed
Update the order of an item

Each of these events represent something meaningful happening, in the language of the domain.

To implement this—while keeping the API semantics as an "update"—I created a UpdateTodo command that produces many events:

def execute(%Todo{} = todo, %UpdateTodo{} = update) do
    completion_command =
        if todo.completed != update.completed and not is_nil(update.completed) do
            if update.completed do
                %TodoCompleted{uuid: todo.uuid}
            else
                %TodoUncompleted{uuid: todo.uuid}
            end
        end

    title_command =
        if todo.title != update.title and not is_nil(update.title),
            do: %TodoTitleUpdated{uuid: todo.uuid, title: update.title}

    order_command =
        if todo.order != update.order and not is_nil(update.order),
            do: %TodoOrderUpdated{uuid: todo.uuid, order: update.order}

    [completion_command, title_command, order_command] |> Enum.filter(&Function.identity/1)
end

The execute/2 method is invoked once per command, one at a time per instance. There are no data races, so we can put our business logic in this method and decide how to translate the command into events. If the todo's completed value is updated, we can emit a meaningful event. If more than one field is updated, we create more than one event.

Quick Note

Filtering on the Function.identity/1 method is a neat little trick to remove falsy (nil and false) entries from an enumerable.

And we're back

To update the state of the aggregate, I implemented simple apply/2 methods:

def apply(%Todo{} = todo, %TodoCompleted{}) do
    %Todo{todo | completed: true}
end

def apply(%Todo{} = todo, %TodoUncompleted{}) do
    %Todo{todo | completed: false}
end

def apply(%Todo{} = todo, %TodoTitleUpdated{title: title}) do
    %Todo{todo | title: title}
end

def apply(%Todo{} = todo, %TodoOrderUpdated{order: order}) do
    %Todo{todo | order: order}
end

Now, we have the ability to articulate things that happened in the language of the application's domain!

Projecting state into DB

Related commits:

So far, three commands can produce six events. There is an aggregate that tracks the events. Its state can be fetched by directly querying TodoBackend.App.aggregate_state/4. This is great as a domain exploration exercise, but it isn't queryable.

Enter: the read model—a representation of state, produced as a function of the event log. Unlike the original model, we only can use the read model for read operations. Aside from this one restriction, we are able to use all of the functionality available in the Ecto ORM—selecting, aggregating, and even joining.

The process of creating a read model from an event log is called "projecting." To simplify the creation and maintenance of projections, I opted to use the commanded_ecto_projections hex package. This library uses a little bit of metaprogramming magic to add the project macro to a module, causing it to add operations to an Ecto Multi—which eventually is executed against the application's database.

One housekeeping item that I had to take care of is replacing the numeric primary key of the table with a uuid, since we won't be relying on the sequential generated identifiers within our read model. This migration is awful as-written since it changes the schema by dropping all of the data in the existing table, but it got the job done for this learning exercise.

The projector for the Todo model looks like this (with some parts omitted for brevity):

defmodule TodoBackend.Todos.Projectors.Todo do
  use Commanded.Projections.Ecto,
    # Register a name for the handler's subscription in the event store
    name: "Todos.Projectors.Todo",
    application: TodoBackend.App,
    # Ensure the database operation completes before allowing the command to be considered completed.
    consistency: :strong

  project(%TodoCreated{} = created, _, fn multi ->
    Ecto.Multi.insert(multi, :todo, %Todo{
      uuid: created.uuid,
      title: created.title,
      completed: created.completed,
      order: created.order
    })
  end)

  project(%TodoDeleted{uuid: uuid}, _, fn multi ->
    Ecto.Multi.delete(multi, :todo, fn _ -> %Todo{uuid: uuid} end)
  end)

  project(%TodoCompleted{uuid: uuid}, _, fn multi ->
    case Repo.get(Todo, uuid) do
      nil -> multi
      todo -> Ecto.Multi.update(multi, :todo, Todo.update_changeset(todo, %{completed: true}))
    end
  end)

  # …more project calls below
end

Once the projector is registered with the application's supervision tree, the database should start populating with todo items from the log events.

This is one of the most important properties of an event-sourced system: a read model can easily be reconstructed from the event log. Rather than performing complicated migrations and backfills, a team can produce a new database table and replay events to produce the necessary representation.

Wire into the API

Link to relevant commit

Finally, it's time to connect the commands and read model up to the API!

As noted earlier, Phoenix produces a module for each context of an application. This context serves as an API for any application code—API, background job, or script—to interact with the domain without needing to know the persistence implementation details.

To bring it all together, the commit for this section shows all of the changes that were required to migrate the context over to the CQRS solution.

I'll only show one method here, but they're all present in the code.

Before

def create_todo(attrs \\ %{}) do
  %Todo{}
  |> Todo.changeset(attrs)
  |> Repo.insert()
end

After

def create_todo(attrs \\ %{}) do
  uuid = Ecto.UUID.generate()

  command =
  attrs
  |> CreateTodo.new()
  |> CreateTodo.assign_uuid(uuid)

  with :ok <- App.dispatch(command, consistency: :strong) do
    {:ok, get_todo!(uuid)}
  else
    reply -> reply
  end
end

One key aspect of this invocation is the consistency: :strong, which ensures that all handlers and projectors with strong consistency enabled have committed before returning. This means the read model is ready to query after dispatch has completed.

By changing the context implementation, the controller didn't have to be updated (aside from the omission of a deleted todo in the return value of delete_todo, which I thought was a pretty bad idea to begin with) in order to start using a CQRS solution.

Every change is tracked through a log of meaningful domain events. The read model can be migrated and updated as our requirements grow. New read models can be created from the same events, in case we start to have different views or different consumers. Arbitrary handlers can be written to extract the changes into some other system. Data is never truly deleted, and can be resurrected for new use cases in the future (possibly the topic of a sequel to this post).

Where to go from here

With this experience kicking the tires of Commanded, I'm inspired to try building something real and useful with CQRS.

As for this project, I have a few ideas for improvements:

Soft delete read model: Create a projection that populates a deleted_at column, to show how events can be re-used to enable new use cases like restoring deleted items
Validation: Determine how to apply validation to the commands, enforcing schemas and mandatory fields
Error handling: Test out the rejection of commands, returning the error messages to the API's caller
Taking on a meaningful domain, not just "another todo app"

How to Use the Firestore Export API Before it is Released

Christian Alexander — Fri, 20 Jul 2018 23:48:46 +0000

TL;DR

Check out FirestoreRestore.
It uses a brand new API to backup and restore your Firestore database!

Background

There have been many requests for Google to add backup and restore functionality to their Cloud Firestore offering. Until this week, the only method for backing up a Firestore database has been to run utilities that traverse the tree as an admin client, saving each document's JSON representation on disk.

Discovery

This morning, I opened the Google APIs Explorer and found two new methods listed under Firestore v1beta1: exportDocuments and importDocuments.

Exploration

As the API documentation states, firestore.projects.databases.exportDocuments is able to export Firestore data to Google Cloud Storage.

To test this out, I opened up the Firebase Console and created a new project. In order for there to be data to back up, I created a user document from the console.

In the same project, I opened the Google Cloud Storage Console and created a bucket in which to store my backups.

From the Google APIs Explorer's exportDocuments page, I entered the Firebase project name, as well as a path to the storage bucket, and hit Execute.

The following is the API's response:

{
 "name": "projects/firestore-restore/databases/(default)/operations/<REDACTED>",
 "metadata": {
  "@type": "type.googleapis.com/google.firestore.admin.v1beta1.ExportDocumentsMetadata",
  "startTime": "2018-07-18T23:20:26.535130Z",
  "operationState": "PROCESSING",
  "outputUriPrefix": "gs://firestore-restore/backups/2018-07-18"
 }
}

A few moments later, I opened the bucket. Sure enough, the data was present!

Restore

Now, to test the restore functionality. I deleted the contents of the Firestore database, and ran importDocuments with the same parameters as the export request, and sure enough, the data came back!

Is the Backup Complete?

Unfortunately, there is currently no "documented" way to check the status of a Firestore export in progress. Luckily, with a little URL guessing, I was able to find the endpoint for the operation resource. It was there all along; the name field in the export response is the status path!

A Temporary Tool: FirestoreRestore

As I don't know when Google will officially support and build tooling for backing up and restoring Cloud Firestore databases, I developed a simple command line utility to interact with the service.

It requires the creation of a service account with the "Cloud Datastore Import Export Admin" role. With a JSON key for the service account, this tool is able to perform a backup and restore.

Check out the README for more info 😀

Speculation

Google will likely launch backup and restore functionality within the Firebase UI. It'll probably expose the per-collection backup capability that is present in the API. It'll probably be much easier to use than some command line program you found on the internet.

Until then, enjoy the CLI!

DEV Community: Christian Alexander

Async Elixir Telemetry

Intent: Capture Product Analytics

Sample Code, Synchronous Approach

PostHog Client Module

The Synchronous Telemetry Handler

Sending an Event

Reflecting on the Approach

Making it Better: Async

Outcome

Random Cut Forests

How RCF Works

Making a tree

The Forest

Evaluation

Sine Wave Anomaly Exercise

Shingling the Data

Final Results

Summary

Citations

Mitigating Server-Side Request Forgery

The attack

Cloud credential leak

Internal endpoints

Filesystem access

Mitigating the vulnerability

Preventing non-HTTP(S) URIs

Preventing the instance metadata IP address

Looking up DNS hostnames

Perform validation on every request

Using SafeURL

Similar libraries

Taking it further

Don't follow redirects

Security groups

Proxy

Wrapping up

Efficient bidirectional infinite scroll in Phoenix LiveView

Context

Setup

Attempt 1: The Memory Hog

Attempt 2: Out of Order

Attempt 3: It Works Again

Bonus: Automatically Load More

Conclusion

Adding soft delete to a Phoenix Commanded (CQRS) API

Goal

Context

Updating the aggregate

Aside: aggregates sound slow

Lifespans

Snapshots

Effective dating the delete event

Read your writes

What about existing events?

Updating the read model

Preparing the database

Updating the projector

Forcefully re-populating the database

Hiding deleted items

Where are we, now?

New feature: restore deleted items

Implementation

Create a RestoreTodo command

Create a TodoRestored event

Update the aggregate

Route the command to the aggregate

Update the projector

Adding the API method

Add a method to the context module

Add a method to the controller

Register the route

Validation

Summary

Using CQRS in a simple Phoenix API with Commanded

Context

What did I do?

Commanded

Initializing

First command: create todo

Create a `RestoreTodo` command

Create a `TodoRestored` event

Add the `execute` and `apply` methods to the aggregate