//! Core service behind chj.disks (port of `disks/diskService.js`). Linux only; other platforms
//! get an explicit "unsupported" result and no Linux tool is ever started on them. Reads can run
//! without root; every change goes through the PrivilegedExecutor with the fixed
//! `remoteDisks.sh`, as a detached remote job with persistent state and log.

use super::inventory::{build_inventory, split_sections};
use super::js::{number_str, truthy};
use super::planner::{plan_operations, suggestions};
use crate::core::error::{CoreError, CoreResult};
use crate::core::logging::Logger;
use crate::core::ssh::platform::detect_platform;
use crate::core::ssh::privileged::{Frame, PrivilegedExecutor, RunOptions, RunResult};
use crate::core::ssh::session::SessionManager;
use indexmap::IndexMap;
use once_cell::sync::Lazy;
use parking_lot::Mutex;
use regex::Regex;
use serde_json::{json, Map, Value};
use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;

pub const SCRIPT: &str = include_str!("../../../resources/remote/remoteDisks.sh");
const PLAN_TTL_MS: i64 = 30 * 60 * 1000;
const MAX_PLANS: usize = 20;
static JOB_ID: Lazy<Regex> = Lazy::new(|| Regex::new(r"^[a-z0-9-]{8,64}$").unwrap());
static DEVICE: Lazy<Regex> = Lazy::new(|| Regex::new(r"^/dev/[A-Za-z0-9][A-Za-z0-9._/:+-]{0,90}$").unwrap());
static STEP_MARKER: Lazy<Regex> = Lazy::new(|| Regex::new(r"\x1eSTEP (\d+)\|(\d+)\|([a-z]+)\|([^\n]*)").unwrap());
static STEP_LINE: Lazy<Regex> = Lazy::new(|| Regex::new(r"\x1eSTEP [^\n]*\n?").unwrap());
static CONTROL_KEEP_LF: Lazy<Regex> = Lazy::new(|| Regex::new(r"[\x00-\x08\x0b-\x1f\x7f]").unwrap());

fn fail(code: &str, message: impl Into<String>) -> CoreError {
    CoreError::new(&format!("DISKS_{code}"), message)
}

fn now_ms() -> i64 {
    chrono::Utc::now().timestamp_millis()
}

fn iso(ms: i64) -> String {
    chrono::DateTime::<chrono::Utc>::from_timestamp_millis(ms).map(|t| t.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)).unwrap_or_default()
}

/// diskService `text()`: control characters except TAB/LF → space, no trim.
fn text(value: &str, limit: usize) -> String {
    super::js::cut(&CONTROL_KEEP_LF.replace_all(value, " "), limit)
}

fn key_values(section: &str) -> Map<String, Value> {
    let mut values = Map::new();
    for line in section.split('\n') {
        let line = line.strip_suffix('\r').unwrap_or(line);
        if let Some(i) = line.find('=').filter(|i| *i > 0) {
            values.insert(line[..i].to_string(), json!(&line[i + 1..]));
        }
    }
    values
}

pub fn public_job(values: &Map<String, Value>) -> Value {
    let get = |k: &str| values.get(k).and_then(Value::as_str).unwrap_or("").to_string();
    let error = get("error");
    let mut parts = error.split('|');
    let code = parts.next().unwrap_or("").to_string();
    let message = parts.collect::<Vec<_>>().join("|");
    let state = get("state");
    let alive = get("alive") == "1";
    let num = |k: &str| {
        let n = number_str(&get(k));
        if n.is_finite() && n != 0.0 { super::js::num(n) } else { json!(0) }
    };
    let or_null = |s: String| if s.is_empty() { Value::Null } else { json!(s) };
    let running = state == "running" || state == "queued";
    json!({
        "id": text(&get("id"), 64),
        "state": if text(&state, 20).is_empty() { "unknown".to_string() } else { text(&state, 20) },
        "step": num("step"), "total": num("total"),
        "created": or_null(text(&get("created"), 30)), "started": or_null(text(&get("started"), 30)), "finished": or_null(text(&get("finished"), 30)),
        "runner": or_null(text(&get("runner"), 16)), "title": text(&get("title"), 300),
        "alive": alive,
        "error": if code.is_empty() { Value::Null } else { json!({ "code": format!("DISKS_{}", text(&code, 64)), "message": text(&message, 600) }) },
        // "running" without a live runner means the outcome is unknown (crash/reboot): never restart it automatically.
        "outcome": if ["succeeded", "failed", "pending-reboot"].contains(&state.as_str()) { state.clone() } else if running && !alive { "unknown".into() } else if running { "running".into() } else { "unknown".into() }
    })
}

