From f7b5e63429c9e596a315fc19c417663fe375ee3c Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Mon, 9 Feb 2026 12:58:54 -0600 Subject: [PATCH] live update agent status --- lib/towerops_web/channels/agent_channel.ex | 31 ++++++++++++++ lib/towerops_web/live/agent_live/index.ex | 49 ++++++++++++++++++++++ lib/towerops_web/live/device_live/show.ex | 43 +++++++++++++++++++ 3 files changed, 123 insertions(+) diff --git a/lib/towerops_web/channels/agent_channel.ex b/lib/towerops_web/channels/agent_channel.ex index 08676257..40cd3bd9 100644 --- a/lib/towerops_web/channels/agent_channel.ex +++ b/lib/towerops_web/channels/agent_channel.ex @@ -89,6 +89,14 @@ defmodule ToweropsWeb.AgentChannel do remote_ip = get_remote_ip(socket) _ = Agents.update_agent_token_heartbeat(agent_token.id, remote_ip, %{}) + # Broadcast agent connection for real-time UI updates + _ = + Phoenix.PubSub.broadcast( + Towerops.PubSub, + "agents:health", + {:agent_connected, agent_token.id, agent_token.organization_id} + ) + # Send initial job list send(self(), :send_jobs) @@ -103,6 +111,21 @@ defmodule ToweropsWeb.AgentChannel do {:error, %{reason: "missing token"}} end + @impl true + def terminate(_reason, socket) do + # Broadcast agent disconnection for real-time UI updates + if Map.has_key?(socket.assigns, :agent_token_id) do + _ = + Phoenix.PubSub.broadcast( + Towerops.PubSub, + "agents:health", + {:agent_disconnected, socket.assigns.agent_token_id, socket.assigns.organization_id} + ) + end + + :ok + end + @impl true @spec handle_info(:send_jobs, Phoenix.Socket.t()) :: {:noreply, Phoenix.Socket.t()} def handle_info(:send_jobs, socket) do @@ -292,6 +315,14 @@ defmodule ToweropsWeb.AgentChannel do metadata ) + # Broadcast heartbeat for real-time UI updates (especially important for stale agents coming back online) + _ = + Phoenix.PubSub.broadcast( + Towerops.PubSub, + "agents:health", + {:agent_heartbeat, socket.assigns.agent_token_id, socket.assigns.organization_id} + ) + {:noreply, socket} else {:error, {type, message}} -> diff --git a/lib/towerops_web/live/agent_live/index.ex b/lib/towerops_web/live/agent_live/index.ex index 0937648a..85fb2f3b 100644 --- a/lib/towerops_web/live/agent_live/index.ex +++ b/lib/towerops_web/live/agent_live/index.ex @@ -18,6 +18,11 @@ defmodule ToweropsWeb.AgentLive.Index do is_superuser = Scope.superuser?(current_scope) agent_tokens = Agents.list_organization_agent_tokens(organization.id) + # Subscribe to agent health updates for real-time status changes + if connected?(socket) do + _ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agents:health") + end + # If superadmin (including when impersonating), also load cloud pollers and global default {cloud_pollers, global_default_cloud_poller_id} = if is_superuser do @@ -347,4 +352,48 @@ defmodule ToweropsWeb.AgentLive.Index do |> assign(:selected_global_default, agent_token_id || "") |> put_flash(:info, message)} end + + # Handle real-time agent status updates + @impl true + def handle_info({:agent_connected, _agent_token_id, _organization_id}, socket) do + {:noreply, reload_agent_data(socket)} + end + + @impl true + def handle_info({:agent_disconnected, _agent_token_id, _organization_id}, socket) do + {:noreply, reload_agent_data(socket)} + end + + @impl true + def handle_info({:agent_heartbeat, _agent_token_id, _organization_id}, socket) do + {:noreply, reload_agent_data(socket)} + end + + @impl true + def handle_info({:agents_stale, _stale_agents}, socket) do + {:noreply, reload_agent_data(socket)} + end + + defp reload_agent_data(socket) do + organization = socket.assigns.current_scope.organization + agent_tokens = Agents.list_organization_agent_tokens(organization.id) + + cloud_pollers = load_cloud_pollers_if_superuser(socket.assigns.current_scope) + + equipment_counts = calculate_device_counts(agent_tokens) + cloud_poller_counts = calculate_device_counts(cloud_pollers) + + agent_health_stats = Stats.get_organization_agent_health(organization.id) + assignment_breakdown = Stats.get_device_assignment_breakdown(organization.id) + offline_agents = Stats.get_offline_agents(organization.id) + + socket + |> assign(:agent_tokens, agent_tokens) + |> assign(:cloud_pollers, cloud_pollers) + |> assign(:device_counts, equipment_counts) + |> assign(:cloud_poller_counts, cloud_poller_counts) + |> assign(:agent_health_stats, agent_health_stats) + |> assign(:assignment_breakdown, assignment_breakdown) + |> assign(:offline_agents, offline_agents) + end end diff --git a/lib/towerops_web/live/device_live/show.ex b/lib/towerops_web/live/device_live/show.ex index a8c35205..3ed9c5a8 100644 --- a/lib/towerops_web/live/device_live/show.ex +++ b/lib/towerops_web/live/device_live/show.ex @@ -165,12 +165,55 @@ defmodule ToweropsWeb.DeviceLive.Show do {:noreply, load_equipment_data(socket, socket.assigns.device.id)} end + @impl true + def handle_info({:agent_connected, agent_token_id, _organization_id}, socket) do + # Reload if this device's agent just connected + if socket.assigns.agent_info.agent_token_id == agent_token_id do + {:noreply, load_equipment_data(socket, socket.assigns.device.id)} + else + {:noreply, socket} + end + end + + @impl true + def handle_info({:agent_disconnected, agent_token_id, _organization_id}, socket) do + # Reload if this device's agent just disconnected + if socket.assigns.agent_info.agent_token_id == agent_token_id do + {:noreply, load_equipment_data(socket, socket.assigns.device.id)} + else + {:noreply, socket} + end + end + + @impl true + def handle_info({:agent_heartbeat, agent_token_id, _organization_id}, socket) do + # Reload if this device's agent sent a heartbeat (important for stale agents coming back) + if socket.assigns.agent_info.agent_token_id == agent_token_id do + {:noreply, load_equipment_data(socket, socket.assigns.device.id)} + else + {:noreply, socket} + end + end + + @impl true + def handle_info({:agents_stale, stale_agents}, socket) do + # Reload if this device's agent went stale + device_agent_id = socket.assigns.agent_info.agent_token_id + + if device_agent_id && Enum.any?(stale_agents, &(&1.id == device_agent_id)) do + {:noreply, load_equipment_data(socket, socket.assigns.device.id)} + else + {:noreply, socket} + end + end + # Private functions defp maybe_subscribe_and_schedule_refresh(socket, device_id) do _ = if connected?(socket) do _ = Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device_id}") + _ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agents:health") Process.send_after(self(), :refresh_data, 10_000) end