//! Fleet control plane: an in-memory hub of live daemon control channels plus
//! the WebSocket channel runner. Each connected `clawmates-node` daemon dials
//! `GET /api/nodes/agent?token=…` (outbound), and we drive that socket to
//! receive host-health heartbeats and to run commands on the node.

use std::collections::{HashMap, HashSet};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

use axum::extract::ws::{Message, WebSocket};
use base64::Engine;
use cm_db::repo::nodes::{self, NodeHealth};
use cm_domain::NodeId;
use cm_sandbox::{
    ExecResult, ManagedSandbox, PtySession, SandboxDriver, SandboxError, SandboxHandle, SandboxSpec,
};
use futures::{SinkExt, StreamExt};
use serde::Deserialize;
use serde_json::{json, Value};
use sqlx::PgPool;
use tokio::sync::{mpsc, oneshot, Mutex};

const B64: base64::engine::general_purpose::GeneralPurpose =
    base64::engine::general_purpose::STANDARD;

/// Monotonic per-connection id. A reconnecting daemon gets a fresh epoch so a
/// stale channel's teardown can't clobber the newer connection's online status.
static CONN_EPOCH: AtomicU64 = AtomicU64::new(1);

/// Server keepalive: ping the daemon this often; if no inbound frame (incl. the
/// matching pong) arrives within the deadline, treat the socket as dead.
const PING_EVERY: Duration = Duration::from_secs(15);
const IDLE_DEADLINE: Duration = Duration::from_secs(35);

/// Backstop sweeper: flip any node whose heartbeat stopped (without a clean
/// channel close) to offline. Heartbeats refresh `last_seen`, so a live node
/// (5s heartbeat) is never older than the window and is never swept; a vanished
/// node goes offline within `stale_secs` + one tick even if its channel hangs.
pub fn spawn_node_sweeper(pool: PgPool, interval: Duration, stale_secs: i64) {
    tokio::spawn(async move {
        let mut tick = tokio::time::interval(interval);
        loop {
            tick.tick().await;
            if let Err(e) = nodes::mark_stale_offline(&pool, stale_secs).await {
                eprintln!("node_sweeper: mark_stale_offline failed: {e}");
            }
        }
    });
}

/// The result of running a command on a node.
#[derive(Debug, Clone)]
pub struct ExecOutput {
    pub ok: bool,
    pub output: String,
}

struct NodeConn {
    tx: mpsc::UnboundedSender<String>,
    pending: Mutex<HashMap<u64, oneshot::Sender<ExecOutput>>>,
    /// Live terminal sessions: sid → byte sink for the browser bridge (WS-relay
    /// PTY output). Bounded: a runaway PTY (say `cat /var/log/huge`) with a
    /// stalled browser must not accumulate megabytes here. On overflow the
    /// session is closed instead of holding output indefinitely.
    pty_sinks: Mutex<HashMap<u64, mpsc::Sender<Vec<u8>>>>,
    /// WebRTC signaling: sid → text sink delivering the daemon's answer/ICE to
    /// the browser bridge.
    signal_sinks: Mutex<HashMap<u64, mpsc::UnboundedSender<String>>>,
    next_id: AtomicU64,
    /// Unique per physical connection — see `CONN_EPOCH`.
    epoch: u64,
}

/// Registry of live daemon channels, keyed by node id.
#[derive(Default)]
pub struct NodeHub {
    conns: Mutex<HashMap<NodeId, Arc<NodeConn>>>,
    /// Sync-readable set of connected node ids (so placement can check liveness
    /// without an await — `conns` is behind an async mutex).
    online: std::sync::Mutex<HashSet<NodeId>>,
    /// Short-lived single-use terminal tickets (browser WS can't send a bearer).
    tickets: Mutex<HashMap<String, (NodeId, Instant)>>,
}

impl NodeHub {
    pub fn new() -> Self {
        Self::default()
    }

    pub async fn is_online(&self, id: NodeId) -> bool {
        self.conns.lock().await.contains_key(&id)
    }

    async fn get(&self, id: NodeId) -> Option<Arc<NodeConn>> {
        self.conns.lock().await.get(&id).cloned()
    }

    /// Run the node's built-in verification (host + docker check) and await its
    /// output. The gateway never sends arbitrary shell — only typed ops the
    /// daemon vets and runs itself (verify today; container ops later).
    pub async fn verify(&self, id: NodeId) -> Result<ExecOutput, String> {
        self.request(id, |req_id| {
            json!({ "t": "verify", "id": req_id }).to_string()
        })
        .await
    }

