Fleet: Beszel hub integration — rich per-node metrics + per-node monitor (Phase 1)

Tap each node's Beszel metrics (GPU/temps/disk-IO/network/per-container — beyond
our basic heartbeat) by reading the workspace's Beszel hub. The agents run in
WS-only mode with no locally-readable socket, so (per the de-risk) the server taps
the hub's PocketBase API instead of the daemon reading agents — no daemon changes.

- migrations: workspace_beszel (BYO hub URL + login, server-side only, mirrors the
  Tailscale BYO pattern) + node_metrics (latest scalar columns + JSONB blob).
- cm-db: repo/fleet_beszel.rs, repo/node_metrics.rs; nodes SELECT joins node_metrics
  (gpu_pct/temp_max surfaced on node_json for the live cards).
- cm-api: beszel.rs client (auth-with-password, poll `systems`, map to nodes by
  hostname, upsert metrics) + a 15s spawn_poller; routes/beszel.rs (connect/status/
  disconnect + GET /api/nodes/{id}/metrics with history proxied live from the hub).
- frontend: HostCard gains a GPU/temp readout + a Monitor button; NodeMonitor is a
  full-width per-node page (current panel + CPU/mem/GPU/temp/net/disk charts from the
  hub's 1m history); a "Beszel monitoring" connect form in the Local view.

Reachability confirmed: gw-04 → the hub over the tailnet (100.123.224.84:8090). Needs
the user to connect their hub login to activate the poller.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

crates/cm-db/src/repo/fleet_beszel.rs

@@ -0,0 +1,85 @@
+//! Per-workspace Beszel hub connection (BYO monitoring): the hub URL + a login we
+//! use to read rich per-node metrics via the hub's PocketBase API. Server-side only.
+
+use cm_domain::WorkspaceId;
+use sqlx::{PgPool, Row};
+
+use crate::DbError;
+
+/// A stored Beszel hub connection.
+#[derive(Debug, Clone)]
+pub struct BeszelConn {
+    pub hub_url: String,
+    pub username: String,
+    pub password: String,
+}
+
+/// Store (or replace) a workspace's Beszel hub URL + credentials.
+pub async fn set(
+    pool: &PgPool,
+    workspace_id: WorkspaceId,
+    hub_url: &str,
+    username: &str,
+    password: &str,
+) -> Result<(), DbError> {
+    sqlx::query(
+        "INSERT INTO workspace_beszel (workspace_id, hub_url, username, password, connected_at)
+         VALUES ($1, $2, $3, $4, now())
+         ON CONFLICT (workspace_id) DO UPDATE SET
+            hub_url = excluded.hub_url, username = excluded.username,
+            password = excluded.password, connected_at = now()",
+    )
+    .bind(workspace_id.as_uuid())
+    .bind(hub_url)
+    .bind(username)
+    .bind(password)
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+/// Get a workspace's stored Beszel connection, if any.
+pub async fn get(
+    pool: &PgPool,
+    workspace_id: WorkspaceId,
+) -> Result<Option<BeszelConn>, DbError> {
+    let row =
+        sqlx::query("SELECT hub_url, username, password FROM workspace_beszel WHERE workspace_id = $1")
+            .bind(workspace_id.as_uuid())
+            .fetch_optional(pool)
+            .await?;
+    Ok(row.map(|r| BeszelConn {
+        hub_url: r.get("hub_url"),
+        username: r.get("username"),
+        password: r.get("password"),
+    }))
+}
+
+/// Every workspace with a Beszel hub connected (for the background poll task).
+pub async fn all(pool: &PgPool) -> Result<Vec<(WorkspaceId, BeszelConn)>, DbError> {
+    let rows = sqlx::query("SELECT workspace_id, hub_url, username, password FROM workspace_beszel")
+        .fetch_all(pool)
+        .await?;
+    Ok(rows
+        .into_iter()
+        .map(|r| {
+            (
+                WorkspaceId::from(r.get::<uuid::Uuid, _>("workspace_id")),
+                BeszelConn {
+                    hub_url: r.get("hub_url"),
+                    username: r.get("username"),
+                    password: r.get("password"),
+                },
+            )
+        })
+        .collect())
+}
+
+/// Disconnect a workspace's Beszel hub.
+pub async fn delete(pool: &PgPool, workspace_id: WorkspaceId) -> Result<(), DbError> {
+    sqlx::query("DELETE FROM workspace_beszel WHERE workspace_id = $1")
+        .bind(workspace_id.as_uuid())
+        .execute(pool)
+        .await?;
+    Ok(())
+}

crates/cm-db/src/repo/mod.rs

@@ -6,8 +6,10 @@ pub mod companies;
 pub mod connections;
 pub mod credits;
 pub mod files;
+pub mod fleet_beszel;
 pub mod fleet_tailscale;
 pub mod messages;
+pub mod node_metrics;
 pub mod nodes;
 pub mod orgs;
 pub mod outbox;

crates/cm-db/src/repo/node_metrics.rs

@@ -0,0 +1,75 @@
+//! Rich per-node metrics (from the workspace's Beszel hub). One latest snapshot
+//! per node: scalar columns for the fleet cards + rules engine, plus a JSONB blob
+//! for the per-node monitor page.
+
+use cm_domain::NodeId;
+use serde_json::Value;
+use sqlx::{PgPool, Row};
+
+use crate::DbError;
+
+/// The latest metric snapshot for a node (nullable scalars + the full blob).
+#[derive(Debug, Clone, Default)]
+pub struct NodeMetrics {
+    pub cpu_pct: Option<f64>,
+    pub mem_pct: Option<f64>,
+    pub disk_pct: Option<f64>,
+    pub gpu_pct: Option<f64>,
+    pub temp_max: Option<f64>,
+    pub net_sent_ps: Option<i64>,
+    pub net_recv_ps: Option<i64>,
+    pub disk_read_ps: Option<i64>,
+    pub disk_write_ps: Option<i64>,
+    pub load1: Option<f64>,
+    pub container_count: Option<i32>,
+    pub data: Value,
+}
+
+/// Upsert a node's latest metrics snapshot.
+pub async fn upsert(pool: &PgPool, node_id: NodeId, m: &NodeMetrics) -> Result<(), DbError> {
+    sqlx::query(
+        "INSERT INTO node_metrics
+            (node_id, updated_at, cpu_pct, mem_pct, disk_pct, gpu_pct, temp_max,
+             net_sent_ps, net_recv_ps, disk_read_ps, disk_write_ps, load1, container_count, data)
+         VALUES ($1, now(), $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
+         ON CONFLICT (node_id) DO UPDATE SET
+            updated_at = now(), cpu_pct = excluded.cpu_pct, mem_pct = excluded.mem_pct,
+            disk_pct = excluded.disk_pct, gpu_pct = excluded.gpu_pct, temp_max = excluded.temp_max,
+            net_sent_ps = excluded.net_sent_ps, net_recv_ps = excluded.net_recv_ps,
+            disk_read_ps = excluded.disk_read_ps, disk_write_ps = excluded.disk_write_ps,
+            load1 = excluded.load1, container_count = excluded.container_count, data = excluded.data",
+    )
+    .bind(node_id.as_uuid())
+    .bind(m.cpu_pct)
+    .bind(m.mem_pct)
+    .bind(m.disk_pct)
+    .bind(m.gpu_pct)
+    .bind(m.temp_max)
+    .bind(m.net_sent_ps)
+    .bind(m.net_recv_ps)
+    .bind(m.disk_read_ps)
+    .bind(m.disk_write_ps)
+    .bind(m.load1)
+    .bind(m.container_count)
+    .bind(&m.data)
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+/// The latest metrics blob for a node (the full snapshot for the monitor page).
+pub async fn latest(pool: &PgPool, node_id: NodeId) -> Result<Option<Value>, DbError> {
+    let row = sqlx::query(
+        "SELECT data, extract(epoch from updated_at)::bigint AS updated FROM node_metrics WHERE node_id = $1",
+    )
+    .bind(node_id.as_uuid())
+    .fetch_optional(pool)
+    .await?;
+    Ok(row.map(|r| {
+        let mut data: Value = r.get("data");
+        if let Some(obj) = data.as_object_mut() {
+            obj.insert("updatedAt".into(), serde_json::json!(r.get::<i64, _>("updated")));
+        }
+        data
+    }))
+}

crates/cm-db/src/repo/nodes.rs

@@ -37,6 +37,9 @@ pub struct NodeRow {
     pub last_seen: Option<OffsetDateTime>,
     pub created_at: OffsetDateTime,
     pub health: Option<NodeHealth>,
+    /// Latest Beszel metrics scalars (the rich bits beyond basic health).
+    pub gpu_pct: Option<f64>,
+    pub temp_max: Option<f64>,
 }
 
 /// Register a new (pending) node with its control-channel token.
@@ -73,8 +76,10 @@ pub async fn auth(pool: &PgPool, token: &str) -> Result<Option<(NodeId, Workspac
 
 const SELECT_WITH_HEALTH: &str = "SELECT n.id, n.name, n.hostname, n.local_ip, n.status, n.agent_version, n.tailscale_ip, n.last_seen, n.created_at,
         h.node_id AS health_node, h.cpu_pct, h.mem_total, h.mem_used, h.mem_pressure, h.swap_used,
-        h.disk_total, h.disk_free, h.load1, h.load5, h.load15, h.container_count
-   FROM nodes n LEFT JOIN node_health h ON h.node_id = n.id";
+        h.disk_total, h.disk_free, h.load1, h.load5, h.load15, h.container_count,
+        m.gpu_pct AS m_gpu_pct, m.temp_max AS m_temp_max
+   FROM nodes n LEFT JOIN node_health h ON h.node_id = n.id
+                LEFT JOIN node_metrics m ON m.node_id = n.id";
 
 /// List a workspace's nodes (oldest first) with their latest health.
 pub async fn list(pool: &PgPool, workspace_id: WorkspaceId) -> Result<Vec<NodeRow>, DbError> {
@@ -220,5 +225,7 @@ fn map_node(r: sqlx::postgres::PgRow) -> NodeRow {
         last_seen: r.get("last_seen"),
         created_at: r.get("created_at"),
         health,
+        gpu_pct: r.get("m_gpu_pct"),
+        temp_max: r.get("m_temp_max"),
     }
 }