fn parse_step_markers(log: &str) -> Vec<Value> {
    STEP_MARKER.captures_iter(log).map(|c| json!({ "index": super::js::num(number_str(&c[1])), "total": super::js::num(number_str(&c[2])), "status": &c[3], "step": text(&c[4], 400) })).collect()
}

pub fn smart_summary(sections: &std::collections::HashMap<String, String>) -> Value {
    let smart: Value = sections.get("smartctl").and_then(|s| serde_json::from_str(s).ok()).unwrap_or(Value::Null);
    let (mut nvme_log, mut nvme_id) = (Value::Null, Value::Null);
    if let Some(nvme) = sections.get("nvme").filter(|s| !s.is_empty()) {
        for part in split_at_line_start(nvme, "@@") {
            let body = part.find('\n').map(|i| &part[i + 1..]).unwrap_or(part);
            if let Ok(value) = serde_json::from_str::<Value>(body) {
                if part.starts_with("smart") {
                    nvme_log = value;
                } else if part.starts_with("id") {
                    nvme_id = value;
                }
            }
        }
    }
    let messages: Vec<String> = smart["smartctl"]["messages"].as_array().map(|m| m.iter().map(|e| text(&super::js::string(e.get("string")), 200)).collect()).unwrap_or_default();
    let nvme_health = if truthy(&smart["nvme_smart_health_information_log"]) { smart["nvme_smart_health_information_log"].clone() } else { nvme_log.clone() };
    let attributes = smart["ata_smart_attributes"]["table"].as_array().cloned().unwrap_or_default();
    let attribute = |id: i64| attributes.iter().find(|e| e["id"] == id).map(|e| e["raw"]["value"].clone()).filter(|v| !v.is_null()).unwrap_or(Value::Null);
    let available = truthy(&smart["smart_status"]) || truthy(&nvme_health) || smart["smart_support"]["enabled"] == true;
    let first = |values: &[&Value]| values.iter().find(|v| !v.is_null()).map(|v| (*v).clone()).unwrap_or(Value::Null);
    let pick_text = |a: &Value, b: &Value, limit: usize| {
        let source = if truthy(a) { super::js::string(Some(a)) } else if truthy(b) { super::js::string(Some(b)) } else { String::new() };
        let t = text(&source, limit).trim().to_string();
        if t.is_empty() { Value::Null } else { json!(t) }
    };
    let temperature = if !smart["temperature"]["current"].is_null() {
        smart["temperature"]["current"].clone()
    } else if nvme_health.get("temperature").is_some() {
        let t = super::js::number(&nvme_health["temperature"]);
        super::js::num(if t > 200.0 { t - 273.0 } else { t })
    } else {
        Value::Null
    };
    let passed = if smart["smart_status"]["passed"].is_boolean() {
        smart["smart_status"]["passed"].clone()
    } else if truthy(&nvme_health) {
        json!(super::js::number(&nvme_health["critical_warning"]) == 0.0)
    } else {
        Value::Null
    };
    json!({
        "available": available,
        "tool": if truthy(&smart) && !truthy(&smart["chj_unavailable"]) { json!("smartctl") } else if truthy(&nvme_log) { json!("nvme-cli") } else { Value::Null },
        "reason": if available { Value::Null } else if truthy(&smart["chj_unavailable"]) { json!("SMARTCTL_MISSING") } else { json!(messages.first().filter(|m| !m.is_empty()).cloned().unwrap_or_else(|| "SMART_UNAVAILABLE".into())) },
        "messages": messages,
        "model": pick_text(&smart["model_name"], &nvme_id["mn"], 80),
        "serial": pick_text(&smart["serial_number"], &nvme_id["sn"], 80),
        "firmware": pick_text(&smart["firmware_version"], &nvme_id["fr"], 40),
        "capacity": smart["user_capacity"]["bytes"].clone(),
        "passed": passed,
        "temperatureC": temperature,
        "powerOnHours": first(&[&smart["power_on_time"]["hours"], &nvme_health["power_on_hours"]]),
        "powerCycles": first(&[&smart["power_cycle_count"], &nvme_health["power_cycles"]]),
        "reallocatedSectors": attribute(5), "pendingSectors": attribute(197), "offlineUncorrectable": attribute(198), "crcErrors": attribute(199),
        "percentageUsed": first(&[&nvme_health["percentage_used"], &nvme_health["percent_used"]]),
        "availableSpare": nvme_health["available_spare"].clone(),
        "mediaErrors": nvme_health["media_errors"].clone(),
        "errorLogEntries": first(&[&nvme_health["num_err_log_entries"], &smart["ata_smart_error_log"]["summary"]["count"]]),
        "dataUnitsWritten": nvme_health["data_units_written"].clone(),
        "selfTestPassed": smart["ata_smart_self_test_log"]["standard"]["table"][0]["status"]["passed"].clone()
    })
}