    /// Provision + run + tear down a fully hardened throwaway container on the
    /// node (readiness check that it can host agent workloads).
    pub async fn sandbox_check(&self, id: NodeId) -> Result<ExecOutput, String> {
        self.request(id, |req_id| {
            json!({ "t": "sb_check", "id": req_id }).to_string()
        })
        .await
    }

    /// Sync liveness check (no await) — used by placement.
    pub fn is_connected(&self, id: NodeId) -> bool {
        self.online.lock().map(|s| s.contains(&id)).unwrap_or(false)
    }

    /// Every currently-connected node id. Sync (no await), like `is_connected`,
    /// so the container reapers can enumerate nodes to sweep.
    pub fn online_ids(&self) -> Vec<NodeId> {
        self.online
            .lock()
            .map(|s| s.iter().copied().collect())
            .unwrap_or_default()
    }

    /// Send a typed op with JSON args and await its result (20s default).
    pub async fn call(&self, id: NodeId, op: &str, args: Value) -> Result<ExecOutput, String> {
        self.call_timeout(id, op, args, 20).await
    }

    /// Like `call` but with a custom result timeout — for long ops (e.g. tool
    /// updates) whose result legitimately takes longer than the default.
    pub async fn call_timeout(
        &self,
        id: NodeId,
        op: &str,
        args: Value,
        secs: u64,
    ) -> Result<ExecOutput, String> {
        let op = op.to_owned();
        self.request_timeout(
            id,
            move |req_id| {
                let mut o = args.as_object().cloned().unwrap_or_default();
                o.insert("t".to_owned(), Value::String(op));
                o.insert("id".to_owned(), Value::from(req_id));
                Value::Object(o).to_string()
            },
            std::time::Duration::from_secs(secs),
        )
        .await
    }

    /// Send a typed request frame and await the node's matching result (20s).
    async fn request(
        &self,
        id: NodeId,
        frame: impl FnOnce(u64) -> String,
    ) -> Result<ExecOutput, String> {
        self.request_timeout(id, frame, std::time::Duration::from_secs(20))
            .await
    }

    async fn request_timeout(
        &self,
        id: NodeId,
        frame: impl FnOnce(u64) -> String,
        dur: std::time::Duration,
    ) -> Result<ExecOutput, String> {
        let conn = self.get(id).await.ok_or("node is not connected")?;
        let req_id = conn.next_id.fetch_add(1, Ordering::Relaxed);
        let (tx, rx) = oneshot::channel();
        conn.pending.lock().await.insert(req_id, tx);
        conn.tx
            .send(frame(req_id))
            .map_err(|_| "node channel closed".to_string())?;
        match tokio::time::timeout(dur, rx).await {
            Ok(Ok(out)) => Ok(out),
            Ok(Err(_)) => Err("node dropped before responding".into()),
            Err(_) => {
                conn.pending.lock().await.remove(&req_id);
                Err("node timed out".into())
            }
        }
    }

    /// Allocate a terminal session: a sid, a byte stream (WS-relay PTY output),
    /// and a text stream (WebRTC answer/ICE). The PTY is NOT opened yet — the
    /// browser picks the transport (WebRTC direct, or `open_pty` fallback).
    pub async fn open_session(
        &self,
        id: NodeId,
    ) -> Option<(
        u64,
        mpsc::Receiver<Vec<u8>>,
        mpsc::UnboundedReceiver<String>,
    )> {
        let conn = self.get(id).await?;
        let sid = conn.next_id.fetch_add(1, Ordering::Relaxed);
        // 256 × ~4KB PTY frames = ~1 MB per stalled session before we close it.
        let (ptx, prx) = mpsc::channel(256);
        let (stx, srx) = mpsc::unbounded_channel();
        conn.pty_sinks.lock().await.insert(sid, ptx);
        conn.signal_sinks.lock().await.insert(sid, stx);
        Some((sid, prx, srx))
    }

