Function bodies 699 total

    fn spawn_refresh_and_emit(self: &Arc<Self>, event_type: &str, name: &str) {
        let manager = Arc::clone(self);
        let event_type = event_type.to_string();
        let node_name = name.to_string();
        tokio::spawn(async move {
            if let Err(e) = manager.refresh_all().await {
                log::warn!("Failed to refresh after {}: {}", event_type, e);
            }
            manager.emit_event(&event_type, &node_name).await;
        });
    }

    async fn start_node(self: &Arc<Self>, name: &str) -> Result<String> {
        let service_name = systemd::get_service_name(name);
        self.systemd.start_unit(&service_name).await?;
        self.spawn_refresh_and_emit("started", name);
        Ok(format!("Started {}", name))
    }

    async fn stop_node(self: &Arc<Self>, name: &str) -> Result<String> {
        let service_name = systemd::get_service_name(name);
        self.systemd.stop_unit(&service_name).await?;
        self.spawn_refresh_and_emit("stopped", name);
        Ok(format!("Stopped {}", name))
    }

    async fn restart_node(self: &Arc<Self>, name: &str) -> Result<String> {
        let service_name = systemd::get_service_name(name);
        self.systemd.restart_unit(&service_name).await?;
        self.spawn_refresh_and_emit("restarted", name);
        Ok(format!("Restarted {}", name))
    }

    async fn install_node(self: &Arc<Self>, name: &str) -> Result<String> {
        // Look up by effective name (the HashMap key)
        let nodes = self.nodes.read().await;
        let node = nodes
            .get(name)
            .ok_or_else(|| NodeManagerError::NodeNotFound(name.to_string()))?;

        let path = node.path.clone();
        let manifest = node
            .manifest
            .as_ref()
            .ok_or_else(|| NodeManagerError::NodeNotFound(name.to_string()))?;

        // If config_override is set, append -c <config> to the command
        let command = if let Some(ref config_path) = node.config_override {
            let base_cmd = manifest
                .command
                .as_deref()
                .unwrap_or("./target/release/unknown");
            Some(format!("{} -c {}", base_cmd, config_path))
        } else {
            manifest.command.clone()
        };

        systemd::install_service(
            &path,
            name,
            &ma

    async fn stop_if_running(&self, name: &str) {
        let service_name = systemd::get_service_name(name);
        match self.systemd.get_active_state(&service_name).await {
            Ok(ActiveState::Active | ActiveState::Activating) => {
                log::info!("Stopping {} before build/clean", name);
                let _ = self.systemd.stop_unit(&service_name).await;
                tokio::time::sleep(Duration::from_millis(500)).await;
            }
            Err(e) => {
                log::warn!(
                    "Could not check state of {} ({}), proceeding anyway",
                    name,
                    e
                );
            }
            _ => {}
        }
    }

    async fn begin_build_activity(
        &self,
        name: &str,
        status: BuildStatus,
        start_event: &str,
    ) -> Result<String> {
        let path = self.find_node_path(name).await?;
        self.stop_if_running(name).await;

        let mut building = self.building_nodes.lock().await;
        if !building.insert(name.to_string()) {
            return Err(NodeManagerError::AlreadyBuilding(name.to_string()));
        }
        drop(building);

        let mut nodes = self.nodes.write().await;
        if let Some(node) = nodes.get_mut(name) {
            node.build_state.status = status;
            node.build_state.output.clear();
            node.status = NodeStatus::Building;
        }
        drop(nodes);

        self.emit_event(start_event, name).await;
        Ok(path)
    }

    async fn build_node(self: &Arc<Self>, manager: Arc<Self>, name: &str) -> Result<String> {
        let path = self
            .begin_build_activity(name, BuildStatus::Building, "building")
            .await?;

        // Get build command
        let build_cmd = {
            let nodes = self.nodes.read().await;
            let node = nodes
                .get(name)
                .ok_or_else(|| NodeManagerError::NodeNotFound(name.to_string()))?;
            node.manifest
                .as_ref()
                .and_then(|m| m.build.clone())
                .ok_or_else(|| {
                    NodeManagerError::BuildError("No build command defined".to_string())
                })?
        };

        let name_clone = name.to_string();
        let path_clone = path.clone();

        tokio::spawn(async move {
            let result = run_with_timeout(&manager, &path_clone, &build_cmd, &name_clone).await;

            finish_build_activity(&manager, &name_clone, &result, "Build")

    async fn clean_node(self: &Arc<Self>, manager: Arc<Self>, name: &str) -> Result<String> {
        let path = self
            .begin_build_activity(name, BuildStatus::Cleaning, "cleaning")
            .await?;

        let name_clone = name.to_string();
        let path_clone = path.clone();

        tokio::spawn(async move {
            let result =
                run_with_timeout(&manager, &path_clone, "pixi run clean", &name_clone).await;

            finish_build_activity(&manager, &name_clone, &result, "Clean").await;

            let _ = manager.refresh_all().await;

            // Force is_built = false after refresh for successful cleans.
            // refresh_all() re-checks the filesystem, which may still find
            // artifacts if the clean process hasn't fully flushed yet.
            {
                let mut nodes = manager.nodes.write().await;
                if let Some(node) = nodes.get_mut(&name_clone) {
                    node.is_built = false;
         

    async fn enable_autostart(&self, name: &str) -> Result<String> {
        let service_name = systemd::get_service_name(name);
        self.systemd.enable_unit(&service_name).await?;

        self.refresh_all().await?;
        self.emit_event("autostart_enabled", name).await;

        Ok(format!("Enabled autostart for {}", name))
    }

    async fn disable_autostart(&self, name: &str) -> Result<String> {
        let service_name = systemd::get_service_name(name);
        self.systemd.disable_unit(&service_name).await?;

        self.refresh_all().await?;
        self.emit_event("autostart_disabled", name).await;

        Ok(format!("Disabled autostart for {}", name))
    }

    async fn add_node(
        &self,
        path: &str,
        name_override: Option<&str>,
        config_override: Option<&str>,
    ) -> Result<String> {
        let (_manifest, eff_name) = registry::register_node(path, name_override, config_override)?;

        self.refresh_all().await?;
        self.emit_event("added", &eff_name).await;

        Ok(format!("Added node: {}", eff_name))
    }

    async fn remove_node(&self, name: &str) -> Result<String> {
        // Verify the node exists
        let _path = self.find_node_path(name).await?;

        // Uninstall service if installed before removing from registry
        if systemd::is_service_installed(name) {
            log::info!("Uninstalling service {} before removal", name);
            let _ = systemd::uninstall_service(name).await;
        }

        // Unregister by effective name (handles multi-instance correctly)
        registry::unregister_node(name)?;

        self.refresh_all().await?;
        self.emit_event("removed", name).await;

        Ok(format!("Removed node: {}", name))
    }

    async fn emit_event(&self, event_type: &str, node_name: &str) {
        if let Some(state) = self.get_node(node_name).await {
            let event = NodeEvent {
                event_type: event_type.to_string(),
                node_name: node_name.to_string(),
                state: Some(state),
                timestamp_ms: Self::now_ms(),
            };

            // Ignore send errors (no subscribers)
            let _ = self.event_tx.send(event);
        }
    }

async fn run_with_timeout(
    manager: &Arc<NodeManager>,
    path: &str,
    cmd: &str,
    name: &str,
) -> Result<()> {
    let timeout_duration = Duration::from_secs(BUILD_TIMEOUT_SECS);
    match tokio::time::timeout(
        timeout_duration,
        run_build_command(manager, path, cmd, name),
    )
    .await
    {
        Ok(result) => result,
        Err(_) => Err(NodeManagerError::BuildTimeout(name.to_string())),
    }
}

fn validate_build_command(cmd: &str) -> Result<()> {
    // Allowlist of permitted build command prefixes
    const ALLOWED_PREFIXES: &[&str] = &["cargo ", "pixi ", "npm ", "make ", "python ", "pip "];

    let cmd_lower = cmd.to_lowercase();
    let has_allowed_prefix = ALLOWED_PREFIXES
        .iter()
        .any(|prefix| cmd_lower.starts_with(prefix));

    if !has_allowed_prefix {
        return Err(NodeManagerError::BuildError(format!(
            "Build command must start with one of: cargo, pixi, npm, make, python, pip. Got: {}",
            cmd.chars().take(50).collect::<String>()
        )));
    }

    // Reject dangerous shell metacharacters
    const DANGEROUS_CHARS: &[char] = &[
        '$', '`', '|', ';', '&', '>', '<', '(', ')', '{', '}', '!', '\\', '\n', '\r',
    ];
    if let Some(bad_char) = cmd.chars().find(|c| DANGEROUS_CHARS.contains(c)) {
        return Err(NodeManagerError::BuildError(format!(
            "Build command contains dangerous character '{}': {}",
 

async fn run_build_command(
    manager: &Arc<NodeManager>,
    path: &str,
    cmd: &str,
    name: &str,
) -> Result<()> {
    // Validate command before execution to prevent command injection
    validate_build_command(cmd)?;

    // Build a PATH that includes user tool directories (pixi, cargo, etc.)
    // so build commands like "pixi run build" work under systemd's minimal env.
    let home = dirs::home_dir().unwrap_or_else(|| std::path::PathBuf::from("/home/user"));
    let build_path = format!(
        "{}:{}:/usr/local/bin:/usr/bin:/bin",
        home.join(".cargo/bin").display(),
        home.join(".pixi/bin").display(),
    );

    let mut child = Command::new("sh")
        .args(["-c", cmd])
        .current_dir(path)
        .env("PATH", &build_path)
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .kill_on_drop(true) // Kill child process if future is dropped (e.g., on timeout)
        .spawn()?;

    let stdout = child.s

fn extract_health_node_name(key: &str) -> Option<String> {
    let parts: Vec<&str> = key.split('/').collect();
    if parts.is_empty() || parts[0] != "bubbaloop" {
        return None;
    }

    // Legacy format: bubbaloop/nodes/{name}/health
    if parts.len() >= 4 && parts[1] == "nodes" && parts[3] == "health" {
        return Some(parts[2].to_string());
    }

    // Scoped format: bubbaloop/{scope}/{machine}/health/{name}
    if parts.len() >= 5 && parts[3] == "health" {
        return Some(parts[4].to_string());
    }

    None
}

    fn test_validate_build_command_allowed_prefixes() {
        assert!(validate_build_command("cargo build --release").is_ok());
        assert!(validate_build_command("pixi run build").is_ok());
        assert!(validate_build_command("npm run build").is_ok());
        assert!(validate_build_command("make all").is_ok());
        assert!(validate_build_command("python setup.py build").is_ok());
        assert!(validate_build_command("pip install .").is_ok());
    }

    fn test_validate_build_command_rejects_make_without_space() {
        // "make" without a trailing space should not match arbitrary commands
        // like "makefile-exploit" or "makeover"
        assert!(validate_build_command("makefile-exploit").is_err());
        assert!(validate_build_command("makeover something").is_err());
    }

    fn test_validate_build_command_rejects_shell_metacharacters() {
        assert!(validate_build_command("cargo build; rm -rf /").is_err());
        assert!(validate_build_command("cargo build && evil").is_err());
        assert!(validate_build_command("cargo build | evil").is_err());
        assert!(validate_build_command("cargo build > /etc/passwd").is_err());
        assert!(validate_build_command("cargo build $(evil)").is_err());
        assert!(validate_build_command("cargo build `evil`").is_err());
    }

    fn test_extract_health_node_name_legacy() {
        assert_eq!(
            extract_health_node_name("bubbaloop/nodes/my-node/health"),
            Some("my-node".to_string())
        );
    }

    fn test_extract_health_node_name_scoped() {
        assert_eq!(
            extract_health_node_name("bubbaloop/scope/machine1/health/my-node"),
            Some("my-node".to_string())
        );
    }

    fn test_cached_node_base_node_with_override() {
        let node = CachedNode {
            path: "/path/to/rtsp-camera".to_string(),
            manifest: Some(NodeManifest {
                name: "rtsp-camera".to_string(),
                version: "0.1.0".to_string(),
                description: "RTSP camera node".to_string(),
                node_type: "rust".to_string(),
                ..Default::default()
            }),
            status: NodeStatus::Stopped,
            installed: false,
            autostart_enabled: false,
            is_built: false,
            build_state: BuildState::default(),
            last_updated_ms: 0,
            health_status: HealthStatus::Unknown,
            last_health_check_ms: 0,
            name_override: Some("rtsp-camera-terrace".to_string()),
            config_override: None,
        };

        let proto = node.to_proto("machine1", "host1", &[]);
        assert_eq!(proto.name, "rtsp-camera-terrace");
        assert_eq!(proto.bas

    fn test_cached_node_base_node_without_override() {
        let node = CachedNode {
            path: "/path/to/openmeteo".to_string(),
            manifest: Some(NodeManifest {
                name: "openmeteo".to_string(),
                version: "0.1.0".to_string(),
                description: "Weather node".to_string(),
                node_type: "rust".to_string(),
                ..Default::default()
            }),
            status: NodeStatus::Running,
            installed: true,
            autostart_enabled: false,
            is_built: true,
            build_state: BuildState::default(),
            last_updated_ms: 0,
            health_status: HealthStatus::Unknown,
            last_health_check_ms: 0,
            name_override: None,
            config_override: None,
        };

        let proto = node.to_proto("machine1", "host1", &[]);
        assert_eq!(proto.name, "openmeteo");
        assert_eq!(proto.base_node, "");
    }

    fn test_multi_instance_cameras_base_node_tracking() {
        let camera_manifest = NodeManifest {
            name: "rtsp-camera".to_string(),
            version: "0.2.0".to_string(),
            description: "RTSP camera node".to_string(),
            node_type: "rust".to_string(),
            build: Some("cargo build --release".to_string()),
            command: Some("./target/release/cameras_node".to_string()),
            ..Default::default()
        };

        let instances = vec![
            (
                "rtsp-camera-terrace",
                Some("rtsp-camera-terrace"),
                Some("/etc/bubbaloop/terrace.yaml"),
            ),
            (
                "rtsp-camera-garage",
                Some("rtsp-camera-garage"),
                Some("/etc/bubbaloop/garage.yaml"),
            ),
            ("rtsp-camera-entrance", Some("rtsp-camera-entrance"), None),
        ];

        let mut protos = Vec::new();

        for (expected_name, name_override, confi

    fn test_cached_node_effective_name_with_override() {
        let node = CachedNode {
            path: "/path/to/rtsp-camera".to_string(),
            manifest: Some(NodeManifest {
                name: "rtsp-camera".to_string(),
                version: "0.1.0".to_string(),
                description: "RTSP camera node".to_string(),
                node_type: "rust".to_string(),
                ..Default::default()
            }),
            status: NodeStatus::Stopped,
            installed: false,
            autostart_enabled: false,
            is_built: false,
            build_state: BuildState::default(),
            last_updated_ms: 0,
            health_status: HealthStatus::Unknown,
            last_health_check_ms: 0,
            name_override: Some("rtsp-camera-terrace".to_string()),
            config_override: None,
        };
        assert_eq!(node.effective_name(), "rtsp-camera-terrace");
    }

    fn test_cached_node_effective_name_without_override() {
        let node = CachedNode {
            path: "/path/to/openmeteo".to_string(),
            manifest: Some(NodeManifest {
                name: "openmeteo".to_string(),
                version: "0.1.0".to_string(),
                description: "Weather".to_string(),
                node_type: "rust".to_string(),
                ..Default::default()
            }),
            status: NodeStatus::Running,
            installed: true,
            autostart_enabled: false,
            is_built: true,
            build_state: BuildState::default(),
            last_updated_ms: 0,
            health_status: HealthStatus::Unknown,
            last_health_check_ms: 0,
            name_override: None,
            config_override: None,
        };
        assert_eq!(node.effective_name(), "openmeteo");
    }

    fn test_cached_node_effective_name_no_manifest() {
        let node = CachedNode {
            path: "/path/to/unknown".to_string(),
            manifest: None,
            status: NodeStatus::Stopped,
            installed: false,
            autostart_enabled: false,
            is_built: false,
            build_state: BuildState::default(),
            last_updated_ms: 0,
            health_status: HealthStatus::Unknown,
            last_health_check_ms: 0,
            name_override: None,
            config_override: None,
        };
        assert_eq!(node.effective_name(), "unknown");
    }

    pub fn validate(&self) -> Result<()> {
        // Validate name: 1-64 chars, alphanumeric + hyphen + underscore
        if self.name.is_empty() || self.name.len() > 64 {
            return Err(RegistryError::InvalidNode(format!(
                "Node name must be 1-64 characters, got: {}",
                self.name.len()
            )));
        }
        if !self
            .name
            .chars()
            .all(|c| c.is_alphanumeric() || c == '-' || c == '_')
        {
            return Err(RegistryError::InvalidNode(format!(
                "Node name contains invalid characters: {}",
                self.name
            )));
        }

        // Validate version: basic semver check (contains digits and dots)
        if self.version.is_empty() {
            return Err(RegistryError::InvalidNode(
                "Node version cannot be empty".to_string(),
            ));
        }
        let has_digit = self.version.chars().any(|c| c.is_ascii_digit());
        if !has_d

pub fn load_registry() -> Result<NodesRegistry> {
    let path = get_nodes_file();
    if !path.exists() {
        return Ok(NodesRegistry::default());
    }

    let content = fs::read_to_string(&path)?;
    let registry: NodesRegistry = serde_json::from_str(&content)?;
    Ok(registry)
}

pub fn save_registry(registry: &NodesRegistry) -> Result<()> {
    let home = get_bubbaloop_home();
    fs::create_dir_all(&home)?;

    let path = get_nodes_file();
    let content = serde_json::to_string_pretty(registry)?;
    fs::write(path, content)?;
    Ok(())
}

pub fn read_manifest(node_path: &Path) -> Result<NodeManifest> {
    let manifest_path = node_path.join("node.yaml");
    if !manifest_path.exists() {
        return Err(RegistryError::InvalidNode(format!(
            "No node.yaml found in {}",
            node_path.display()
        )));
    }

    let content = fs::read_to_string(&manifest_path)?;
    let manifest: NodeManifest = serde_yaml::from_str(&content)?;

    // Validate the manifest fields
    manifest.validate()?;

    Ok(manifest)
}

pub fn effective_name(entry: &NodeEntry, manifest: &NodeManifest) -> String {
    entry
        .name_override
        .as_deref()
        .unwrap_or(&manifest.name)
        .to_string()
}

pub fn register_node(
    node_path: &str,
    name_override: Option<&str>,
    config_override: Option<&str>,
) -> Result<(NodeManifest, String)> {
    let path = Path::new(node_path);

    // Check directory exists
    if !path.exists() {
        return Err(RegistryError::InvalidNode(format!(
            "Directory not found: {}",
            node_path
        )));
    }

    // Read and validate manifest
    let manifest = read_manifest(path)?;

    // Determine effective name
    let eff_name = name_override.unwrap_or(&manifest.name).to_string();

    // Validate override name if provided (same rules as manifest name)
    if let Some(override_name) = name_override {
        if override_name.is_empty() || override_name.len() > 64 {
            return Err(RegistryError::InvalidNode(format!(
                "Instance name must be 1-64 characters, got: {}",
                override_name.len()
            )));
        }
        if !override_name
            .chars()
            .all(|c|

pub fn unregister_node(name_or_path: &str) -> Result<()> {
    let mut registry = load_registry()?;

    let initial_len = registry.nodes.len();

    // Try matching by effective name first
    registry.nodes.retain(|entry| {
        let entry_path = Path::new(&entry.path);

        // Try to match by effective name (manifest-based or name_override)
        if let Ok(manifest) = read_manifest(entry_path) {
            let eff = effective_name(entry, &manifest);
            if eff == name_or_path {
                return false; // remove this entry
            }
        } else {
            // Directory deleted - still try to match by name_override or directory name
            if let Some(ref name_ov) = entry.name_override {
                if name_ov == name_or_path {
                    log::warn!(
                        "Node directory '{}' was already deleted, removing registry entry",
                        entry.path
                    );
                    return false; // r

pub fn list_nodes() -> Result<Vec<(NodeEntry, Option<NodeManifest>)>> {
    let registry = load_registry()?;

    let nodes = registry
        .nodes
        .iter()
        .map(|entry| {
            let path = Path::new(&entry.path);
            let manifest = read_manifest(path).ok();
            (entry.clone(), manifest)
        })
        .collect();

    Ok(nodes)
}

pub fn check_is_built(node_path: &str, manifest: &NodeManifest) -> bool {
    let path = Path::new(node_path);

    if let Some(ref command) = manifest.command {
        let binary_path = path.join(command);
        binary_path.exists()
    } else if manifest.node_type == "rust" {
        // Check for target/release or target/debug binary
        let release_path = path.join("target/release").join(&manifest.name);
        let debug_path = path.join("target/debug").join(&manifest.name);
        release_path.exists() || debug_path.exists()
    } else {
        // Python - check for main.py and venv
        let main_py = path.join("main.py");
        main_py.exists()
    }
}

fn chrono_now() -> String {
    use std::time::{SystemTime, UNIX_EPOCH};
    let duration = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default();
    let secs = duration.as_secs();

    // Simple ISO format without external dependencies
    // This is good enough for our purposes
    format!("{}000", secs)
}

    fn test_manifest_validation_valid() {
        let manifest = NodeManifest {
            name: "test-node".to_string(),
            version: "1.0.0".to_string(),
            node_type: "rust".to_string(),
            description: "A test node".to_string(),
            author: Some("Test Author".to_string()),
            build: Some("cargo build --release".to_string()),
            command: Some("./target/release/test-node".to_string()),
            ..Default::default()
        };
        assert!(manifest.validate().is_ok());
    }

    fn test_manifest_validation_empty_name() {
        let manifest = NodeManifest {
            name: "".to_string(),
            version: "1.0.0".to_string(),
            node_type: "rust".to_string(),
            description: "Test".to_string(),
            ..Default::default()
        };
        assert!(manifest.validate().is_err());
    }

    fn test_manifest_validation_name_too_long() {
        let manifest = NodeManifest {
            name: "a".repeat(65),
            version: "1.0.0".to_string(),
            node_type: "rust".to_string(),
            description: "Test".to_string(),
            ..Default::default()
        };
        assert!(manifest.validate().is_err());
    }

    fn test_manifest_validation_invalid_name_chars() {
        let manifest = NodeManifest {
            name: "test node!".to_string(),
            version: "1.0.0".to_string(),
            node_type: "rust".to_string(),
            description: "Test".to_string(),
            ..Default::default()
        };
        assert!(manifest.validate().is_err());
    }

    fn test_manifest_validation_valid_name_with_underscores_hyphens() {
        let manifest = NodeManifest {
            name: "test_node-123".to_string(),
            version: "1.0.0".to_string(),
            node_type: "rust".to_string(),
            description: "Test".to_string(),
            ..Default::default()
        };
        assert!(manifest.validate().is_ok());
    }

    fn test_manifest_validation_empty_version() {
        let manifest = NodeManifest {
            name: "test-node".to_string(),
            version: "".to_string(),
            node_type: "rust".to_string(),
            description: "Test".to_string(),
            ..Default::default()
        };
        assert!(manifest.validate().is_err());
    }

    fn test_manifest_validation_version_no_digits() {
        let manifest = NodeManifest {
            name: "test-node".to_string(),
            version: "alpha".to_string(),
            node_type: "rust".to_string(),
            description: "Test".to_string(),
            ..Default::default()
        };
        assert!(manifest.validate().is_err());
    }

    fn test_manifest_validation_invalid_node_type() {
        let manifest = NodeManifest {
            name: "test-node".to_string(),
            version: "1.0.0".to_string(),
            node_type: "javascript".to_string(),
            description: "Test".to_string(),
            ..Default::default()
        };
        assert!(manifest.validate().is_err());
    }

    fn test_manifest_validation_python_type() {
        let manifest = NodeManifest {
            name: "test-node".to_string(),
            version: "1.0.0".to_string(),
            node_type: "python".to_string(),
            description: "Test".to_string(),
            ..Default::default()
        };
        assert!(manifest.validate().is_ok());
    }

    fn test_manifest_validation_description_too_long() {
        let manifest = NodeManifest {
            name: "test-node".to_string(),
            version: "1.0.0".to_string(),
            node_type: "rust".to_string(),
            description: "a".repeat(501),
            ..Default::default()
        };
        assert!(manifest.validate().is_err());
    }

    fn test_manifest_validation_null_bytes_in_name() {
        let manifest = NodeManifest {
            name: "test\0node".to_string(),
            version: "1.0.0".to_string(),
            node_type: "rust".to_string(),
            description: "Test".to_string(),
            ..Default::default()
        };
        assert!(manifest.validate().is_err());
    }