mecha10-core 0.6.3

//! 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"] }
//! ```
//!
//! [`derive_relay_topics`] walks every node scaffolded into a project's
//! `configs/nodes/@mecha10/` tree, collects each node's `relay_topics` (nodes that don't declare
//! the field silently contribute 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`]).
//!
//! [`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 scaffolded project's node config directory is not guaranteed to be in one single shape -
//! see [`load_resolved_node_config`] for the three layouts this module resolves.

use crate::error::{Mecha10Error, Result};
use crate::topics::system::{HEALTH, LOG};
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";

/// 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, 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 node's own `relay_topics` field, resolved from whichever config shape that node uses
///   and whichever environment (`dev`/`production`) is currently active (`MECHA10_ENV`, default
///   `dev` - mirrors [`crate::context::Context::resolve_node_config`]'s convention).
///
/// A project with no `configs/nodes/@mecha10/` 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 nodes_dir = project_root.join(NODES_MECHA10_DIR);
    if !nodes_dir.is_dir() {
        return Ok(topics.into_iter().collect());
    }

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

    let mut node_dirs: Vec<PathBuf> = std::fs::read_dir(&nodes_dir)
        .map_err(|e| Mecha10Error::Configuration(format!("Failed to read {}: {}", nodes_dir.display(), e)))?
        .filter_map(|entry| entry.ok())
        .map(|entry| entry.path())
        .filter(|path| path.is_dir())
        .collect();
    // Deterministic ordering - directory iteration order isn't guaranteed, and stable output
    // matters for a generated file that gets diffed in PRs.
    node_dirs.sort();

    for node_dir in node_dirs {
        let Some(resolved) = load_resolved_node_config(&node_dir, &environment)? else {
            continue;
        };

        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())
}

/// Resolve a single node's config object for `environment`, handling all three config-file
/// layouts observed across nodes today:
///
/// 1. Single `config.json` with `"dev"`/`"production"` keys wrapping the actual config (e.g.
///    `llm-command`, `object-detector`, `image-classifier`, `websocket-bridge`).
/// 2. Split `dev/config.json` + `production/config.json` files, each already the resolved config
///    for that environment (e.g. `imu`, `motor`, `camera`, `video-streamer`).
/// 3. `dev/config.json` only, no `production/config.json` (e.g. `simulator`) - falls back to the
///    `dev` file regardless of requested environment rather than treating it as "no config"; a
///    node not yet given a production config shouldn't silently lose its relay topics.
///
/// Returns `Ok(None)` when a node directory has no config file in any recognized shape.
fn load_resolved_node_config(node_dir: &Path, environment: &str) -> Result<Option<Value>> {
    let single_config = node_dir.join("config.json");
    if single_config.is_file() {
        let value = read_json(&single_config)?;
        return Ok(Some(resolve_environment_section(value, environment)));
    }

    let dev_config = node_dir.join("dev").join("config.json");
    let production_config = node_dir.join("production").join("config.json");

    let wants_production = matches!(environment, "production" | "prod");
    let chosen = if wants_production && production_config.is_file() {
        Some(production_config)
    } else if dev_config.is_file() {
        // Either dev was requested, or production was requested but doesn't exist for this
        // node (e.g. `simulator`) - reuse dev rather than dropping the node's topics.
        Some(dev_config)
    } else if production_config.is_file() {
        // dev was requested but only a production file exists - the mirror-image fallback.
        Some(production_config)
    } else {
        None
    };

    match chosen {
        Some(path) => Ok(Some(read_json(&path)?)),
        None => Ok(None),
    }
}

/// 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.
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)
}

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() {
        // Flat, unwrapped config (no "dev"/"production" envelope) - union directly into it.
        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();

        // Only the two framework topics - nothing from the un-migrated node, and no error.
        assert_eq!(topics.len(), 2);
        assert!(topics.contains(&"/system/health".to_string()));
        assert!(topics.contains(&"/system/log".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();

        // A node contributing a new topic not yet in websocket-bridge's config.
        write_json(
            &nodes_dir(project.path()).join("image-classifier").join("config.json"),
            r#"{ "relay_topics": ["/vision/classification"] }"#,
        );

        // websocket-bridge's own config, with a pre-existing hand-authored entry that duplicates
        // one of the derived framework topics (by value) under a friendly-name key, plus an
        // unrelated hand-authored entry that isn't derivable at all.
        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();

            // Never-derived, hand-authored entry survives untouched.
            assert!(values.contains("/legacy/only/topic"));
            // Framework topics are present (additively, not duplicated).
            assert!(values.contains("/system/health"));
            assert!(values.contains("/system/log"));
            // The newly migrated node's topic was added.
            assert!(values.contains("/vision/classification"));
        }

        // The pre-existing friendly-name key for the topic that overlapped with a derived
        // framework topic was preserved, not replaced with a path-as-key duplicate entry.
        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();

        // Should not error just because the project hasn't scaffolded websocket-bridge.
        regenerate_websocket_bridge_relay_config(project.path()).unwrap();
    }
}