/// `text.split(/^@@/m)`.
fn split_at_line_start<'a>(text: &'a str, marker: &str) -> Vec<&'a str> {
    let mut parts = Vec::new();
    let mut start = 0;
    let mut search = 0;
    while let Some(found) = text[search..].find(marker) {
        let at = search + found;
        if at == 0 || text.as_bytes()[at - 1] == b'\n' {
            parts.push(&text[start..at]);
            start = at + marker.len();
        }
        search = at + marker.len();
    }
    parts.push(&text[start..]);
    parts
}

struct StoredPlan {
    session_id: String,
    identity: String,
    operations: Value,
    plan_text: String,
    title: Value,
    created_at: i64,
}

impl StoredPlan {
    fn check_session(&self, session_id: &str, identity: &str) -> CoreResult<()> {
        if self.session_id != session_id || self.identity != identity {
            return Err(fail("PLAN_SERVER_CHANGED", "The plan belongs to a different server or connection."));
        }
        Ok(())
    }
}

#[derive(Default)]
pub struct RunAuth {
    pub privileged: bool,
    pub sudo_password: Option<String>,
    pub remember: bool,
}

fn auth(payload: &Value, privileged: bool) -> RunAuth {
    RunAuth { privileged, sudo_password: payload["sudoPassword"].as_str().filter(|s| !s.is_empty()).map(str::to_string), remember: payload["rememberSudo"] == true }
}

pub struct DiskService {
    sessions: Arc<SessionManager>,
    executor: Arc<PrivilegedExecutor>,
    logger: Arc<Logger>,
    plans: Mutex<IndexMap<String, StoredPlan>>,
    busy: Mutex<HashSet<String>>,
}

impl DiskService {
    pub fn new(sessions: Arc<SessionManager>, executor: Arc<PrivilegedExecutor>, logger: Arc<Logger>) -> Self {
        Self { sessions, executor, logger, plans: Mutex::new(IndexMap::new()), busy: Mutex::new(HashSet::new()) }
    }

    fn identity(&self, session_id: &str) -> CoreResult<String> {
        let record = self.sessions.require_connected(session_id)?;
        Ok(json!([record.profile_id, record.host, record.port, record.username, record.public_status()["jump"]]).to_string())
    }

