sausage/cli/src/activation.rs

//! Serialized candidate-copy activation and revert under PROTOCOL.md section 10.3.
use crate::{
    Host, RpcResult, error,
    graph::{Interface, io_error, problems, read_hashed},
};
use serde_json::{Value, json};
use std::{collections::BTreeMap, path::PathBuf};

/// Method descriptions plus their reachable local references, without expanding cycles.
fn methods(document: &Value) -> BTreeMap<String, Value> {
    fn references(value: &Value, document: &Value, found: &mut BTreeMap<String, Value>) {
        match value {
            Value::Object(object) => {
                if let Some(reference) = object.get("$ref").and_then(Value::as_str)
                    && let Some(pointer) = reference.strip_prefix('#')
                    && !found.contains_key(reference)
                    && let Some(target) = document.pointer(pointer)
                {
                    found.insert(reference.into(), target.clone());
                    references(target, document, found);
                }
                for child in object.values() {
                    references(child, document, found);
                }
            }
            Value::Array(array) => {
                for child in array {
                    references(child, document, found);
                }
            }
            _ => {}
        }
    }
    document["methods"]
        .as_array()
        .into_iter()
        .flatten()
        .filter_map(|method| {
            let name = method["name"].as_str()?;
            let mut found = BTreeMap::new();
            references(method, document, &mut found);
            Some((name.into(), json!({"method":method,"references":found})))
        })
        .collect()
}