    /// Open the WS-relay PTY for an allocated session (the fallback path).
    /// `container` (+ `session`) targets `docker exec` into an agent container on
    /// the node (the node-placed agent terminal); both `None` ⇒ the host shell.
    /// `command`, when set to a non-empty argv, wins over both — spawns
    /// the program directly (used by the Herdr Live Pane to attach xterm.js
    /// straight to `herdr`).
    // 8 params is at the target-shape ceiling: (id, sid, cols, rows) address
    // the session, (container, session, command) address the target. Wrapping
    // in a struct would add ceremony without collapsing dimensions.
    #[allow(clippy::too_many_arguments)]
    pub async fn open_pty(
        &self,
        id: NodeId,
        sid: u64,
        cols: u16,
        rows: u16,
        container: Option<&str>,
        session: Option<&str>,
        command: Option<&[String]>,
    ) {
        if let Some(conn) = self.get(id).await {
            let mut frame = json!({ "t": "pty_open", "sid": sid, "cols": cols, "rows": rows });
            if let Some(cmd) = command.filter(|c| !c.is_empty()) {
                frame["command"] = json!(cmd);
            } else {
                if let Some(c) = container {
                    frame["container"] = json!(c);
                }
                if let Some(s) = session {
                    frame["session"] = json!(s);
                }
            }
            let _ = conn.tx.send(frame.to_string());
        }
    }

    /// Relay a browser SDP offer to the daemon (it answers + trickles ICE back).
    /// `container`/`session` target a node-placed agent container as in `open_pty`.
    pub async fn webrtc_offer(
        &self,
        id: NodeId,
        sid: u64,
        sdp: &str,
        container: Option<&str>,
        session: Option<&str>,
    ) {
        if let Some(conn) = self.get(id).await {
            let mut frame = json!({ "t": "webrtc_offer", "sid": sid, "sdp": sdp });
            if let Some(c) = container {
                frame["container"] = json!(c);
            }
            if let Some(s) = session {
                frame["session"] = json!(s);
            }
            let _ = conn.tx.send(frame.to_string());
        }
    }

    /// Relay a browser ICE candidate to the daemon.
    pub async fn webrtc_ice(
        &self,
        id: NodeId,
        sid: u64,
        candidate: &str,
        sdp_mid: Option<&str>,
        sdp_mline_index: Option<u16>,
    ) {
        if let Some(conn) = self.get(id).await {
            let _ = conn.tx.send(
                json!({ "t": "webrtc_ice", "sid": sid, "candidate": candidate,
                        "sdp_mid": sdp_mid, "sdp_mline_index": sdp_mline_index })
                .to_string(),
            );
        }
    }

    /// Tell the daemon to tear down a WebRTC peer for a session.
    pub async fn webrtc_close(&self, id: NodeId, sid: u64) {
        if let Some(conn) = self.get(id).await {
            let _ = conn
                .tx
                .send(json!({ "t": "webrtc_close", "sid": sid }).to_string());
        }
    }

    pub async fn terminal_input(&self, id: NodeId, sid: u64, bytes: &[u8]) {
        if let Some(conn) = self.get(id).await {
            let _ = conn
                .tx
                .send(json!({ "t": "pty_in", "sid": sid, "data": B64.encode(bytes) }).to_string());
        }
    }

    pub async fn terminal_resize(&self, id: NodeId, sid: u64, cols: u16, rows: u16) {
        if let Some(conn) = self.get(id).await {
            let _ = conn.tx.send(
                json!({ "t": "pty_resize", "sid": sid, "cols": cols, "rows": rows }).to_string(),
            );
        }
    }

    pub async fn terminal_close(&self, id: NodeId, sid: u64) {
        if let Some(conn) = self.get(id).await {
            conn.pty_sinks.lock().await.remove(&sid);
            conn.signal_sinks.lock().await.remove(&sid);
            let _ = conn
                .tx
                .send(json!({ "t": "pty_close", "sid": sid }).to_string());
            let _ = conn
                .tx
                .send(json!({ "t": "webrtc_close", "sid": sid }).to_string());
        }
    }

    /// Mint a single-use terminal ticket for a node (60s TTL) — the browser WS
    /// handshake can't carry a bearer header.
    pub async fn mint_ticket(&self, id: NodeId) -> String {
        let token = format!(
            "{}{}",
            uuid::Uuid::now_v7().simple(),
            uuid::Uuid::now_v7().simple()
        );
        let mut t = self.tickets.lock().await;
        let now = Instant::now();
        t.retain(|_, (_, exp)| *exp > now);
        t.insert(token.clone(), (id, now + Duration::from_secs(60)));
        token
    }

    /// Redeem a terminal ticket (single use).
    pub async fn redeem_ticket(&self, token: &str) -> Option<NodeId> {
        match self.tickets.lock().await.remove(token) {
            Some((id, exp)) if exp > Instant::now() => Some(id),
            _ => None,
        }
    }
}

