defmodule ExESDBDashboard.ClusterLive do @moduledoc """ Main cluster dashboard LiveView component. Displays real-time cluster information including: - Connected nodes and their status - Available stores and statistics - Overall cluster health - Real-time updates via PubSub """ use Phoenix.LiveView alias ExESDBDashboard alias ExESDBGater.Messages.{HealthMessages, LifecycleMessages, SystemMessages} @impl true def mount(_params, _session, socket) do # Subscribe to structured message topics if connected?(socket) do # Subscribe to health updates Phoenix.PubSub.subscribe(:ex_esdb_health, "cluster_health") Phoenix.PubSub.subscribe(:ex_esdb_health, "node_health") # Subscribe to lifecycle events Phoenix.PubSub.subscribe(:ex_esdb_lifecycle, "cluster_membership") Phoenix.PubSub.subscribe(:ex_esdb_lifecycle, "node_lifecycle") # Subscribe to system events Phoenix.PubSub.subscribe(:ex_esdb_system, "lifecycle") # Set up periodic refresh as fallback :timer.send_interval(30_000, self(), :refresh_data) end # Load initial cluster data cluster_data = ExESDBDashboard.get_cluster_data() socket = socket |> assign(:cluster_data, cluster_data) |> assign(:loading, false) |> assign(:last_updated, DateTime.utc_now()) |> assign(:page_title, "Cluster Dashboard") {:ok, socket} end @impl true def handle_params(params, _url, socket) do action = Map.get(params, "live_action", :cluster) {:noreply, assign(socket, :live_action, action)} end @impl true def handle_info(:refresh_data, socket) do cluster_data = ExESDBDashboard.get_cluster_data() socket = socket |> assign(:cluster_data, cluster_data) |> assign(:last_updated, DateTime.utc_now()) {:noreply, socket} end # Handle structured health messages @impl true def handle_info({:secure_message, _signature, {:cluster_health_updated, _payload}} = message, socket) do case HealthMessages.validate_secure_message(message) do {:ok, _} -> handle_cluster_update(socket) {:error, _} -> {:noreply, socket} end end @impl true def handle_info({:secure_message, _signature, {:node_health_updated, _payload}} = message, socket) do case HealthMessages.validate_secure_message(message) do {:ok, _} -> handle_cluster_update(socket) {:error, _} -> {:noreply, socket} end end # Handle structured lifecycle messages @impl true def handle_info({:secure_message, _signature, {:cluster_membership_changed, _payload}} = message, socket) do case LifecycleMessages.validate_secure_message(message) do {:ok, _} -> handle_cluster_update(socket) {:error, _} -> {:noreply, socket} end end @impl true def handle_info({:secure_message, _signature, {:node_lifecycle_event, _payload}} = message, socket) do case LifecycleMessages.validate_secure_message(message) do {:ok, _} -> handle_cluster_update(socket) {:error, _} -> {:noreply, socket} end end # Handle structured system messages @impl true def handle_info({:secure_message, _signature, {:system_lifecycle_event, _payload}} = message, socket) do case SystemMessages.validate_secure_message(message) do {:ok, _} -> handle_cluster_update(socket) {:error, _} -> {:noreply, socket} end end # Handle unsecured messages (dev/test environments) @impl true def handle_info({:unsecured_message, {:cluster_health_updated, _payload}} = message, socket) do case HealthMessages.validate_secure_message(message) do {:ok, _} -> handle_cluster_update(socket) {:error, _} -> {:noreply, socket} end end @impl true def handle_info({:unsecured_message, {:node_health_updated, _payload}} = message, socket) do case HealthMessages.validate_secure_message(message) do {:ok, _} -> handle_cluster_update(socket) {:error, _} -> {:noreply, socket} end end @impl true def handle_info({:unsecured_message, {:cluster_membership_changed, _payload}} = message, socket) do case LifecycleMessages.validate_secure_message(message) do {:ok, _} -> handle_cluster_update(socket) {:error, _} -> {:noreply, socket} end end @impl true def handle_info({:unsecured_message, {:node_lifecycle_event, _payload}} = message, socket) do case LifecycleMessages.validate_secure_message(message) do {:ok, _} -> handle_cluster_update(socket) {:error, _} -> {:noreply, socket} end end @impl true def handle_info({:unsecured_message, {:system_lifecycle_event, _payload}} = message, socket) do case SystemMessages.validate_secure_message(message) do {:ok, _} -> handle_cluster_update(socket) {:error, _} -> {:noreply, socket} end end # Handle legacy messages during transition period @impl true def handle_info({:cluster_state_changed, _new_state}, socket) do handle_cluster_update(socket) end # Ignore unknown messages @impl true def handle_info(_msg, socket) do {:noreply, socket} end # Common cluster update handler defp handle_cluster_update(socket) do cluster_data = ExESDBDashboard.get_cluster_data() socket = socket |> assign(:cluster_data, cluster_data) |> assign(:last_updated, DateTime.utc_now()) {:noreply, socket} end @impl true def handle_event("refresh", _params, socket) do cluster_data = ExESDBDashboard.get_cluster_data() socket = socket |> assign(:cluster_data, cluster_data) |> assign(:last_updated, DateTime.utc_now()) |> put_flash(:info, "Cluster data refreshed") {:noreply, socket} end @impl true def render(assigns) do ~H"""
<.header> <:title>ExESDB Cluster Dashboard <:subtitle>Real-time cluster monitoring and management <:actions>
<.cluster_health_card cluster_health={@cluster_data.cluster_health} /> <.stats_cards nodes={@cluster_data.nodes} stores={@cluster_data.stores} total_streams={@cluster_data.total_streams} /> <.nodes_section nodes={@cluster_data.nodes} /> <.stores_section stores={@cluster_data.stores} />
""" end # Component functions defp header(assigns) do ~H"""

