mecha10-core 0.6.5

//! 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, unioning with
//! whatever `topics.subscribes` 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).
//!
//! The friendly-name key used in existing `topics.subscribes` entries (e.g.
//! `{ "detections": "/vision/object/detections" }`) is purely cosmetic - nothing downstream
//! (`relay_client.rs`, the relay protocol, the dashboard) ever reads it, only the map's values
//! are used. Entries this module adds use the topic path as both the map key and value (e.g.
//! `{ "/vision/classification": "/vision/classification" }`) rather than inventing a "smart"
//! friendly name.
//!
//! # 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};

/// Node-config field a node optionally declares, listing Redis topic paths this node publishes
/// that `websocket-bridge`'s relay allowlist should include. Absent = this node contributes
/// nothing (not yet migrated onto this mechanism - not an error).
pub const RELAY_TOPICS_FIELD: &str = "relay_topics";

/// The shared, canonical shape of a `relay_topics` declaration. A node's own config struct
/// embeds this via `#[serde(flatten)] pub relay: RelayTopicsField` - see the module docs above
/// for why that's the whole story (no other per-node code needed): `relay_topics` becomes a
/// real, typed field on the node's config the same as any other, `derive_relay_topics` (which
/// reads the raw JSON off disk directly, independent of any node's Rust type) doesn't need this
/// type at all, and `#[serde(deny_unknown_fields)]` on the outer struct still works correctly
/// for every other field.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
pub struct RelayTopicsField {
    /// Which of this node's *published* topics should be forwarded to the dashboard by
    /// `websocket-bridge`. A topic this node only subscribes to (a control/command input)
    /// doesn't belong here - inbound (dashboard -> robot) forwarding has no allowlist, so
    /// there's nothing for it to enable.
    #[serde(default)]
    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()]
}

/// Project-relative path to the scaffolded framework nodes' config directory.
///
/// Mirrors `mecha10_cli_core::paths::config::NODES_MECHA10_DIR` - this module can't reference
/// that constant directly since `mecha10-core` is a dependency *of* `mecha10-cli-core`, not the
/// reverse (same constraint already documented on `Context::resolve_node_config`), but it must
/// stay in sync with it.
const NODES_MECHA10_DIR: &str = "configs/nodes/@mecha10";

/// Project-relative path [`derive_relay_topics`] recursively scans for every node's
/// `config.json` - covers both the framework tier (`configs/nodes/@mecha10/{name}/`) and the
/// plain project tier a custom node scaffolds into (`configs/nodes/{name}/`, no `@mecha10`
/// prefix - see `mecha10-templates`' `NodeGenerator`). A custom node's author has no way to edit
/// anything under this framework's own `packages/nodes/` source, so scanning only the framework
/// tier would make it structurally impossible for a project-authored node to ever reach the
/// dashboard.
pub(crate) const NODES_PROJECT_DIR: &str = "configs/nodes";

/// Project-relative path, under [`NODES_MECHA10_DIR`], to `websocket-bridge`'s own scaffolded
/// config file.
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.
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);

        let Some(relay_topics) = resolved.get(RELAY_TOPICS_FIELD).and_then(Value::as_array) else {
            continue;
        };

        for topic in relay_topics.iter().filter_map(Value::as_str) {
            topics.insert(topic.to_string());
        }
    }

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

/// Recursively find every file named `config.json` under `dir`, in any of the shapes node
/// configs come in (`{node}/config.json`, `{node}/dev/config.json`,
/// `{node}/production/config.json`, at any nesting depth - e.g. both `configs/nodes/{name}/...`
/// and `configs/nodes/@mecha10/{name}/...`). A missing `dir` yields an empty list, not an error.
// duplicate: false positive — matches `cli::handlers::run::locomotion_env_for_nodes` purely on
// the "make an empty collection, pass it by &mut to a helper that fills it, return it" AST
// shape; unrelated domains (recursive config-file discovery vs locomotion env var injection)
// with no shared behavior to extract.
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` (falling back to whichever of the two is present if the requested one isn't).
/// Otherwise, return `value` unchanged - it's already a flat, unwrapped config object.
pub(crate) fn resolve_environment_section(value: Value, environment: &str) -> Value {
    let env_key = match environment {
        "production" | "prod" => "production",
        _ => "dev",
    };

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

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

    let fallback_key = if env_key == "dev" { "production" } else { "dev" };
    obj.get(env_key)
        .or_else(|| obj.get(fallback_key))
        .cloned()
        .unwrap_or(value)
}

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
/// `topics.subscribes` allowlist reflects [`derive_relay_topics`]'s current output.
///
/// **Additive, never destructive**: the derived topics are *unioned* into whatever
/// `topics.subscribes` 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 `topics.subscribes` 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_subscribes_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_subscribes_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["topics"]["subscribes"]`, appending only the ones
/// not already present as a value anywhere in the existing list. Existing entries (including
/// their cosmetic friendly-name keys) are left completely untouched. Silently leaves a malformed
/// `topics`/`subscribes` field (present but not the expected object/array shape) alone rather
/// than erroring - regeneration should never be the thing that breaks an otherwise-loadable
/// config file.
fn union_subscribes_into(section: &mut serde_json::Map<String, Value>, derived: &[String]) {
    let topics_entry = section
        .entry("topics")
        .or_insert_with(|| Value::Object(serde_json::Map::new()));
    let Some(topics_obj) = topics_entry.as_object_mut() else {
        return;
    };

    let subscribes_entry = topics_obj
        .entry("subscribes")
        .or_insert_with(|| Value::Array(Vec::new()));
    let Some(subscribes_arr) = subscribes_entry.as_array_mut() else {
        return;
    };

    let mut seen: HashSet<String> = subscribes_arr
        .iter()
        .filter_map(Value::as_object)
        .flat_map(|entry| entry.values())
        .filter_map(Value::as_str)
        .map(|s| s.to_string())
        .collect();

    for topic in derived {
        if seen.insert(topic.clone()) {
            let mut entry = serde_json::Map::new();
            entry.insert(topic.clone(), Value::String(topic.clone()));
            subscribes_arr.push(Value::Object(entry));
        }
    }
}

#[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()));
    }

    #[test]
    #[serial]
    fn derives_from_dev_only_config_falling_back_when_production_requested() {
        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()));
    }

    #[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"},
                    "topics": {
                        "subscribes": [
                            { "health": "/system/health" },
                            { "legacyOnly": "/legacy/only/topic" }
                        ]
                    }
                },
                "production": {
                    "relay": {"url": "wss://x", "robot_id": "r", "api_key": "k"},
                    "topics": {
                        "subscribes": [
                            { "legacyOnly": "/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 subscribes = updated[env_key]["topics"]["subscribes"].as_array().unwrap();
            let values: HashSet<String> = subscribes
                .iter()
                .filter_map(|entry| entry.as_object())
                .flat_map(|m| m.values())
                .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_subscribes = updated["dev"]["topics"]["subscribes"].as_array().unwrap();
        let health_entries: Vec<&Value> = dev_subscribes
            .iter()
            .filter(|entry| entry.get("health").and_then(Value::as_str) == Some("/system/health"))
            .collect();
        assert_eq!(health_entries.len(), 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();
    }
}