Implementing a Phoenix PubSub Adapter with EventStore

Distributed systems need async message delivery across nodes. Phoenix provides Phoenix PubSub for this, with pluggable adapters for different backends — officially PG2 and Redis.

This post walks through implementing a Phoenix PubSub adapter backed by EventStore, an Elixir event sourcing library that persists events to PostgreSQL as an append-only log.

Using EventStore as a PubSub backend has a few advantages over the default PG2 adapter:

• No Erlang distribution required: nodes communicate through the shared database rather than through the Erlang cluster, so you can run multiple nodes without configuring Erlang node connectivity.
• Persistence: every broadcast is stored and can be replayed or audited later.

The tradeoffs are the need for storage and the additional latency of a database round-trip per broadcast, making it best suited for lower-throughput messaging where persistence and cross-node decoupling matter more than raw speed. This implementation is a proof of concept — no load tests were performed.

A full implementation of the adapter can be found on Github.

Phoenix.PubSub.Adapter in a nutshell

A Phoenix PubSub adapter must implement a few callbacks specified in Phoenix.PubSub.Adapter:

node_name(adapter_name)

Returns the node name as an atom or binary. Used mainly by Phoenix.Tracker. In most cases:

def node_name(nil), do: node()
def node_name(configured_name), do: configured_name

child_spec(keyword)

Generates the child spec for the adapter. GenServer provides a default; this rarely needs overriding.

broadcast(adapter_name, topic, message, dispatcher)

Called when a message is broadcast through Phoenix.PubSub.broadcast. The adapter_name is the PubSub name with .Adapter appended (e.g. MyApp.PubSub → MyApp.PubSub.Adapter). The dispatcher module handles local delivery via dispatch/3.

direct_broadcast(adapter_name, node_name, topic, message, dispatcher)

Same as broadcast/4 with an additional node_name — the message should only reach subscribers on that node.

The EventStore adapter

This section walks through a possible implementation of a Phoenix PubSub adapter that uses EventStore to distribute messages between nodes. This gives a solution that does not depend on Erlang/Elixir distribution, and an event log is stored in case further analysis is needed.

How Phoenix.PubSub works

Phoenix.PubSub uses Elixir’s Registry for subscriptions — each subscribe call registers an entry under the topic key. When broadcast is called, the framework invokes the adapter callback to distribute the message, then handles local dispatch.

The adapter’s job is to get the message to other nodes. For direct_broadcast, only subscribers on the target node should receive it.

The implementation

The adapter is a GenServer that joins the PubSub supervision tree. An eventstore option selects which EventStore module to use (in case you have multiple):

{Phoenix.PubSub,
  [name: MyApp.PubSub,
   adapter: Phoenix.PubSub.EventStore,
   eventstore: MyApp.EventStore]
}

The GenServer stores the EventStore module and the PubSub name in state — both are needed later:

defmodule Phoenix.PubSub.EventStore do
  @behaviour Phoenix.PubSub.Adapter
  use GenServer

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: opts[:adapter_name])
  end

  def init(opts) do
    {:ok,
     %{
       eventstore: opts[:eventstore],
       pubsub_name: opts[:name]
     }}
  end
  #... implementation will come here ...#
end

Note the difference between opts[:name] and opts[:adapter_name]. The former is the name of the PubSub as a whole and is reserved for the Registry. Publishers use it when broadcasting messages. opts[:adapter_name] can be used as the name of the GenServer.

Distributing a message as an event

The GenServer appends a new event to the EventStore when broadcast is called:

def broadcast(server, topic, message, dispatcher, metadata \\ %{}) do
  metadata = Map.put(metadata, :dispatcher, dispatcher)
  GenServer.call(server, {:broadcast, topic, message, metadata})
end

def handle_call(
      {:broadcast, topic, message, metadata},
      _from_pid,
      %{id: id, eventstore: eventstore, serializer: serializer, pubsub_name: pubsub_name} = state
    ) do
  event = %EventStore.EventData{
    # ... constructed below
  }

  res = eventstore.append_to_stream(topic, :any_version, [event])

  # For direct_broadcast targeting the current node, the framework does not
  # call local dispatch, so the adapter must do it. For regular broadcast,
  # the framework handles local dispatch after adapter.broadcast returns :ok.
  current_node = to_string(node())
  destination_node = Map.get(metadata, :destination_node)

  if destination_node == current_node do
    dispatcher = Map.get(metadata, :dispatcher, Phoenix.PubSub)
    Phoenix.PubSub.local_broadcast(pubsub_name, topic, message, dispatcher)
  end

  {:reply, res, state}
