liminal_server/console/

topology.rs

//! The topology document (SRV-011 R1), built from in-memory state only.

use liminal::channel::ChannelMode;

use crate::ServerError;
use crate::server::connection::channel_registry::{ChannelOrigin, ChannelStatus};
use crate::server::connection::services::LiminalConnectionServices;

/// One channel (a line) as the console draws it.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ChannelNode {
    /// The channel name.
    pub name: String,
    /// `durable` or `ephemeral`.
    pub mode: &'static str,
    /// `live` (admitting) or `held` (quiesced).
    pub state: &'static str,
    /// `config` (boot) or `runtime` (registered live).
    pub origin: &'static str,
    /// The quiesce cause, when held.
    pub reason: Option<String>,
}

/// The topology document.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct TopologySnapshot {
    /// Every channel on the roster, sorted by name.
    pub channels: Vec<ChannelNode>,
    /// Every registered conversation responder subject, sorted.
    pub services: Vec<String>,
}

/// Builds the topology from the roster, read once (pinned by
/// `the_snapshot_reads_the_roster_once`), and the responder registry. Reads
/// no store row.
///
/// # Errors
/// Returns [`ServerError`] when the roster or responder lock is poisoned.
pub fn snapshot(services: &LiminalConnectionServices) -> Result<TopologySnapshot, ServerError> {
    let roster_error = |error: &dyn std::fmt::Display| ServerError::ListenerAccept {
        message: format!("console topology: {error}"),
    };
    let census = services
        .channel_census()
        .map_err(|error| roster_error(&error))?;
    let mut channels = Vec::with_capacity(census.len());
    for (name, status) in census {
        let node = match status {
            ChannelStatus::NotRegistered => continue,
            ChannelStatus::Active { origin, mode, .. } => ChannelNode {
                name,
                mode: mode_word(mode),
                state: "live",
                origin: origin_word(origin),
                reason: None,
            },
            ChannelStatus::Quiesced {
                reason,
                origin,
                mode,
            } => ChannelNode {
                name,
                mode: mode_word(mode),
                state: "held",
                origin: origin_word(origin),
                reason: Some(reason),
            },
        };
        channels.push(node);
    }
    Ok(TopologySnapshot {
        channels,
        services: services.responder_subjects()?,
    })
}

const fn mode_word(mode: ChannelMode) -> &'static str {
    match mode {
        ChannelMode::Durable => "durable",
        ChannelMode::Ephemeral => "ephemeral",
    }
}

const fn origin_word(origin: ChannelOrigin) -> &'static str {
    match origin {
        ChannelOrigin::BootConfigured => "config",
        ChannelOrigin::RuntimeRegistered => "runtime",
    }
}

/// Adds the proved far stations after local agent and construction projections.
///
/// # Errors
/// Discovery bounds, invalid documents or a poisoned far registry.
pub fn with_far_stations(
    payload: &[u8],
    far: &crate::link::federated::FarTopology,
) -> Result<Vec<u8>, crate::link::federated::DiscoveryError> {
    far.project(payload)
}

/// Adds construction services after the agent projection.
///
/// # Errors
/// Invalid topology shape, poisoned registry or serialization failure.
pub fn with_sites(
    payload: &[u8],
    lifecycle: &crate::services::lifecycle::Lifecycle,
    device: &str,
) -> Result<Vec<u8>, crate::services::lifecycle::site::SiteError> {
    use crate::services::lifecycle::site::SiteError;
    let mut document: serde_json::Value =
        serde_json::from_slice(payload).map_err(|_| SiteError::Invalid("topology JSON"))?;
    let mut stations = document["stations"]
        .as_array()
        .ok_or(SiteError::Invalid("topology stations"))?
        .clone();
    for station in lifecycle.site_stations(device)? {
        let id = station["id"]
            .as_str()
            .ok_or(SiteError::Invalid("site station id"))?;
        let resource = station["service"]
            .as_str()
            .ok_or(SiteError::Invalid("site resource id"))?;
        let subjects = station["subjects"]
            .as_array()
            .ok_or(SiteError::Invalid("site subjects"))?;
        let mut removed = Vec::new();
        stations.retain(|old| {
            let local = old["device"].as_str() == Some(device);
            let worker = local
                && old["kind"] == "worker"
                && old["subjects"].as_array().is_some_and(|old_subjects| {
                    old_subjects
                        .iter()
                        .any(|subject| subjects.contains(subject))
                });
            let replaced = local
                && (old["id"].as_str() == Some(id)
                    || (old["kind"] == "service" && old["id"].as_str() == Some(resource)));
            if worker || replaced {
                if let Some(name) = old["id"].as_str() {
                    removed.push(name.to_owned());
                }
                false
            } else {
                true
            }
        });
        if let Some(lines) = document["lines"].as_array_mut() {
            for line in lines {
                if line["kind"] == "call" && subjects.contains(&line["id"]) {
                    let stops = line["stops"]
                        .as_array_mut()
                        .ok_or(SiteError::Invalid("site line stops"))?;
                    for stop in stops.iter_mut() {
                        if stop
                            .as_str()
                            .is_some_and(|name| removed.iter().any(|old| old == name))
                        {
                            *stop = id.into();
                        }
                    }
                }
            }
        }
        stations.push(station);
    }
    document["stations"] = stations.into();
    liminal::channel::canonical::to_vec(document)
        .map_err(|_| SiteError::Invalid("site topology encoding"))
}