#[derive(Deserialize)]
#[serde(tag = "t")]
enum Uplink {
    #[serde(rename = "heartbeat")]
    Heartbeat {
        version: Option<String>,
        tailscale_ip: Option<String>,
        hostname: Option<String>,
        local_ip: Option<String>,
        health: HealthMsg,
    },
    #[serde(rename = "result")]
    Result { id: u64, ok: bool, output: String },
    #[serde(rename = "pty_out")]
    PtyOut { sid: u64, data: String },
    #[serde(rename = "pty_exit")]
    PtyExit { sid: u64 },
    #[serde(rename = "webrtc_answer")]
    WebRtcAnswer { sid: u64, sdp: String },
    #[serde(rename = "webrtc_ice")]
    WebRtcIce {
        sid: u64,
        candidate: String,
        sdp_mid: Option<String>,
        sdp_mline_index: Option<u16>,
    },
    #[serde(rename = "webrtc_failed")]
    WebRtcFailed { sid: u64 },
    #[serde(rename = "node_tools")]
    NodeTools {
        tools: std::collections::HashMap<String, String>,
    },
    /// What the node can HOST, as opposed to what it has installed — the
    /// inputs to placement predicates. Free-form so a new predicate does not
    /// need a migration; see `migrations/0065_microvm_placement.sql`.
    #[serde(rename = "node_capabilities")]
    NodeCapabilities { capabilities: serde_json::Value },
}

#[derive(Deserialize)]
struct HealthMsg {
    cpu_pct: f64,
    mem_total: i64,
    mem_used: i64,
    mem_pressure: f64,
    swap_used: i64,
    disk_total: i64,
    disk_free: i64,
    load1: f64,
    load5: f64,
    load15: f64,
    container_count: i32,
}

