mecha10_core/

relay_topics.rs

//! Relay topics auto-derivation for `websocket-bridge`.
//!
//! # Problem this replaces
//!
//! For a Redis topic published by some node to reach the dashboard, two independent things
//! today have to happen: (1) the node publishes to some topic path, and (2) a human manually
//! adds that exact same path to `websocket-bridge`'s own `configs/config.json`
//! `topics.subscribes` allowlist. Nothing enforces these stay in sync - the two lists live in
//! unrelated files, maintained by unrelated people, at unrelated times. This has already caused
//! a real bug (a fix PR edited the wrong file entirely, unnoticed, because there was no
//! structural link between "a node publishes X" and "the relay knows about X").
//!
//! # Design
//!
//! Nodes now optionally declare a `relay_topics` array in their own resolved config object
//! (same level as any other node-specific config field, e.g. `model`, `topics`, ...):
//!
//! ```json
//! { "relay_topics": ["/vision/classification"] }
//! ```
//!
//! A node's own config struct picks this up by embedding [`RelayTopicsField`] with
//! `#[serde(flatten)]` - one line, reusing a single canonical type, rather than each node
//! hand-declaring its own `relay_topics: Option<Vec<String>>` field (the fragile pattern this
//! replaced: easy to forget entirely, as `behavior-executor` once did, silently crashing that
//! node at startup with "unknown field: relay_topics" the moment its `config.json` declared the
//! key without a matching struct field). `#[serde(flatten)]` composes cleanly with
//! `#[serde(deny_unknown_fields)]` on the outer struct - real typos in *other* fields still fail
//! loudly, `relay_topics` just isn't one of them anymore.
//!
//! [`derive_relay_topics`] recursively walks every `config.json` found anywhere under a
//! project's `configs/nodes/` tree - both the framework tier (`configs/nodes/@mecha10/{name}/`)
//! and the project tier a custom node scaffolds into (`configs/nodes/{name}/`, no `@mecha10`
//! prefix) - collects each one's `relay_topics` (a config that doesn't declare the field silently
//! contributes nothing - this is how not-yet-migrated nodes are skipped, no error), and unions
//! that with a small fixed set of framework-level topics that no single node owns
//! (`/system/health`, `/system/log` - see [`framework_relay_topics`]).
//!
//! This is a plain recursive scan, not a precedence-aware resolution: if a node somehow has
//! `config.json` files in more than one place (e.g. both tiers, or split `dev`/`production`
//! files), every one of them is read and their `relay_topics` are unioned together, even ones
//! that aren't the copy actually in effect for a given node/environment. That's deliberately
//! unlike `Context::resolve_node_config`'s tier precedence - this module only ever produces an
//! *allowlist*, so including an extra topic from an inactive config is harmless (nothing
//! downstream is required to actually use it), whereas replicating full precedence-resolution
//! logic here just to avoid that is not worth the complexity.
//!
//! [`regenerate_websocket_bridge_relay_config`] is the write half: it re-derives the topic list
//! and merges it *additively* into `websocket-bridge`'s own scaffolded config file's top-level
//! `relay_topics` array (the same flat shape every other node declares its own `relay_topics` in -
//! see [`RelayTopicsField`]), unioning with whatever entries are already present rather than
//! replacing them outright. This is deliberate - it guarantees regeneration can never regress an
//! already-working topic, even before every node has been migrated to declare `relay_topics`
//! (that migration is scoped as separate, parallel PRs per node/node-group).
//!
//! `websocket-bridge` previously carried its own hand-maintained `topics.subscribes` allowlist
//! (a list of `{ friendlyName: topicPath }` entries) as a second, independently-edited place the
//! same information lived - this duplicated `relay_topics` and was a frequent source of drift
//! (an entry added to one list but not the other). `websocket-bridge`'s own `relay_topics` field
//! (inherited from [`NodeConfigBase`](crate::node_config_base::NodeConfigBase) like every other
//! node) is now the only place this is persisted.
//!
//! # Node config file layout
//!
//! A node's `config.json` isn't guaranteed to be in one single shape - it may be a single file
//! with `"dev"`/`"production"` keys wrapping the actual config, or split into separate
//! `dev/config.json` + `production/config.json` files. [`find_node_config_files`] finds every
//! `config.json` regardless of shape or nesting depth; [`resolve_environment_section`] then picks
//! the right section back out of whichever shape it turns out to be.

