liminal-server 0.25.4

//! Peer data is projected only while the same local document names its link up.

use super::{DiscoveryError, FarLink, FarTopology, Value, reaches};
use std::collections::{BTreeMap, BTreeSet};

pub(super) fn validate(document: &Value, bound: usize) -> Result<u64, DiscoveryError> {
    if document["v"] != 1 || document["type"] != "topology" {
        return Err(DiscoveryError::Invalid);
    }
    let revision = document["rev"]
        .as_u64()
        .filter(|revision| *revision > 0)
        .ok_or(DiscoveryError::Invalid)?;
    let device = document["this_device"]
        .as_str()
        .filter(|device| !device.is_empty())
        .ok_or(DiscoveryError::Invalid)?;
    let devices = document["devices"]
        .as_array()
        .ok_or(DiscoveryError::Invalid)?;
    if devices
        .iter()
        .filter(|item| item["id"].as_str() == Some(device))
        .count()
        != 1
    {
        return Err(DiscoveryError::Invalid);
    }
    let stations = document["stations"]
        .as_array()
        .ok_or(DiscoveryError::Invalid)?;
    if stations.len() > bound {
        return Err(DiscoveryError::Bound);
    }
    let mut identifiers = BTreeSet::new();
    for station in stations {
        let id = station["id"]
            .as_str()
            .filter(|id| !id.is_empty())
            .ok_or(DiscoveryError::Invalid)?;
        if !identifiers.insert(id)
            || station["name"].as_str().is_none()
            || station["device"].as_str() != Some(device)
        {
            return Err(DiscoveryError::Invalid);
        }
        if let Some(subjects) = station.get("subjects") {
            for subject in subjects.as_array().ok_or(DiscoveryError::Invalid)? {
                let subject = subject.as_str().ok_or(DiscoveryError::Invalid)?;
                crate::auth_lys::SubjectBinding::new("wire".into(), subject.into(), subject.into())
                    .map_err(|_| DiscoveryError::Invalid)?;
            }
        }
        if let Some(activities) = station.get("activities") {
            for activity in activities.as_array().ok_or(DiscoveryError::Invalid)? {
                if activity["name"].as_str().is_none()
                    || activity["input_schema_json"].as_str().is_none()
                    || activity["output_schema_json"].as_str().is_none()
                {
                    return Err(DiscoveryError::Invalid);
                }
            }
        }
    }
    document["lines"]
        .as_array()
        .ok_or(DiscoveryError::Invalid)?;
    Ok(revision)
}

fn identity(link: &str, original: &str) -> String {
    format!("far.{}.{}", hex::encode(link), hex::encode(original))
}

pub(super) fn stations(link: &str, far: &FarLink) -> Result<Vec<Value>, DiscoveryError> {
    let Some(document) = &far.document else {
        return Ok(Vec::new());
    };
    let mut projected = Vec::new();
    for station in document["stations"]
        .as_array()
        .ok_or(DiscoveryError::Invalid)?
    {
        let id = station["id"].as_str().ok_or(DiscoveryError::Invalid)?;
        if !matches!(
            station["kind"].as_str(),
            Some("worker" | "service" | "agent")
        ) {
            continue;
        }
        let subjects = match station.get("subjects") {
            Some(subjects) => subjects
                .as_array()
                .ok_or(DiscoveryError::Invalid)?
                .iter()
                .map(|value| value.as_str().ok_or(DiscoveryError::Invalid))
                .collect::<Result<Vec<_>, _>>()?,
            None if station["kind"] == "service" => vec![id],
            None => Vec::new(),
        };
        let exported: Vec<_> = subjects
            .into_iter()
            .filter(|subject| far.prefixes.iter().any(|prefix| reaches(prefix, subject)))
            .collect();
        if exported.is_empty() {
            continue;
        }
        let mut station = station.clone();
        station["source_id"] = id.into();
        station["source_device"] = document["this_device"].clone();
        station["id"] = identity(link, id).into();
        station["device"] = far.peer.clone().into();
        station["link"] = link.into();
        station["subjects"] =
            serde_json::to_value(exported).map_err(|_| DiscoveryError::Invalid)?;
        projected.push(station);
    }
    Ok(projected)
}

