//! Bring-your-own Beszel hub client: authenticate to a workspace's Beszel hub
//! (PocketBase) and poll rich per-node metrics, mapping its `systems` to our fleet
//! nodes by hostname and upserting `node_metrics`. Server-side only.

use std::collections::HashMap;
use std::time::Duration;

use cm_db::repo::fleet_beszel::BeszelConn;
use cm_db::repo::node_metrics::NodeMetrics;
use cm_db::repo::{fleet_beszel, node_metrics, nodes};
use cm_domain::{NodeId, WorkspaceId};
use serde_json::{json, Value};
use sqlx::PgPool;

/// Authenticate to a Beszel hub → a PocketBase token. Tries the regular users
/// collection first, then superusers.
pub async fn authenticate(client: &reqwest::Client, conn: &BeszelConn) -> Result<String, String> {
    let base = conn.hub_url.trim_end_matches('/');
    for coll in ["users", "_superusers"] {
        let url = format!("{base}/api/collections/{coll}/auth-with-password");
        let resp = client
            .post(&url)
            .json(&json!({ "identity": conn.username, "password": conn.password }))
            .send()
            .await
            .map_err(|e| e.to_string())?;
        if resp.status().is_success() {
            let body: Value = resp.json().await.map_err(|e| e.to_string())?;
            if let Some(tok) = body.get("token").and_then(Value::as_str) {
                return Ok(tok.to_owned());
            }
        }
    }
    Err("auth failed (check hub URL + credentials)".into())
}

/// Fetch all `systems` records (current status + the `info` snapshot).
pub async fn fetch_systems(
    client: &reqwest::Client,
    conn: &BeszelConn,
    token: &str,
) -> Result<Vec<Value>, String> {
    let base = conn.hub_url.trim_end_matches('/');
    let resp = client
        .get(format!("{base}/api/collections/systems/records"))
        .query(&[("perPage", "500")])
        .header("Authorization", token)
        .send()
        .await
        .map_err(|e| e.to_string())?;
    if !resp.status().is_success() {
        return Err(format!("systems list HTTP {}", resp.status()));
    }
    let body: Value = resp.json().await.map_err(|e| e.to_string())?;
    Ok(body
        .get("items")
        .and_then(Value::as_array)
        .cloned()
        .unwrap_or_default())
}

/// Proxy a system's recent 1m time-series (for the monitor-page charts).
pub async fn fetch_history(
    client: &reqwest::Client,
    conn: &BeszelConn,
    token: &str,
    system_id: &str,
) -> Result<Value, String> {
    let base = conn.hub_url.trim_end_matches('/');
    let filter = format!("system='{system_id}' && type='1m'");
    let resp = client
        .get(format!("{base}/api/collections/system_stats/records"))
        .query(&[
            ("perPage", "120"),
            ("sort", "-created"),
            ("filter", filter.as_str()),
        ])
        .header("Authorization", token)
        .send()
        .await
        .map_err(|e| e.to_string())?;
    if !resp.status().is_success() {
        return Err(format!("system_stats HTTP {}", resp.status()));
    }
    resp.json().await.map_err(|e| e.to_string())
}

fn f(v: &Value, k: &str) -> Option<f64> {
    v.get(k).and_then(Value::as_f64)
}

/// Map a Beszel `systems` record (its `info` snapshot) into our NodeMetrics.
fn metrics_from_system(system: &Value) -> NodeMetrics {
    let info = system.get("info").cloned().unwrap_or_else(|| json!({}));
    let load1 = info
        .get("la")
        .and_then(Value::as_array)
        .and_then(|a| a.first())
        .and_then(Value::as_f64);
    NodeMetrics {
        cpu_pct: f(&info, "cpu"),
        mem_pct: f(&info, "mp"),
        disk_pct: f(&info, "dp"),
        gpu_pct: f(&info, "g"),
        temp_max: f(&info, "dt"),
        net_sent_ps: None,
        net_recv_ps: None,
        disk_read_ps: None,
        disk_write_ps: None,
        load1,
        container_count: None,
        data: json!({
            "beszelSystemId": system.get("id").and_then(Value::as_str),
            "status": system.get("status").and_then(Value::as_str),
            "name": system.get("name").and_then(Value::as_str),
            "host": system.get("host").and_then(Value::as_str),
            "info": info,
        }),
    }
}

/// Poll one workspace's hub: auth, fetch systems, match to nodes by hostname,
/// upsert metrics. Returns how many nodes were updated.
pub async fn poll_workspace(
    client: &reqwest::Client,
    pool: &PgPool,
    ws: WorkspaceId,
    conn: &BeszelConn,
) -> Result<usize, String> {
    let token = authenticate(client, conn).await?;
    let systems = fetch_systems(client, conn, &token).await?;
    let node_rows = nodes::list(pool, ws).await.map_err(|e| e.to_string())?;
    // hostname/name (lowercased) → node id.
    let mut by_host: HashMap<String, NodeId> = HashMap::new();
    for n in &node_rows {
        if let Some(h) = n.hostname.as_deref() {
            by_host.insert(h.to_lowercase(), n.id);
        }
        by_host.entry(n.name.to_lowercase()).or_insert(n.id);
    }
    let mut updated = 0;
    for sys in &systems {
        let key = sys
            .get("name")
            .and_then(Value::as_str)
            .or_else(|| sys.get("host").and_then(Value::as_str))
            .map(str::to_lowercase);
        let Some(node_id) = key.as_deref().and_then(|k| by_host.get(k).copied()) else {
            continue;
        };
        if node_metrics::upsert(pool, node_id, &metrics_from_system(sys))
            .await
            .is_ok()
        {
            updated += 1;
        }
    }
    Ok(updated)
}

/// Spawn the background poller: every `interval`, refresh node_metrics for each
/// workspace with a Beszel hub connected.
pub fn spawn_poller(pool: PgPool, interval: Duration) {
    tokio::spawn(async move {
        let client = reqwest::Client::new();
        let mut tick = tokio::time::interval(interval);
        loop {
            tick.tick().await;
            let workspaces = match fleet_beszel::all(&pool).await {
                Ok(w) => w,
                Err(_) => continue,
            };
            for (ws, conn) in workspaces {
                if let Err(e) = poll_workspace(&client, &pool, ws, &conn).await {
                    eprintln!("beszel poll (ws {}): {e}", ws.as_uuid());
                }
            }
        }
    });
}