<%= render_slot(@title) %>

<%= render_slot(@subtitle) %>

<%= render_slot(@actions) %>
""" end defp cluster_health_card(assigns) do health_class = health_to_class(assigns.cluster_health) health_text = health_to_text(assigns.cluster_health) health_emoji = health_to_emoji(assigns.cluster_health) assigns = assign(assigns, :health_class, health_class) assigns = assign(assigns, :health_text, health_text) assigns = assign(assigns, :health_emoji, health_emoji) ~H"""

Cluster Health

<%= @health_emoji %> <%= @health_text %>
""" end defp stats_cards(assigns) do ex_esdb_nodes = Enum.count(assigns.nodes, & &1.is_ex_esdb) total_nodes = length(assigns.nodes) total_stores = length(assigns.stores) assigns = assign(assigns, :ex_esdb_nodes, ex_esdb_nodes) assigns = assign(assigns, :total_nodes, total_nodes) assigns = assign(assigns, :total_stores, total_stores) ~H"""
<%= @total_nodes %>
Total Nodes
<%= @ex_esdb_nodes %>
ExESDB Nodes
<%= @total_stores %>
Stores
<%= @total_streams %>
Total Streams
""" end defp nodes_section(assigns) do ~H"""

Cluster Nodes

<%= length(@nodes) %> nodes
<.node_card :for={node <- @nodes} node={node} />
""" end defp node_card(assigns) do node_class = node_status_class(assigns.node.status, assigns.node.is_ex_esdb) node_emoji = node_status_emoji(assigns.node.status, assigns.node.is_ex_esdb) uptime_text = format_uptime(assigns.node.uptime) assigns = assign(assigns, :node_class, node_class) assigns = assign(assigns, :node_emoji, node_emoji) assigns = assign(assigns, :uptime_text, uptime_text) ~H"""
<%= @node_emoji %> <%= @node.name %>
<%= if @node.is_ex_esdb, do: "ExESDB Node", else: "Gater Node" %>
Uptime: <%= @uptime_text %>
""" end defp stores_section(assigns) do ~H"""

Event Stores

<%= length(@stores) %> stores
<%= if length(@stores) > 0 do %>
<.store_card :for={store <- @stores} store={store} />
<% else %>

No stores available. Ensure ExESDB nodes are connected and running.

<% end %>
""" end defp store_card(assigns) do ~H"""

<%= @store.name %>

<%= if @store.status == :healthy, do: "✅", else: "⚠️" %>
<%= @store.stream_count %> Streams
<%= @store.subscription_count %> Subscriptions
<%= length(@store.nodes) %> Nodes
""" end # Helper functions defp health_to_class(:healthy), do: "healthy" defp health_to_class(:degraded), do: "degraded" defp health_to_class(:unhealthy), do: "unhealthy" defp health_to_class(:no_cluster), do: "no-cluster" defp health_to_class(:no_stores), do: "no-stores" defp health_to_class(_), do: "unknown" defp health_to_text(:healthy), do: "Healthy" defp health_to_text(:degraded), do: "Degraded" defp health_to_text(:unhealthy), do: "Unhealthy" defp health_to_text(:no_cluster), do: "No Cluster" defp health_to_text(:no_stores), do: "No Stores" defp health_to_text(_), do: "Unknown" defp health_to_emoji(:healthy), do: "✅" defp health_to_emoji(:degraded), do: "⚠️" defp health_to_emoji(:unhealthy), do: "❌" defp health_to_emoji(:no_cluster), do: "🚫" defp health_to_emoji(:no_stores), do: "📭" defp health_to_emoji(_), do: "❓" defp node_status_class(:self, _), do: "self-node" defp node_status_class(:connected, true), do: "ex-esdb-node connected" defp node_status_class(:connected, false), do: "gater-node connected" defp node_status_class(_, _), do: "disconnected" defp node_status_emoji(:self, _), do: "🏠" defp node_status_emoji(:connected, true), do: "🗄️" defp node_status_emoji(:connected, false), do: "🌐" defp node_status_emoji(_, _), do: "❌" defp format_uptime(uptime_ms) when is_integer(uptime_ms) and uptime_ms > 0 do seconds = div(uptime_ms, 1000) minutes = div(seconds, 60) hours = div(minutes, 60) days = div(hours, 24) cond do days > 0 -> "#{days}d #{rem(hours, 24)}h" hours > 0 -> "#{hours}h #{rem(minutes, 60)}m" minutes > 0 -> "#{minutes}m #{rem(seconds, 60)}s" true -> "#{seconds}s" end end defp format_uptime(_), do: "Unknown" end