pub(super) fn project(registry: &FarTopology, bytes: &[u8]) -> Result<Vec<u8>, DiscoveryError> {
    if bytes.len() > registry.limits.bytes.get() {
        return Err(DiscoveryError::Bound);
    }
    let mut local: Value = serde_json::from_slice(bytes).map_err(|_| DiscoveryError::Invalid)?;
    validate_local(&local)?;
    let links: BTreeMap<String, String> = local["links"]
        .as_array()
        .ok_or(DiscoveryError::Invalid)?
        .iter()
        .filter(|link| link["state"] == "up" && link["kind"] == "federated")
        .map(|link| {
            Ok((
                link["id"]
                    .as_str()
                    .ok_or(DiscoveryError::Invalid)?
                    .to_owned(),
                link["name"]
                    .as_str()
                    .ok_or(DiscoveryError::Invalid)?
                    .to_owned(),
            ))
        })
        .collect::<Result<_, DiscoveryError>>()?;
    let mut state = registry.lock()?;
    let mut station_ids: BTreeSet<String> = local["stations"]
        .as_array()
        .ok_or(DiscoveryError::Invalid)?
        .iter()
        .map(|station| {
            station["id"]
                .as_str()
                .map(str::to_owned)
                .ok_or(DiscoveryError::Invalid)
        })
        .collect::<Result<_, _>>()?;
    for (link, far) in &state.links {
        if links.get(link) != Some(&far.peer) {
            continue;
        }
        if local["this_device"].as_str() == Some(far.peer.as_str()) {
            return Err(DiscoveryError::Invalid);
        }
        let projected = stations(link, far)?;
        for station in &projected {
            let id = station["id"].as_str().ok_or(DiscoveryError::Invalid)?;
            if !station_ids.insert(id.into()) {
                return Err(DiscoveryError::Invalid);
            }
        }
        project_device(&mut local, far)?;
        project_lines(&mut local, link, far, &projected)?;
        local["stations"]
            .as_array_mut()
            .ok_or(DiscoveryError::Invalid)?
            .extend(projected);
    }
    let projected = serde_json::to_vec(&local).map_err(|_| DiscoveryError::Invalid)?;
    if projected.len() > registry.limits.bytes.get() {
        return Err(DiscoveryError::Bound);
    }
    // Retire only after the whole replacement document has passed every check.
    state
        .links
        .retain(|link, far| links.get(link) == Some(&far.peer));
    drop(state);
    Ok(projected)
}

fn validate_local(document: &Value) -> Result<(), DiscoveryError> {
    if document["v"] != 1
        || document["type"] != "topology"
        || document["rev"]
            .as_u64()
            .is_none_or(|revision| revision == 0)
        || document["this_device"].as_str().is_none_or(str::is_empty)
    {
        return Err(DiscoveryError::Invalid);
    }
    for collection in ["devices", "stations", "links", "lines"] {
        let mut identifiers = BTreeSet::new();
        for record in document[collection]
            .as_array()
            .ok_or(DiscoveryError::Invalid)?
        {
            let id = record["id"]
                .as_str()
                .filter(|id| !id.is_empty())
                .ok_or(DiscoveryError::Invalid)?;
            if !identifiers.insert(id) {
                return Err(DiscoveryError::Invalid);
            }
            if collection == "lines"
                && !record["stops"]
                    .as_array()
                    .is_some_and(|stops| stops.iter().all(Value::is_string))
            {
                return Err(DiscoveryError::Invalid);
            }
        }
    }
    Ok(())
}

fn project_device(local: &mut Value, far: &FarLink) -> Result<(), DiscoveryError> {
    let Some(document) = &far.document else {
        return Ok(());
    };
    let original = document["this_device"]
        .as_str()
        .ok_or(DiscoveryError::Invalid)?;
    let mut device = document["devices"]
        .as_array()
        .ok_or(DiscoveryError::Invalid)?
        .iter()
        .find(|device| device["id"].as_str() == Some(original))
        .ok_or(DiscoveryError::Invalid)?
        .clone();
    device["id"] = far.peer.clone().into();
    // A far device's drawing coordinates and parent are local to its map.
    let object = device.as_object_mut().ok_or(DiscoveryError::Invalid)?;
    object.remove("box");
    object.remove("parent");
    let devices = local["devices"]
        .as_array_mut()
        .ok_or(DiscoveryError::Invalid)?;
    devices.retain(|device| device["id"].as_str() != Some(&far.peer));
    devices.push(device);
    Ok(())
}

fn project_lines(
    local: &mut Value,
    link: &str,
    far: &FarLink,
    stations: &[Value],
) -> Result<(), DiscoveryError> {
    let Some(document) = &far.document else {
        return Ok(());
    };
    let ids: BTreeMap<_, _> = stations
        .iter()
        .map(|station| (station["source_id"].as_str(), station["id"].clone()))
        .collect();
    let lines = local["lines"]
        .as_array_mut()
        .ok_or(DiscoveryError::Invalid)?;
    for line in document["lines"]
        .as_array()
        .ok_or(DiscoveryError::Invalid)?
    {
        let subject = line["id"].as_str().ok_or(DiscoveryError::Invalid)?;
        if !far.prefixes.iter().any(|prefix| reaches(prefix, subject)) {
            continue;
        }
        let stops: Vec<_> = line["stops"]
            .as_array()
            .ok_or(DiscoveryError::Invalid)?
            .iter()
            .filter_map(|stop| ids.get(&stop.as_str()).cloned())
            .collect();
        if let Some(own) = lines
            .iter_mut()
            .find(|own| own["id"].as_str() == Some(subject))
        {
            let current = own["stops"].as_array_mut().ok_or(DiscoveryError::Invalid)?;
            for stop in stops {
                if !current.contains(&stop) {
                    current.push(stop);
                }
            }
        } else {
            let mut projected = line.clone();
            projected["stops"] = stops.into();
            projected["link"] = link.into();
            projected["device"] = far.peer.clone().into();
            lines.push(projected);
        }
    }
    Ok(())
}