use crate::error::{Mecha10Error, Result};
use crate::topics::system::{HEALTH, LOG};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::{BTreeSet, HashSet};
use std::path::{Path, PathBuf};

pub const RELAY_TOPICS_FIELD: &str = "relay_topics";

#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(description = "Declares which topics this node publishes should be relayed to the dashboard.")]
pub struct RelayTopicsField {
    #[serde(default)]
    #[schemars(example = Some(vec!["/vision/classification".to_string()]))]
    pub relay_topics: Option<Vec<String>>,
}

/// Framework-level topics no single node owns, so they can't live in any node's `relay_topics`
/// field:
///
/// - `/system/health`, published by [`crate::health::HealthReporter`], wired into every node's
///   shared runtime (`crate::node::run_node`).
/// - `/system/log`, published by [`crate::log_publisher::attach`], also called from
///   `run_node()` for every node.
///
/// Reuses the existing typed topic constants from [`crate::topics::system`] rather than
/// duplicating the literal path strings.
fn framework_relay_topics() -> [&'static str; 2] {
    [HEALTH.path(), LOG.path()]
}

const NODES_MECHA10_DIR: &str = "configs/nodes/@mecha10";

pub(crate) const NODES_PROJECT_DIR: &str = "configs/nodes";

const WEBSOCKET_BRIDGE_NODE_NAME: &str = "websocket-bridge";

/// Derive the full relay allowlist (topic paths `websocket-bridge` should subscribe to) for the
/// project rooted at `project_root`.
///
/// Returns the deduplicated union of:
/// - The fixed framework-level topics (always present, see [`framework_relay_topics`]).
/// - Every `relay_topics` field found in any `config.json` anywhere under `configs/nodes/`
///   (see [`find_node_config_files`]), resolved for whichever environment (`dev`/`production`)
///   is currently active (`MECHA10_ENV`, default `dev`).
///
/// A project with no `configs/nodes/` directory at all yields just the framework-level topics
/// rather than an error - there's nothing to walk, which is a valid (if minimal) state.
///
/// A multi-instance node (LAB-4131 - see [`crate::context::node_config`]'s module docs) resolves
/// its environment section to an instance-keyed object (e.g.
/// `{"real": {"relay_topics": [...]}, "sim": {"relay_topics": [...]}}`) rather than a flat config
/// with a top-level `relay_topics` key. To still pick those up, this also checks one level of
/// nesting: every value of the resolved section that is itself a JSON object is inspected for its
/// own top-level `relay_topics` array, and any found are unioned in alongside the top-level one.
/// This is a harmless no-op for a flat config whose own field values happen to be objects (e.g.
/// `topics`, `source`) since those don't have a `relay_topics` key themselves.
pub fn derive_relay_topics(project_root: &Path) -> Result<Vec<String>> {
    let mut topics: BTreeSet<String> = framework_relay_topics().iter().map(|t| t.to_string()).collect();

    let environment = std::env::var("MECHA10_ENV").unwrap_or_else(|_| "dev".to_string());

    for config_path in find_node_config_files(&project_root.join(NODES_PROJECT_DIR)) {
        let value = read_json(&config_path)?;
        let resolved = resolve_environment_section(value, &environment, &config_path)?;

        if let Some(relay_topics) = resolved.get(RELAY_TOPICS_FIELD).and_then(Value::as_array) {
            for topic in relay_topics.iter().filter_map(Value::as_str) {
                topics.insert(topic.to_string());
            }
        }

        if let Some(obj) = resolved.as_object() {
            for nested in obj.values().filter_map(Value::as_object) {
                if let Some(relay_topics) = nested.get(RELAY_TOPICS_FIELD).and_then(Value::as_array) {
                    for topic in relay_topics.iter().filter_map(Value::as_str) {
                        topics.insert(topic.to_string());
                    }
                }
            }
        }
    }

    Ok(topics.into_iter().collect())
}