end

direct_broadcast/5 is a thin wrapper that sets destination_node in the metadata before delegating to broadcast/5:

def direct_broadcast(server, node_name, topic, message, dispatcher) do
  metadata = %{
    destination_node: to_string(node_name),
    source_node: to_string(node())
  }
  broadcast(server, topic, message, dispatcher, metadata)
end

source_node is stored in the event metadata for auditing. Routing is handled downstream by comparing destination_node against the current node.

The key decision is how to wrap the message inside %EventStore.EventData{}. Serialization is handled by a pluggable module (defaulting to Phoenix.PubSub.EventStore.Serializer.Base64) so the adapter is not tied to a specific encoding. The default serializer base64-encodes :erlang.term_to_binary/1 output — this is necessary because EventStore stores data as JSON and raw binaries would be invalid, and because JSON cannot distinguish atoms from strings so a round-trip through term serialization preserves type fidelity.

event = %EventStore.EventData{
  event_type: to_string(serializer),
  data: serializer.serialize(message)
}

A custom serializer can be provided via the serializer option as long as it implements serialize/1 and deserialize/1.

Handling events, local distribution

Now that events are in the event store, any subscribed process will receive them. The GenServer must subscribe to all topics ("$all"). If the event store is also used for another purpose, it’s best to have a separate one for PubSub. The subscription is set up via handle_continue/2, which runs immediately after init/1 completes, before any other messages can be processed.

def handle_continue(:subscribe, %{eventstore: eventstore} = state) do
  eventstore.subscribe("$all")

  {:noreply, state}
end

def handle_info({:subscribed, _subscription}, state), do: {:noreply, state}

A transient subscription is used since previous messages are not needed. The event store replies with a {:subscribed, subscription} message, which must also be handled. After this, the server will start receiving {:events, events} messages.

To avoid dispatching a local message twice (once from broadcast and once when the event arrives back from EventStore), a unique ID is added to the process state:

def init(opts) do
  {:ok,
   %{
     id: generate_unique_id(opts),
     eventstore: opts[:eventstore],
     pubsub_name: opts[:name],
     serializer: opts[:serializer] || Phoenix.PubSub.EventStore.Serializer.Base64
   }, {:continue, :subscribe}}
end

defp generate_unique_id(opts) do
  unique_id_fn = opts[:unique_id_fn] || fn _name -> UUID.uuid4() end
  unique_id_fn.(opts[:name])
end

A custom ID generator can be provided via unique_id_fn — a function that receives the PubSub name and returns a unique string. Useful when UUID is unavailable or when a deterministic ID is needed for testing.

The id is added to the event’s metadata field as source_id, keeping it separate from the message data. Serialization is delegated to the configurable serializer module. The handle_call for :broadcast becomes:

event = %EventStore.EventData{
  event_type: to_string(serializer),
  data: serializer.serialize(message),
  metadata: Map.put(metadata, :source_id, id)
}

Where the value of id and serializer come from the state, and metadata already contains dispatcher and any destination_node for direct broadcasts. When an event arrives back, source_id identifies the origin node so duplicates can be skipped:

def handle_info({:events, events}, state) do
  Enum.each(events, &local_broadcast_event(&1, state))

  {:noreply, state}
end

defp local_broadcast_event(
       %EventStore.RecordedEvent{
         data: data,
         metadata: metadata,
         stream_uuid: topic,
         eventbies_type: event_type
       },
       %{id: id, serializer: serializer, pubsub_name: pubsub_name} = _state
     ) do
  current_node = to_string(node())

  %{source_id: source_id, destination_node: destination_node, dispatcher: dispatcher} =
    convert_metadata_keys_to_atoms(metadata)

  is_destination? = is_nil(destination_node) or destination_node == current_node

  if not is_nil(dispatcher) and is_destination? and source_id != id and
       event_type == to_string(serializer) do
    Phoenix.PubSub.local_broadcast(
      pubsub_name,
      topic,
      serializer.deserialize(data),
      maybe_convert_to_existing_atom(dispatcher)
    )
  end
end

That’s it — a complete implementation of Phoenix PubSub using EventStore, including support for direct_broadcast via the destination_node metadata field and pluggable serialization.

The complete implementation can be found at esl/phoenix_pubsub_eventstore.
Need help building reliable distributed systems with Elixir? Get in touch with our team.