    async fn platform(&self, session_id: &str) -> CoreResult<(String, Value)> {
        let record = self.sessions.require_connected(session_id)?;
        let identity = self.identity(session_id)?;
        let platform = detect_platform(&self.sessions, &record, false).await?;
        Ok((identity, platform))
    }

    async fn run(&self, session_id: &str, args: Vec<String>, auth: RunAuth, frames: Vec<Frame>, timeout_ms: u64) -> CoreResult<RunResult> {
        let purpose = format!("disks.{}", args[0]);
        self.executor
            .run(
                session_id,
                RunOptions {
                    script: SCRIPT.into(),
                    args,
                    frames,
                    privileged: auth.privileged,
                    sudo_password: auth.sudo_password,
                    remember_password: auth.remember,
                    use_cached_password: auth.privileged,
                    purpose,
                    error_prefix: "DISKS_".into(),
                    timeout: Duration::from_millis(timeout_ms),
                    output_limit: 8 * 1024 * 1024,
                },
            )
            .await
    }

    /// `options`: `{ privileged?, sudoPassword?, rememberSudo? }`.
    pub async fn inspect(&self, session_id: &str, options: &Value) -> CoreResult<Value> {
        let (identity, platform) = self.platform(session_id).await?;
        if platform["os"] != "linux" {
            let caption = if truthy(&platform["caption"]) { super::js::string(Some(&platform["caption"])) } else { super::js::string(platform.get("kernel")) };
            return Ok(json!({
                "platform": platform["os"], "supported": false, "identity": identity,
                "os": super::js::text(&caption, 120),
                "reason": match platform["os"].as_str() { Some("windows") => "WINDOWS", Some("posix") => "POSIX_NOT_LINUX", _ => "UNKNOWN" },
                "collectedAt": iso(now_ms())
            }));
        }
        let privilege = self.executor.probe(session_id, false).await?;
        let cached = self.executor.has_cached_password(session_id);
        let privileged = options["privileged"] == true || truthy(&options["sudoPassword"]) || ["root", "sudo-nopasswd"].contains(&privilege["mode"].as_str().unwrap_or("")) || cached;
        let run = self.run(session_id, vec!["inspect".into()], auth(options, privileged), vec![], 120_000).await?;
        let inventory = build_inventory(&run.stdout);
        let suggestions = suggestions(&inventory);
        let mut out = inventory.as_object().cloned().unwrap_or_default();
        out.insert("identity".into(), json!(identity));
        out.insert("privilege".into(), self.executor.probe(session_id, false).await?);
        out.insert("canManage".into(), json!(privilege["mode"] != "none"));
        out.insert("sudoCached".into(), json!(self.executor.has_cached_password(session_id)));
        out.insert("suggestions".into(), Value::Array(suggestions));
        out.insert("collectedAt".into(), json!(iso(now_ms())));
        Ok(Value::Object(out))
    }

    pub async fn smart(&self, session_id: &str, payload: &Value) -> CoreResult<Value> {
        let (_, platform) = self.platform(session_id).await?;
        if platform["os"] != "linux" {
            return Err(fail("PLATFORM_UNSUPPORTED", "SMART is read only on Linux servers."));
        }
        let device = super::js::string(payload.get("device").filter(|v| truthy(v)));
        if !DEVICE.is_match(&device) || device.contains("..") {
            return Err(fail("ARGUMENTS_INVALID", "Invalid device."));
        }
        let run = self.run(session_id, vec!["smart".into(), device.clone()], auth(payload, true), vec![], 60_000).await?;
        let mut out = Map::new();
        out.insert("device".into(), json!(device));
        for (k, v) in smart_summary(&split_sections(&run.stdout)).as_object().cloned().unwrap_or_default() {
            out.insert(k, v);
        }
        out.insert("collectedAt".into(), json!(iso(now_ms())));
        Ok(Value::Object(out))
    }