pub(crate) fn find_node_config_files(dir: &Path) -> Vec<PathBuf> {
    let mut found = Vec::new();
    collect_config_files(dir, &mut found);
    found
}

fn collect_config_files(dir: &Path, found: &mut Vec<PathBuf>) {
    let Ok(entries) = std::fs::read_dir(dir) else {
        return;
    };

    for entry in entries.filter_map(|entry| entry.ok()) {
        let path = entry.path();
        if path.is_dir() {
            collect_config_files(&path, found);
        } else if path.file_name().and_then(|n| n.to_str()) == Some("config.json") {
            found.push(path);
        }
    }
}

/// If `value` is a JSON object with a `"dev"` or `"production"` key, return the section for
/// `environment`. Otherwise, return `value` unchanged - it's already a flat, unwrapped config
/// object.
///
/// Errors if `value` is an envelope (has a `"dev"` and/or `"production"` key) but is missing the
/// section for the *requested* `environment` - this must fail loudly rather than silently run
/// with the other environment's values (`LAB-4129`).
pub(crate) fn resolve_environment_section(value: Value, environment: &str, config_path: &Path) -> Result<Value> {
    let env_key = match environment {
        "production" | "prod" => "production",
        _ => "dev",
    };

    let Some(obj) = value.as_object() else {
        return Ok(value);
    };

    if !obj.contains_key("dev") && !obj.contains_key("production") {
        return Ok(value);
    }

    obj.get(env_key).cloned().ok_or_else(|| {
        Mecha10Error::Configuration(format!(
            "Config is missing its '{}' environment section (MECHA10_ENV='{}'): {}",
            env_key,
            environment,
            config_path.display()
        ))
    })
}

pub(crate) fn read_json(path: &Path) -> Result<Value> {
    let content = std::fs::read_to_string(path)
        .map_err(|e| Mecha10Error::Configuration(format!("Failed to read {}: {}", path.display(), e)))?;
    serde_json::from_str(&content)
        .map_err(|e| Mecha10Error::Configuration(format!("Failed to parse JSON from {}: {}", path.display(), e)))
}

/// Regenerate `websocket-bridge`'s own scaffolded config file
/// (`<project_root>/configs/nodes/@mecha10/websocket-bridge/config.json`) so its own
/// `relay_topics` array reflects [`derive_relay_topics`]'s current output.
///
/// **Additive, never destructive**: the derived topics are *unioned* into whatever
/// `relay_topics` entries already exist (per environment section) rather than replacing the list
/// outright. This guarantees regeneration can never regress an already-working topic, which
/// matters because this lands independently of the separate, parallel PRs that migrate
/// individual nodes onto declaring `relay_topics` - some nodes may not have been migrated yet
/// when this runs, and their existing hand-authored entries must survive.
///
/// No-ops (not an error) if the project has no `websocket-bridge` config file scaffolded at all.
pub fn regenerate_websocket_bridge_relay_config(project_root: &Path) -> Result<()> {
    let config_path = project_root
        .join(NODES_MECHA10_DIR)
        .join(WEBSOCKET_BRIDGE_NODE_NAME)
        .join("config.json");

    if !config_path.is_file() {
        return Ok(());
    }

    let derived = derive_relay_topics(project_root)?;

    let content = std::fs::read_to_string(&config_path)
        .map_err(|e| Mecha10Error::Configuration(format!("Failed to read {}: {}", config_path.display(), e)))?;
    let mut root: Value = serde_json::from_str(&content).map_err(|e| {
        Mecha10Error::Configuration(format!("Failed to parse JSON from {}: {}", config_path.display(), e))
    })?;

    let root_obj = root.as_object_mut().ok_or_else(|| {
        Mecha10Error::Configuration(format!(
            "websocket-bridge config must be a JSON object: {}",
            config_path.display()
        ))
    })?;

    let env_keys: Vec<&'static str> = ["dev", "production"]
        .into_iter()
        .filter(|key| root_obj.contains_key(*key))
        .collect();

    if env_keys.is_empty() {
        union_relay_topics_into(root_obj, &derived);
    } else {
        for key in env_keys {
            if let Some(section) = root_obj.get_mut(key).and_then(Value::as_object_mut) {
                union_relay_topics_into(section, &derived);
            }
        }
    }

    let updated = serde_json::to_string_pretty(&root)
        .map_err(|e| Mecha10Error::Configuration(format!("Failed to serialize {}: {}", config_path.display(), e)))?;
    std::fs::write(&config_path, updated)
        .map_err(|e| Mecha10Error::Configuration(format!("Failed to write {}: {}", config_path.display(), e)))?;

    Ok(())
}

