sausage/cli/src/host.rs

//! Host startup, boot calls, and the public library API.
use crate::{RpcResult, components, journal, runtime::Runtime, world::World};
use crate::{
    locks::Lock,
    native::NativePlugin,
    process::{Process, ProcessInfo},
};
use serde_json::{Value, json};
use std::{
    collections::HashMap,
    path::Path,
    sync::{Arc, Mutex},
};
use tokio::sync::watch;
use uuid::Uuid;
pub(crate) struct State {
    pub(crate) processes: HashMap<String, Process>,
    pub(crate) locks: HashMap<String, Lock>,
    pub(crate) native: HashMap<String, NativePlugin>,
    pub(crate) services: Vec<crate::services::Service>,
}
pub(crate) struct Inner {
    pub(crate) world: World,
    pub(crate) runtime: Arc<Runtime>,
    pub(crate) state: Mutex<State>,
    pub(crate) journal: Arc<journal::Journal>,
    pub(crate) notifications: tokio::sync::broadcast::Sender<Value>,
    pub(crate) exit: watch::Sender<bool>,
    pub(crate) loop_supervised: bool,
    pub(crate) restart_pending: std::sync::atomic::AtomicBool,
    pub(crate) components: components::Components,
    /// Calls hold it for reading from the choice of a version until their
    /// process is in the table. Activation holds it for writing while it
    /// installs a version, so that it sees every process of the old version.
    pub(crate) admission: tokio::sync::RwLock<()>,
    /// Set when the host starts to stop. After that, the core starts no new
    /// processes, and upgrade tasks do nothing.
    pub(crate) shutdown: std::sync::atomic::AtomicBool,
    pub(crate) run: sausage_vfs::Memory,
    pub(crate) watches: crate::watches::Watches,
    pub(crate) observer: sausage_vfs::observe::Observer,
    pub(crate) graph: Mutex<crate::graph::Graph>,
    pub(crate) activations: tokio::sync::Mutex<()>,
    pub(crate) watcher: Mutex<Option<notify::RecommendedWatcher>>,
    pub(crate) retirements: Mutex<std::collections::HashSet<(String, String)>>,
}
#[derive(Clone)]
/// A cloneable handle to a world and its running processes.
pub struct Host(pub(crate) Arc<Inner>);

impl Host {
    /// Start a world with the default compile cache and its ordered boot calls.
    pub async fn start(path: &Path) -> anyhow::Result<Self> {
        let base = std::env::var_os("XDG_CACHE_HOME")
            .map(std::path::PathBuf::from)
            .unwrap_or_else(|| {
                std::path::PathBuf::from(std::env::var_os("HOME").unwrap_or_default())
                    .join(".cache")
            });
        Self::start_with_cache(path, &base.join("sausage")).await
    }
    /// Start a world with an explicit compile cache folder.
    pub async fn start_with_cache(path: &Path, cache: &Path) -> anyhow::Result<Self> {
        let path = path.to_path_buf();
        let cargo_cache = cache.to_path_buf();
        let cache = cache.to_path_buf();
        let (world, runtime) = tokio::task::spawn_blocking(move || -> anyhow::Result<_> {
            Ok((World::open(&path)?, Runtime::new(&cache)?))
        })
        .await??;
        let journal = journal::Journal::open_with_legacy(
            world.root.join("logs/core/journal.sqlite"),
            Some(world.state_dir("core").join("journal.sqlite")),
        )
        .await?;
        let host = Self::with_parts(world, runtime, journal);
        let cargo_world = host.0.world.clone();
        let cargo = tokio::task::spawn_blocking(move || {
            crate::cargo::Cargo::new(cargo_world, &cargo_cache)
        })
        .await??;
        cargo.register(&host);
        host.boot().await?;
        Ok(host)
    }
    /// Creates an empty test world with a shared engine and an in-memory journal.
    pub async fn test_world(path: &Path, runtime: Arc<Runtime>) -> anyhow::Result<Self> {
        std::fs::create_dir_all(path)?;
        Ok(Self::with_parts(
            World::open(path)?,
            runtime,
            journal::Journal::memory().await?,
        ))
    }
    fn with_parts(world: World, runtime: Arc<Runtime>, journal: journal::Journal) -> Self {
        let watches = crate::watches::Watches::default();
        let events = watches.clone();
        let observer =
            sausage_vfs::observe::Observer::new(move |path| crate::watches::changed(&events, path));
        Self(Arc::new(Inner {
            world,
            runtime,
            journal: Arc::new(journal),
            state: Mutex::new(State {
                processes: HashMap::new(),
                locks: HashMap::new(),
                native: HashMap::new(),
                services: Vec::new(),
            }),
            notifications: tokio::sync::broadcast::channel(1024).0,
            exit: watch::channel(false).0,
            loop_supervised: crate::restart::enabled(std::env::var_os("SAUSAGE_LOOP").as_deref()),
            restart_pending: std::sync::atomic::AtomicBool::new(false),
            components: components::Components::default(),
            admission: tokio::sync::RwLock::new(()),
            shutdown: std::sync::atomic::AtomicBool::new(false),
            run: sausage_vfs::Memory::default(),
            watches,
            observer,
            graph: Mutex::new(crate::graph::Graph::default()),
            activations: tokio::sync::Mutex::new(()),
            watcher: Mutex::new(None),
            retirements: Mutex::new(std::collections::HashSet::new()),
        }))
    }
    /// Register a native plugin for subsequent calls.
    pub fn register_native(&self, name: &str, plugin: NativePlugin) {
        self.0
            .state
            .lock()
            .unwrap()
            .native
            .insert(name.into(), plugin);
    }
    /// Subscribe to notifications addressed to the console.
    pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<Value> {
        self.0.notifications.subscribe()
    }
    /// Snapshot the metadata of each process that has not ended.
    pub fn processes(&self) -> HashMap<String, ProcessInfo> {
        self.0
            .state
            .lock()
            .unwrap()
            .processes
            .iter()
            .map(|(p, s)| (p.clone(), s.info.clone()))
            .collect()
    }
    /// Make a console-side call and wait for its JSON result.
    pub async fn call(&self, method: &str, params: Value) -> RpcResult {
        self.route(
            "console".into(),
            Some(json!(Uuid::now_v7().to_string())),
            method.into(),
            params,
            false,
        )
        .await
    }
    /// Wait until core.restart (or its alias core.exit) requests shutdown after its response grace.
    pub async fn exit_requested(&self) {
        let mut exit = self.0.exit.subscribe();
        let _ = exit.wait_for(|requested| *requested).await;
    }
    /// Return whether a restart has reached the normal shutdown path.
    pub fn restart_requested(&self) -> bool {
        *self.0.exit.borrow()
    }
}