    /// Validates and simulates a queue of operations against a fresh privileged inventory. Nothing changes on disk.
    pub async fn plan(&self, session_id: &str, payload: &Value) -> CoreResult<Value> {
        let (identity, platform) = self.platform(session_id).await?;
        if platform["os"] != "linux" {
            return Err(fail("PLATFORM_UNSUPPORTED", "Disk changes are supported only on Linux servers."));
        }
        let operations = if payload["operations"].is_array() { payload["operations"].clone() } else { json!([]) };
        let inventory = self.inspect(session_id, &json!({ "privileged": true, "sudoPassword": payload["sudoPassword"], "rememberSudo": payload["rememberSudo"] })).await?;
        let plan = plan_operations(&inventory, &operations, payload.get("title"))?;
        let plan_id = uuid::Uuid::new_v4().to_string();
        {
            let mut plans = self.plans.lock();
            let now = now_ms();
            plans.retain(|_, p| now - p.created_at <= PLAN_TTL_MS);
            while plans.len() >= MAX_PLANS {
                plans.shift_remove_index(0);
            }
            plans.insert(
                plan_id.clone(),
                StoredPlan { session_id: session_id.into(), identity, operations: plan["operations"].clone(), plan_text: super::js::string(plan.get("planText")), title: plan["title"].clone(), created_at: now },
            );
        }
        let mut out = Map::new();
        out.insert("planId".into(), json!(plan_id));
        for (k, v) in plan.as_object().cloned().unwrap_or_default() {
            out.insert(k, v);
        }
        out.insert("expiresAt".into(), json!(iso(now_ms() + PLAN_TTL_MS)));
        Ok(Value::Object(out))
    }

    pub async fn auto_expand(&self, session_id: &str, payload: &Value) -> CoreResult<Value> {
        let inventory = self.inspect(session_id, &json!({ "privileged": true, "sudoPassword": payload["sudoPassword"], "rememberSudo": payload["rememberSudo"] })).await?;
        if !truthy(&inventory["supported"]) {
            return Err(fail("PLATFORM_UNSUPPORTED", "Disk changes are supported only on Linux servers."));
        }
        let wanted = super::js::string(payload.get("suggestionId").filter(|v| truthy(v)));
        let suggestion = inventory["suggestions"].as_array().and_then(|s| s.iter().find(|e| e["id"] == wanted.as_str()).cloned()).ok_or_else(|| fail("SUGGESTION_GONE", "The expansion is no longer possible with the current layout; the view was refreshed."))?;
        if !truthy(&suggestion["available"]) {
            let code = if truthy(&suggestion["blocked"]["code"]) { super::js::string(Some(&suggestion["blocked"]["code"])) } else { "DISKS_FAILED".into() };
            let message = if truthy(&suggestion["blocked"]["message"]) { super::js::string(Some(&suggestion["blocked"]["message"])) } else { "The expansion is not possible.".into() };
            return Err(fail(code.strip_prefix("DISKS_").unwrap_or(&code), message));
        }
        self.plan(session_id, &json!({ "operations": suggestion["operations"], "title": format!("Auto-expand {}", super::js::string(suggestion.get("target"))), "sudoPassword": payload["sudoPassword"] })).await
    }

    pub fn cancel_plan(&self, plan_id: &Value) -> Value {
        let id = super::js::string(Some(plan_id).filter(|v| truthy(v)));
        json!({ "cancelled": self.plans.lock().shift_remove(&id).is_some() })
    }