/// Union `derived` topic paths into `section["relay_topics"]`, appending only the ones not
/// already present in the existing array. Existing entries are left completely untouched.
/// Silently leaves a malformed `relay_topics` field (present but not an array) alone rather than
/// erroring - regeneration should never be the thing that breaks an otherwise-loadable config
/// file.
fn union_relay_topics_into(section: &mut serde_json::Map<String, Value>, derived: &[String]) {
    let relay_topics_entry = section
        .entry(RELAY_TOPICS_FIELD)
        .or_insert_with(|| Value::Array(Vec::new()));
    let Some(relay_topics_arr) = relay_topics_entry.as_array_mut() else {
        return;
    };

    let mut seen: HashSet<String> = relay_topics_arr
        .iter()
        .filter_map(Value::as_str)
        .map(|s| s.to_string())
        .collect();

    for topic in derived {
        if seen.insert(topic.clone()) {
            relay_topics_arr.push(Value::String(topic.clone()));
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serial_test::serial;
    use std::fs;

    fn write_json(path: &Path, content: &str) {
        fs::create_dir_all(path.parent().unwrap()).unwrap();
        fs::write(path, content).unwrap();
    }

    fn nodes_dir(project_root: &Path) -> PathBuf {
        project_root.join(NODES_MECHA10_DIR)
    }

    #[test]
    fn framework_topics_always_present_even_with_no_nodes_dir() {
        let project = tempfile::tempdir().unwrap();

        let topics = derive_relay_topics(project.path()).unwrap();

        assert!(topics.contains(&"/system/health".to_string()));
        assert!(topics.contains(&"/system/log".to_string()));
    }

    #[test]
    #[serial]
    fn derives_from_single_config_json_with_dev_production_envelope() {
        std::env::remove_var("MECHA10_ENV");

        let project = tempfile::tempdir().unwrap();
        write_json(
            &nodes_dir(project.path()).join("llm-command").join("config.json"),
            r#"{
                "dev": { "relay_topics": ["/ai/command", "/ai/response"] },
                "production": { "relay_topics": ["/ai/command", "/ai/response"] }
            }"#,
        );

        let topics = derive_relay_topics(project.path()).unwrap();

        assert!(topics.contains(&"/ai/command".to_string()));
        assert!(topics.contains(&"/ai/response".to_string()));
    }

    #[test]
    #[serial]
    fn derives_from_split_dev_and_production_config_files() {
        std::env::remove_var("MECHA10_ENV");

        let project = tempfile::tempdir().unwrap();
        write_json(
            &nodes_dir(project.path()).join("imu").join("dev").join("config.json"),
            r#"{ "relay_topics": ["/imu/data"] }"#,
        );
        write_json(
            &nodes_dir(project.path())
                .join("imu")
                .join("production")
                .join("config.json"),
            r#"{ "relay_topics": ["/imu/data"] }"#,
        );

        let topics = derive_relay_topics(project.path()).unwrap();

        assert!(topics.contains(&"/imu/data".to_string()));
    }

    /// Split `dev/config.json` + `production/config.json` files are flat (no `"dev"`/
    /// `"production"` envelope key), so [`resolve_environment_section`] passes them through
    /// unchanged regardless of which directory they live under - `find_node_config_files` finds
    /// every `config.json` under a node's directory regardless of the currently active
    /// environment, by design (see the module docs' "plain recursive scan" note). This is
    /// unrelated to the dev/production envelope fallback `LAB-4129` removed below.
    #[test]
    #[serial]
    fn derives_from_dev_only_split_file_regardless_of_requested_environment() {
        std::env::set_var("MECHA10_ENV", "production");

        let project = tempfile::tempdir().unwrap();
        write_json(
            &nodes_dir(project.path())
                .join("simulator")
                .join("dev")
                .join("config.json"),
            r#"{ "relay_topics": ["/camera/rgb"] }"#,
        );

        let topics = derive_relay_topics(project.path()).unwrap();
        std::env::remove_var("MECHA10_ENV");

        assert!(topics.contains(&"/camera/rgb".to_string()));
    }

    /// `LAB-4129`: a single-file envelope config (`{"dev": {...}, "production": {...}}`) missing
    /// its *requested* environment key must fail to load, not silently fall back to whichever
    /// other environment's section happens to exist.
    #[test]
    #[serial]
    fn derives_from_envelope_config_errors_when_requested_environment_section_is_missing() {
        std::env::set_var("MECHA10_ENV", "production");

        let project = tempfile::tempdir().unwrap();
        write_json(
            &nodes_dir(project.path()).join("simulator").join("config.json"),
            r#"{ "dev": { "relay_topics": ["/camera/rgb"] } }"#,
        );

        let err = derive_relay_topics(project.path()).unwrap_err();
        std::env::remove_var("MECHA10_ENV");

        assert!(err.to_string().contains("'production'"));
    }

    #[test]
    #[serial]
    fn node_without_relay_topics_field_is_silently_skipped() {
        std::env::remove_var("MECHA10_ENV");

        let project = tempfile::tempdir().unwrap();
        write_json(
            &nodes_dir(project.path()).join("motor").join("dev").join("config.json"),
            r#"{ "update_rate_hz": 50.0 }"#,
        );

        let topics = derive_relay_topics(project.path()).unwrap();

        assert_eq!(topics.len(), 2);
        assert!(topics.contains(&"/system/health".to_string()));
        assert!(topics.contains(&"/system/log".to_string()));
    }

    /// LAB-2270 follow-up: a project-authored custom node (scaffolded at `configs/nodes/{name}/`,
    /// no `@mecha10` prefix) must be able to reach the dashboard the same way a framework node
    /// does, since its author can't edit anything under this framework's own `packages/nodes/`
    /// source to register it any other way.
    #[test]
    #[serial]
    fn derives_from_a_project_authored_custom_node_with_no_mecha10_prefix() {
        std::env::remove_var("MECHA10_ENV");

        let project = tempfile::tempdir().unwrap();
        write_json(
            &project.path().join("configs/nodes/my-custom-node/config.json"),
            r#"{ "relay_topics": ["/custom/status"] }"#,
        );

        let topics = derive_relay_topics(project.path()).unwrap();

        assert!(topics.contains(&"/custom/status".to_string()));
    }

    /// This is a plain recursive scan, not a precedence-aware resolution: if a node somehow has
    /// `config.json` in both tiers (e.g. a leftover directory from before a node was migrated),
    /// both contribute - unlike `Context::resolve_node_config`, which would only ever load one.
    /// That's fine here since an extra allowlisted topic is harmless.
    #[test]
    #[serial]
    fn both_tiers_contribute_even_for_the_same_node_name() {
        std::env::remove_var("MECHA10_ENV");

        let project = tempfile::tempdir().unwrap();
        write_json(
            &nodes_dir(project.path()).join("camera").join("config.json"),
            r#"{ "relay_topics": ["/camera/rgb"] }"#,
        );
        write_json(
            &project.path().join("configs/nodes/camera/config.json"),
            r#"{ "relay_topics": ["/camera/other"] }"#,
        );

        let topics = derive_relay_topics(project.path()).unwrap();

        assert!(topics.contains(&"/camera/rgb".to_string()));
        assert!(topics.contains(&"/camera/other".to_string()));
    }

    #[test]
    #[serial]
    fn dedups_topics_declared_by_multiple_nodes() {
        std::env::remove_var("MECHA10_ENV");

        let project = tempfile::tempdir().unwrap();
        write_json(
            &nodes_dir(project.path()).join("node-a").join("config.json"),
            r#"{ "relay_topics": ["/shared/topic"] }"#,
        );
        write_json(
            &nodes_dir(project.path()).join("node-b").join("config.json"),
            r#"{ "relay_topics": ["/shared/topic"] }"#,
        );

        let topics = derive_relay_topics(project.path()).unwrap();

        assert_eq!(topics.iter().filter(|t| *t == "/shared/topic").count(), 1);
    }

    #[test]
    #[serial]
    fn regenerate_unions_derived_topics_with_existing_entries_without_removing_any() {
        std::env::remove_var("MECHA10_ENV");

        let project = tempfile::tempdir().unwrap();

        write_json(
            &nodes_dir(project.path()).join("image-classifier").join("config.json"),
            r#"{ "relay_topics": ["/vision/classification"] }"#,
        );

        let bridge_config = nodes_dir(project.path())
            .join(WEBSOCKET_BRIDGE_NODE_NAME)
            .join("config.json");
        write_json(
            &bridge_config,
            r#"{
                "dev": {
                    "relay": {"url": "wss://x", "robot_id": "r", "api_key": "k"},
                    "relay_topics": ["/system/health", "/legacy/only/topic"]
                },
                "production": {
                    "relay": {"url": "wss://x", "robot_id": "r", "api_key": "k"},
                    "relay_topics": ["/legacy/only/topic"]
                }
            }"#,
        );

        regenerate_websocket_bridge_relay_config(project.path()).unwrap();

        let updated: Value = serde_json::from_str(&fs::read_to_string(&bridge_config).unwrap()).unwrap();

        for env_key in ["dev", "production"] {
            let relay_topics = updated[env_key]["relay_topics"].as_array().unwrap();
            let values: HashSet<String> = relay_topics
                .iter()
                .filter_map(Value::as_str)
                .map(|s| s.to_string())
                .collect();

            assert!(values.contains("/legacy/only/topic"));
            assert!(values.contains("/system/health"));
            assert!(values.contains("/system/log"));
            assert!(values.contains("/vision/classification"));
        }

        let dev_relay_topics = updated["dev"]["relay_topics"].as_array().unwrap();
        let health_entries = dev_relay_topics
            .iter()
            .filter(|t| t.as_str() == Some("/system/health"))
            .count();
        assert_eq!(health_entries, 1);
    }

    #[test]
    fn regenerate_is_a_no_op_when_websocket_bridge_has_no_config_file() {
        let project = tempfile::tempdir().unwrap();

        regenerate_websocket_bridge_relay_config(project.path()).unwrap();
    }

    /// LAB-4131 follow-up: a multi-instance node's resolved environment section is
    /// instance-keyed (e.g. `{"real": {...}, "sim": {...}}`) rather than a flat config with a
    /// top-level `relay_topics` key, so `derive_relay_topics` must also check one level of
    /// nesting for each instance's own `relay_topics`.
    #[test]
    #[serial]
    fn derives_nested_relay_topics_from_an_instance_keyed_multi_instance_node() {
        std::env::remove_var("MECHA10_ENV");

        let project = tempfile::tempdir().unwrap();
        write_json(
            &nodes_dir(project.path()).join("camera").join("config.json"),
            r#"{
                "dev": {
                    "real": { "relay_topics": ["/camera/rgb/real"] },
                    "sim": { "relay_topics": ["/camera/rgb/sim"] }
                },
                "production": {
                    "real": { "relay_topics": ["/camera/rgb/real"] },
                    "sim": { "relay_topics": ["/camera/rgb/sim"] }
                }
            }"#,
        );

        let topics = derive_relay_topics(project.path()).unwrap();

        assert!(topics.contains(&"/camera/rgb/real".to_string()));
        assert!(topics.contains(&"/camera/rgb/sim".to_string()));
    }
}