/// Adds approved agents to the hub's whole document, preserving its revision
///
/// and every existing station. Presence supplies the state; the existing line
/// stops supply the subjects the ticket actually listens to.
///
/// # Errors
/// A document or agent record cannot be read or encoded.
pub fn with_agents(
    payload: &[u8],
    agents: &crate::agents::runtime::Agents,
    device: &str,
) -> Result<Vec<u8>, crate::agents::error::AgentError> {
    use crate::agents::error::AgentError;
    use crate::console::records::{AgentStation, StationState};
    let mut document: serde_json::Value =
        serde_json::from_slice(payload).map_err(|_| AgentError::Invalid("topology JSON"))?;
    let records = agents.registry().station_records()?;
    for record in records {
        let connection = agents.presence().connection(&record.name)?;
        let id = format!("agents.{}.inbox", record.name);
        if !record.active() {
            document["stations"]
                .as_array_mut()
                .ok_or(AgentError::Invalid("topology stations"))?
                .retain(|station| station["id"].as_str() != Some(id.as_str()));
            continue;
        }
        let mut stops = connection.map_or(Ok(Vec::new()), |connection| {
            agents.listened_subjects(connection)
        })?;
        if let Some(connection) = connection {
            let connection = connection.to_string();
            if let Some(lines) = document["lines"].as_array_mut() {
                for line in lines {
                    let line_name = line["id"].as_str().map(str::to_owned);
                    if let Some(stations) = line["stops"].as_array_mut() {
                        let listened = stations
                            .iter()
                            .any(|stop| stop.as_str() == Some(&connection));
                        if listened {
                            for stop in stations {
                                if stop.as_str() == Some(&connection) {
                                    *stop = id.clone().into();
                                }
                            }
                            if let Some(name) = line_name {
                                stops.push(name);
                            }
                        }
                    }
                }
            }
        }
        if let Some(lines) = document["lines"].as_array_mut() {
            for line in lines {
                if line["id"]
                    .as_str()
                    .is_some_and(|subject| stops.iter().any(|own| own == subject))
                {
                    let stations = line["stops"]
                        .as_array_mut()
                        .ok_or(AgentError::Invalid("line stops"))?;
                    if !stations
                        .iter()
                        .any(|station| station.as_str() == Some(id.as_str()))
                    {
                        stations.push(id.clone().into());
                    }
                }
                if line["id"].as_str() == Some(id.as_str()) {
                    line["kind"] = "stream".into();
                    line["life"] = "kept".into();
                }
            }
        }
        stops.sort();
        stops.dedup();
        let state = agents
            .retained()
            .value(&format!("agents.{}.state", record.name))?;
        let state = match state.as_ref().and_then(|state| state["state"].as_str()) {
            Some("waiting") => StationState::Waiting,
            Some("present") if connection.is_some() => StationState::Present,
            _ => StationState::Gone,
        };
        let station = serde_json::to_value(AgentStation::new(&record, device, state, stops))
            .map_err(|_| AgentError::Invalid("agent station encoding"))?;
        let stations = document["stations"]
            .as_array_mut()
            .ok_or(AgentError::Invalid("topology stations"))?;
        stations.retain(|station| station["id"].as_str() != Some(&id));
        stations.push(station);
    }
    liminal::channel::canonical::to_vec(document)
        .map_err(|_| AgentError::Invalid("agent topology encoding"))
}