Add real-time updates to admin agents page via PubSub

Broadcast agent lifecycle events (created/deleted) on a separate
"admin:agents" topic so the admin page updates dynamically when
agents are added or removed from any organization.

lib/towerops/agents.ex

@@ -35,8 +35,12 @@ defmodule Towerops.Agents do
     {token, changeset} = AgentToken.build_token(organization_id, name)
 
     case Repo.insert(changeset) do
-      {:ok, agent_token} -> {:ok, agent_token, token}
-      {:error, changeset} -> {:error, changeset}
+      {:ok, agent_token} ->
+        broadcast_agent_change(:agent_created, agent_token)
+        {:ok, agent_token, token}
+
+      {:error, changeset} ->
+        {:error, changeset}
     end
   end
 
@@ -60,8 +64,12 @@ defmodule Towerops.Agents do
     {token, changeset} = AgentToken.build_token(nil, name, is_cloud_poller: true)
 
     case Repo.insert(changeset) do
-      {:ok, agent_token} -> {:ok, agent_token, token}
-      {:error, changeset} -> {:error, changeset}
+      {:ok, agent_token} ->
+        broadcast_agent_change(:agent_created, agent_token)
+        {:ok, agent_token, token}
+
+      {:error, changeset} ->
+        {:error, changeset}
     end
   end
 
@@ -333,23 +341,35 @@ defmodule Towerops.Agents do
       :token_disabled
     )
 
-    Repo.transaction(fn ->
-      agent_token = Repo.get!(AgentToken, id)
+    result =
+      Repo.transaction(fn ->
+        agent_token = Repo.get!(AgentToken, id)
 
-      # 1. Remove all direct device assignments
-      Repo.delete_all(from(a in AgentAssignment, where: a.agent_token_id == ^id))
+        # 1. Remove all direct device assignments
+        Repo.delete_all(from(a in AgentAssignment, where: a.agent_token_id == ^id))
 
-      # 2. Remove this agent from any sites using it as default
-      Repo.update_all(from(s in Towerops.Sites.Site, where: s.agent_token_id == ^id), set: [agent_token_id: nil])
+        # 2. Remove this agent from any sites using it as default
+        Repo.update_all(from(s in Towerops.Sites.Site, where: s.agent_token_id == ^id), set: [agent_token_id: nil])
 
-      # 3. Remove this agent from any organizations using it as default
-      Repo.update_all(from(o in Towerops.Organizations.Organization, where: o.default_agent_token_id == ^id),
-        set: [default_agent_token_id: nil]
-      )
+        # 3. Remove this agent from any organizations using it as default
+        Repo.update_all(from(o in Towerops.Organizations.Organization, where: o.default_agent_token_id == ^id),
+          set: [default_agent_token_id: nil]
+        )
 
-      # 4. Delete the agent token
-      Repo.delete!(agent_token)
-    end)
+        # 4. Delete the agent token
+        Repo.delete!(agent_token)
+
+        agent_token
+      end)
+
+    case result do
+      {:ok, agent_token} ->
+        broadcast_agent_change(:agent_deleted, agent_token)
+        {:ok, agent_token}
+
+      error ->
+        error
+    end
   end
 
   ## Assignment management
@@ -793,4 +813,19 @@ defmodule Towerops.Agents do
   end
 
   def broadcast_assignment_change(nil, _event), do: :ok
+
+  @doc """
+  Broadcasts agent lifecycle changes (created/deleted) to the admin agents topic.
+
+  Only the admin agents page subscribes to this topic, keeping these events
+  separate from the org-scoped "agents:health" topic.
+  """
+  @spec broadcast_agent_change(atom(), AgentToken.t()) :: :ok
+  def broadcast_agent_change(event, %AgentToken{} = agent_token) when event in [:agent_created, :agent_deleted] do
+    Phoenix.PubSub.broadcast(
+      Towerops.PubSub,
+      "admin:agents",
+      {event, agent_token.id, agent_token.is_cloud_poller}
+    )
+  end
 end

lib/towerops_web/live/admin/agent_live/index.ex

@@ -10,6 +10,7 @@ defmodule ToweropsWeb.Admin.AgentLive.Index do
   def mount(_params, _session, socket) do
     if connected?(socket) do
       Phoenix.PubSub.subscribe(Towerops.PubSub, "agents:health")
+      Phoenix.PubSub.subscribe(Towerops.PubSub, "admin:agents")
       :timer.send_interval(1000, :tick)
     end
 
@@ -54,16 +55,30 @@ defmodule ToweropsWeb.Admin.AgentLive.Index do
     {:noreply, reload_all(socket)}
   end
 