struct Candidate {
    plugin: String,
    copy: PathBuf,
    interface: Interface,
}
impl Host {
    /// Copies first, discovers that copy, checks the combined graph, then installs it.
    pub(crate) async fn activation(&self, _caller: &str, method: &str, params: Value) -> RpcResult {
        let host = self.clone();
        let method = method.to_owned();
        tokio::spawn(async move { host.activation_body(&method, params).await })
            .await
            .map_err(|failure| error(-32603, &failure.to_string()))?
    }
    async fn activation_body(&self, method: &str, params: Value) -> RpcResult {
        let _activation = self.0.activations.lock().await;
        let mut candidates = Vec::new();
        let result = self
            .activate_candidates(method, &params, &mut candidates)
            .await;
        for candidate in candidates {
            let _ = tokio::fs::remove_file(candidate.copy).await;
        }
        result
    }
    /// Prepares every member before graph checks or changes to installed files.
    async fn activate_candidates(
        &self,
        method: &str,
        params: &Value,
        candidates: &mut Vec<Candidate>,
    ) -> RpcResult {
        let requests = params
            .get("plugins")
            .and_then(Value::as_array)
            .cloned()
            .unwrap_or_else(|| vec![params.clone()]);
        if requests.is_empty() {
            return Err(error(-32602, "plugins must not be empty"));
        }
        for request in requests {
            let plugin = request["plugin"]
                .as_str()
                .ok_or_else(|| error(-32602, "plugin must be a string"))?;
            if plugin.is_empty()
                || !plugin.bytes().all(|byte| {
                    byte.is_ascii_lowercase()
                        || byte.is_ascii_digit()
                        || byte == b'_'
                        || byte == b'-'
                })
                || candidates
                    .iter()
                    .any(|candidate| candidate.plugin == plugin)
            {
                return Err(error(-32602, "invalid or duplicate plugin"));
            }
            let directory = self.0.world.plugin_dir(plugin);
            tokio::fs::create_dir_all(&directory)
                .await
                .map_err(io_error)?;
            let source = if method == "revert" {
                directory.join("plugin.prev.wasm")
            } else {
                let source = PathBuf::from(
                    request["wasm"]
                        .as_str()
                        .ok_or_else(|| error(-32602, "wasm must be a string"))?,
                );
                if source.starts_with(&self.0.world.root) {
                    source
                } else {
                    self.0
                        .world
                        .root
                        .join(source.strip_prefix("/").unwrap_or(&source))
                }
            };
            let copy = directory.join(format!("activate-{}.wasm", uuid::Uuid::now_v7()));
            // Read once. Discovery and installation both use the same private bytes.
            let (bytes, hash) = read_hashed(source).await.map_err(io_error)?;
            tokio::fs::write(&copy, &bytes).await.map_err(io_error)?;
            self.0
                .journal
                .write("core", plugin, "core", "activation copied candidate");
            let discovery = self.discover_copy(plugin, &copy, hash).await;
            let interface = match discovery {
                Ok(interface) => interface,
                Err(error) => {
                    let _ = tokio::fs::remove_file(copy).await;
                    return Err(error);
                }
            };
            self.0
                .journal
                .write("core", plugin, "core", "activation discovered candidate");
            candidates.push(Candidate {
                plugin: plugin.into(),
                copy,
                interface,
            });
        }
        let names: Vec<_> = candidates
            .iter()
            .map(|candidate| candidate.plugin.clone())
            .collect();
        self.refresh_graph_except(&names).await?;
        let mut previous = self.0.graph.lock().unwrap().plugins.clone();
        // Startup discovery may not yet have visited a replaced plugin. Discover
        // its immutable old copy for the diff, but never prevent repairing it.
        for candidate in candidates.iter() {
            if let Some(old) = previous.get_mut(&candidate.plugin)
                && old.document.is_null()
                && old.error.is_none()
            {
                if old.hash == candidate.interface.hash {
                    old.document = candidate.interface.document.clone();
                    continue;
                }
                let directory = self.0.world.plugin_dir(&candidate.plugin);
                if let Ok((bytes, hash)) = read_hashed(directory.join("plugin.wasm")).await {
                    let copy = directory.join(format!("discover-{}.wasm", uuid::Uuid::now_v7()));
                    if tokio::fs::write(&copy, bytes).await.is_ok()
                        && let Ok(interface) =
                            self.discover_copy(&candidate.plugin, &copy, hash).await
                    {
                        *old = interface;
                    }
                    let _ = tokio::fs::remove_file(copy).await;
                }
            }
        }
        let mut proposed = previous.clone();
        for candidate in candidates.iter() {
            proposed.insert(candidate.plugin.clone(), candidate.interface.clone());
        }
        if params["force"] != true {
            let failures = problems(&proposed, Some(&names));
            if !failures.is_empty() {
                self.0.journal.write(
                    "core",
                    "core",
                    "core",
                    &format!("activation rejected {}", json!(failures)),
                );
                let details = failures
                    .iter()
                    .map(|problem| {
                        format!(
                            "- {} imports {}: {}",
                            problem["consumer"],
                            problem["provider"],
                            problem["display"]
                                .as_str()
                                .unwrap_or_else(|| problem["message"]
                                    .as_str()
                                    .unwrap_or("incompatible import"))
                        )
                    })
                    .collect::<Vec<_>>()
                    .join("\n");
                return Err(
                    json!({"code":-32003,"message":format!("{} was not activated: consumers break.\n{details}\nUpdate consumers first, or activate both together.",names.join(", ")),"data":failures}),
                );
            }
        }
        self.install_candidates(candidates, previous, proposed, params["force"] != true)
            .await
    }
    /// Installs staged files and snapshots all old admissions under the shared write gate.
    async fn install_candidates(
        &self,
        candidates: &[Candidate],
        previous: BTreeMap<String, Interface>,
        proposed: BTreeMap<String, Interface>,
        checked: bool,
    ) -> RpcResult {
        let admission = self.0.admission.write().await;
        // Every candidate already has a temporary name before any installed file changes.
        for candidate in candidates {
            if previous
                .get(&candidate.plugin)
                .is_some_and(|old| old.hash == candidate.interface.hash)
            {
                continue;
            }
            let directory = self.0.world.plugin_dir(&candidate.plugin);
            let current = directory.join("plugin.wasm");
            if tokio::fs::try_exists(&current).await.map_err(io_error)? {
                let backup = candidate.copy.with_extension("previous");
                tokio::fs::hard_link(&current, &backup)
                    .await
                    .map_err(io_error)?;
                tokio::fs::rename(&backup, directory.join("plugin.prev.wasm"))
                    .await
                    .map_err(io_error)?;
            }
            tokio::fs::rename(&candidate.copy, &current)
                .await
                .map_err(io_error)?;
        }
        let mut upgrades = Vec::new();
        let mut summaries = serde_json::Map::new();
        for candidate in candidates {
            let old = previous.get(&candidate.plugin);
            let before = methods(old.map(|old| &old.document).unwrap_or(&Value::Null));
            let after = methods(&candidate.interface.document);
            let added: Vec<_> = after
                .keys()
                .filter(|name| !before.contains_key(*name))
                .collect();
            let removed: Vec<_> = before
                .keys()
                .filter(|name| !after.contains_key(*name))
                .collect();
            let changed: Vec<_> = after
                .iter()
                .filter(|(name, method)| before.get(*name).is_some_and(|old| old != *method))
                .map(|(name, _)| name)
                .collect();
            summaries.insert(candidate.plugin.clone(), json!({
                "plugin": candidate.plugin,
                "wasm": format!("/plugins/{}/plugin.wasm", candidate.plugin),
                "hash": candidate.interface.hash,
                "previous": old.map(|old| &old.hash).filter(|hash| !hash.is_empty()),
                "method_count": candidate.interface.document["methods"].as_array().map_or(0, Vec::len),
                "added": added,
                "removed": removed,
                "changed": changed,
                "interface_check": if checked { "passed" } else { "skipped" },
            }));
            if let Some(old) = previous
                .get(&candidate.plugin)
                .filter(|old| !old.hash.is_empty() && old.hash != candidate.interface.hash)
            {
                let grace = old.document["x-sausage-grace"]
                    .as_f64()
                    .filter(|value| value.is_finite() && *value >= 0.0)
                    .unwrap_or(60.0);
                upgrades.push(self.snapshot_upgrade(
                    &candidate.plugin,
                    old.hash.clone(),
                    &candidate.interface.hash,
                    grace,
                ));
            }
            self.0.journal.write(
                "core",
                &candidate.plugin,
                "core",
                &format!("activation installed {}", candidate.interface.hash),
            );
        }
        self.publish_graph(proposed);
        drop(admission);
        for upgrade in upgrades {
            self.begin_upgrade(upgrade);
        }
        Ok(Value::Object(summaries))
    }
}