/// Drive a daemon's control channel: register it, pump outbound command frames,
/// and apply uplink heartbeats/results until the socket closes.
pub async fn run_channel(pool: PgPool, hub: Arc<NodeHub>, node_id: NodeId, socket: WebSocket) {
    let (mut ws_tx, mut ws_rx) = socket.split();
    let (tx, mut rx) = mpsc::unbounded_channel::<String>();
    let epoch = CONN_EPOCH.fetch_add(1, Ordering::Relaxed);
    let conn = Arc::new(NodeConn {
        tx,
        pending: Mutex::new(HashMap::new()),
        pty_sinks: Mutex::new(HashMap::new()),
        signal_sinks: Mutex::new(HashMap::new()),
        next_id: AtomicU64::new(0),
        epoch,
    });
    // Inserting overwrites any stale conn for this node — dropping the old conn's
    // `tx`, so its writer's `rx.recv()` returns None and that channel tears down.
    hub.conns.lock().await.insert(node_id, conn.clone());
    hub.online.lock().unwrap().insert(node_id);

    // One select loop carries outbound frames, inbound frames, and a keepalive
    // ping. `last_inbound` tracks liveness: any frame (incl. Pong) refreshes it;
    // if it goes stale past the deadline, the socket is dead and we tear down.
    let mut ping_tick = tokio::time::interval(PING_EVERY);
    ping_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    let mut last_inbound = Instant::now();
    loop {
        tokio::select! {
            frame = rx.recv() => match frame {
                Some(f) => {
                    if ws_tx.send(Message::Text(f.into())).await.is_err() {
                        break;
                    }
                }
                None => break,
            },
            _ = ping_tick.tick() => {
                if last_inbound.elapsed() > IDLE_DEADLINE {
                    break;
                }
                if ws_tx.send(Message::Ping(Vec::new().into())).await.is_err() {
                    break;
                }
            },
            msg = ws_rx.next() => {
                let Some(Ok(msg)) = msg else { break };
                last_inbound = Instant::now();
                let Message::Text(t) = msg else { continue };
                match serde_json::from_str::<Uplink>(t.as_str()) {
                    Ok(Uplink::Heartbeat {
                        version,
                        tailscale_ip,
                        hostname,
                        local_ip,
                        health,
                    }) => {
                        let h = NodeHealth {
                            cpu_pct: health.cpu_pct,
                            mem_total: health.mem_total,
                            mem_used: health.mem_used,
                            mem_pressure: health.mem_pressure,
                            swap_used: health.swap_used,
                            disk_total: health.disk_total,
                            disk_free: health.disk_free,
                            load1: health.load1,
                            load5: health.load5,
                            load15: health.load15,
                            container_count: health.container_count,
                        };
                        let _ = nodes::heartbeat(
                            &pool,
                            node_id,
                            version.as_deref(),
                            tailscale_ip.as_deref(),
                            hostname.as_deref(),
                            local_ip.as_deref(),
                            &h,
                        )
                        .await;
                    }
                    Ok(Uplink::Result { id, ok, output }) => {
                        if let Some(s) = conn.pending.lock().await.remove(&id) {
                            let _ = s.send(ExecOutput { ok, output });
                        }
                    }
                    Ok(Uplink::PtyOut { sid, data }) => {
                        if let Ok(bytes) = B64.decode(&data) {
                            let sink = conn.pty_sinks.lock().await.get(&sid).cloned();
                            if let Some(s) = sink {
                                // try_send so a stalled browser can't grow the
                                // per-session buffer without bound. On Full, the
                                // session is torn down: drop both sinks and tell
                                // the node to close its side, preventing an
                                // orphan PTY.
                                if let Err(err) = s.try_send(bytes) {
                                    if matches!(err, mpsc::error::TrySendError::Full(_)) {
                                        conn.pty_sinks.lock().await.remove(&sid);
                                        conn.signal_sinks.lock().await.remove(&sid);
                                        let _ = conn.tx.send(
                                            json!({ "t": "pty_close", "sid": sid })
                                                .to_string(),
                                        );
                                    }
                                }
                            }
                        }
                    }
                    Ok(Uplink::PtyExit { sid }) => {
                        conn.pty_sinks.lock().await.remove(&sid);
                    }
                    // WebRTC signaling from the daemon → forward to the browser
                    // bridge (re-tagged `type` for the browser) via signal_sinks.
                    Ok(Uplink::WebRtcAnswer { sid, sdp }) => {
                        if let Some(s) = conn.signal_sinks.lock().await.get(&sid) {
                            let _ = s.send(
                                json!({ "type": "webrtc_answer", "sdp": sdp }).to_string(),
                            );
                        }
                    }
                    Ok(Uplink::WebRtcIce {
                        sid,
                        candidate,
                        sdp_mid,
                        sdp_mline_index,
                    }) => {
                        if let Some(s) = conn.signal_sinks.lock().await.get(&sid) {
                            let _ = s.send(
                                json!({ "type": "webrtc_ice", "candidate": candidate,
                                        "sdp_mid": sdp_mid, "sdp_mline_index": sdp_mline_index })
                                .to_string(),
                            );
                        }
                    }
                    Ok(Uplink::WebRtcFailed { sid }) => {
                        if let Some(s) = conn.signal_sinks.lock().await.get(&sid) {
                            let _ = s.send(json!({ "type": "webrtc_failed" }).to_string());
                        }
                    }
                    Ok(Uplink::NodeTools { tools }) => {
                        let pairs: Vec<(String, String)> = tools.into_iter().collect();
                        let _ = cm_db::repo::node_tools::upsert(&pool, node_id, &pairs).await;
                    }
                    Ok(Uplink::NodeCapabilities { capabilities }) => {
                        if let Err(e) = nodes::set_capabilities(&pool, node_id, &capabilities).await
                        {
                            // Loud: a node whose capabilities never land looks
                            // exactly like a node that has none, and will be
                            // passed over for every microVM mission forever
                            // while appearing perfectly healthy.
                            eprintln!(
                                "fleet: could not record capabilities for node {node_id} ({e}) — \
                                 it will not be selected for microvm placement"
                            );
                        }
                    }
                    Err(_) => {}
                }
            },
        }
    }

    // Teardown — only clear if we're STILL the registered connection. A daemon
    // that reconnected has a newer epoch; a stale channel must not flip it offline.
    let mut conns = hub.conns.lock().await;
    if conns.get(&node_id).map(|c| c.epoch) == Some(conn.epoch) {
        conns.remove(&node_id);
        hub.online.lock().unwrap().remove(&node_id);
        let _ = nodes::set_status(&pool, node_id, "offline").await;
    }
}

// ── RemoteDriver: run agent sandboxes on a connected node over the channel ────

/// A `SandboxDriver` that proxies container ops to a connected fleet node's
/// daemon, which drives the REAL `DockerDriver` — so the hardening is identical
/// to a local sandbox. Interactive PTY is unsupported on remote nodes for now
/// (agents don't use it; the Terminal app stays local).
pub struct RemoteDriver {
    hub: Arc<NodeHub>,
    node_id: NodeId,
}