    /// Starts a confirmed plan. The plan is rebuilt from a fresh inventory and must produce exactly the
    /// reviewed steps; a plan is single-use, so a retry after a lost connection can never start the
    /// same destructive job twice (the remote side also refuses an existing job ID).
    pub async fn apply(&self, session_id: &str, payload: &Value) -> CoreResult<Value> {
        let plan_id = super::js::string(payload.get("planId").filter(|v| truthy(v)));
        {
            let mut plans = self.plans.lock();
            let expired = plans.get(&plan_id).is_none_or(|p| now_ms() - p.created_at > PLAN_TTL_MS);
            if expired {
                plans.shift_remove(&plan_id);
                return Err(fail("PLAN_EXPIRED", "The plan expired or was already used; plan the operations again."));
            }
        }
        if payload["confirm"] != true {
            return Err(fail("CONFIRMATION_REQUIRED", "The plan must be confirmed explicitly."));
        }
        let identity = self.identity(session_id)?;
        {
            let plans = self.plans.lock();
            let stored = plans.get(&plan_id).ok_or_else(|| fail("PLAN_EXPIRED", "The plan expired or was already used; plan the operations again."))?;
            stored.check_session(session_id, &identity)?;
        }
        if !self.busy.lock().insert(session_id.to_string()) {
            return Err(fail("BUSY", "Another disk job is being started for this server."));
        }
        let result = self.apply_inner(session_id, &plan_id, payload).await;
        self.busy.lock().remove(session_id);
        result
    }

    async fn apply_inner(&self, session_id: &str, plan_id: &str, payload: &Value) -> CoreResult<Value> {
        let Some(stored) = self.plans.lock().shift_remove(plan_id) else { return Err(fail("PLAN_EXPIRED", "The plan expired or was already used; plan the operations again.")) };
        stored.check_session(session_id, &self.identity(session_id)?)?;
        let remember = payload["rememberSudo"] != false;
        let inventory = self.inspect(session_id, &json!({ "privileged": true, "sudoPassword": payload["sudoPassword"], "rememberSudo": remember })).await?;
        let fresh = plan_operations(&inventory, &stored.operations, Some(&stored.title)).map_err(|e| fail("PLAN_STALE", format!("The layout changed since the plan was reviewed: {}", e.message)))?;
        let plan_text = super::js::string(fresh.get("planText"));
        if plan_text != stored.plan_text {
            return Err(fail("PLAN_STALE", "The layout changed since the plan was reviewed; review the new plan."));
        }
        let job_id = uuid::Uuid::new_v4().to_string();
        let steps = fresh["steps"].clone();
        let title = super::js::string(Some(&stored.title));
        self.logger.warn("Disk job submitted.", json!({ "sessionId": session_id, "jobId": job_id, "title": title, "steps": steps.as_array().map(Vec::len) }));
        let frames = vec![Frame::text("PLAN", &format!("{plan_text}\n")), Frame::text("HELPER", SCRIPT), Frame::text("TITLE", &title)];
        let mut run_auth = auth(payload, true);
        run_auth.remember = remember;
        let run = self.run(session_id, vec!["job-start".into(), job_id.clone()], run_auth, frames, 90_000).await?;
        let values = key_values(split_sections(&run.stdout).get("job").map(String::as_str).unwrap_or(""));
        let get = |k: &str| values.get(k).and_then(Value::as_str).unwrap_or("").to_string();
        Ok(json!({ "jobId": job_id, "state": text(&get("state"), 20), "runner": text(&get("runner"), 16), "steps": steps }))
    }

    pub async fn jobs(&self, session_id: &str, payload: &Value) -> CoreResult<Value> {
        let (_, platform) = self.platform(session_id).await?;
        if platform["os"] != "linux" {
            return Ok(json!({ "jobs": [] }));
        }
        let run = self.run(session_id, vec!["job-list".into()], RunAuth { privileged: true, sudo_password: auth(payload, true).sudo_password, remember: false }, vec![], 30_000).await?;
        let section = split_sections(&run.stdout).get("jobs").cloned().unwrap_or_default();
        let jobs: Vec<Value> = section
            .split('\n')
            .map(|l| l.strip_suffix('\r').unwrap_or(l))
            .filter(|l| !l.is_empty())
            .map(|line| {
                let f: Vec<&str> = line.split('|').collect();
                let mut values = Map::new();
                for (i, key) in ["id", "state", "step", "total", "created", "finished", "alive", "title"].iter().enumerate() {
                    if let Some(v) = f.get(i) {
                        values.insert((*key).into(), json!(v));
                    }
                }
                public_job(&values)
            })
            .filter(|job| JOB_ID.is_match(job["id"].as_str().unwrap_or("")))
            .collect();
        Ok(json!({ "jobs": jobs }))
    }

