LiveView Dashboard

http://localhost:4000/dashboard

http://localhost:4000/dashboard/{profile}

PubSub Events → EventStream (per profile)
                      ↓
            Batch aggregation (175ms)
                      ↓
            {:dashboard_batch, ...}
                      ↓
                 LiveViews

%{
  health: %{},                    # Provider health counters
  circuit_states: %{},            # Latest circuit states
  block_states: %{},              # Block heights and lag
  cluster: %{},                   # Cluster topology
  metrics: %{},                   # Per-provider RPS/latency/success
  routing_events: [],             # RPC routing decisions
  circuit_events: [],             # Circuit transitions
  provider_events: [],            # Provider health events
  subscription_events: [],        # WebSocket subscription events
  block_events: [],               # New block notifications
  heartbeat: false                # Staleness prevention
}

{:dashboard_snapshot, %{
  metrics: provider_metrics,
  circuits: circuit_states,
  health_counters: health_counters,
  cluster: cluster_state,
  events: recent_events
}}

@cache_ttl_ms 15_000           # 15-second cache lifetime
@refresh_timeout_ms 5_000      # RPC timeout per node

# Provider leaderboard (all nodes)
MetricsStore.get_provider_leaderboard("default", "ethereum")
# => %{
#   data: [
#     %{
#       provider_id: "ethereum_llamarpc",
#       score: 0.95,
#       avg_latency_ms: 120,
#       success_rate: 0.99,
#       total_calls: 1500,
#       node_count: 3,
#       latency_by_node: %{
#         "us-east-1" => %{p50: 100, p95: 180, p99: 250},
#         "eu-west-1" => %{p50: 140, p95: 220, p99: 300}
#       }
#     }
#   ],
#   coverage: %{responding: 3, total: 3},
#   stale: false
# }

# Real-time stats (all nodes)
MetricsStore.get_realtime_stats("default", "ethereum")
# => %{
#   data: %{
#     rpc_methods: ["eth_blockNumber", "eth_getLogs"],
#     providers: ["ethereum_llamarpc", "ethereum_alchemy"],
#     total_entries: 5000,
#     node_count: 3
#   },
#   coverage: %{responding: 3, total: 3},
#   stale: false
# }

# Weighted average latency
weighted_avg = 
  Enum.map(entries, fn entry ->
    entry.avg_latency_ms * entry.total_calls
  end)
  |> Enum.sum()
  |> safe_divide(total_calls_across_all_nodes)

Phoenix.PubSub.subscribe(Lasso.PubSub, "cluster:topology")

def handle_info({:topology_event, %{event: event}}, state)
    when event in [:node_connected, :node_disconnected] do
  :ets.delete_all_objects(@cache_table)
  {:noreply, %{state | generation: state.generation + 1}}
end

%{
  provider_id: "ethereum_llamarpc",
  rps: 15.3,                    # Requests per second (rolling 60s)
  avg_latency_ms: 125,          # Weighted average
  success_rate: 0.98,           # Success / total
  total_calls: 920,             # Rolling window
  failovers: 3,                 # Failover count
  last_request_ms: 1234567890   # Timestamp
}

┌─────────────────────────┐
│  Cluster: 3/3 nodes     │  ← All nodes responding
│  Updated 2s ago         │
└─────────────────────────┘

# lib/lasso_web/network_topology.ex
TopologyConfig.canvas_center()  # => {960, 540}

// Client-side hook: ProviderRequestAnimator
window.addEventListener('routing_decision', (event) => {
  animateRequest(event.detail.chain, event.detail.provider_id);
});

@batch_interval_ms 175
@max_batch_size 100

def handle_cast({:subscribe, pid, account_id}, state) do
  subscribers = MapSet.put(state.subscribers, pid)
  Process.monitor(pid)
  
  # Start ticking if first subscriber
  state = if MapSet.size(subscribers) == 1 do
    schedule_tick(state)
  else
    state
  end
  
  send(pid, {:dashboard_snapshot, build_snapshot(state)})
  {:noreply, %{state | subscribers: subscribers}}
end

@seen_request_id_ttl_ms 120_000
@max_seen_request_ids 50_000

# O(1) dedup check before processing
if Map.has_key?(state.seen_request_ids, event.request_id) do
  {:noreply, state}
else
  # Process event...
end

# config/config.exs
config :lasso_web, LassoWeb.Dashboard,
  event_history_size: 500,        # Max events in memory
  recent_blocks_limit: 100,       # Max block events
  provider_events_limit: 200      # Max provider events

config :lasso_web, LassoWeb.Dashboard.EventStream,
  batch_interval_ms: 175,         # Flush frequency
  max_batch_size: 100,            # Immediate flush threshold
  window_duration_ms: 60_000,     # Metrics rolling window
  idle_timeout_ms: 30_000         # Terminate when no subscribers

config :lasso_web, LassoWeb.Dashboard.MetricsStore,
  cache_ttl_ms: 15_000,           # Cache lifetime
  refresh_timeout_ms: 5_000,      # RPC timeout
  min_calls_threshold: 10         # Minimum calls for aggregation

iex> Registry.lookup(Lasso.Dashboard.StreamRegistry, {:stream, "default"})
[{#PID<0.1234.0>, nil}]  # Should return a PID

iex> Phoenix.PubSub.subscribers(Lasso.PubSub, "routing_decision:default")
[#PID<0.1234.0>]  # EventStream should be subscribed

iex> Lasso.Cluster.Topology.get_responding_nodes()
[:node2@host, :node3@host]  # Should list remote nodes

iex> Lasso.Cluster.Topology.get_coverage()
%{connected: 3, responding: 3}  # All nodes healthy

iex> :rpc.multicall([node()], Lasso.Benchmarking.BenchmarkStore, :get_provider_leaderboard, ["default", "ethereum"], 5000)
{[results], []}  # Empty bad_nodes list = success

iex> :sys.get_state(pid)
%EventStream{
  seen_request_ids: %{...},  # Should be < 50,000 entries
  circuit_states: %{...},    # Should be < 500 entries
  block_heights: %{...}      # Should be < 500 entries
}