impl RemoteDriver {
    pub fn new(hub: Arc<NodeHub>, node_id: NodeId) -> Self {
        Self { hub, node_id }
    }
    async fn call(&self, op: &str, args: Value) -> Result<Value, SandboxError> {
        let out = self
            .hub
            .call(self.node_id, op, args)
            .await
            .map_err(SandboxError::Engine)?;
        if !out.ok {
            return Err(SandboxError::Engine(out.output));
        }
        Ok(serde_json::from_str(&out.output).unwrap_or(Value::Null))
    }
}

#[async_trait::async_trait]
impl SandboxDriver for RemoteDriver {
    async fn provision(&self, spec: &SandboxSpec) -> Result<SandboxHandle, SandboxError> {
        let v = self.call("sb_provision", json!({ "spec": spec })).await?;
        Ok(SandboxHandle {
            id: v
                .get("id")
                .and_then(Value::as_str)
                .unwrap_or_default()
                .to_owned(),
            name: v
                .get("name")
                .and_then(Value::as_str)
                .unwrap_or_default()
                .to_owned(),
        })
    }
    async fn exec(&self, handle: &SandboxHandle, cmd: &[&str]) -> Result<ExecResult, SandboxError> {
        let v = self
            .call(
                "sb_exec",
                json!({ "cid": handle.id, "cname": handle.name, "cmd": cmd }),
            )
            .await?;
        Ok(ExecResult {
            exit_code: v.get("exit").and_then(Value::as_i64).unwrap_or(-1),
            stdout: v
                .get("stdout")
                .and_then(Value::as_str)
                .unwrap_or_default()
                .to_owned(),
            stderr: v
                .get("stderr")
                .and_then(Value::as_str)
                .unwrap_or_default()
                .to_owned(),
        })
    }
    async fn attach_pty(
        &self,
        _h: &SandboxHandle,
        _cmd: &[&str],
        _c: u16,
        _r: u16,
        _e: &[String],
    ) -> Result<PtySession, SandboxError> {
        Err(SandboxError::Engine(
            "interactive PTY is not supported on remote nodes yet".into(),
        ))
    }
    async fn resize_pty(&self, _exec_id: &str, _c: u16, _r: u16) -> Result<(), SandboxError> {
        Err(SandboxError::Engine(
            "pty resize is not supported on remote nodes".into(),
        ))
    }
    async fn destroy(&self, handle: &SandboxHandle) -> Result<(), SandboxError> {
        self.call(
            "sb_destroy",
            json!({ "cid": handle.id, "cname": handle.name }),
        )
        .await?;
        Ok(())
    }
    async fn health(&self, handle: &SandboxHandle) -> Result<bool, SandboxError> {
        let v = self
            .call(
                "sb_health",
                json!({ "cid": handle.id, "cname": handle.name }),
            )
            .await?;
        Ok(v.get("alive").and_then(Value::as_bool).unwrap_or(false))
    }
    async fn list_managed(&self, kind: &str) -> Result<Vec<ManagedSandbox>, SandboxError> {
        let v = self.call("sb_list", json!({ "kind": kind })).await?;
        Ok(v.as_array()
            .map(|a| {
                a.iter()
                    .map(|m| ManagedSandbox {
                        id: m
                            .get("id")
                            .and_then(Value::as_str)
                            .unwrap_or_default()
                            .to_owned(),
                        created_unix: m.get("created_unix").and_then(Value::as_i64).unwrap_or(0),
                    })
                    .collect()
            })
            .unwrap_or_default())
    }
}

/// Bridges the runtime's placement to live node channels: hands out a
/// `RemoteDriver` for any currently-connected fleet node (None ⇒ offline).
pub struct HubDriverProvider {
    hub: Arc<NodeHub>,
}

impl HubDriverProvider {
    pub fn new(hub: Arc<NodeHub>) -> Self {
        Self { hub }
    }
}

impl cm_runtime::NodeDriverProvider for HubDriverProvider {
    fn driver(&self, node_id: &str) -> Option<Arc<dyn SandboxDriver>> {
        let nid = NodeId::from(node_id.parse::<uuid::Uuid>().ok()?);
        if self.hub.is_connected(nid) {
            Some(Arc::new(RemoteDriver::new(self.hub.clone(), nid)))
        } else {
            None
        }
    }

    fn node_ids(&self) -> Vec<String> {
        self.hub
            .online_ids()
            .into_iter()
            .map(|id| id.to_string())
            .collect()
    }
}