    pub async fn job_status(&self, session_id: &str, payload: &Value) -> CoreResult<Value> {
        let job_id = super::js::string(payload.get("jobId").filter(|v| truthy(v)));
        if !JOB_ID.is_match(&job_id) {
            return Err(fail("ARGUMENTS_INVALID", "Invalid job ID."));
        }
        let requested = super::js::number(&payload["offset"]);
        let offset = if requested.is_finite() && requested != 0.0 { requested } else { 0.0 }.clamp(0.0, 1e12).floor();
        let run = self.run(session_id, vec!["job-status".into(), job_id, super::js::num_string(offset)], RunAuth { privileged: true, sudo_password: auth(payload, true).sudo_password, remember: false }, vec![], 30_000).await?;
        let sections = split_sections(&run.stdout);
        let values = key_values(sections.get("job").map(String::as_str).unwrap_or(""));
        let log = sections.get("log").cloned().unwrap_or_default();
        let mut out = public_job(&values).as_object().cloned().unwrap_or_default();
        out.insert("plan".into(), json!(sections.get("plan").map(|p| p.split('\n').map(|l| l.strip_suffix('\r').unwrap_or(l)).filter(|l| !l.is_empty()).map(|l| text(l, 400)).collect::<Vec<_>>()).unwrap_or_default()));
        out.insert("log".into(), json!(CONTROL_KEEP_LF.replace_all(&STEP_LINE.replace_all(&log, ""), "").to_string()));
        out.insert("steps".into(), json!(parse_step_markers(&log)));
        out.insert("logOffset".into(), super::js::num(offset + log.len() as f64));
        let size = number_str(values.get("logSize").and_then(Value::as_str).unwrap_or(""));
        out.insert("logSize".into(), super::js::num(if size.is_finite() { size } else { 0.0 }));
        Ok(Value::Object(out))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn plan_is_bound_to_connection_and_identity() {
        let stored = StoredPlan { session_id: "A".into(), identity: "server-A".into(), operations: json!([]), plan_text: String::new(), title: Value::Null, created_at: now_ms() };
        assert!(stored.check_session("A", "server-A").is_ok());
        assert_eq!(stored.check_session("B", "server-A").unwrap_err().code, "DISKS_PLAN_SERVER_CHANGED");
        assert_eq!(stored.check_session("A", "server-B").unwrap_err().code, "DISKS_PLAN_SERVER_CHANGED");
        assert!(stored.check_session("B", "server-B").is_err());
    }

    #[test]
    fn job_and_smart_parsing() {
        let job = public_job(&key_values("id=abcd-1234\nstate=running\nstep=2\ntotal=5\nalive=0\nerror=SFDISK_FAILED|bad|table\n"));
        assert_eq!(job["outcome"], "unknown");
        assert_eq!(job["error"]["code"], "DISKS_SFDISK_FAILED");
        assert_eq!(job["error"]["message"], "bad|table");
        let mut sections = std::collections::HashMap::new();
        sections.insert("smartctl".to_string(), r#"{"smart_status":{"passed":true},"model_name":"QEMU","temperature":{"current":35},"ata_smart_attributes":{"table":[{"id":5,"raw":{"value":0}}]}}"#.to_string());
        let smart = smart_summary(&sections);
        assert_eq!(smart["passed"], true);
        assert_eq!(smart["tool"], "smartctl");
        assert_eq!(smart["reallocatedSectors"], 0);
        assert_eq!(smart["temperatureC"], 35);
        let steps = parse_step_markers("x\n\x1eSTEP 1|3|ok|create-part /dev/vdb\nmore");
        assert_eq!(steps[0]["status"], "ok");
    }
}