liminal-server 0.25.4

//! Far discovery is generation-bound, prefix-filtered and never guessed.

use serde_json::Value;
use std::collections::BTreeMap;
use std::num::NonZeroUsize;
use std::sync::{Mutex, MutexGuard};

#[path = "link_federated/projection.rs"]
mod projection;

/// Bounds checked before any peer-controlled discovery data is retained.
#[derive(Clone, Copy, Debug)]
pub struct ProjectionLimits {
    /// Maximum snapshot bytes, bounding JSON parsing and retained contracts.
    pub bytes: NonZeroUsize,
    /// Maximum stations per snapshot, bounding directory indexing work.
    pub stations: NonZeroUsize,
    /// Maximum distinct links, bounding total discovery memory.
    pub peers: NonZeroUsize,
}

/// Discovery refuses without leaking schema or message bytes.
#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
pub enum DiscoveryError {
    /// Input or retained discovery exceeded an operator bound.
    #[error("far_topology_bound")]
    Bound,
    /// The far document is not a whole topology from its own device.
    #[error("far_topology_invalid")]
    Invalid,
    /// A stale link generation or topology revision cannot replace current data.
    #[error("far_topology_stale")]
    Stale,
    /// A poisoned registry cannot supply trusted discovery.
    #[error("far_topology_unavailable")]
    Unavailable,
    /// The monotonic link generation is exhausted.
    #[error("far_topology_generation_exhausted")]
    Exhausted,
}

#[derive(Default)]
struct State {
    generation: u64,
    links: BTreeMap<String, FarLink>,
}
struct FarLink {
    generation: u64,
    peer: String,
    prefixes: Vec<String>,
    revision: Option<u64>,
    document: Option<Value>,
}

/// One authoritative index used for topology, describe and service routing.
pub struct FarTopology {
    limits: ProjectionLimits,
    state: Mutex<State>,
}

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

impl FarTopology {
    /// Creates an empty index with required positive bounds.
    #[must_use]
    pub fn new(limits: ProjectionLimits) -> Self {
        Self {
            limits,
            state: Mutex::default(),
        }
    }

    fn lock(&self) -> Result<MutexGuard<'_, State>, DiscoveryError> {
        self.state.lock().map_err(|_| DiscoveryError::Unavailable)
    }

    /// Binds this generation to the proved peer and configured export prefixes.
    ///
    /// # Errors
    /// Invalid identities/prefixes, peer bound, poisoned state or exhausted ids.
    pub fn open(
        &self,
        link: &str,
        peer: &str,
        prefixes: Vec<String>,
    ) -> Result<u64, DiscoveryError> {
        if link.is_empty()
            || peer.is_empty()
            || prefixes.is_empty()
            || link
                .chars()
                .chain(peer.chars())
                .any(|character| character.is_control() || character.is_whitespace())
        {
            return Err(DiscoveryError::Invalid);
        }
        for prefix in &prefixes {
            crate::auth_lys::SubjectBinding::new("wire".into(), prefix.clone(), prefix.clone())
                .map_err(|_| DiscoveryError::Invalid)?;
            if prefix == "liminal"
                || prefix.starts_with("liminal.")
                || prefix == "console"
                || prefix.starts_with("console.")
            {
                return Err(DiscoveryError::Invalid);
            }
        }
        let mut state = self.lock()?;
        if !state.links.contains_key(link) && state.links.len() >= self.limits.peers.get() {
            return Err(DiscoveryError::Bound);
        }
        state.generation = state
            .generation
            .checked_add(1)
            .ok_or(DiscoveryError::Exhausted)?;
        let generation = state.generation;
        state.links.insert(
            link.into(),
            FarLink {
                generation,
                peer: peer.into(),
                prefixes,
                revision: None,
                document: None,
            },
        );
        drop(state);
        Ok(generation)
    }

    /// Accepts a whole far snapshot only from the current keyed generation.
    ///
    /// # Errors
    /// Oversize, malformed or forged document, stale revision/generation, or poisoned state.
    pub fn update(
        &self,
        link: &str,
        generation: u64,
        bytes: &[u8],
    ) -> Result<bool, DiscoveryError> {
        if bytes.len() > self.limits.bytes.get() {
            return Err(DiscoveryError::Bound);
        }
        let document: Value = serde_json::from_slice(bytes).map_err(|_| DiscoveryError::Invalid)?;
        let revision = projection::validate(&document, self.limits.stations.get())?;
        let mut state = self.lock()?;
        let far = state
            .links
            .get_mut(link)
            .filter(|far| far.generation == generation)
            .ok_or(DiscoveryError::Stale)?;
        if far.revision.is_some_and(|previous| previous > revision) {
            return Err(DiscoveryError::Stale);
        }
        if far.revision == Some(revision) {
            return if far.document.as_ref() == Some(&document) {
                Ok(false)
            } else {
                Err(DiscoveryError::Stale)
            };
        }
        far.revision = Some(revision);
        far.document = Some(document);
        drop(state);
        Ok(true)
    }

    /// Retires only this generation; a replaced carrier cannot remove its successor.
    ///
    /// # Errors
    /// The registry lock is poisoned.
    pub fn close(&self, link: &str, generation: u64) -> Result<bool, DiscoveryError> {
        let mut state = self.lock()?;
        if state
            .links
            .get(link)
            .is_some_and(|far| far.generation == generation)
        {
            state.links.remove(link);
            drop(state);
            Ok(true)
        } else {
            Ok(false)
        }
    }

    /// Number of retained peer generations, including peers awaiting discovery.
    ///
    /// # Errors
    /// The registry lock is poisoned.
    pub fn pending_peers(&self) -> Result<usize, DiscoveryError> {
        Ok(self.lock()?.links.len())
    }

    /// Describes the actual far station serving an exported subject.
    ///
    /// # Errors
    /// The registry or a retained document is invalid.
    pub fn describe(&self, subject: &str) -> Result<Option<Value>, DiscoveryError> {
        let state = self.lock()?;
        for (link, far) in &state.links {
            for station in projection::stations(link, far)? {
                if station["subjects"].as_array().is_some_and(|subjects| {
                    subjects.iter().any(|item| item.as_str() == Some(subject))
                }) {
                    return Ok(Some(station));
                }
            }
        }
        drop(state);
        Ok(None)
    }

    /// Projects only currently up links, preserving the supplied local revision.
    /// A down record and the removal of its far stations are one atomic document.
    ///
    /// # Errors
    /// Malformed local/far topology, oversize output, encoding or registry failure.
    pub fn project(&self, bytes: &[u8]) -> Result<Vec<u8>, DiscoveryError> {
        projection::project(self, bytes)
    }
}

fn reaches(prefix: &str, subject: &str) -> bool {
    subject == prefix
        || subject
            .strip_prefix(prefix)
            .is_some_and(|suffix| suffix.starts_with('.'))
}