liminal_server/services/site/

server.rs

//! Construction routes borrow the admitted session and the approval owner.

use super::super::state::FrameAction;
use liminal::protocol::{CONVERSATION_REPLY_REQUESTED_FLAG, Frame, MessageEnvelope};
use serde::Deserialize;
use serde_json::Value;
use std::sync::Arc;

use super::LiminalConnectionServices;
use crate::auth_lys::session::LysSession;
use crate::auth_lys::{AccessAction, AccessResource, HotAction, SubjectBinding};
use crate::services::lifecycle::{Lifecycle, plan, site};
use site::{SiteChange, SiteError};

/// The approval owner derives `SiteAct` from verified claims, performs the exact
///
/// Lys check with immutable words, persists the prepared request after holding,
/// and calls the one waiting writer. Outright never executes a construction act.
pub trait SiteDraftHolder: Send + Sync + std::fmt::Debug {
    /// Holds one validated act with the caller's own admitted pass.
    ///
    /// # Errors
    /// Missing held-mode right, issuer refusal, or prepared/waiting store failure.
    fn hold(
        &self,
        lifecycle: &Lifecycle,
        session: &dyn LysSession,
        operation: &str,
        target: &str,
        change: SiteChange,
        at_ms: u64,
    ) -> Result<Value, SiteError>;
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct Request {
    operation: String,
    payload: Value,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct PlanSet {
    service: String,
    steps: Vec<Title>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct Title {
    title: String,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct PlanStep {
    service: String,
    step: usize,
    status: plan::StepStatus,
    note: Option<String>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct Propose {
    service: String,
    command: std::path::PathBuf,
    args: Vec<String>,
    env: Vec<String>,
    stop_timeout_ms: u64,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct Bind {
    service: String,
    agent: String,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct Target {
    service: String,
}

/// Closed built-in route inventory, before ordinary worker/channel routing.
#[must_use]
pub fn is_site_route(subject: &str) -> bool {
    matches!(
        subject,
        "services.create_site"
            | "services.bind_builder"
            | "services.unbind_builder"
            | "services.plan_set"
            | "services.plan_step"
            | "services.propose_template"
            | "services.complete"
            | "templates.update"
            | "templates.retire"
    )
}

fn parse<T: serde::de::DeserializeOwned>(value: Value) -> Result<T, SiteError> {
    serde_json::from_value(value).map_err(|_| SiteError::Invalid("construction payload"))
}

/// Maps the plan's exact subject to its site's plan-only right. All other
/// resources remain on the already installed agent/ordinary pass evaluator.
///
/// # Errors
/// The verifier's own expiry, audience or rights error.
pub fn evaluate_subject<'a>(
    pass: &'a lys_pass::VerifiedPass,
    resource: &AccessResource,
    action: AccessAction,
    now: u64,
) -> Result<lys_pass::Decision<'a>, lys_pass::Error> {
    if resource.kind == crate::auth_lys::ResourceKind::Subject && action == AccessAction::Publish {
        if let Some(name) = plan_subject(&resource.id) {
            return pass.evaluate("liminal.service", name, "plan", now);
        }
    }
    crate::agents::binding::evaluate_subject(pass, resource, action, now)
}

/// Only exact service-name plan subjects are server-owned retained streams.
#[must_use]
pub fn plan_subject(subject: &str) -> Option<&str> {
    let name = subject.strip_prefix("services.")?.strip_suffix(".plan")?;
    crate::config::templates::is_segment(name).then_some(name)
}

fn authorize_plan(
    session: &dyn LysSession,
    lifecycle: &Lifecycle,
    service: &str,
    at_ms: u64,
) -> Result<(), SiteError> {
    let record = lifecycle.record(service)?;
    if session.holder_kind() != Some("agent") {
        return Err(SiteError::Refused(format!(
            "pass holder {} is not the builder of {}",
            session.holder(),
            record.name
        )));
    }
    lifecycle.check_plan_holder(&record, session.holder())?;
    let subject = format!("services.{}.plan", record.name);
    let binding = SubjectBinding::new(lifecycle.network_id().into(), subject.clone(), subject)
        .map_err(|_| SiteError::Invalid("plan binding"))?;
    session
        .decide(&binding, HotAction::Publish, at_ms / 1000)
        .map_err(|error| SiteError::Pass(error.to_string()))?;
    Ok(())
}

fn prepare_change(
    lifecycle: &Lifecycle,
    subject: &str,
    payload: Value,
    limits: &site::SiteLimits,
) -> Result<(String, SiteChange), SiteError> {
    Ok(match subject {
        "services.create_site" => {
            let spec: site::SiteSpec = parse(payload)?;
            site::validate_contracts(&spec, limits)?;
            (spec.name.clone(), SiteChange::CreateSite(spec))
        }
        "services.bind_builder" => {
            let body: Bind = parse(payload)?;
            (body.service, SiteChange::BindBuilder { agent: body.agent })
        }
        "services.unbind_builder" => {
            let body: Target = parse(payload)?;
            (body.service, SiteChange::UnbindBuilder)
        }
        "services.complete" => {
            let body: Target = parse(payload)?;
            // Refused before a draft is held, so no person is asked to approve what cannot run.
            let record = lifecycle
                .record(&body.service)
                .map_err(|_| SiteError::Refused(format!("unknown_service: {}", body.service)))?;
            plan::refuse_unfinished(&record.plan)?;
            (body.service, SiteChange::Complete)
        }
        "services.propose_template" | "templates.update" => {
            let body: Propose = parse(payload)?;
            let proposal = if subject == "templates.update" {
                lifecycle.prepare_template_update(
                    &body.service,
                    &body.command,
                    body.args,
                    body.env,
                    body.stop_timeout_ms,
                )?
            } else {
                lifecycle.propose_template(
                    &body.service,
                    &body.command,
                    body.args,
                    body.env,
                    body.stop_timeout_ms,
                )?
            };
            let target = proposal.name.clone();
            (
                target,
                if subject == "templates.update" {
                    SiteChange::UpdateTemplate(proposal)
                } else {
                    SiteChange::CreateTemplate(proposal)
                },
            )
        }
        "templates.retire" => {
            let body: Target = parse(payload)?;
            (body.service, SiteChange::RetireTemplate)
        }
        _ => return Err(SiteError::Invalid("construction route")),
    })
}

impl LiminalConnectionServices {
    /// Installs the lifecycle projection once, using the same channel scheduler
    /// and the configured agent watcher budgets. No second scheduler is built.
    ///
    /// # Errors
    /// Missing construction bounds, machine pass, agent registry or issuer.
    pub fn install_construction(
        &self,
        lifecycle: Arc<Lifecycle>,
        lys: &crate::config::types::LysConfig,
        http: Arc<dyn crate::services::lys_http::LysHttp>,
        drafts: Arc<dyn SiteDraftHolder>,
    ) -> Result<(), SiteError> {
        let config = lys
            .construction
            .as_ref()
            .ok_or(SiteError::Invalid("[auth.lys.construction]"))?;
        let agents = self
            .agents
            .as_ref()
            .ok_or(SiteError::Invalid("[auth.lys.agents]"))?;
        let pass = lys
            .machine
            .as_ref()
            .map(crate::link_pass::MachinePass::configured)
            .transpose()
            .map_err(|_| SiteError::Invalid("construction machine pass"))?;
        let effects = Arc::new(site::effects::LysSiteEffects::new(
            http,
            pass,
            Arc::clone(agents),
            site::effects::GrantSources {
                placement: lys
                    .app_credential_path
                    .clone()
                    .map(|path| (path, lys.network_id.clone())),
                ..config.grants.clone()
            },
            config.limits.text_bytes,
        ));
        let mut bounds = *agents.bounds();
        bounds.max_event_bytes = config.limits.request_bytes.get();
        let retained = crate::agents::retained::Retained::new(
            bounds,
            self.cluster.supervisor().clone(),
            self.console.device(),
        );
        lifecycle.configure_sites(
            config.limits.clone(),
            self.durable_store(),
            lys.templates.policy.clone(),
            effects,
            retained,
        )?;
        lifecycle.validate_site_storage()?;
        lifecycle.bind_site_console(&self.console)?;
        self.site_drafts
            .set(drafts)
            .map_err(|_| SiteError::Invalid("construction drafts already installed"))?;
        self.site_lifecycle
            .set(lifecycle)
            .map_err(|_| SiteError::Invalid("construction already installed"))?;
        Ok(())
    }

    /// Executes the ticket route using exactly the connection's admitted session.
    ///
    /// # Errors
    /// Unknown route, invalid boundary, grant refusal or target-store failure.
    pub fn serve_site_tool(
        &self,
        session: Option<&dyn LysSession>,
        subject: &str,
        bytes: &[u8],
        at_ms: u64,
    ) -> Result<Value, SiteError> {
        if !is_site_route(subject) {
            return Err(SiteError::Invalid("construction route"));
        }
        let lifecycle = self
            .site_lifecycle
            .get()
            .ok_or(SiteError::Invalid("construction not installed"))?;
        let session = session.ok_or(SiteError::Invalid(
            "construction needs an admitted Lys pass",
        ))?;
        let limits = &lifecycle.construction_limits()?;
        if bytes.len() > limits.request_bytes.get() {
            return Err(SiteError::Bound("request_bytes"));
        }
        let request: Request = serde_json::from_slice(bytes)
            .map_err(|_| SiteError::Invalid("construction request"))?;
        if request.operation.is_empty()
            || request.operation.len() > limits.text_bytes.get()
            || request.operation.chars().any(char::is_control)
        {
            return Err(SiteError::Invalid("operation"));
        }
        let (target, change) = match subject {
            "services.plan_set" => {
                let body: PlanSet = parse(request.payload)?;
                authorize_plan(session, lifecycle, &body.service, at_ms)?;
                let titles = body
                    .steps
                    .into_iter()
                    .map(|step| step.title)
                    .collect::<Vec<_>>();
                let plan = lifecycle.plan_set(&body.service, session.holder(), &titles, at_ms)?;
                self.construction_changed()?;
                return Ok(plan::current_value(&plan));
            }
            "services.plan_step" => {
                let body: PlanStep = parse(request.payload)?;
                authorize_plan(session, lifecycle, &body.service, at_ms)?;
                let plan = lifecycle.plan_step(
                    &body.service,
                    session.holder(),
                    body.step,
                    body.status,
                    body.note,
                    at_ms,
                )?;
                self.construction_changed()?;
                return Ok(plan::current_value(&plan));
            }
            _ => prepare_change(lifecycle, subject, request.payload, limits)?,
        };
        self.site_drafts
            .get()
            .ok_or(SiteError::Invalid("construction approval owner absent"))?
            .hold(
                lifecycle,
                session,
                &request.operation,
                &target,
                change,
                at_ms,
            )
    }

    fn construction_changed(&self) -> Result<(), SiteError> {
        self.site_lifecycle
            .get()
            .ok_or(SiteError::Invalid("construction not installed"))?
            .refresh_sites()
    }

    pub(super) fn subscribe_site_plan(
        &self,
        session: Option<&dyn LysSession>,
        channel: &str,
        schemas: &[super::ProtocolSchemaId],
        install: Option<liminal::channel::InboxInstall>,
    ) -> Result<super::ConnectionSubscription, crate::ServerError> {
        let name =
            plan_subject(channel).ok_or_else(|| site_error(SiteError::Invalid("plan subject")))?;
        let lifecycle = self
            .site_lifecycle
            .get()
            .ok_or_else(|| site_error(SiteError::Invalid("construction not installed")))?;
        lifecycle
            .record(name)
            .map_err(|error| site_error(error.into()))?;
        let session =
            session.ok_or_else(|| site_error(SiteError::Invalid("plan watch needs a pass")))?;
        let binding = SubjectBinding::new(
            lifecycle.network_id().into(),
            channel.into(),
            channel.into(),
        )
        .map_err(|_| site_error(SiteError::Invalid("plan watch binding")))?;
        session
            .decide(
                &binding,
                HotAction::Subscribe,
                super::super::apply::unix_seconds_now(),
            )
            .map_err(|error| site_error(SiteError::Pass(error.to_string())))?;
        let id = self
            .next_subscription_id
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        lifecycle
            .subscribe_plan(id, channel, schemas, install)
            .map_err(site_error)
    }
}

/// Maps named construction refusals without retaining their payload bytes.
pub fn site_error(error: SiteError) -> crate::ServerError {
    error.into()
}
impl From<SiteError> for crate::ServerError {
    fn from(error: SiteError) -> Self {
        Self::ListenerAccept {
            message: error.to_string(),
        }
    }
}

/// Uses the existing conversation reply and refusal shapes for construction.
pub(in super::super) fn site_frame_response(
    frame: SiteFrame<'_>,
) -> Option<super::super::state::FrameAction> {
    let SiteFrame {
        services,
        session,
        flags,
        stream_id,
        conversation_id,
        subject,
        envelope,
        at_ms,
    } = frame;
    let answer = services.construction_request(session, subject, &envelope.payload, at_ms)?;
    let refused = |message| {
        FrameAction::Respond(Frame::ConversationError {
            flags: 0,
            stream_id,
            conversation_id,
            reason_code: super::super::service_route::SERVICE_REFUSED_CODE,
            message: Some(message),
        })
    };
    Some(match answer {
        Err(crate::ServerError::ListenerAccept { message }) => refused(message),
        Err(error) => refused(error.to_string()),
        Ok(answer) => {
            if flags & CONVERSATION_REPLY_REQUESTED_FLAG == 0 {
                return Some(FrameAction::NoResponse);
            }
            serde_json::to_vec(&answer).map_or_else(
                |_| refused("construction_reply_encoding_failed".into()),
                |bytes| {
                    FrameAction::Respond(Frame::ConversationMessage {
                        flags: CONVERSATION_REPLY_REQUESTED_FLAG,
                        stream_id,
                        conversation_id,
                        envelope: MessageEnvelope::new(
                            envelope.schema_id,
                            envelope.causal_context.clone(),
                            bytes,
                        ),
                    })
                },
            )
        }
    })
}

/// Borrowed wire metadata for the existing conversation response shape.
#[derive(Clone, Copy)]
pub(in super::super) struct SiteFrame<'a> {
    pub services: &'a dyn super::ConnectionServices,
    pub session: Option<&'a dyn LysSession>,
    pub flags: u8,
    pub stream_id: u32,
    pub conversation_id: u64,
    pub subject: &'a str,
    pub envelope: &'a liminal::protocol::MessageEnvelope,
    pub at_ms: u64,
}