liminal-server 0.25.4

//! Contracts-first construction, with immutable acts and atomic receipts.

use std::os::unix::fs::PermissionsExt;
use std::sync::Arc;

use serde::{Deserialize, Serialize};

use super::{Lifecycle, LifecycleError, ServiceRecord, ServiceState, Transition};
use crate::config::templates::is_segment;

#[path = "site/acts.rs"]
mod acts;
#[path = "site/effects.rs"]
pub mod effects;
#[path = "site/executor.rs"]
pub mod executor;
#[path = "site/outright.rs"]
mod outright;
#[path = "site/preparation.rs"]
pub mod preparation;
#[path = "site/records.rs"]
mod records;
#[path = "site/storage.rs"]
pub mod storage;
#[path = "site/topology.rs"]
mod topology;
pub use records::{
    Approval, BuilderGrants, ConstructionConfig, Contract, SiteAct, SiteChange, SiteLimits,
    SiteSpec,
};

/// All construction refusals contain boundary names, never rejected payloads.
#[derive(Debug, thiserror::Error)]
pub enum SiteError {
    /// The boundary named here is missing or invalid.
    #[error("site_invalid: {0}")]
    Invalid(&'static str),
    /// A configured count or byte bound was reached.
    #[error("site_bound_exceeded: {0}")]
    Bound(&'static str),
    /// Safe business words generated from validated routing metadata.
    #[error("{0}")]
    Refused(String),
    /// Admission is poisoned after an uncertain durable operation.
    #[error("site_store_unavailable")]
    Store,
    /// The admitted pass's own refusal, without token or payload bytes.
    #[error("site_pass_refused: {0}")]
    Pass(String),
    /// The existing lifecycle failed.
    #[error("{0}")]
    Lifecycle(#[from] LifecycleError),
}

/// The schema strings are carried exactly; equivalence compares parsed shapes.
///
/// # Errors
/// Missing or invalid schema, reserved subject, or configured boundary refusal.
pub fn validate_contracts(spec: &SiteSpec, limits: &SiteLimits) -> Result<(), SiteError> {
    if !is_segment(&spec.name) || spec.name.len() > limits.text_bytes.get() {
        return Err(SiteError::Invalid("name"));
    }
    if spec.purpose.trim().is_empty()
        || spec.purpose.contains(['\n', '\r'])
        || spec.purpose.len() > limits.text_bytes.get()
    {
        return Err(SiteError::Invalid("purpose: one sentence"));
    }
    if spec.subjects.is_empty() || spec.subjects.len() > limits.subjects.get() {
        return Err(SiteError::Bound("subjects"));
    }
    let mut seen = std::collections::BTreeSet::new();
    for subject in &spec.subjects {
        if subject.len() > limits.text_bytes.get()
            || !subject.split('.').all(is_segment)
            || subject.starts_with("console.")
            || subject.starts_with("agents.")
            || subject.starts_with("services.")
            || crate::server::connection::services::is_site_route(subject)
            || !seen.insert(subject)
        {
            return Err(SiteError::Invalid("subject"));
        }
        let contract = spec
            .contracts
            .get(subject)
            .ok_or_else(|| SiteError::Refused(format!("subject {subject} has no schema")))?;
        if contract.input_schema_json.is_empty() || contract.output_schema_json.is_empty() {
            return Err(SiteError::Refused(format!(
                "subject {subject} has no schema"
            )));
        }
        for (kind, bytes) in [
            ("input", &contract.input_schema_json),
            ("output", &contract.output_schema_json),
        ] {
            if bytes.len() > limits.schema_bytes.get() {
                return Err(SiteError::Bound("schema_bytes"));
            }
            let value = serde_json::from_str(bytes).map_err(|_| {
                SiteError::Refused(format!("subject {subject} has no valid {kind} schema"))
            })?;
            liminal::channel::Schema::new(value).map_err(|_| {
                SiteError::Refused(format!("subject {subject} has no valid {kind} schema"))
            })?;
        }
    }
    if spec.contracts.len() != seen.len() {
        return Err(SiteError::Invalid("contracts name undeclared subjects"));
    }
    Ok(())
}

/// Creates `dir` private to the server (mode 0700) when it does not exist,
/// whatever the process umask; an existing directory keeps its mode and is
/// judged by the check that follows, since it is the operator's (SRV-036).
pub(super) fn create_private_dir(dir: &std::path::Path) -> std::io::Result<()> {
    use std::os::unix::fs::DirBuilderExt;
    if dir.try_exists()? {
        return Ok(());
    }
    if let Some(parent) = dir.parent() {
        std::fs::create_dir_all(parent)?;
    }
    std::fs::DirBuilder::new().mode(0o700).create(dir)
}

impl Lifecycle {
    /// Opens construction with explicit migration bounds before parsing any old
    /// file. Ordinary configuration-only registries retain their existing path.
    ///
    /// # Errors
    /// Invalid old snapshot, nonregular file, count/byte bound, or migration write.
    pub fn open_bounded(
        templates: &[crate::config::types::ServiceTemplateConfig],
        dir: std::path::PathBuf,
        network: String,
        placer: Option<Arc<dyn super::Placer>>,
        limits: &SiteLimits,
    ) -> Result<Arc<Self>, LifecycleError> {
        create_private_dir(&dir).map_err(|error| super::store(&dir, &error))?;
        if std::fs::metadata(&dir)
            .map_err(|error| super::store(&dir, &error))?
            .permissions()
            .mode()
            & 0o022
            != 0
        {
            return Err(LifecycleError::Store(
                "construction_registry_writable_by_others".into(),
            ));
        }
        let snapshot = storage::load_bounded(&dir, limits)?;
        Ok(Self::open_snapshot(
            templates, dir, network, placer, snapshot,
        ))
    }

    /// Installs operator bounds and the grant effect once, before admission.
    /// The retained projection shares the existing scheduler and inbox bounds.
    ///
    /// # Errors
    /// Repeated installation or invalid restored construction records.
    pub fn configure_sites(
        &self,
        limits: SiteLimits,
        store: Arc<dyn liminal::durability::DurableStore>,
        templates: Option<super::templates::TemplatePolicy>,
        effects: Arc<dyn effects::SiteEffects>,
        retained: Arc<crate::agents::retained::Retained>,
    ) -> Result<(), SiteError> {
        if std::fs::metadata(&self.dir)
            .map_err(|_| SiteError::Store)?
            .permissions()
            .mode()
            & 0o022
            != 0
        {
            return Err(SiteError::Invalid(
                "construction registry must not be writable by group or others",
            ));
        }
        self.site_config
            .set(storage::SiteBindings {
                store,
                limits,
                templates,
                effects,
                retained,
            })
            .map_err(|_| SiteError::Invalid("construction already configured"))?;
        let mut registry = self.lock()?;
        storage::restore(self, &mut registry)?;
        drop(registry);
        self.validate_site_storage()?;
        let registry = self.lock()?;
        for record in registry
            .records
            .values()
            .filter(|record| record.purpose.is_some())
        {
            validate_contracts(
                &SiteSpec {
                    name: record.name.clone(),
                    purpose: record.purpose.clone().ok_or(SiteError::Store)?,
                    subjects: record.subjects.clone(),
                    contracts: record.contracts.clone(),
                },
                &self.site_bindings()?.limits,
            )?;
            self.publish_plan(record)?;
        }
        drop(registry);
        Ok(())
    }

    pub(super) fn site_bindings(&self) -> Result<&storage::SiteBindings, SiteError> {
        self.site_config.get().ok_or(SiteError::Invalid(
            "[auth.lys.construction] is not configured",
        ))
    }

    pub(super) fn building(&self, id: &str) -> Result<ServiceRecord, SiteError> {
        let record = self.record(id)?;
        if record.state != ServiceState::Building {
            return Err(SiteError::Refused(format!(
                "service {} is not building",
                record.name
            )));
        }
        Ok(record)
    }

    pub(super) fn publish_plan(&self, record: &ServiceRecord) -> Result<(), SiteError> {
        let retained = &self.site_bindings()?.retained;
        let subject = format!("services.{}.plan", record.name);
        let value = super::plan::current_value(&record.plan);
        if retained
            .value(&subject)
            .map_err(|_| SiteError::Store)?
            .as_ref()
            == Some(&value)
        {
            return Ok(());
        }
        retained
            .publish(&[(subject, value)])
            .map_err(|_| SiteError::Store)
    }

    pub(super) fn site_transition(
        record: &mut ServiceRecord,
        holder: &str,
        at_ms: u64,
        words: &str,
    ) {
        record.transitions.push(Transition {
            from: record.state,
            to: record.state,
            holder: holder.into(),
            at_ms,
            note: Some(words.into()),
        });
    }
}

/// A construction snapshot never exposes its free text through Debug.
#[derive(Clone, Default, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub(super) struct Construction {
    pub templates: std::collections::BTreeMap<String, super::templates::TemplateRecord>,
    pub drafts: std::collections::BTreeMap<String, executor::Prepared>,
    pub outcomes: std::collections::BTreeMap<String, executor::Outcome>,
}

#[cfg(test)]
#[path = "site/test_support.rs"]
pub(crate) mod test_support;
#[cfg(test)]
#[path = "site_tests.rs"]
mod tests;

impl Lifecycle {
    /// Network placement used for the caller's exact permission binding.
    #[must_use]
    pub fn network_id(&self) -> &str {
        &self.network
    }
    /// Returns the explicitly configured positive construction limits.
    ///
    /// # Errors
    /// Construction is not configured.
    pub fn construction_limits(&self) -> Result<SiteLimits, SiteError> {
        Ok(self.site_bindings()?.limits.clone())
    }
    /// Subscribes to the current plan before any later changes, under its writer lock.
    ///
    /// # Errors
    /// Missing subject, watcher bound, schema or inbox refusal.
    pub fn subscribe_plan(
        &self,
        id: u64,
        subject: &str,
        schemas: &[liminal::protocol::SchemaId],
        install: Option<liminal::channel::InboxInstall>,
    ) -> Result<crate::server::connection::services::ConnectionSubscription, SiteError> {
        self.site_bindings()?
            .retained
            .subscribe(id, subject, schemas, install)
            .map_err(|error| SiteError::Refused(error.to_string()))
    }
}

impl Lifecycle {
    /// Installs the existing hub's event signal without another publisher or timer.
    ///
    /// # Errors
    /// The signal was already bound.
    pub fn bind_site_console(
        &self,
        console: &Arc<crate::console::ConsoleHub>,
    ) -> Result<(), SiteError> {
        self.site_console
            .set(Arc::downgrade(console))
            .map_err(|_| SiteError::Invalid("site console already bound"))
    }
    /// Repairs retained current values after a durable outcome or restart.
    ///
    /// # Errors
    /// A target, retained stream or hub signal failed.
    pub fn refresh_sites(&self) -> Result<(), SiteError> {
        let registry = self.lock()?;
        for record in registry
            .records
            .values()
            .filter(|record| record.purpose.is_some())
        {
            self.publish_plan(record)?;
        }
        drop(registry);
        self.notify_sites()
    }
    pub(super) fn notify_sites(&self) -> Result<(), SiteError> {
        if let Some(console) = self.site_console.get().and_then(std::sync::Weak::upgrade) {
            console
                .agent_changed()
                .map_err(|_| SiteError::Invalid("site topology signal failed"))?;
        }
        Ok(())
    }
}