liminal-server 0.25.4

//! Installs the services registry once the listener is bound (SRV-028 R5):
//! the registry is read by one scan, and the services last recorded live
//! are started off the start path, so the start line does not grow with
//! them.

use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;

use crate::ServerError;
use crate::config::ServerConfig;
use crate::services::lifecycle::{Lifecycle, LysPlacer, Placer};
use crate::services::lys_http::{HttpLys, LysHttp};

/// The startup draft pull (SRV-041 R4), made once the listener binds.
type StartupPull = Box<dyn FnOnce() + Send>;

/// Stops every service process when the server's run ends; their states
/// stay, so a live service starts again with the next start.
pub struct Services {
    lifecycle: Option<Arc<Lifecycle>>,
    /// Pending until the listener binds; then pulling on its own thread,
    /// joined when the stop begins, before the Lys runtime it asks on goes.
    pull: Mutex<(Option<StartupPull>, Option<JoinHandle<()>>)>,
}

impl std::fmt::Debug for Services {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("Services")
            .field("lifecycle", &self.lifecycle)
            .finish_non_exhaustive()
    }
}

impl Services {
    fn new(lifecycle: Option<Arc<Lifecycle>>, pull: Option<StartupPull>) -> Self {
        Self {
            lifecycle,
            pull: Mutex::new((pull, None)),
        }
    }

    /// Gives the registry the address the listener bound and starts every
    /// service last recorded live, off the start path; begins the startup
    /// draft pull on its own thread.
    ///
    /// # Errors
    /// The restart or pull thread could not start.
    pub fn listening(&self, address: std::net::SocketAddr) -> Result<(), ServerError> {
        self.begin_pull()?;
        let Some(lifecycle) = self.lifecycle.as_ref() else {
            return Ok(());
        };
        lifecycle.listening(address.to_string());
        let restarting = Arc::clone(lifecycle);
        std::thread::Builder::new()
            .name("liminal-services-restart".to_owned())
            .spawn(move || {
                let now_ms = std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .map_or(0, |since| {
                        u64::try_from(since.as_millis()).unwrap_or(u64::MAX)
                    });
                for (service, error) in restarting.restart_live(now_ms) {
                    tracing::warn!(%service, %error, "a live service could not be restarted");
                }
            })
            .map_err(|error| refused(format!("the services restart could not start: {error}")))?;
        Ok(())
    }
}

impl Services {
    fn begin_pull(&self) -> Result<(), ServerError> {
        let mut pull = self
            .pull
            .lock()
            .map_err(|_| refused("the startup draft pull lock is poisoned".to_owned()))?;
        if let Some(pending) = pull.0.take() {
            pull.1 = Some(
                std::thread::Builder::new()
                    .name("liminal-draft-startup".to_owned())
                    .spawn(pending)
                    .map_err(|error| {
                        refused(format!("the startup draft pull could not start: {error}"))
                    })?,
            );
        }
        drop(pull);
        Ok(())
    }

    /// The server's stop was asked: an exit from here keeps its service's
    /// recorded state (see [`Lifecycle::shutdown_begun`]). A startup draft
    /// pull still asking is waited for, its asks bounded by their timeouts.
    pub fn shutdown_begun(&self) {
        let pulling = self.pull.lock().map_or_else(
            |_| {
                tracing::error!("the startup draft pull lock is poisoned; the pull is not joined");
                None
            },
            |mut pull| pull.1.take(),
        );
        if let Some(pulling) = pulling
            && pulling.join().is_err()
        {
            tracing::error!("the startup draft pull panicked");
        }
        if let Some(lifecycle) = self.lifecycle.as_ref() {
            lifecycle.shutdown_begun();
        }
    }
}

impl Drop for Services {
    fn drop(&mut self) {
        if let Some(lifecycle) = self.lifecycle.take() {
            lifecycle.shutdown();
        }
    }
}

const fn refused(message: String) -> ServerError {
    ServerError::ConfigValidation { message }
}

