abstraction over pubsub to include OpenTelemetry context

lib/acme/orders.ex

@@ -3,8 +3,6 @@ defmodule Acme.Orders do
   The Orders context.
   """
 
-  require OpenTelemetry.Tracer
-
   import Ecto.Query, warn: false
   alias Acme.Repo
 
@@ -70,15 +68,12 @@ defmodule Acme.Orders do
 
   """
   def update_order(%Order{} = order, attrs) do
-    OpenTelemetry.Tracer.with_span "orders:update_order" do
-      order_changeset = Order.changeset(order, attrs)
+    order_changeset = Order.changeset(order, attrs)
 
-      with {:ok, order} <- Repo.update(order_changeset) do
-        ctx = OpenTelemetry.Tracer.current_span_ctx()
-        Phoenix.PubSub.broadcast(Acme.PubSub, "orders:#{order.id}", {:order_updated, order, ctx})
+    with {:ok, order} <- Repo.update(order_changeset) do
+      Acme.PubSub.broadcast("orders:#{order.id}", {:order_updated, order})
 
-        {:ok, order}
-      end
+      {:ok, order}
     end
   end
 

lib/acme/pub_sub.ex

@@ -0,0 +1,23 @@
+defmodule Acme.PubSub do
+  defmodule Event do
+    defstruct [:message, :span_ctx]
+  end
+
+  defmacro __using__(_opts) do
+    quote do
+      def handle_info(%Acme.PubSub.Event{} = event, socket) do
+        OpenTelemetry.Tracer.set_current_span(event.span_ctx)
+        handle_info(event.message, socket)
+      end
+    end
+  end
+
+  def broadcast(topic, message) do
+    require OpenTelemetry.Tracer
+
+    OpenTelemetry.Tracer.with_span "acme.pubsub:broadcast" do
+      event = %Event{message: message, span_ctx: OpenTelemetry.Tracer.current_span_ctx()}
+      Phoenix.PubSub.broadcast(__MODULE__, topic, event)
+    end
+  end
+end

lib/acme_web/live/order_live/show.ex

@@ -1,5 +1,6 @@
 defmodule AcmeWeb.OrderLive.Show do
   use AcmeWeb, :live_view
+  use Acme.PubSub
 
   require OpenTelemetry.Tracer
 
@@ -21,17 +22,12 @@ defmodule AcmeWeb.OrderLive.Show do
   end
 
   @impl true
-  def handle_info({:order_updated, order, ctx}, socket) do
-    OpenTelemetry.Tracer.set_current_span(ctx)
+  def handle_info({:order_updated, order}, socket) do
+    # expensive operation like DB call, service call.. etc.
+    # for the example we'll do a function call that results in DB query
+    order = Orders.get_order!(order.id)
 
-    opts = %{attributes: %{user: inspect(self())}}
-
-    OpenTelemetry.Tracer.with_span "order_live.show:order_updated", opts do
-      # expensive operation like DB call, service call.. etc.
-      Process.sleep(70)
-
-      {:noreply, assign(socket, :order, order)}
-    end
+    {:noreply, assign(socket, :order, order)}
   end
 
   defp page_title(:show), do: "Show Order"