+  @impl true
+  def handle_info({:agent_created, agent_token_id, _is_cloud_poller}, socket) do
+    {:noreply, refresh_agent(socket, agent_token_id)}
+  end
+
+  @impl true
+  def handle_info({:agent_deleted, _agent_token_id, _is_cloud_poller}, socket) do
+    {:noreply, reload_all(socket)}
+  end
+
   defp refresh_agent(socket, agent_token_id) do
     agent_token = agent_token_id |> Agents.get_agent_token!() |> Towerops.Repo.preload(:organization)
 
     direct = Agents.count_assigned_devices(agent_token_id)
     total = length(Agents.list_agent_polling_targets(agent_token_id))
 
-    stream_name = if agent_token.is_cloud_poller, do: :cloud_pollers, else: :org_agents
+    {stream_name, section_assign} =
+      if agent_token.is_cloud_poller,
+        do: {:cloud_pollers, :has_cloud_pollers},
+        else: {:org_agents, :has_org_agents}
 
     socket
     |> stream_insert(stream_name, agent_token)
+    |> assign(section_assign, true)
     |> assign(:device_counts, Map.put(socket.assigns.device_counts, agent_token_id, %{direct: direct, total: total}))
   rescue
     Ecto.NoResultsError -> socket

test/towerops/agents_test.exs

@@ -1628,6 +1628,37 @@ defmodule Towerops.AgentsTest do
     end
   end
 
+  describe "broadcast_agent_change/2" do
+    test "broadcasts agent_created on admin:agents topic", %{organization: org} do
+      Phoenix.PubSub.subscribe(Towerops.PubSub, "admin:agents")
+
+      {:ok, agent_token, _token} = Agents.create_agent_token(org.id, "Broadcast Test Agent")
+
+      assert_receive {:agent_created, agent_token_id, false}
+      assert agent_token_id == agent_token.id
+    end
+
+    test "broadcasts agent_created for cloud pollers", %{} do
+      Phoenix.PubSub.subscribe(Towerops.PubSub, "admin:agents")
+
+      {:ok, cloud_poller, _token} = Agents.create_cloud_poller("Broadcast Cloud Poller")
+
+      assert_receive {:agent_created, agent_token_id, true}
+      assert agent_token_id == cloud_poller.id
+    end
+
+    test "broadcasts agent_deleted when agent is removed", %{organization: org} do
+      {:ok, agent_token, _token} = Agents.create_agent_token(org.id, "Delete Broadcast Agent")
+
+      Phoenix.PubSub.subscribe(Towerops.PubSub, "admin:agents")
+
+      {:ok, _} = Agents.delete_agent_token(agent_token.id)
+
+      assert_receive {:agent_deleted, agent_token_id, false}
+      assert agent_token_id == agent_token.id
+    end
+  end
+
   describe "should_phoenix_poll_device?/1" do
     setup %{organization: org} do
       {:ok, site} =

test/towerops_web/live/admin/agent_live/index_test.exs

@@ -68,6 +68,43 @@ defmodule ToweropsWeb.Admin.AgentLive.IndexTest do
     end
   end
 
+  describe "real-time updates" do
+    setup [:register_and_log_in_superuser]
+
+    test "updates when a new agent is created", %{conn: conn, user: user} do
+      {:ok, view, _html} = live(conn, ~p"/admin/agents")
+
+      {:ok, org} = Towerops.Organizations.create_organization(%{name: "New Org"}, user.id)
+      {:ok, _agent, _} = Agents.create_agent_token(org.id, "Dynamic Agent")
+
+      html = render(view)
+      assert html =~ "Dynamic Agent"
+      assert html =~ "New Org"
+    end
+
+    test "updates when a new cloud poller is created", %{conn: conn} do
+      {:ok, view, _html} = live(conn, ~p"/admin/agents")
+
+      {:ok, _cloud, _} = Agents.create_cloud_poller("Dynamic Cloud Poller")
+
+      html = render(view)
+      assert html =~ "Dynamic Cloud Poller"
+    end
+
+    test "updates when an agent is deleted", %{conn: conn, user: user} do
+      {:ok, org} = Towerops.Organizations.create_organization(%{name: "Delete Org"}, user.id)
+      {:ok, agent, _} = Agents.create_agent_token(org.id, "Doomed Agent")
+
+      {:ok, view, html} = live(conn, ~p"/admin/agents")
+      assert html =~ "Doomed Agent"
+
+      {:ok, _} = Agents.delete_agent_token(agent.id)
+
+      html = render(view)
+      refute html =~ "Doomed Agent"
+    end
+  end
+
   describe "non-superuser access" do
     setup [:register_and_log_in_user]