/// Opens and installs the registry on the supervisor before any connection
/// is admitted, so no request finds it missing; holds nothing without
/// templates. [`Services::listening`] then gives it the bound address and
/// starts the restart of live services on its own thread.
///
/// # Errors
/// The registry, legacy placement journal or app credential cannot be read,
/// or Lys cannot be reached as configured.
pub fn install(
    config: &ServerConfig,
    supervisor: &crate::server::connection::ConnectionSupervisor,
    lys_ask: Option<&tokio::runtime::Runtime>,
    services: Option<&Arc<crate::server::connection::services::LiminalConnectionServices>>,
    health: &crate::health::HealthServerHandle,
) -> Result<Services, ServerError> {
    let Some(lys) = config.auth.as_ref().and_then(|auth| auth.lys.as_ref()) else {
        return Ok(Services::new(None, None));
    };
    let placement_path = config
        .persistence_path
        .as_ref()
        .map(|store| tap_placement_journal_path(store, &lys.network_id))
        .transpose()
        .map_err(refused)?;
    let client = match (lys.app_credential_path.as_ref(), lys_ask) {
        (Some(path), Some(runtime)) => {
            let credential = std::fs::read_to_string(path).map_err(|error| {
                refused(format!(
                    "auth.lys.app_credential_path '{}' cannot be read: {}",
                    path.display(),
                    error.kind()
                ))
            })?;
            let http: Arc<dyn LysHttp> = Arc::new(
                HttpLys::new(
                    lys.api_url.as_deref().unwrap_or(&lys.issuer),
                    runtime.handle().clone(),
                )
                .map_err(|error| refused(error.to_string()))?,
            );
            Some((http, credential.trim().to_owned()))
        }
        _ => None,
    };
    if config.services.profile()? == crate::config::types::ServiceProfile::Full
        && let (Some(path), Some((http, credential))) = (placement_path.as_ref(), client.as_ref())
    {
        let installed = crate::console::tap_placement::TapPlacements::start(
            path,
            &lys.network_id,
            Arc::clone(http),
            credential.clone(),
        )
        .and_then(|placements| supervisor.install_tap_placements(placements));
        if let Err(error) = installed {
            tracing::warn!(%error, "TAP_SUBJECT_PLACEMENT_INSTALL_FAILED: taps still serve; retry at next boot");
        }
    }
    if client.is_none() || config.persistence_path.is_none() {
        tracing::warn!(
            "TAP_SUBJECT_PLACEMENT_UNAVAILABLE: app_credential_path, persistence_path and a Lys runtime are required"
        );
    }
    if lys.templates.is_empty() && lys.construction.is_none() && lys.drafts.is_none() {
        return Ok(Services::new(None, None));
    }
    let store = config
        .persistence_path
        .as_ref()
        .ok_or_else(|| refused("auth.lys.templates: persistence_path is required".to_owned()))?;
    let placer: Option<Arc<dyn Placer>> = client.as_ref().map(|(http, credential)| {
        Arc::new(LysPlacer::new(http.clone(), credential.clone())) as Arc<dyn Placer>
    });
    let lifecycle = if let Some(construction) = &lys.construction {
        Lifecycle::open_bounded(
            &lys.templates,
            registry_dir(store),
            lys.network_id.clone(),
            placer,
            &construction.limits,
        )
    } else {
        Lifecycle::open(
            &lys.templates,
            registry_dir(store),
            lys.network_id.clone(),
            placer,
        )
    }
    .map_err(|error| refused(error.to_string()))?;
    let mut pull = None;
    if let Some(services) = services {
        if lys.construction.is_some() || lys.drafts.is_some() {
            let (http, credential) = client
                .as_ref()
                .ok_or_else(|| refused("construction needs an app connector".into()))?;
            let handle = lys_ask
                .ok_or_else(|| refused("construction needs a Lys runtime".into()))?
                .handle()
                .clone();
            pull = Some(services.install_approval_runtime(
                lys,
                lifecycle.clone(),
                http.clone(),
                credential.clone(),
                handle,
                health,
            )?);
        }
    }
    supervisor.install_lifecycle(Arc::clone(&lifecycle))?;
    Ok(Services::new(Some(lifecycle), pull))
}

/// The registry sits beside the store, never inside it: the store directory
/// belongs to its engine.
pub fn registry_dir(store: &std::path::Path) -> PathBuf {
    let name = store.file_name().map_or_else(
        || "liminal".into(),
        |name| name.to_string_lossy().into_owned(),
    );
    store.with_file_name(format!("{name}-services"))
}

/// Keeps the journal beside both persistence directories and migrates it
/// before the registry's service-record scan.
pub fn tap_placement_journal_path(
    store: &std::path::Path,
    network: &str,
) -> Result<PathBuf, String> {
    let name = store.file_name().map_or_else(
        || "liminal".into(),
        |name| name.to_string_lossy().into_owned(),
    );
    let path = store.with_file_name(format!("{name}-console-tap-placements.json"));
    crate::console::tap_placement::PlacementBook::relocate_legacy(
        &registry_dir(store).join("console-tap-placements.json"),
        &path,
        network,
    )?;
    Ok(path)
}

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

    #[test]
    fn the_registry_sits_beside_the_store() {
        assert_eq!(
            registry_dir(std::path::Path::new("/data/liminal/store")),
            std::path::PathBuf::from("/data/liminal/store-services")
        );
    }
}